PIP-121: Introduce ServiceInfoProvider to update service info dynamically

include/pulsar/Client.h

@@ -29,9 +29,12 @@
 #include <pulsar/Reader.h>
 #include <pulsar/Result.h>
 #include <pulsar/Schema.h>
+#include <pulsar/ServiceInfo.h>
+#include <pulsar/ServiceInfoProvider.h>
 #include <pulsar/TableView.h>
 #include <pulsar/defines.h>
 
+#include <memory>
 #include <string>
 
 namespace pulsar {
@@ -68,6 +71,20 @@ class PULSAR_PUBLIC Client {
      */
     Client(const std::string& serviceUrl, const ClientConfiguration& clientConfiguration);
 
+    /**
+     * Create a Pulsar client object using the specified ServiceInfoProvider.
+     *
+     * The ServiceInfoProvider is responsible for providing the service information (such as service URL)
+     * dynamically. For example, if it detects a primary Pulsar service is down, it can switch to a secondary
+     * service and update the client with the new service information.
+     *
+     * The Client instance takes ownership of the given ServiceInfoProvider. The provider will be destroyed
+     * as part of the client's shutdown lifecycle, for example when `Client::close()` or
+     * `Client::closeAsync()` is called, ensuring that its lifetime is properly managed.
+     */
+    static Client create(std::unique_ptr<ServiceInfoProvider> serviceInfoProvider,
+                         const ClientConfiguration& clientConfiguration);
+
     /**
      * Create a producer with default configuration
      *
@@ -414,6 +431,13 @@ class PULSAR_PUBLIC Client {
     void getSchemaInfoAsync(const std::string& topic, int64_t version,
                             std::function<void(Result, const SchemaInfo&)> callback);
 
+    /**
+     * Get the current service information of the client.
+     *
+     * @return the current service information
+     */
+    ServiceInfo getServiceInfo() const;
+
    private:
     Client(const std::shared_ptr<ClientImpl>&);
 

include/pulsar/ClientConfiguration.h

@@ -70,15 +70,12 @@ class PULSAR_PUBLIC ClientConfiguration {
     /**
      * Set the authentication method to be used with the broker
      *
+     * You can get the configured authentication data in `ServiceInfo` returned by `Client::getServiceInfo`.
+     *
      * @param authentication the authentication data to use
      */
     ClientConfiguration& setAuth(const AuthenticationPtr& authentication);
 
-    /**
-     * @return the authentication data
-     */
-    Authentication& getAuth() const;
-
     /**
      * Set timeout on client operations (subscribe, create producer, close, unsubscribe)
      * Default is 30 seconds.
@@ -202,20 +199,6 @@ class PULSAR_PUBLIC ClientConfiguration {
      */
     ClientConfiguration& setLogger(LoggerFactory* loggerFactory);
 
-    /**
-     * Configure whether to use the TLS encryption on the connections.
-     *
-     * The default value is false.
-     *
-     * @param useTls
-     */
-    ClientConfiguration& setUseTls(bool useTls);
-
-    /**
-     * @return whether the TLS encryption is used on the connections
-     */
-    bool isUseTls() const;
-
     /**
      * Set the path to the TLS private key file.
      *
@@ -243,15 +226,13 @@ class PULSAR_PUBLIC ClientConfiguration {
     /**
      * Set the path to the trusted TLS certificate file.
      *
+     * You can get the configured trusted TLS certificate file path in `ServiceInfo` returned by
+     * `Client::getServiceInfo`.
+     *
      * @param tlsTrustCertsFilePath
      */
     ClientConfiguration& setTlsTrustCertsFilePath(const std::string& tlsTrustCertsFilePath);
 
-    /**
-     * @return the path to the trusted TLS certificate file
-     */
-    const std::string& getTlsTrustCertsFilePath() const;
-
     /**
      * Configure whether the Pulsar client accepts untrusted TLS certificates from brokers.
      *

include/pulsar/ServiceInfo.h

@@ -0,0 +1,57 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#ifndef PULSAR_SERVICE_INFO_H_
+#define PULSAR_SERVICE_INFO_H_
+
+#include <pulsar/Authentication.h>
+
+#include <optional>
+#include <string>
+
+namespace pulsar {
+
+/**
+ * ServiceInfo encapsulates the information of a Pulsar service, which is used by the client to connect to the
+ * service. It includes the service URL, authentication information, and TLS configuration.
+ */
+class PULSAR_PUBLIC ServiceInfo final {
+   public:
+    ServiceInfo(std::string serviceUrl, AuthenticationPtr authentication = AuthFactory::Disabled(),
+                std::optional<std::string> tlsTrustCertsFilePath = std::nullopt);
+
+    auto& serviceUrl() const noexcept { return serviceUrl_; }
+    auto useTls() const noexcept { return useTls_; }
+    auto& authentication() const noexcept { return authentication_; }
+    auto& tlsTrustCertsFilePath() const noexcept { return tlsTrustCertsFilePath_; }
+
+    bool operator==(const ServiceInfo& other) const noexcept {
+        return serviceUrl_ == other.serviceUrl_ && useTls_ == other.useTls_ &&
+               authentication_ == other.authentication_ &&
+               tlsTrustCertsFilePath_ == other.tlsTrustCertsFilePath_;
+    }
+
+   private:
+    std::string serviceUrl_;
+    bool useTls_;
+    AuthenticationPtr authentication_;
+    std::optional<std::string> tlsTrustCertsFilePath_;
+};
+
+}  // namespace pulsar
+#endif

include/pulsar/ServiceInfoProvider.h

@@ -0,0 +1,62 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#ifndef PULSAR_SERVICE_INFO_PROVIDER_H_
+#define PULSAR_SERVICE_INFO_PROVIDER_H_
+
+#include <pulsar/ServiceInfo.h>
+
+#include <functional>
+
+namespace pulsar {
+
+class PULSAR_PUBLIC ServiceInfoProvider {
+   public:
+    /**
+     * The destructor will be called when `Client::close()` is invoked, and the provider should stop any
+     * ongoing work and release the resources in the destructor.
+     */
+    virtual ~ServiceInfoProvider() = default;
+
+    /**
+     * Get the initial `ServiceInfo` connection for the client.
+     * This method is called **only once** internally in `Client::create()` to get the initial `ServiceInfo`
+     * for the client to connect to the Pulsar service, typically before {@link initialize} is invoked.
+     * Since it's only called once, it's legal to return a moved `ServiceInfo` object to avoid unnecessary
+     * copying.
+     */
+    virtual ServiceInfo initialServiceInfo() = 0;
+
+    /**
+     * Initialize the ServiceInfoProvider.
+     *
+     * After the client has obtained the initial `ServiceInfo` via {@link initialServiceInfo}, this method is
+     * called to allow the provider to start any background work (for example, service discovery or watching
+     * configuration changes) and to report subsequent updates to the service information.
+     *
+     * @param onServiceInfoUpdate the callback to deliver updated `ServiceInfo` values to the client after
+     *                            the initial connection has been established
+     *
+     * Implementations may choose not to invoke `onServiceInfoUpdate` if the `ServiceInfo` never changes.
+     */
+    virtual void initialize(std::function<void(ServiceInfo)> onServiceInfoUpdate) = 0;
+};
+
+};  // namespace pulsar
+
+#endif

include/pulsar/c/client_configuration.h

@@ -147,16 +147,9 @@ PULSAR_PUBLIC void pulsar_client_configuration_set_logger(pulsar_client_configur
 PULSAR_PUBLIC void pulsar_client_configuration_set_logger_t(pulsar_client_configuration_t *conf,
                                                             pulsar_logger_t logger);
 
-PULSAR_PUBLIC void pulsar_client_configuration_set_use_tls(pulsar_client_configuration_t *conf, int useTls);
-
-PULSAR_PUBLIC int pulsar_client_configuration_is_use_tls(pulsar_client_configuration_t *conf);
-
 PULSAR_PUBLIC void pulsar_client_configuration_set_tls_trust_certs_file_path(
     pulsar_client_configuration_t *conf, const char *tlsTrustCertsFilePath);
 
-PULSAR_PUBLIC const char *pulsar_client_configuration_get_tls_trust_certs_file_path(
-    pulsar_client_configuration_t *conf);
-
 PULSAR_PUBLIC void pulsar_client_configuration_set_tls_allow_insecure_connection(
     pulsar_client_configuration_t *conf, int allowInsecure);
 

lib/AtomicSharedPtr.h

@@ -0,0 +1,41 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#pragma once
+
+#include <memory>
+namespace pulsar {
+
+// C++17 does not have std::atomic<std::shared_ptr<T>>, so we have to manually implement it.
+template <typename T>
+class AtomicSharedPtr {
+   public:
+    using Pointer = std::shared_ptr<const T>;
+
+    AtomicSharedPtr() = default;
+    explicit AtomicSharedPtr(T value) : ptr_(std::make_shared<const T>(std::move(value))) {}
+
+    auto load() const { return std::atomic_load(&ptr_); }
+
+    void store(Pointer&& newPtr) { std::atomic_store(&ptr_, std::move(newPtr)); }
+
+   private:
+    std::shared_ptr<const T> ptr_;
+};
+
+}  // namespace pulsar

lib/BinaryProtoLookupService.h

@@ -22,6 +22,7 @@
 #include <pulsar/Authentication.h>
 #include <pulsar/ClientConfiguration.h>
 #include <pulsar/Schema.h>
+#include <pulsar/ServiceInfo.h>
 
 #include <mutex>
 
@@ -38,9 +39,9 @@ using GetSchemaPromisePtr = std::shared_ptr<Promise<Result, SchemaInfo>>;
 
 class PULSAR_PUBLIC BinaryProtoLookupService : public LookupService {
    public:
-    BinaryProtoLookupService(const std::string& serviceUrl, ConnectionPool& pool,
+    BinaryProtoLookupService(const ServiceInfo& serviceInfo, ConnectionPool& pool,
                              const ClientConfiguration& clientConfiguration)
-        : serviceNameResolver_(serviceUrl),
+        : serviceNameResolver_(serviceInfo.serviceUrl()),
           cnxPool_(pool),
           listenerName_(clientConfiguration.getListenerName()),
           maxLookupRedirects_(clientConfiguration.getMaxLookupRedirects()) {}

lib/Client.cc

@@ -17,6 +17,7 @@
  * under the License.
  */
 #include <pulsar/Client.h>
+#include <pulsar/ServiceInfoProvider.h>
 
 #include <iostream>
 #include <memory>
@@ -33,13 +34,17 @@ DECLARE_LOG_OBJECT()
 
 namespace pulsar {
 
-Client::Client(const std::shared_ptr<ClientImpl>& impl) : impl_(impl) {}
+Client::Client(const std::shared_ptr<ClientImpl>& impl) : impl_(impl) { impl_->initialize(); }
 
-Client::Client(const std::string& serviceUrl)
-    : impl_(std::make_shared<ClientImpl>(serviceUrl, ClientConfiguration())) {}
+Client::Client(const std::string& serviceUrl) : Client(serviceUrl, ClientConfiguration()) {}
 
 Client::Client(const std::string& serviceUrl, const ClientConfiguration& clientConfiguration)
-    : impl_(std::make_shared<ClientImpl>(serviceUrl, clientConfiguration)) {}
+    : Client(std::make_shared<ClientImpl>(serviceUrl, clientConfiguration)) {}
+
+Client Client::create(std::unique_ptr<ServiceInfoProvider> serviceInfoProvider,
+                      const ClientConfiguration& clientConfiguration) {
+    return Client(std::make_shared<ClientImpl>(std::move(serviceInfoProvider), clientConfiguration));
+}
 
 Result Client::createProducer(const std::string& topic, Producer& producer) {
     return createProducer(topic, ProducerConfiguration(), producer);
@@ -193,8 +198,10 @@ uint64_t Client::getNumberOfConsumers() { return impl_->getNumberOfConsumers();
 
 void Client::getSchemaInfoAsync(const std::string& topic, int64_t version,
                                 std::function<void(Result, const SchemaInfo&)> callback) {
-    impl_->getLookup()
-        ->getSchema(TopicName::get(topic), (version >= 0) ? toBigEndianBytes(version) : "")
+    impl_->getSchema(TopicName::get(topic), (version >= 0) ? toBigEndianBytes(version) : "")
         .addListener(std::move(callback));
 }
+
+ServiceInfo Client::getServiceInfo() const { return impl_->getServiceInfo(); }
+
 }  // namespace pulsar

lib/ClientConfiguration.cc

@@ -57,8 +57,6 @@ ClientConfiguration& ClientConfiguration::setAuth(const AuthenticationPtr& authe
     return *this;
 }
 
-Authentication& ClientConfiguration::getAuth() const { return *impl_->authenticationPtr; }
-
 const AuthenticationPtr& ClientConfiguration::getAuthPtr() const { return impl_->authenticationPtr; }
 
 ClientConfiguration& ClientConfiguration::setOperationTimeoutSeconds(int timeout) {
@@ -94,13 +92,6 @@ ClientConfiguration& ClientConfiguration::setMessageListenerThreads(int threads)
 
 int ClientConfiguration::getMessageListenerThreads() const { return impl_->messageListenerThreads; }
 
-ClientConfiguration& ClientConfiguration::setUseTls(bool useTls) {
-    impl_->useTls = useTls;
-    return *this;
-}
-
-bool ClientConfiguration::isUseTls() const { return impl_->useTls; }
-
 ClientConfiguration& ClientConfiguration::setValidateHostName(bool validateHostName) {
     impl_->validateHostName = validateHostName;
     return *this;
@@ -131,10 +122,6 @@ ClientConfiguration& ClientConfiguration::setTlsTrustCertsFilePath(const std::st
     return *this;
 }
 
-const std::string& ClientConfiguration::getTlsTrustCertsFilePath() const {
-    return impl_->tlsTrustCertsFilePath;
-}
-
 ClientConfiguration& ClientConfiguration::setTlsAllowInsecureConnection(bool allowInsecure) {
     impl_->tlsAllowInsecureConnection = allowInsecure;
     return *this;