add reauthn

Signed-off-by: Emelia Lei <[email protected]>

Author: emelialei88

src/groups/bmq/bmqimp/bmqimp_application.cpp

@@ -720,7 +720,21 @@ int Application::start(const bsls::TimeInterval& timeout)
                   << "::: START (SYNC) << [state: " << d_brokerSession.state()
                   << "] :::";
 
-    return d_brokerSession.start(timeout);
+    int rc = d_initialConnectionChannelFactory.start();
+    if (rc != 0) {
+        BALL_LOG_ERROR << id()
+                       << "Failed to start initialConnectionChannelFactory "
+                       << "[rc: " << rc << "]";
+        return rc;  // RETURN
+    }
+
+    rc = d_brokerSession.start(timeout);
+    if (rc != 0) {
+        d_initialConnectionChannelFactory.stop();
+        return rc;  // RETURN
+    }
+
+    return rc;
 }
 
 int Application::startAsync(const bsls::TimeInterval& timeout)
@@ -755,6 +769,9 @@ void Application::stop()
 
     // Stop the brokerSession
     d_brokerSession.stop();
+
+    // Stop the channel factories
+    d_initialConnectionChannelFactory.stop();
 }
 
 void Application::stopAsync()

src/groups/bmq/bmqimp/bmqimp_initialconnectionchannelfactory.cpp

@@ -21,6 +21,7 @@
 #include <bmqp_event.h>
 #include <bmqp_protocol.h>
 #include <bmqp_schemaeventbuilder.h>
+#include <bmqsys_time.h>
 
 #include <bmqio_channelutil.h>
 #include <bmqu_blob.h>
@@ -32,12 +33,14 @@
 #include <bdlf_bind.h>
 #include <bdlf_placeholder.h>
 #include <bdlma_localsequentialallocator.h>
+#include <bsl_algorithm.h>
 #include <bsl_iostream.h>
 #include <bsl_memory.h>
 #include <bsla_annotations.h>
 #include <bsla_unused.h>
 #include <bslma_default.h>
 #include <bsls_assert.h>
+#include <bsls_timeinterval.h>
 
 namespace BloombergLP {
 namespace bmqimp {
@@ -277,7 +280,7 @@ void InitialConnectionChannelFactory::readPacketsCb(
     const ResultCallback&                  cb,
     const bmqio::Status&                   status,
     int*                                   numNeeded,
-    bdlbb::Blob*                           blob) const
+    bdlbb::Blob*                           blob)
 {
     BALL_LOG_INFO << "At readPacketsCb";
 
@@ -441,7 +444,7 @@ int InitialConnectionChannelFactory::decodeInitialConnectionMessage(
 void InitialConnectionChannelFactory::onBrokerAuthenticationResponse(
     const bmqp_ctrlmsg::AuthenticationMessage& response,
     const ResultCallback&                      cb,
-    const bsl::shared_ptr<bmqio::Channel>&     channel) const
+    const bsl::shared_ptr<bmqio::Channel>&     channel)
 {
     const bmqp_ctrlmsg::AuthenticateResponse& authenticateResponse =
         response.authenticateResponse();
@@ -461,7 +464,22 @@ void InitialConnectionChannelFactory::onBrokerAuthenticationResponse(
     // Authentication SUCCEEDED
     BALL_LOG_INFO << "Authentication with broker was successful: " << response;
 
-    // TODO: reauthenticate
+    // Schedule recurring reauthentication events if lifetime is specified in
+    // the response.
+    if (authenticateResponse.lifetimeMs()) {
+        int lifetimeMs = authenticateResponse.lifetimeMs().value();
+        int intervalMs = bsl::min(lifetimeMs - k_REAUTHN_EARLY_BUFFER,
+                                  lifetimeMs * k_REAUTHN_EARLY_RATIO);
+        d_scheduler.scheduleRecurringEvent(
+            &d_authnEventHandle,
+            bsls::TimeInterval(intervalMs),
+            bdlf::BindUtil::bind(
+                bmqu::WeakMemFnUtil::weakMemFn(
+                    &InitialConnectionChannelFactory::authenticate,
+                    d_self.acquireWeak()),
+                channel,
+                cb));
+    }
 
     negotiate(channel, cb);
 }
@@ -549,6 +567,8 @@ InitialConnectionChannelFactory::InitialConnectionChannelFactory(
     const Config&     config,
     bslma::Allocator* basicAllocator)
 : d_config(config, basicAllocator)
+, d_scheduler(bsls::SystemClockType::e_MONOTONIC, basicAllocator)
+, d_authnEventHandle()
 , d_self(this)  // use default allocator
 {
     // NOTHING
@@ -561,6 +581,22 @@ InitialConnectionChannelFactory::~InitialConnectionChannelFactory()
     d_self.invalidate();
 }
 
+int InitialConnectionChannelFactory::start()
+{
+    int rc = d_scheduler.start();
+    if (rc != 0) {
+        BALL_LOG_ERROR << "Failed to start event scheduler [rc: " << rc << "]";
+    }
+    return rc;
+}
+
+void InitialConnectionChannelFactory::stop()
+{
+    // Cancel any scheduled reauthentication events
+    d_scheduler.cancelEvent(&d_authnEventHandle);
+    d_scheduler.stop();
+}
+
 // MANIPULATORS
 void InitialConnectionChannelFactory::listen(
     BSLA_UNUSED bmqio::Status* status,

src/groups/bmq/bmqimp/bmqimp_initialconnectionchannelfactory.h

@@ -41,6 +41,7 @@
 
 // BDE
 #include <bdlbb_blob.h>
+#include <bdlmt_eventscheduler.h>
 #include <bsl_functional.h>
 #include <bslma_allocator.h>
 #include <bslma_usesbslmaallocator.h>
@@ -104,6 +105,8 @@ class InitialConnectionChannelFactory : public bmqio::ChannelFactory {
     // TYPES
     typedef InitialConnectionChannelFactoryConfig Config;
 
+    typedef bdlmt::EventScheduler::RecurringEventHandle EventHandle;
+
     // CONSTANTS
 
     /// Name of a property set on the channel representing the broker's
@@ -118,13 +121,25 @@ class InitialConnectionChannelFactory : public bmqio::ChannelFactory {
 
     static const char* k_CHANNEL_PROPERTY_MAX_MISSED_HEARTBEATS;
 
+    // Minimum buffer to subtract from lifetimeMs to avoid cutting too close
+    const int k_REAUTHN_EARLY_BUFFER = 1000;  // 1 second
+
+    // Proportion of lifetimeMs after which to initiate reauthentication.
+    const int k_REAUTHN_EARLY_RATIO = 0.9;
+
   private:
     // TYPES
     enum ACTION { AUTHENTICATION = 0, NEGOTIATION = 1 };
 
     // PRIVATE DATA
     Config d_config;
 
+    // Used to schedule events for sending reauthentication requests.
+    bdlmt::EventScheduler d_scheduler;
+
+    // Event handle for reauthentication events.
+    EventHandle d_authnEventHandle;
+
     // Used to make sure no callback is invoked on a destroyed object.
     mutable bmqu::SharedResource<InitialConnectionChannelFactory> d_self;
 
@@ -167,7 +182,7 @@ class InitialConnectionChannelFactory : public bmqio::ChannelFactory {
                        const ResultCallback&                  cb,
                        const bmqio::Status&                   status,
                        int*                                   numNeeded,
-                       bdlbb::Blob*                           blob) const;
+                       bdlbb::Blob*                           blob);
 
     int decodeInitialConnectionMessage(
         const bdlbb::Blob&                                  packet,
@@ -179,7 +194,7 @@ class InitialConnectionChannelFactory : public bmqio::ChannelFactory {
     void onBrokerAuthenticationResponse(
         const bmqp_ctrlmsg::AuthenticationMessage& response,
         const ResultCallback&                      cb,
-        const bsl::shared_ptr<bmqio::Channel>&     channel) const;
+        const bsl::shared_ptr<bmqio::Channel>&     channel);
 
     void onBrokerNegotiationResponse(
         const bmqp_ctrlmsg::NegotiationMessage& response,
@@ -196,6 +211,10 @@ class InitialConnectionChannelFactory : public bmqio::ChannelFactory {
 
   public:
     // MANIPULATORS
+    int start();
+
+    void stop();
+
     void listen(bmqio::Status*               status,
                 bslma::ManagedPtr<OpHandle>* handle,
                 const bmqio::ListenOptions&  options,