bmqt_sessionoptions.h

Go to the documentation of this file.

// Copyright 2014-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.
 
// bmqt_sessionoptions.h                                              -*-C++-*-
#ifndef INCLUDED_BMQT_SESSIONOPTIONS
#define INCLUDED_BMQT_SESSIONOPTIONS
 
 
// BMQ
 
// BDE
#include <bdlt_timeunitratio.h>
#include <bsl_iosfwd.h>
#include <bsl_memory.h>
#include <bsl_string.h>
#include <bslma_allocator.h>
#include <bslma_usesbslmaallocator.h>
#include <bslmf_nestedtraitdeclaration.h>
#include <bsls_annotation.h>
#include <bsls_assert.h>
#include <bsls_timeinterval.h>
#include <bsls_types.h>
 
namespace BloombergLP {
 
// FORWARD DECLARATIONS
namespace bmqpi {
class DTContext;
}
namespace bmqpi {
class DTTracer;
}
namespace bmqpi {
class HostHealthMonitor;
}
 
namespace bmqt {
 
// ====================
// class SessionOptions
// ====================
 
class SessionOptions {
  public:
    // CONSTANTS
 
    static const char k_BROKER_DEFAULT_URI[];
 
    static const int k_BROKER_DEFAULT_PORT = 30114;
 
    static const int k_QUEUE_OPERATION_DEFAULT_TIMEOUT =
        5 * bdlt::TimeUnitRatio::k_SECONDS_PER_MINUTE;
 
  private:
    // DATA
 
    bsl::string d_brokerUri;
 
    bsl::string d_processNameOverride;
 
    int d_numProcessingThreads;
 
    int d_blobBufferSize;
 
    bsls::Types::Int64 d_channelHighWatermark;
 
    bsls::TimeInterval d_statsDumpInterval;
 
    bsls::TimeInterval d_connectTimeout;
 
    bsls::TimeInterval d_disconnectTimeout;
 
    bsls::TimeInterval d_openQueueTimeout;
 
    bsls::TimeInterval d_configureQueueTimeout;
 
    bsls::TimeInterval d_closeQueueTimeout;
 
    int d_eventQueueLowWatermark;
 
    int d_eventQueueHighWatermark;
 
    int d_eventQueueSize;
 
    bsl::shared_ptr<bmqpi::HostHealthMonitor> d_hostHealthMonitor_sp;
 
    bsl::shared_ptr<bmqpi::DTContext> d_dtContext_sp;
    bsl::shared_ptr<bmqpi::DTTracer>  d_dtTracer_sp;
 
  public:
    // TRAITS
    BSLMF_NESTED_TRAIT_DECLARATION(SessionOptions, bslma::UsesBslmaAllocator)
 
    // CREATORS
 
    
    explicit SessionOptions(bslma::Allocator* allocator = 0);
 
    SessionOptions(const SessionOptions& other,
                   bslma::Allocator*     allocator = 0);
 
    // MANIPULATORS
 
    SessionOptions& setBrokerUri(const bslstl::StringRef& value);
 
    SessionOptions& setProcessNameOverride(const bslstl::StringRef& value);
 
    SessionOptions& setNumProcessingThreads(int value);
 
    SessionOptions& setBlobBufferSize(int value);
 
    SessionOptions& setChannelHighWatermark(bsls::Types::Int64 value);
 
    SessionOptions& setStatsDumpInterval(const bsls::TimeInterval& value);
 
    SessionOptions& setConnectTimeout(const bsls::TimeInterval& value);
 
    SessionOptions& setDisconnectTimeout(const bsls::TimeInterval& value);
 
    SessionOptions& setOpenQueueTimeout(const bsls::TimeInterval& value);
 
    SessionOptions& setConfigureQueueTimeout(const bsls::TimeInterval& value);
 
    SessionOptions& setCloseQueueTimeout(const bsls::TimeInterval& value);
 
    SessionOptions& setHostHealthMonitor(
        const bsl::shared_ptr<bmqpi::HostHealthMonitor>& monitor);
 
    SessionOptions&
    setTraceOptions(const bsl::shared_ptr<bmqpi::DTContext>& dtContext,
                    const bsl::shared_ptr<bmqpi::DTTracer>&  dtTracer);
 
    SessionOptions&
    configureEventQueue(int queueSize, int lowWatermark, int highWatermark);
 
    SessionOptions& configureEventQueue(int lowWatermark, int highWatermark);
 
    // ACCESSORS
 
    const bsl::string& brokerUri() const;
 
    const bsl::string& processNameOverride() const;
 
    int numProcessingThreads() const;
 
    int blobBufferSize() const;
 
    bsls::Types::Int64 channelHighWatermark() const;
 
    const bsls::TimeInterval& statsDumpInterval() const;
 
    const bsls::TimeInterval& connectTimeout() const;
 
    const bsls::TimeInterval& disconnectTimeout() const;
 
    const bsls::TimeInterval& openQueueTimeout() const;
 
    const bsls::TimeInterval& configureQueueTimeout() const;
 
    const bsls::TimeInterval& closeQueueTimeout() const;
 
    const bsl::shared_ptr<bmqpi::HostHealthMonitor>& hostHealthMonitor() const;
 
    const bsl::shared_ptr<bmqpi::DTContext>& traceContext() const;
 
    const bsl::shared_ptr<bmqpi::DTTracer>& tracer() const;
 
    int eventQueueLowWatermark() const;
    int eventQueueHighWatermark() const;
 
    int eventQueueSize() const;
 
    bsl::ostream&
    print(bsl::ostream& stream, int level = 0, int spacesPerLevel = 4) const;
};
 
// FREE OPERATORS
 
bool operator==(const SessionOptions& lhs, const SessionOptions& rhs);
 
bool operator!=(const SessionOptions& lhs, const SessionOptions& rhs);
 
bsl::ostream& operator<<(bsl::ostream& stream, const SessionOptions& rhs);
 
// ============================================================================
//                             INLINE DEFINITIONS
// ============================================================================
 
// --------------------
// class SessionOptions
// --------------------
 
// MANIPULATORS
inline SessionOptions&
SessionOptions::setBrokerUri(const bslstl::StringRef& value)
{
    d_brokerUri.assign(value.data(), value.length());
    return *this;
}
 
inline SessionOptions&
SessionOptions::setProcessNameOverride(const bslstl::StringRef& value)
{
    d_processNameOverride.assign(value.data(), value.length());
    return *this;
}
 
inline SessionOptions& SessionOptions::setNumProcessingThreads(int value)
{
    d_numProcessingThreads = value;
    return *this;
}
 
inline SessionOptions& SessionOptions::setBlobBufferSize(int value)
{
    d_blobBufferSize = value;
    return *this;
}
 
inline SessionOptions&
SessionOptions::setChannelHighWatermark(bsls::Types::Int64 value)
{
    // PRECONDITIONS
    BSLS_ASSERT_OPT(8 * 1024 * 1024 < value);
    // We reserve 4MB for control message (see
    // 'bmqimp::BrokerSession::k_CONTROL_DATA_WATERMARK_EXTRA') so make
    // sure the provided value is greater than it.
 
    d_channelHighWatermark = value;
    return *this;
}
 
inline SessionOptions&
SessionOptions::setStatsDumpInterval(const bsls::TimeInterval& value)
{
    // PRECONDITIONS
    BSLS_ASSERT_OPT(value.seconds() % 30 == 0 &&
                    "value must be a multiple of 30s");
    BSLS_ASSERT_OPT(value.seconds() <
                        bdlt::TimeUnitRatio::k_SECONDS_PER_HOUR &&
                    "value must be < 10 min");
 
    d_statsDumpInterval = value;
    return *this;
}
 
inline SessionOptions&
SessionOptions::setConnectTimeout(const bsls::TimeInterval& value)
{
    d_connectTimeout = value;
    return *this;
}
 
inline SessionOptions&
SessionOptions::setDisconnectTimeout(const bsls::TimeInterval& value)
{
    d_disconnectTimeout = value;
    return *this;
}
 
inline SessionOptions&
SessionOptions::setOpenQueueTimeout(const bsls::TimeInterval& value)
{
    d_openQueueTimeout = value;
    return *this;
}
 
inline SessionOptions&
SessionOptions::setConfigureQueueTimeout(const bsls::TimeInterval& value)
{
    d_configureQueueTimeout = value;
    return *this;
}
 
inline SessionOptions&
SessionOptions::setCloseQueueTimeout(const bsls::TimeInterval& value)
{
    d_closeQueueTimeout = value;
    return *this;
}
 
inline SessionOptions& SessionOptions::setHostHealthMonitor(
    const bsl::shared_ptr<bmqpi::HostHealthMonitor>& monitor)
{
    d_hostHealthMonitor_sp = monitor;
    return *this;
}
 
inline SessionOptions& SessionOptions::setTraceOptions(
    const bsl::shared_ptr<bmqpi::DTContext>& context,
    const bsl::shared_ptr<bmqpi::DTTracer>&  tracer)
{
    BSLS_ASSERT_OPT((context && tracer) || (!context && !tracer));
 
    d_dtContext_sp = context;
    d_dtTracer_sp  = tracer;
    return *this;
}
 
inline SessionOptions& SessionOptions::configureEventQueue(int queueSize,
                                                           int lowWatermark,
                                                           int highWatermark)
{
    d_eventQueueSize = queueSize;
    return configureEventQueue(lowWatermark, highWatermark);
}
 
inline SessionOptions& SessionOptions::configureEventQueue(int lowWatermark,
                                                           int highWatermark)
{
    // PRECONDITIONS
    BSLS_ASSERT_OPT(lowWatermark < highWatermark);
 
    d_eventQueueLowWatermark  = lowWatermark;
    d_eventQueueHighWatermark = highWatermark;
 
    return *this;
}
 
// ACCESSORS
inline const bsl::string& SessionOptions::brokerUri() const
{
    return d_brokerUri;
}
 
inline const bsl::string& SessionOptions::processNameOverride() const
{
    return d_processNameOverride;
}
 
inline int SessionOptions::numProcessingThreads() const
{
    return d_numProcessingThreads;
}
 
inline int SessionOptions::blobBufferSize() const
{
    return d_blobBufferSize;
}
 
inline bsls::Types::Int64 SessionOptions::channelHighWatermark() const
{
    return d_channelHighWatermark;
}
 
inline const bsls::TimeInterval& SessionOptions::statsDumpInterval() const
{
    return d_statsDumpInterval;
}
 
inline const bsls::TimeInterval& SessionOptions::connectTimeout() const
{
    return d_connectTimeout;
}
 
inline const bsls::TimeInterval& SessionOptions::disconnectTimeout() const
{
    return d_disconnectTimeout;
}
 
inline const bsls::TimeInterval& SessionOptions::openQueueTimeout() const
{
    return d_openQueueTimeout;
}
 
inline const bsls::TimeInterval& SessionOptions::configureQueueTimeout() const
{
    return d_configureQueueTimeout;
}
 
inline const bsls::TimeInterval& SessionOptions::closeQueueTimeout() const
{
    return d_closeQueueTimeout;
}
 
inline const bsl::shared_ptr<bmqpi::HostHealthMonitor>&
SessionOptions::hostHealthMonitor() const
{
    return d_hostHealthMonitor_sp;
}
 
inline const bsl::shared_ptr<bmqpi::DTContext>&
SessionOptions::traceContext() const
{
    return d_dtContext_sp;
}
 
inline const bsl::shared_ptr<bmqpi::DTTracer>& SessionOptions::tracer() const
{
    return d_dtTracer_sp;
}
 
inline int SessionOptions::eventQueueLowWatermark() const
{
    return d_eventQueueLowWatermark;
}
 
inline int SessionOptions::eventQueueHighWatermark() const
{
    return d_eventQueueHighWatermark;
}
 
inline int SessionOptions::eventQueueSize() const
{
    return d_eventQueueSize;
}
 
}  // close package namespace
 
// --------------------
// class SessionOptions
// --------------------
 
inline bool bmqt::operator==(const bmqt::SessionOptions& lhs,
                             const bmqt::SessionOptions& rhs)
{
    return lhs.brokerUri() == rhs.brokerUri() &&
           lhs.numProcessingThreads() == rhs.numProcessingThreads() &&
           lhs.blobBufferSize() == rhs.blobBufferSize() &&
           lhs.channelHighWatermark() == rhs.channelHighWatermark() &&
           lhs.statsDumpInterval() == rhs.statsDumpInterval() &&
           lhs.connectTimeout() == rhs.connectTimeout() &&
           lhs.openQueueTimeout() == rhs.openQueueTimeout() &&
           lhs.configureQueueTimeout() == rhs.configureQueueTimeout() &&
           lhs.closeQueueTimeout() == rhs.closeQueueTimeout() &&
           lhs.eventQueueLowWatermark() == rhs.eventQueueLowWatermark() &&
           lhs.eventQueueHighWatermark() == rhs.eventQueueHighWatermark() &&
           lhs.hostHealthMonitor() == rhs.hostHealthMonitor() &&
           lhs.traceContext() == rhs.traceContext() &&
           lhs.tracer() == rhs.tracer();
}
 
inline bool bmqt::operator!=(const bmqt::SessionOptions& lhs,
                             const bmqt::SessionOptions& rhs)
{
    return lhs.brokerUri() != rhs.brokerUri() ||
           lhs.numProcessingThreads() != rhs.numProcessingThreads() ||
           lhs.blobBufferSize() != rhs.blobBufferSize() ||
           lhs.channelHighWatermark() != rhs.channelHighWatermark() ||
           lhs.statsDumpInterval() != rhs.statsDumpInterval() ||
           lhs.connectTimeout() != rhs.connectTimeout() ||
           lhs.openQueueTimeout() != rhs.openQueueTimeout() ||
           lhs.configureQueueTimeout() != rhs.configureQueueTimeout() ||
           lhs.closeQueueTimeout() != rhs.closeQueueTimeout() ||
           lhs.eventQueueLowWatermark() != rhs.eventQueueLowWatermark() ||
           lhs.eventQueueHighWatermark() != rhs.eventQueueHighWatermark() ||
           lhs.hostHealthMonitor() != rhs.hostHealthMonitor() ||
           lhs.traceContext() != rhs.traceContext() ||
           lhs.tracer() != rhs.tracer();
}
 
inline bsl::ostream& bmqt::operator<<(bsl::ostream&               stream,
                                      const bmqt::SessionOptions& rhs)
{
    return rhs.print(stream, 0, -1);
}
 
}  // close enterprise namespace
 
#endif