clean up implementation

Author: BewareMyPower

include/pulsar/ServiceInfoProvider.h

@@ -19,10 +19,9 @@
 #ifndef PULSAR_SERVICE_INFO_PROVIDER_H_
 #define PULSAR_SERVICE_INFO_PROVIDER_H_
 
+#include <pulsar/ClientConfiguration.h>
 #include <pulsar/ServiceInfo.h>
 
-#include <functional>
-
 namespace pulsar {
 
 class PULSAR_PUBLIC ServiceInfoProvider {
@@ -34,11 +33,12 @@ class PULSAR_PUBLIC ServiceInfoProvider {
     virtual ~ServiceInfoProvider() = default;
 
     /**
-     * Get the current `ServiceInfo` connection for the client.
-     * This method is called **only once** internally when the `Client` is being initialized, and the client
-     * will use the returned `ServiceInfo` to establish the initial connection to the Pulsar cluster.
+     * 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. Since it's only called once, it's legal to return a
+     * moved `ServiceInfo` object to avoid unnecessary copying.
      */
-    virtual ServiceInfo getServiceInfo() const = 0;
+    virtual ServiceInfo initialServiceInfo() = 0;
 
     /**
      * Initialize the ServiceInfoProvider.

lib/AtomicSharedPtr.h

@@ -27,6 +27,9 @@ 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)); }

lib/ClientConnection.cc

@@ -182,8 +182,6 @@ static bool file_exists(const std::string& path) {
 
 std::atomic<int32_t> ClientConnection::maxMessageSize_{Commands::DefaultMaxMessageSize};
 
-// NOTE: we should get the connection info from `serviceInfo` rather than `clientConfiguration` because it can
-// be modified dynamically.
 ClientConnection::ClientConnection(const std::string& logicalAddress, const std::string& physicalAddress,
                                    const ServiceInfo& serviceInfo, const ExecutorServicePtr& executor,
                                    const ClientConfiguration& clientConfiguration,

lib/ClientImpl.cc

@@ -111,6 +111,7 @@ ClientImpl::ClientImpl(std::unique_ptr<ServiceInfoProvider> serviceInfoProvider,
     : serviceInfoProvider_(std::move(serviceInfoProvider)),
       state_(Open),
       clientConfiguration_(clientConfiguration),
+      serviceInfo_(serviceInfoProvider_->initialServiceInfo()),
       memoryLimitController_(clientConfiguration.getMemoryLimit()),
       ioExecutorProvider_(std::make_shared<ExecutorServiceProvider>(clientConfiguration_.getIOThreads())),
       listenerExecutorProvider_(
@@ -130,7 +131,6 @@ ClientImpl::ClientImpl(std::unique_ptr<ServiceInfoProvider> serviceInfoProvider,
         LogUtils::setLoggerFactory(std::move(loggerFactory));
     }
 
-    serviceInfo_.store(std::make_shared<const ServiceInfo>(serviceInfoProvider_->getServiceInfo()));
     lookupServicePtr_ = createLookup(*serviceInfo_.load());
 }
 
@@ -768,8 +768,8 @@ void ClientImpl::handleClose(Result result, const SharedInt& numberOfOpenHandler
         }
 
         LOG_DEBUG("Shutting down producers and consumers for client");
-        // handleClose() is called in ExecutorService's event loop, while shutdown() tried to wait the
-        // event loop exits. So here we use another thread to call shutdown().
+        // handleClose() is called in ExecutorService's event loop, while shutdown() tried to wait the event
+        // loop exits. So here we use another thread to call shutdown().
         auto self = shared_from_this();
         std::thread shutdownTask{[this, self, callback] {
             shutdown();

lib/DefaultServiceUrlProvider.h

@@ -31,7 +31,7 @@ class DefaultServiceUrlProvider : public ServiceInfoProvider {
     DefaultServiceUrlProvider(const std::string& serviceUrl, const ClientConfigurationImpl& config)
         : serviceInfo_(config.toServiceInfo(serviceUrl)) {}
 
-    ServiceInfo getServiceInfo() const override { return serviceInfo_; }
+    ServiceInfo initialServiceInfo() override { return std::move(serviceInfo_); }
 
     void initialize(std::function<void(ServiceInfo)> onServiceInfoUpdate) override {}
 

tests/ClientTest.cc

@@ -17,13 +17,13 @@
  * under the License.
  */
 #include <gtest/gtest.h>
-#include <pulsar/Authentication.h>
 #include <pulsar/Client.h>
 #include <pulsar/Version.h>
 
 #include <algorithm>
 #include <chrono>
 #include <future>
+#include <sstream>
 
 #include "MockClientImpl.h"
 #include "PulsarAdminHelper.h"
@@ -32,6 +32,7 @@
 #include "lib/ClientConnection.h"
 #include "lib/LogUtils.h"
 #include "lib/checksum/ChecksumProvider.h"
+#include "lib/stats/ProducerStatsImpl.h"
 
 DECLARE_LOG_OBJECT()
 

tests/ServiceInfoProviderTest.cc

@@ -20,11 +20,9 @@
 #include <pulsar/Client.h>
 
 #include <atomic>
-#include <condition_variable>
 #include <memory>
 #include <mutex>
-#include <queue>
-#include <stdexcept>
+#include <optional>
 #include <thread>
 
 #include "PulsarFriend.h"
@@ -36,70 +34,62 @@ DECLARE_LOG_OBJECT()
 using namespace pulsar;
 using namespace std::chrono_literals;
 
-class ServiceInfoQueue {
+class ServiceInfoHolder {
    public:
-    void push(ServiceInfo info) {
-        {
-            std::lock_guard<std::mutex> lock(mutex_);
-            queue_.push(std::move(info));
-        }
-        cond_.notify_all();
-    }
+    ServiceInfoHolder(ServiceInfo info) : serviceInfo_(std::move(info)) {}
 
-    ServiceInfo pop() {
-        std::unique_lock<std::mutex> lock(mutex_);
-        cond_.wait(lock, [this] { return !queue_.empty() || !running_; });
-        if (queue_.empty()) {
-            throw std::runtime_error("Queue is closed");
+    std::optional<ServiceInfo> getUpdatedValue() {
+        std::lock_guard lock(mutex_);
+        if (!owned_) {
+            return std::nullopt;
         }
-
-        ServiceInfo info = std::move(queue_.front());
-        queue_.pop();
-        return info;
+        owned_ = false;
+        return std::move(serviceInfo_);
     }
 
-    void close() {
-        running_ = false;
-        cond_.notify_all();
+    void updateValue(ServiceInfo info) {
+        std::lock_guard lock(mutex_);
+        serviceInfo_ = std::move(info);
+        owned_ = true;
     }
 
    private:
+    ServiceInfo serviceInfo_;
+    bool owned_{true};
+
     mutable std::mutex mutex_;
-    mutable std::condition_variable cond_;
-    std::queue<ServiceInfo> queue_;
-    std::atomic_bool running_{true};
 };
 
 class TestServiceInfoProvider : public ServiceInfoProvider {
    public:
-    TestServiceInfoProvider(ServiceInfoQueue &queue) : queue_(queue) {}
+    TestServiceInfoProvider(ServiceInfoHolder &serviceInfo) : serviceInfo_(serviceInfo) {}
 
-    ServiceInfo getServiceInfo() const override { return queue_.pop(); }
+    ServiceInfo initialServiceInfo() override { return serviceInfo_.getUpdatedValue().value(); }
 
     void initialize(std::function<void(ServiceInfo)> onServiceInfoUpdate) override {
-        onServiceInfoUpdate(queue_.pop());
         thread_ = std::thread([this, onServiceInfoUpdate] {
-            try {
-                while (true) {
-                    ServiceInfo info = queue_.pop();
-                    onServiceInfoUpdate(std::move(info));
+            while (running_) {
+                auto updatedValue = serviceInfo_.getUpdatedValue();
+                if (updatedValue) {
+                    onServiceInfoUpdate(std::move(*updatedValue));
                 }
-            } catch (const std::runtime_error &) {
+                // Use a tight wait loop for tests
+                std::this_thread::sleep_for(10ms);
             }
         });
     }
 
     ~TestServiceInfoProvider() override {
-        queue_.close();
+        running_ = false;
         if (thread_.joinable()) {
             thread_.join();
         }
     }
 
    private:
-    ServiceInfoQueue &queue_;
-
     std::thread thread_;
+    ServiceInfoHolder &serviceInfo_;
+    std::atomic_bool running_{true};
     mutable std::mutex mutex_;
 };
 
@@ -114,10 +104,8 @@ TEST(ServiceInfoProviderTest, testSwitchCluster) {
     // Access "public/default" namespace in cluster 1, which doesn't require authentication
     ServiceInfo info3{"pulsar://localhost:6650"};
 
-    ServiceInfoQueue queue;
-    queue.push(info1);
-
-    auto client = Client::create(std::make_unique<TestServiceInfoProvider>(queue), {});
+    ServiceInfoHolder serviceInfo{info1};
+    auto client = Client::create(std::make_unique<TestServiceInfoProvider>(serviceInfo), {});
 
     const auto topicRequiredAuth = "private/auth/testUpdateConnectionInfo-" + std::to_string(time(nullptr));
     Producer producer;
@@ -141,7 +129,7 @@ TEST(ServiceInfoProviderTest, testSwitchCluster) {
 
     // Switch to cluster 2 (started by ./build-support/start-mim-test-service-inside-container.sh)
     ASSERT_FALSE(PulsarFriend::getConnections(client).empty());
-    queue.push(info2);
+    serviceInfo.updateValue(info2);
     ASSERT_TRUE(waitUntil(1s, [&] {
         return PulsarFriend::getConnections(client).empty() && client.getServiceInfo() == info2;
     }));
@@ -152,7 +140,7 @@ TEST(ServiceInfoProviderTest, testSwitchCluster) {
     // Switch back to cluster 1 without any authentication, the previous authentication info configured for
     // cluster 2 will be cleared.
     ASSERT_FALSE(PulsarFriend::getConnections(client).empty());
-    queue.push(info3);
+    serviceInfo.updateValue(info3);
     ASSERT_TRUE(waitUntil(1s, [&] {
         return PulsarFriend::getConnections(client).empty() && client.getServiceInfo() == info3;
     }));