bmqa_mocksession.h

Go to the documentation of this file.

// Copyright 2016-2023 Bloomberg Finance L.P.
// SPDX-License-Identifier: Apache-2.0
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
 
// bmqa_mocksession.h                                                 -*-C++-*-
#ifndef INCLUDED_BMQA_MOCKSESSION
#define INCLUDED_BMQA_MOCKSESSION
 
 
// BMQ
#include <bmqa_abstractsession.h>
#include <bmqa_closequeuestatus.h>
#include <bmqa_configurequeuestatus.h>
#include <bmqa_messageeventbuilder.h>
#include <bmqa_openqueuestatus.h>
#include <bmqa_queueid.h>
#include <bmqa_session.h>  // for 'bmqa::SessionEventHandler'
#include <bmqt_queueoptions.h>
#include <bmqt_sessionoptions.h>
 
// BDE
#include <ball_log.h>
#include <bdlb_variant.h>
#include <bdlbb_blob.h>
#include <bdlbb_pooledblobbufferfactory.h>
#include <bdlcc_sharedobjectpool.h>
#include <bsl_cstddef.h>
#include <bsl_deque.h>
#include <bsl_functional.h>
#include <bsl_memory.h>
#include <bsl_string.h>
#include <bsl_unordered_map.h>
#include <bsl_unordered_set.h>
#include <bsl_vector.h>
#include <bslma_allocator.h>
#include <bslma_managedptr.h>
#include <bslma_usesbslmaallocator.h>
#include <bslmf_nestedtraitdeclaration.h>
#include <bslmt_mutex.h>
#include <bsls_alignedbuffer.h>
#include <bsls_assert.h>
#include <bsls_timeinterval.h>
#include <bsls_types.h>
 
namespace BloombergLP {
 
// FORWARD DECLARATION
namespace bmqimp {
class Event;
}
namespace bmqimp {
class MessageCorrelationIdContainer;
}
namespace bmqimp {
struct Stat;
}
namespace bmqst {
class StatContext;
}
 
namespace bmqa {
 
// FORWARD DECLARATION
class ConfirmEventBuilder;
 
// Record an expected method 'CALL' into the specified mock session 'OBJ'.
#define BMQA_EXPECT_CALL(OBJ, CALL)                                           \
    (((OBJ).expect_##CALL).fromLocation(__FILE__, __LINE__))
 
// ======================
// struct MockSessionUtil
// ======================
 
struct MockSessionUtil {
  private:
    // PRIVATE TYPES
 
    typedef bdlcc::SharedObjectPool<
        bdlbb::Blob,
        bdlcc::ObjectPoolFunctors::DefaultCreator,
        bdlcc::ObjectPoolFunctors::RemoveAll<bdlbb::Blob> >
        BlobSpPool;
 
    typedef bsl::shared_ptr<bmqimp::Event> EventImplSp;
 
    typedef bsl::shared_ptr<bmqimp::Queue> QueueImplSp;
 
  public:
    // PUBLIC TYPES
 
    struct AckParams {
        // PUBLIC DATA
 
        bmqt::AckResult::Enum d_status;
 
        bmqt::CorrelationId d_correlationId;
 
        bmqt::MessageGUID d_guid;
 
        QueueId d_queueId;
 
        // CREATORS
 
        AckParams(const bmqt::AckResult::Enum status,
                  const bmqt::CorrelationId&  correlationId,
                  const bmqt::MessageGUID&    guid,
                  const bmqa::QueueId&        queueId);
    };
 
    // PUBLIC TYPES
 
    struct PushMessageParams {
        // PUBLIC DATA
 
        bdlbb::Blob d_payload;
 
        QueueId d_queueId;
 
        bmqt::MessageGUID d_guid;
 
        MessageProperties d_properties;
 
        // CREATORS
 
        PushMessageParams(const bdlbb::Blob&       payload,
                          const bmqa::QueueId&     queueId,
                          const bmqt::MessageGUID& guid,
                          const MessageProperties& properties);
    };
 
    // CLASS METHODS
 
    static Event createAckEvent(const bsl::vector<AckParams>& acks,
                                bdlbb::BlobBufferFactory*     bufferFactory,
                                bslma::Allocator*             allocator);
 
    static Event
    createPushEvent(const bsl::vector<PushMessageParams>& pushEventParams,
                    bdlbb::BlobBufferFactory*             bufferFactory,
                    bslma::Allocator*                     allocator);
 
    static Event
    createQueueSessionEvent(bmqt::SessionEventType::Enum sessionEventType,
                            QueueId*                     queueId,
                            const bmqt::CorrelationId&   correlationId,
                            int                          errorCode,
                            const bslstl::StringRef&     errorDescription,
                            bslma::Allocator*            allocator);
 
    static Event
    createSessionEvent(bmqt::SessionEventType::Enum sessionEventType,
                       const bmqt::CorrelationId&   correlationId,
                       const int                    errorCode,
                       const bslstl::StringRef&     errorDescription,
                       bslma::Allocator*            allocator);
 
    static OpenQueueStatus
    createOpenQueueStatus(const QueueId&              queueId,
                          bmqt::OpenQueueResult::Enum statusCode,
                          const bsl::string&          errorDescription,
                          bslma::Allocator*           allocator = 0);
 
    static ConfigureQueueStatus
    createConfigureQueueStatus(const QueueId&                   queueId,
                               bmqt::ConfigureQueueResult::Enum statusCode,
                               const bsl::string& errorDescription,
                               bslma::Allocator*  allocator = 0);
 
    static CloseQueueStatus
    createCloseQueueStatus(const QueueId&               queueId,
                           bmqt::CloseQueueResult::Enum statusCode,
                           const bsl::string&           errorDescription,
                           bslma::Allocator*            allocator = 0);
};
 
// =================
// class MockSession
// =================
 
class MockSession : public AbstractSession {
  public:
    // TYPES
 
    // CLASS METHODS
 
    static void initialize(bslma::Allocator* allocator = 0);
 
    static void shutdown();
 
    // PUBLIC TYPES
 
    typedef bsl::function<
        void(const char* description, const char* file, int line)>
        FailureCb;
 
  private:
    // CLASS-SCOPE CATEGORY
    BALL_LOG_SET_CLASS_CATEGORY("BMQA.MOCKSESSION");
 
  private:
    // CONSTANTS
 
    static const int k_MAX_SIZEOF_BMQC_TWOKEYHASHMAP = 256;
 
    // PRIVATE TYPES
 
    typedef bdlcc::SharedObjectPool<
        bdlbb::Blob,
        bdlcc::ObjectPoolFunctors::DefaultCreator,
        bdlcc::ObjectPoolFunctors::RemoveAll<bdlbb::Blob> >
        BlobSpPool;
 
    typedef bsl::shared_ptr<BlobSpPool> BlobSpPoolSp;
 
    typedef bsls::AlignedBuffer<k_MAX_SIZEOF_BMQC_TWOKEYHASHMAP>
        TwoKeyHashMapBuffer;
 
    typedef bsl::shared_ptr<bmqimp::Queue> QueueImplSp;
 
    typedef bsl::shared_ptr<bmqimp::Event> EventImplSp;
 
    typedef bsl::shared_ptr<bmqimp::Stat> StatImplSp;
 
    typedef bsl::shared_ptr<bmqimp::MessageCorrelationIdContainer>
        CorrelationIdContainerSp;
 
    typedef bsl::function<void()> CallbackFn;
 
    struct Job {
        // PUBLIC DATA
 
        CallbackFn d_callback;
 
        QueueImplSp d_queue;
 
        bmqt::SessionEventType::Enum d_type;
 
        int d_status;
    };
 
    typedef bdlb::Variant<Event, Job> EventOrJob;
 
    enum Method {
        e_START,
        e_START_ASYNC,
        e_STOP,
        e_STOP_ASYNC,
        e_FINALIZE_STOP,
        e_OPEN_QUEUE,
        e_OPEN_QUEUE_SYNC,
        e_OPEN_QUEUE_ASYNC,
        e_OPEN_QUEUE_ASYNC_CALLBACK,
        e_CLOSE_QUEUE,
        e_CLOSE_QUEUE_SYNC,
        e_CLOSE_QUEUE_ASYNC,
        e_CLOSE_QUEUE_ASYNC_CALLBACK,
        e_CONFIGURE_QUEUE,
        e_CONFIGURE_QUEUE_SYNC,
        e_CONFIGURE_QUEUE_ASYNC,
        e_CONFIGURE_QUEUE_ASYNC_CALLBACK,
        e_NEXT_EVENT,
        e_POST,
        e_CONFIRM_MESSAGE,
        e_CONFIRM_MESSAGES
    };
 
    struct Call {
        // PUBLIC TYPES
 
        typedef bsl::vector<EventOrJob> EventsAndJobs;
 
        // PUBLIC DATA
 
        int d_rc;
 
        Method d_method;
 
        int d_line;
 
        bsl::string d_file;
 
        bmqt::Uri d_uri;
 
        bsls::Types::Uint64 d_flags;
 
        bmqt::QueueOptions d_queueOptions;
 
        bsls::TimeInterval d_timeout;
 
        OpenQueueCallback d_openQueueCallback;
 
        ConfigureQueueCallback d_configureQueueCallback;
 
        CloseQueueCallback d_closeQueueCallback;
 
        bmqa::OpenQueueStatus d_openQueueResult;
 
        bmqa::ConfigureQueueStatus d_configureQueueResult;
 
        bmqa::CloseQueueStatus d_closeQueueResult;
 
        EventsAndJobs d_emittedEvents;
 
        Event d_returnEvent;
 
        MessageEvent d_messageEvent;
 
        MessageConfirmationCookie d_cookie;
 
        bslma::Allocator* d_allocator_p;
 
        // TRAITS
        BSLMF_NESTED_TRAIT_DECLARATION(Call, bslma::UsesBslmaAllocator)
 
        // CREATORS
 
        
        Call(Method method, bslma::Allocator* allocator);
 
        Call(const Call& other, bslma::Allocator* allocator);
 
        // MANIPULATORS
 
        Call& fromLocation(const char* file, int line);
 
        Call& returning(int rc);
 
        Call& returning(const bmqa::OpenQueueStatus& result);
 
        Call& returning(const bmqa::ConfigureQueueStatus& result);
 
        Call& returning(const bmqa::CloseQueueStatus& result);
 
        Call& returning(const Event& event);
 
        Call& emitting(const Event& event);
 
        Call& emitting(const OpenQueueStatus& openQueueResult);
 
        Call& emitting(const ConfigureQueueStatus& configureQueueResutl);
 
        Call& emitting(const CloseQueueStatus& closeQueueResult);
 
        // ACCESSORS
 
        const char* methodName() const;
    };
 
    typedef bsl::deque<Call> CallQueue;
 
    typedef bsl::deque<bmqa::MessageEvent> PostedEvents;
 
    // DATA
 
    bdlbb::PooledBlobBufferFactory d_blobBufferFactory;
 
    BlobSpPoolSp d_blobSpPool_sp;
 
    bslma::ManagedPtr<SessionEventHandler> d_eventHandler_mp;
 
    mutable CallQueue d_calls;
 
    mutable bsl::deque<EventOrJob> d_eventsAndJobs;
 
    bsl::unordered_set<bmqt::MessageGUID> d_unconfirmedGUIDs;
 
    TwoKeyHashMapBuffer d_twoKeyHashMapBuffer;
 
    FailureCb d_failureCb;
 
    int d_lastQueueId;
 
    CorrelationIdContainerSp d_corrIdContainer_sp;
 
    PostedEvents d_postedEvents;
 
    mutable bslmt::Mutex d_mutex;
 
    bslma::ManagedPtr<bmqst::StatContext> d_rootStatContext_mp;
 
    StatImplSp d_queuesStats_sp;
 
    bmqt::SessionOptions d_sessionOptions;
 
    bslma::Allocator* d_allocator_p;
 
  private:
    // PRIVATE CLASS METHODS
 
    static const char* toAscii(const Method method);
 
    // PRIVATE MANIPULATORS
 
    void initializeStats();
 
    void openQueueImp(QueueId*                  queueId,
                      const bmqt::QueueOptions& options,
                      const bmqt::Uri&          uri,
                      bsls::Types::Uint64       flags,
                      bool                      async);
 
    void processIfQueueEvent(Event* event);
 
    void processIfQueueJob(Job* job);
 
    void processIfPushEvent(const Event& event);
 
    // PRIVATE ACCESSORS
 
    void assertWrongCall(const Method method) const;
 
    void assertWrongCall(const Method method, const Call& expectedCall) const;
 
    template <class T, class U>
    void assertWrongArg(const T&     expected,
                        const U&     actual,
                        const Method method,
                        const char*  arg,
                        const Call&  call) const;
 
  public:
    // TRAITS
    BSLMF_NESTED_TRAIT_DECLARATION(MockSession, bslma::UsesBslmaAllocator)
 
    // CREATORS
 
    
    explicit MockSession(
        const bmqt::SessionOptions& options   = bmqt::SessionOptions(),
        bslma::Allocator*           allocator = 0);
 
    explicit MockSession(
        bslma::ManagedPtr<SessionEventHandler> eventHandler,
        const bmqt::SessionOptions& options   = bmqt::SessionOptions(),
        bslma::Allocator*           allocator = 0);
 
    ~MockSession() BSLS_KEYWORD_OVERRIDE;
 
    // MANIPULATORS
    //   (specific to 'bmqa::MockSession')
 
    void enqueueEvent(const bmqa::Event& event);
 
    bool emitEvent(int numEvents = 1);
 
    Call&
    expect_start(const bsls::TimeInterval& timeout = bsls::TimeInterval());
    Call& expect_startAsync(
        const bsls::TimeInterval& timeout = bsls::TimeInterval());
    Call& expect_stop();
    Call& expect_stopAsync();
    Call& expect_finalizeStop();
    Call&
          expect_openQueue(QueueId*                  queueId,
                           const bmqt::Uri&          uri,
                           bsls::Types::Uint64       flags,
                           const bmqt::QueueOptions& options = bmqt::QueueOptions(),
                           const bsls::TimeInterval& timeout = bsls::TimeInterval());
    Call& expect_openQueueSync(
        QueueId*                  queueId,
        const bmqt::Uri&          uri,
        bsls::Types::Uint64       flags,
        const bmqt::QueueOptions& options = bmqt::QueueOptions(),
        const bsls::TimeInterval& timeout = bsls::TimeInterval());
 
    Call& expect_openQueueAsync(
        QueueId*                  queueId,
        const bmqt::Uri&          uri,
        bsls::Types::Uint64       flags,
        const bmqt::QueueOptions& options = bmqt::QueueOptions(),
        const bsls::TimeInterval& timeout = bsls::TimeInterval());
    Call& expect_openQueueAsync(
        QueueId*                  queueId,
        const bmqt::Uri&          uri,
        bsls::Types::Uint64       flags,
        const OpenQueueCallback&  callback,
        const bmqt::QueueOptions& options = bmqt::QueueOptions(),
        const bsls::TimeInterval& timeout = bsls::TimeInterval());
 
    Call& expect_closeQueue(
        QueueId*                  queueId,
        const bsls::TimeInterval& timeout = bsls::TimeInterval());
    Call& expect_closeQueueSync(
        QueueId*                  queueId,
        const bsls::TimeInterval& timeout = bsls::TimeInterval());
 
    Call& expect_closeQueueAsync(
        QueueId*                  queueId,
        const bsls::TimeInterval& timeout = bsls::TimeInterval());
    Call& expect_closeQueueAsync(
        QueueId*                  queueId,
        const CloseQueueCallback& callback,
        const bsls::TimeInterval& timeout = bsls::TimeInterval());
    Call& expect_configureQueue(
        QueueId*                  queueId,
        const bmqt::QueueOptions& options = bmqt::QueueOptions(),
        const bsls::TimeInterval& timeout = bsls::TimeInterval());
    Call& expect_configureQueueSync(
        QueueId*                  queueId,
        const bmqt::QueueOptions& options = bmqt::QueueOptions(),
        const bsls::TimeInterval& timeout = bsls::TimeInterval());
    Call& expect_configureQueueAsync(
        QueueId*                  queueId,
        const bmqt::QueueOptions& options = bmqt::QueueOptions(),
        const bsls::TimeInterval& timeout = bsls::TimeInterval());
 
    Call& expect_configureQueueAsync(
        QueueId*                      queueId,
        const bmqt::QueueOptions&     options,
        const ConfigureQueueCallback& callback,
        const bsls::TimeInterval&     timeout = bsls::TimeInterval());
 
    Call&
    expect_nextEvent(const bsls::TimeInterval& timeout = bsls::TimeInterval());
    Call& expect_post(const MessageEvent& messageEvent);
    Call& expect_confirmMessage(const Message& message);
    Call& expect_confirmMessage(const MessageConfirmationCookie& cookie);
 
    Call& expect_confirmMessages(ConfirmEventBuilder* builder);
 
    // MANIPULATORS
    //   (virtual bmqa::AbstractSession)
    int start(const bsls::TimeInterval& timeout = bsls::TimeInterval())
        BSLS_KEYWORD_OVERRIDE;
 
    int startAsync(const bsls::TimeInterval& timeout = bsls::TimeInterval())
        BSLS_KEYWORD_OVERRIDE;
 
    void stop() BSLS_KEYWORD_OVERRIDE;
 
    void stopAsync() BSLS_KEYWORD_OVERRIDE;
 
    void finalizeStop() BSLS_KEYWORD_OVERRIDE;
 
    void loadMessageEventBuilder(MessageEventBuilder* builder)
        BSLS_KEYWORD_OVERRIDE;
 
    void loadConfirmEventBuilder(ConfirmEventBuilder* builder)
        BSLS_KEYWORD_OVERRIDE;
 
    void
    loadMessageProperties(MessageProperties* buffer) BSLS_KEYWORD_OVERRIDE;
 
 
    int getQueueId(QueueId*         queueId,
                   const bmqt::Uri& uri) BSLS_KEYWORD_OVERRIDE;
 
    int getQueueId(QueueId* queueId, const bmqt::CorrelationId& correlationId)
        BSLS_KEYWORD_OVERRIDE;
 
    int openQueue(QueueId*                  queueId,
                  const bmqt::Uri&          uri,
                  bsls::Types::Uint64       flags,
                  const bmqt::QueueOptions& options = bmqt::QueueOptions(),
                  const bsls::TimeInterval& timeout = bsls::TimeInterval())
        BSLS_KEYWORD_OVERRIDE;
 
    OpenQueueStatus
    openQueueSync(QueueId*                  queueId,
                  const bmqt::Uri&          uri,
                  bsls::Types::Uint64       flags,
                  const bmqt::QueueOptions& options = bmqt::QueueOptions(),
                  const bsls::TimeInterval& timeout = bsls::TimeInterval())
        BSLS_KEYWORD_OVERRIDE;
 
    int
    openQueueAsync(QueueId*                  queueId,
                   const bmqt::Uri&          uri,
                   bsls::Types::Uint64       flags,
                   const bmqt::QueueOptions& options = bmqt::QueueOptions(),
                   const bsls::TimeInterval& timeout = bsls::TimeInterval())
        BSLS_KEYWORD_OVERRIDE;
 
    void
    openQueueAsync(QueueId*                  queueId,
                   const bmqt::Uri&          uri,
                   bsls::Types::Uint64       flags,
                   const OpenQueueCallback&  callback,
                   const bmqt::QueueOptions& options = bmqt::QueueOptions(),
                   const bsls::TimeInterval& timeout = bsls::TimeInterval())
        BSLS_KEYWORD_OVERRIDE;
 
    int
    configureQueue(QueueId*                  queueId,
                   const bmqt::QueueOptions& options = bmqt::QueueOptions(),
                   const bsls::TimeInterval& timeout = bsls::TimeInterval())
        BSLS_KEYWORD_OVERRIDE;
 
    ConfigureQueueStatus configureQueueSync(QueueId*                  queueId,
                                            const bmqt::QueueOptions& options,
                                            const bsls::TimeInterval& timeout)
        BSLS_KEYWORD_OVERRIDE;
 
    int configureQueueAsync(
        QueueId*                  queueId,
        const bmqt::QueueOptions& options = bmqt::QueueOptions(),
        const bsls::TimeInterval& timeout = bsls::TimeInterval())
        BSLS_KEYWORD_OVERRIDE;
 
    void configureQueueAsync(QueueId*                      queueId,
                             const bmqt::QueueOptions&     options,
                             const ConfigureQueueCallback& callback,
                             const bsls::TimeInterval&     timeout =
                                 bsls::TimeInterval()) BSLS_KEYWORD_OVERRIDE;
 
    int closeQueue(QueueId*                  queueId,
                   const bsls::TimeInterval& timeout = bsls::TimeInterval())
        BSLS_KEYWORD_OVERRIDE;
 
    CloseQueueStatus
    closeQueueSync(QueueId*                  queueId,
                   const bsls::TimeInterval& timeout = bsls::TimeInterval())
        BSLS_KEYWORD_OVERRIDE;
 
    int closeQueueAsync(QueueId*                  queueId,
                        const bsls::TimeInterval& timeout =
                            bsls::TimeInterval()) BSLS_KEYWORD_OVERRIDE;
 
    void closeQueueAsync(QueueId*                  queueId,
                         const CloseQueueCallback& callback,
                         const bsls::TimeInterval& timeout =
                             bsls::TimeInterval()) BSLS_KEYWORD_OVERRIDE;
 
    Event nextEvent(const bsls::TimeInterval& timeout = bsls::TimeInterval())
        BSLS_KEYWORD_OVERRIDE;
 
    int post(const MessageEvent& messageEvent) BSLS_KEYWORD_OVERRIDE;
 
    int confirmMessage(const Message& message) BSLS_KEYWORD_OVERRIDE;
    int confirmMessage(const MessageConfirmationCookie& cookie)
        BSLS_KEYWORD_OVERRIDE;
 
    int confirmMessages(ConfirmEventBuilder* builder) BSLS_KEYWORD_OVERRIDE;
 
    int configureMessageDumping(const bslstl::StringRef& command)
        BSLS_KEYWORD_OVERRIDE;
    // NOT IMPLEMENTED
 
    void setFailureCallback(const FailureCb& failureCb);
 
    bool popPostedEvent(bmqa::MessageEvent* event);
 
    // ACCESSORS
 
    size_t unconfirmedMessages() const;
};
 
// ============================================================================
//                             INLINE DEFINITIONS
// ============================================================================
 
// -----------------
// class MockSession
// -----------------
 
inline void MockSession::setFailureCallback(const FailureCb& failureCb)
{
    d_failureCb = failureCb;
}
 
inline size_t MockSession::unconfirmedMessages() const
{
    return d_unconfirmedGUIDs.size();
}
 
inline bool MockSession::popPostedEvent(bmqa::MessageEvent* event)
{
    // PRECONDITIONS
    BSLS_ASSERT(event);
 
    bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex);  // LOCKED
 
    if (d_postedEvents.empty()) {
        return false;  // RETURN
    }
 
    *event = d_postedEvents.front();
    d_postedEvents.pop_front();
 
    return true;
}
 
}  // close package namespace
}  // close enterprise namespace
 
#endif