WIP: bmqbrkrcfg.json and authn timeout

Author: emelialei88

src/applications/bmqbrkr/etc/bmqbrkrcfg.json

@@ -92,6 +92,23 @@
         },
         "bmqconfConfig": {
             "cacheTTLSeconds": 30
+        },
+        "plugins": {
+            "libraries": ["/Users/wlei29/Workspace/blazingmq/build/blazingmq/src/plugins/"],
+            "enabled": ["PassAuthenticator"]
+        },
+        "authentication": {
+            "plugins": [
+                {
+                    "name": "FailAuthenticator",
+                    "configs": []
+                },
+                {
+                    "name": "PassAuthenticator",
+                    "configs": []
+                }
+            ],
+            "fallbackPrincipal": "defaultFallbackPrincipal"
         }
     }
 }

src/groups/mqb/mqba/mqba_authenticator.cpp

@@ -14,6 +14,7 @@
 // limitations under the License.
 
 // mqba_authenticator.h                                           -*-C++-*-
+#include <bdlmt_timereventscheduler.h>
 #include <mqba_authenticator.h>
 
 #include <mqbscm_version.h>
@@ -284,6 +285,14 @@ void Authenticator::authenticate(
                 bsl::shared_ptr<mqbnet::Session>());
         }
     }
+
+    // Set off the timer
+    if (result->lifetimeMs().has_value()) {
+        bdlmt::TimerEventScheduler::Handle handle = d_scheduler.scheduleEvent(
+            bsls::TimeInterval(result->lifetimeMs().value()),
+            bdlf::BindUtil::bind(&Authenticator::timeout, this, channel));
+        context->setAuthenticationTimerHandle(handle);
+    }
 }
 
 int Authenticator::reAuthenticateAsync(
@@ -398,6 +407,39 @@ void Authenticator::reAuthenticate(
             channel->close(status);
         }
     }
+
+    // Cancel the timer or reschedule
+    {
+        bslmt::LockGuard<bslmt::Mutex> guard(
+            &context->authenticationTimerHandleMutex());  // MUTEX LOCKED
+
+        bdlmt::TimerEventScheduler::Handle handle =
+            context->authenticationTimerHandle();
+
+        // If the authenticated channel has already timed out, simply return
+        rc = d_scheduler.cancelEvent(handle);
+        if (rc != 0) {
+            return;  // RETURN
+        }
+
+        if (result->lifetimeMs().has_value()) {
+            rc = d_scheduler.rescheduleEvent(
+                handle,
+                bsls::TimeInterval(result->lifetimeMs().value()));
+
+            if (rc != 0) {
+                BALL_LOG_ERROR
+                    << "Failed to reschedule authentication timer for '"
+                    << channel->peerUri() << "' [rc: " << rc
+                    << ", lifetime: " << result->lifetimeMs().value() << "]";
+                bmqio::Status status(bmqio::StatusCategory::e_GENERIC_ERROR,
+                                     "reAuthenticationError",
+                                     rc,
+                                     d_allocator_p);
+                channel->close(status);
+            }
+        }
+    }
 }
 
 // CREATORS
@@ -411,6 +453,8 @@ Authenticator::Authenticator(
                100,                                          // max threads
                bsls::TimeInterval(120).totalMilliseconds(),  // idle time
                allocator)
+, d_scheduler(bdlmt::TimerEventScheduler(bsls::SystemClockType::e_MONOTONIC,
+                                         allocator))
 , d_blobSpPool_p(blobSpPool)
 , d_allocator_p(allocator)
 {
@@ -432,6 +476,15 @@ int Authenticator::start(bsl::ostream& errorDescription)
         return rc;  // RETURN
     }
 
+    rc = d_scheduler.start();
+    if (rc != 0) {
+        errorDescription << "Failed to start TimerEventScheduler for "
+                            "Authenticator [rc: "
+                         << rc << "]";
+        d_threadPool.stop();
+        return rc;  // RETURN
+    }
+
     return 0;
 }
 
@@ -514,6 +567,18 @@ int Authenticator::handleReauthentication(
     return rc;
 }
 
+void Authenticator::timeout(const AuthenticationContextSp&         context,
+                            const bsl::shared_ptr<bmqio::Channel>& channel)
+{
+    bmqio::Status status(bmqio::StatusCategory::e_TIMEOUT,
+                         "authenticationTimeout",
+                         -1,
+                         d_allocator_p);
+    context->setAuthenticationTimerHandle(
+        bdlmt::TimerEventScheduler::Handle());
+    channel->close(status);
+}
+
 int Authenticator::authenticationOutboundOrReverse(
     const AuthenticationContextSp& context)
 {

src/groups/mqb/mqba/mqba_authenticator.h

@@ -47,6 +47,7 @@
 #include <bdlbb_blob.h>
 #include <bdlcc_sharedobjectpool.h>
 #include <bdlmt_threadpool.h>
+#include <bdlmt_timereventscheduler.h>
 #include <bsl_memory.h>
 #include <bsl_ostream.h>
 #include <bslma_allocator.h>
@@ -98,6 +99,8 @@ class Authenticator : public mqbnet::Authenticator {
 
     bdlmt::ThreadPool d_threadPool;
 
+    bdlmt::TimerEventScheduler d_scheduler;
+
     BlobSpPool* d_blobSpPool_p;
 
     /// Allocator to use.
@@ -193,6 +196,9 @@ class Authenticator : public mqbnet::Authenticator {
                                const AuthenticationContextSp& context,
                                const bsl::shared_ptr<bmqio::Channel>& channel);
 
+    void timeout(const AuthenticationContextSp&         context,
+                 const bsl::shared_ptr<bmqio::Channel>& channel);
+
   public:
     // TRAITS
     BSLMF_NESTED_TRAIT_DECLARATION(Authenticator, bslma::UsesBslmaAllocator)

src/groups/mqb/mqbnet/mqbnet_authenticationcontext.cpp

@@ -59,6 +59,23 @@ AuthenticationContext& AuthenticationContext::setAuthenticationResult(
     return *this;
 }
 
+AuthenticationContext& AuthenticationContext::setAuthenticationTimerHandle(
+    bdlmt::TimerEventScheduler::Handle value)
+{
+    d_authenticationTimerHandle = value;
+    return *this;
+}
+
+AuthenticationContext&
+AuthenticationContext::setAuthenticationTimerHandleLocked(
+    bdlmt::TimerEventScheduler::Handle value)
+{
+    bslmt::LockGuard<bslmt::Mutex> guard(&d_timerHandleMutex);  // MUTEX LOCKED
+
+    d_authenticationTimerHandle = value;
+    return *this;
+}
+
 AuthenticationContext& AuthenticationContext::setInitialConnectionContext(
     InitialConnectionContext* value)
 {
@@ -113,6 +130,17 @@ AuthenticationContext::authenticationResult() const
     return d_authenticationResultSp;
 }
 
+bdlmt::TimerEventScheduler::Handle
+AuthenticationContext::authenticationTimerHandle() const
+{
+    return d_authenticationTimerHandle;
+}
+
+bslmt::Mutex& AuthenticationContext::authenticationTimerHandleMutex()
+{
+    return d_timerHandleMutex;
+}
+
 InitialConnectionContext*
 AuthenticationContext::initialConnectionContext() const
 {

src/groups/mqb/mqbnet/mqbnet_authenticationcontext.h

@@ -23,13 +23,14 @@
 ///
 
 // MQB
-#include "bmqp_protocol.h"
 #include <mqbnet_initialconnectioncontext.h>
 
 // BMQ
 #include <bmqp_ctrlmsg_messages.h>
+#include <bmqp_protocol.h>
 
 // BDE
+#include <bdlmt_timereventscheduler.h>
 #include <bslmt_mutex.h>
 #include <bsls_atomic.h>
 
@@ -63,6 +64,12 @@ class AuthenticationContext {
     /// during re-authentication.
     bsl::shared_ptr<mqbplug::AuthenticationResult> d_authenticationResultSp;
 
+    /// The timer handle for the authentication timeout.
+    bdlmt::TimerEventScheduler::Handle d_authenticationTimerHandle;
+
+    /// The mutex to protect the AuthenticationTimerHandle and
+    bslmt::Mutex d_timerHandleMutex;
+
     /// The mutex to protect the AuthenticationResult.
     mutable bslmt::Mutex d_mutex;
 
@@ -105,6 +112,10 @@ class AuthenticationContext {
     AuthenticationContext& setAuthenticationResult(
         const bsl::shared_ptr<mqbplug::AuthenticationResult>& value);
     AuthenticationContext&
+    setAuthenticationTimerHandle(bdlmt::TimerEventScheduler::Handle value);
+    AuthenticationContext& setAuthenticationTimerHandleLocked(
+        bdlmt::TimerEventScheduler::Handle value);
+    AuthenticationContext&
     setInitialConnectionContext(InitialConnectionContext* value);
     AuthenticationContext&
     setAuthenticationMessage(const bmqp_ctrlmsg::AuthenticationMessage& value);
@@ -124,6 +135,12 @@ class AuthenticationContext {
     const bsl::shared_ptr<mqbplug::AuthenticationResult>&
     authenticationResult() const;
 
+    /// This function holds a mutex lock while accessing the
+    /// `d_authenticationTimerHandle` to ensure thread safety.
+    bdlmt::TimerEventScheduler::Handle authenticationTimerHandle() const;
+
+    bslmt::Mutex& authenticationTimerHandleMutex();
+
     InitialConnectionContext* initialConnectionContext() const;
     const bmqp_ctrlmsg::AuthenticationMessage& authenticationMessage() const;
     bmqp::EncodingType::Enum authenticationEncodingType() const;