Feat: startAsync and StopAsync callbacks

src/groups/bmq/bmqa/bmqa_abstractsession.cpp

@@ -54,6 +54,16 @@ int AbstractSession::startAsync(
     return -1;
 }
 
+int AbstractSession::startAsync(
+    BSLA_MAYBE_UNUSED const StartCallback& callback,
+    BSLA_MAYBE_UNUSED const bsls::TimeInterval& timeout)
+{
+    // PRECONDITIONS
+    BSLS_ASSERT_OPT(false && "Method is undefined in base protocol");
+
+    return -1;
+}
+
 void AbstractSession::stop()
 {
     // PRECONDITIONS
@@ -66,6 +76,12 @@ void AbstractSession::stopAsync()
     BSLS_ASSERT_OPT(false && "Method is undefined in base protocol");
 }
 
+void AbstractSession::stopAsync(BSLA_MAYBE_UNUSED const StopCallback& callback)
+{
+    // PRECONDITIONS
+    BSLS_ASSERT_OPT(false && "Method is undefined in base protocol");
+}
+
 void AbstractSession::finalizeStop()
 {
     // PRECONDITIONS

src/groups/bmq/bmqa/bmqa_abstractsession.h

@@ -30,6 +30,8 @@
 #include <bmqa_messageproperties.h>
 #include <bmqa_openqueuestatus.h>
 #include <bmqa_queueid.h>
+#include <bmqa_startstatus.h>
+#include <bmqa_stopstatus.h>
 #include <bmqt_queueoptions.h>
 #include <bmqt_uri.h>
 
@@ -71,6 +73,16 @@ class AbstractSession {
     typedef bsl::function<void(const bmqa::CloseQueueStatus& result)>
         CloseQueueCallback;
 
+    /// An asynchronous session start operation callback
+    /// that takes as an argument the specified `result`, providing the result
+    /// of the start operation.
+    typedef bsl::function<void(const bmqa::StartStatus& result)> StartCallback;
+
+    /// An asynchronous session stop operation callback
+    /// that takes as an argument the specified `result`, providing the result
+    /// and context of the stop operation.
+    typedef bsl::function<void(const bmqa::StopStatus& result)> StopCallback;
+
   public:
     // CREATORS
 
@@ -104,6 +116,25 @@ class AbstractSession {
     virtual int
     startAsync(const bsls::TimeInterval& timeout = bsls::TimeInterval());
 
+    /// Connect to the BlazingMQ broker and start the message processing for
+    /// this `Session`.  This method returns without blocking.  The result of
+    /// the operation is communicated to the specified `callback` via a
+    /// `bmqa::StartStatus`, providing the status of the operation.
+    /// If the optionally specified `timeout` is not populated, use
+    /// the one defined in the session options.  Return 0 on success (this
+    /// doesn't imply the session is connected!), or a non-zero value
+    /// corresponding to the `bmqt::GenericResult::Enum` enum values otherwise.
+    /// If a non-zero value is returned, the `callback` will *not* be invoked.
+    /// The behavior is undefined if this method is called on an already
+    /// started `Session`.
+    ///
+    /// THREAD: The `callback` will *ALWAYS* be invoked from the EventHandler
+    ///         thread(s) (or if a SessionEventHandler was not specified, from
+    ///         the thread invoking `nextEvent`).
+    virtual int
+    startAsync(const StartCallback&      callback,
+               const bsls::TimeInterval& timeout = bsls::TimeInterval());
+
     /// Gracefully disconnect from the BlazingMQ broker and stop the
     /// operation of this `Session`.  This method blocks waiting for all
     /// already invoked event handlers to exit and all session-related
@@ -119,6 +150,18 @@ class AbstractSession {
     /// finished.  No method may be used after this method returns.
     virtual void stopAsync();
 
+    /// Disconnect from the BlazingMQ broker and stop the operation of this
+    /// `Session`.  This method returns without blocking and neither enforce
+    /// nor waits for any already started session-related operation to be
+    /// finished.  The result of the operation is communicated to the specified
+    /// `callback` via a `bmqa::StopStatus`, providing the status the
+    /// operation.  No method may be used after this method returns.
+    ///
+    /// THREAD: The `callback` will *ALWAYS* be invoked from the EventHandler
+    ///         thread(s) (or if a SessionEventHandler was not specified, from
+    ///         the thread invoking `nextEvent`).
+    virtual void stopAsync(const StopCallback& callback);
+
     /// **DEPRECATED**
     ///
     /// This method is only to be used if the session is in synchronous mode

src/groups/bmq/bmqa/bmqa_mocksession.cpp

@@ -562,9 +562,13 @@ MockSession::Call::Call(Method method, bslma::Allocator* allocator)
 , d_openQueueCallback(bsl::allocator_arg, allocator)
 , d_configureQueueCallback(bsl::allocator_arg, allocator)
 , d_closeQueueCallback(bsl::allocator_arg, allocator)
+, d_startCallback(bsl::allocator_arg, allocator)
+, d_stopCallback(bsl::allocator_arg, allocator)
 , d_openQueueResult(allocator)
 , d_configureQueueResult(allocator)
 , d_closeQueueResult(allocator)
+, d_startResult(allocator)
+, d_stopResult(allocator)
 , d_emittedEvents(allocator)
 , d_returnEvent()
 , d_messageEvent()
@@ -590,9 +594,13 @@ MockSession::Call::Call(const Call& other, bslma::Allocator* allocator)
 , d_closeQueueCallback(bsl::allocator_arg,
                        allocator,
                        other.d_closeQueueCallback)
+, d_startCallback(bsl::allocator_arg, allocator, other.d_startCallback)
+, d_stopCallback(bsl::allocator_arg, allocator, other.d_stopCallback)
 , d_openQueueResult(other.d_openQueueResult)
 , d_configureQueueResult(other.d_configureQueueResult, allocator)
 , d_closeQueueResult(other.d_closeQueueResult, allocator)
+, d_startResult(other.d_startResult, allocator)
+, d_stopResult(other.d_stopResult, allocator)
 , d_emittedEvents(other.d_emittedEvents, allocator)
 , d_returnEvent(other.d_returnEvent)
 , d_messageEvent(other.d_messageEvent)
@@ -745,6 +753,56 @@ MockSession::Call::emitting(const CloseQueueStatus& closeQueueResult)
     return *this;
 }
 
+MockSession::Call& MockSession::Call::emitting(const StartStatus& startResult)
+{
+    // PRECONDITIONS
+    BSLS_ASSERT_SAFE(d_method = e_START_ASYNC_CALLBACK);
+    BSLS_ASSERT_SAFE(d_startCallback);
+
+    d_startResult = startResult;
+
+    const CallbackFn callbackFn = bdlf::BindUtil::bindS(d_allocator_p,
+                                                        d_startCallback,
+                                                        startResult);
+
+    // Start is a session-level operation with no associated queue, so
+    // 'processIfQueueJob' will ignore this job (it only acts on queue events).
+    Job job;
+    job.d_callback = callbackFn;
+    job.d_type     = bmqt::SessionEventType::e_CONNECTED;
+    job.d_status   = startResult.result();
+
+    EventOrJob eventOrJob(job, d_allocator_p);
+    d_emittedEvents.push_back(eventOrJob);
+
+    return *this;
+}
+
+MockSession::Call& MockSession::Call::emitting(const StopStatus& stopResult)
+{
+    // PRECONDITIONS
+    BSLS_ASSERT_SAFE(d_method = e_STOP_ASYNC_CALLBACK);
+    BSLS_ASSERT_SAFE(d_stopCallback);
+
+    d_stopResult = stopResult;
+
+    const CallbackFn callbackFn = bdlf::BindUtil::bindS(d_allocator_p,
+                                                        d_stopCallback,
+                                                        stopResult);
+
+    // Stop is a session-level operation with no associated queue, so
+    // 'processIfQueueJob' will ignore this job (it only acts on queue events).
+    Job job;
+    job.d_callback = callbackFn;
+    job.d_type     = bmqt::SessionEventType::e_DISCONNECTED;
+    job.d_status   = stopResult.result();
+
+    EventOrJob eventOrJob(job, d_allocator_p);
+    d_emittedEvents.push_back(eventOrJob);
+
+    return *this;
+}
+
 const char* MockSession::Call::methodName() const
 {
     return MockSession::toAscii(d_method);
@@ -792,8 +850,13 @@ const char* MockSession::toAscii(const Method method)
     switch (method) {
     case e_START: return "start()";
     case e_START_ASYNC: return "startAsync()";
+    case e_START_ASYNC_CALLBACK:
+        return "int startAsync(const StartCallback&      callback,"
+               "const bsls::TimeInterval& timeout)";
     case e_STOP: return "stop()";
     case e_STOP_ASYNC: return "stopAsync()";
+    case e_STOP_ASYNC_CALLBACK:
+        return "void stopAsync(const StopCallback& callback)";
     case e_FINALIZE_STOP: return "finalizeStop()";
     case e_OPEN_QUEUE:
         return "int openQueue(QueueId                   *queueId,"
@@ -1227,6 +1290,21 @@ MockSession::expect_startAsync(const bsls::TimeInterval& timeout)
     return call;
 }
 
+MockSession::Call&
+MockSession::expect_startAsync(const StartCallback&      callback,
+                               const bsls::TimeInterval& timeout)
+{
+    bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex);  // LOCKED
+
+    d_calls.emplace_back(e_START_ASYNC_CALLBACK);
+    Call& call           = d_calls.back();
+    call.d_startCallback = callback;
+    call.d_timeout       = timeout;
+    call.d_allocator_p   = d_allocator_p;
+
+    return call;
+}
+
 MockSession::Call& MockSession::expect_stop()
 {
     bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex);  // LOCKED
@@ -1243,6 +1321,18 @@ MockSession::Call& MockSession::expect_stopAsync()
     return d_calls.back();
 }
 
+MockSession::Call& MockSession::expect_stopAsync(const StopCallback& callback)
+{
+    bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex);  // LOCKED
+
+    d_calls.emplace_back(e_STOP_ASYNC_CALLBACK);
+    Call& call          = d_calls.back();
+    call.d_stopCallback = callback;
+    call.d_allocator_p  = d_allocator_p;
+
+    return call;
+}
+
 MockSession::Call& MockSession::expect_finalizeStop()
 {
     bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex);  // LOCKED
@@ -1584,6 +1674,27 @@ int MockSession::startAsync(const bsls::TimeInterval& timeout)
     return rc;
 }
 
+int MockSession::startAsync(BSLA_MAYBE_UNUSED const StartCallback& callback,
+                            const bsls::TimeInterval&              timeout)
+{
+    bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex);  // LOCKED
+
+    BMQA_CHECK_CALL(e_START_ASYNC_CALLBACK, { return 0; });
+    BMQA_CHECK_ARG(e_START_ASYNC_CALLBACK,
+                   "timeout",
+                   call.d_timeout,
+                   timeout,
+                   call);
+
+    d_eventsAndJobs.insert(d_eventsAndJobs.end(),
+                           call.d_emittedEvents.begin(),
+                           call.d_emittedEvents.end());
+
+    const int rc = call.d_rc;
+    BMQA_ASSERT_AND_POP_FRONT();
+    return rc;
+}
+
 void MockSession::stop()
 {
     bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex);  // LOCKED
@@ -1628,6 +1739,28 @@ void MockSession::stopAsync()
     BMQA_ASSERT_AND_POP_FRONT();
 }
 
+void MockSession::stopAsync(BSLA_MAYBE_UNUSED const StopCallback& callback)
+{
+    bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex);  // LOCKED
+
+    BMQA_CHECK_CALL(e_STOP_ASYNC_CALLBACK, {});
+
+    d_eventsAndJobs.insert(d_eventsAndJobs.end(),
+                           call.d_emittedEvents.begin(),
+                           call.d_emittedEvents.end());
+
+    // Reset all queue's state to 'closed' to mimic the real implementation
+    UriCorrIdToQueueMap& queueMap = uriCorrIdToQueues(d_twoKeyHashMapBuffer);
+    UriCorrIdToQueueMap::iterator qIt = queueMap.begin();
+    while (qIt != queueMap.end()) {
+        QueueImplSp& queueImpl = reinterpret_cast<QueueImplSp&>(qIt->value());
+        queueImpl->setState(bmqimp::QueueState::e_CLOSED);
+        ++qIt;
+    }
+
+    BMQA_ASSERT_AND_POP_FRONT();
+}
+
 void MockSession::finalizeStop()
 {
     bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex);  // LOCKED