Expose keep alive interval for c and c++ client (#457)

* Expose keep alive interval for c and c++ client

* Optimize

Author: shibd

include/pulsar/ClientConfiguration.h

@@ -356,6 +356,19 @@ class PULSAR_PUBLIC ClientConfiguration {
      */
     int getConnectionTimeout() const;
 
+    /**
+     * Set keep alive interval for each client-broker-connection. <i>(default: 30 seconds)</i>.
+     *
+     * @param keepAliveIntervalInSeconds
+     * @return
+     */
+    ClientConfiguration& setKeepAliveIntervalInSeconds(unsigned int keepAliveIntervalInSeconds);
+
+    /**
+     * The getter associated with setKeepAliveIntervalInSeconds().
+     */
+    unsigned int getKeepAliveIntervalInSeconds() const;
+
     friend class ClientImpl;
     friend class PulsarWrapper;
 

include/pulsar/c/client_configuration.h

@@ -204,6 +204,12 @@ PULSAR_PUBLIC const unsigned int pulsar_client_configuration_get_partitions_upda
 PULSAR_PUBLIC const unsigned int pulsar_client_configuration_get_stats_interval_in_seconds(
     pulsar_client_configuration_t *conf);
 
+PULSAR_PUBLIC void pulsar_client_configuration_set_keep_alive_interval_in_seconds(
+    pulsar_client_configuration_t *conf, unsigned int keepAliveIntervalInSeconds);
+
+PULSAR_PUBLIC unsigned int pulsar_client_configuration_get_keep_alive_interval_in_seconds(
+    pulsar_client_configuration_t *conf);
+
 #ifdef __cplusplus
 }
 #endif

lib/ClientConfiguration.cc

@@ -214,6 +214,16 @@ ClientConfiguration& ClientConfiguration::setConnectionTimeout(int timeoutMs) {
 
 int ClientConfiguration::getConnectionTimeout() const { return impl_->connectionTimeoutMs; }
 
+ClientConfiguration& ClientConfiguration::setKeepAliveIntervalInSeconds(
+    unsigned int keepAliveIntervalInSeconds) {
+    impl_->keepAliveIntervalInSeconds = keepAliveIntervalInSeconds;
+    return *this;
+}
+
+unsigned int ClientConfiguration::getKeepAliveIntervalInSeconds() const {
+    return impl_->keepAliveIntervalInSeconds;
+}
+
 ClientConfiguration& ClientConfiguration::setDescription(const std::string& description) {
     if (description.length() > 64) {
         throw std::invalid_argument("The description length exceeds 64");

lib/ClientConfigurationImpl.h

@@ -47,6 +47,7 @@ struct ClientConfigurationImpl {
     unsigned int partitionsUpdateInterval{60};  // 1 minute
     std::string listenerName;
     int connectionTimeoutMs{10000};  // 10 seconds
+    unsigned int keepAliveIntervalInSeconds{30};
     std::string description;
     std::string proxyServiceUrl;
     ClientConfiguration::ProxyProtocol proxyProtocol;

lib/ClientConnection.cc

@@ -51,8 +51,6 @@ using proto::BaseCommand;
 
 static const uint32_t DefaultBufferSize = 64 * 1024;
 
-static const int KeepAliveIntervalInSeconds = 30;
-
 static MessageId toMessageId(const proto::MessageIdData& messageIdData) {
     return MessageIdBuilder::from(messageIdData).build();
 }
@@ -186,6 +184,7 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
       connectTimeoutTask_(
           std::make_shared<PeriodicTask>(*executor_, clientConfiguration.getConnectionTimeout())),
       outgoingBuffer_(SharedBuffer::allocate(DefaultBufferSize)),
+      keepAliveIntervalInSeconds_(clientConfiguration.getKeepAliveIntervalInSeconds()),
       consumerStatsRequestTimer_(executor_->createDeadlineTimer()),
       maxPendingLookupRequest_(clientConfiguration.getConcurrentLookupRequest()),
       clientVersion_(clientVersion),
@@ -310,7 +309,7 @@ void ClientConnection::handlePulsarConnected(const proto::CommandConnected& cmdC
         // Only send keep-alive probes if the broker supports it
         keepAliveTimer_ = executor_->createDeadlineTimer();
         if (keepAliveTimer_) {
-            keepAliveTimer_->expires_from_now(std::chrono::seconds(KeepAliveIntervalInSeconds));
+            keepAliveTimer_->expires_from_now(std::chrono::seconds(keepAliveIntervalInSeconds_));
             auto weakSelf = weak_from_this();
             keepAliveTimer_->async_wait([weakSelf](const ASIO_ERROR&) {
                 auto self = weakSelf.lock();
@@ -1245,7 +1244,7 @@ void ClientConnection::handleKeepAliveTimeout() {
         // be zero And we do not attempt to dereference the pointer.
         Lock lock(mutex_);
         if (keepAliveTimer_) {
-            keepAliveTimer_->expires_from_now(std::chrono::seconds(KeepAliveIntervalInSeconds));
+            keepAliveTimer_->expires_from_now(std::chrono::seconds(keepAliveIntervalInSeconds_));
             auto weakSelf = weak_from_this();
             keepAliveTimer_->async_wait([weakSelf](const ASIO_ERROR&) {
                 auto self = weakSelf.lock();

lib/ClientConnection.h

@@ -388,6 +388,7 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
     // Signals whether we're waiting for a response from broker
     bool havePendingPingRequest_ = false;
     bool isSniProxy_ = false;
+    unsigned int keepAliveIntervalInSeconds_;
     DeadlineTimerPtr keepAliveTimer_;
     DeadlineTimerPtr consumerStatsRequestTimer_;
 

lib/c/c_ClientConfiguration.cc

@@ -208,3 +208,13 @@ const unsigned int pulsar_client_configuration_get_partitions_update_interval(
     pulsar_client_configuration_t *conf) {
     return conf->conf.getPartitionsUpdateInterval();
 }
+
+void pulsar_client_configuration_set_keep_alive_interval_in_seconds(pulsar_client_configuration_t *conf,
+                                                                    unsigned int keepAliveIntervalInSeconds) {
+    conf->conf.setKeepAliveIntervalInSeconds(keepAliveIntervalInSeconds);
+}
+
+unsigned int pulsar_client_configuration_get_keep_alive_interval_in_seconds(
+    pulsar_client_configuration_t *conf) {
+    return conf->conf.getKeepAliveIntervalInSeconds();
+}

tests/c/c_ClientConfigurationTest.cc

@@ -34,4 +34,7 @@ TEST(C_ClientConfigurationTest, testCApiConfig) {
 
     pulsar_client_configuration_set_partitions_update_interval(conf, 10);
     ASSERT_EQ(pulsar_client_configuration_get_partitions_update_interval(conf), 10);
+
+    pulsar_client_configuration_set_keep_alive_interval_in_seconds(conf, 60);
+    ASSERT_EQ(pulsar_client_configuration_get_keep_alive_interval_in_seconds(conf), 60);
 }