bmqt_queueoptions.h

Go to the documentation of this file.

// Copyright 2015-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_queueoptions.h                                                -*-C++-*-
#ifndef INCLUDED_BMQT_QUEUEOPTIONS
#define INCLUDED_BMQT_QUEUEOPTIONS
 
 
// BMQ
 
#include <bmqt_subscription.h>
 
// BDE
#include <bsl_iosfwd.h>
#include <bsl_optional.h>
#include <bsl_unordered_map.h>
#include <bsl_utility.h>
#include <bsl_vector.h>
#include <bslma_allocator.h>
#include <bslma_usesbslmaallocator.h>
#include <bslmf_nestedtraitdeclaration.h>
 
namespace BloombergLP {
 
namespace bmqt {
 
// ==================
// class QueueOptions
// ==================
 
class QueueOptions {
  public:
    // PUBLIC CONSTANTS
 
    static const int k_CONSUMER_PRIORITY_MIN;
 
    static const int k_CONSUMER_PRIORITY_MAX;
 
    static const int  k_DEFAULT_MAX_UNCONFIRMED_MESSAGES;
    static const int  k_DEFAULT_MAX_UNCONFIRMED_BYTES;
    static const int  k_DEFAULT_CONSUMER_PRIORITY;
    static const bool k_DEFAULT_SUSPENDS_ON_BAD_HOST_HEALTH;
 
  private:
    // PRIVATE TYPES
    typedef bsl::unordered_map<SubscriptionHandle, Subscription> Subscriptions;
 
  private:
    // DATA
    Subscription d_info;
 
    bsl::optional<bool> d_suspendsOnBadHostHealth;
 
    Subscriptions d_subscriptions;
 
    bool d_hadSubscriptions;
 
    bslma::Allocator* d_allocator_p;
 
  public:
    // PUBLIC TYPES
 
    typedef bsl::pair<SubscriptionHandle, Subscription> HandleAndSubscription;
 
    typedef bsl::vector<HandleAndSubscription> SubscriptionsSnapshot;
 
  public:
    // TRAITS
    BSLMF_NESTED_TRAIT_DECLARATION(QueueOptions, bslma::UsesBslmaAllocator)
 
    // CREATORS
 
    
    explicit QueueOptions(bslma::Allocator* allocator = 0);
 
    QueueOptions(const QueueOptions& other, bslma::Allocator* allocator = 0);
 
    // MANIPULATORS
 
    QueueOptions& operator=(const QueueOptions& rhs);
 
    QueueOptions& setMaxUnconfirmedMessages(int value);
 
    QueueOptions& setMaxUnconfirmedBytes(int value);
 
    QueueOptions& setConsumerPriority(int value);
 
    QueueOptions& setSuspendsOnBadHostHealth(bool value);
 
    QueueOptions& merge(const QueueOptions& other);
 
    bool addOrUpdateSubscription(bsl::string*              errorDescription,
                                 const SubscriptionHandle& handle,
                                 const Subscription&       subscription);
 
    bool removeSubscription(const SubscriptionHandle& handle);
 
    void removeAllSubscriptions();
 
    // ACCESSORS
 
    int maxUnconfirmedMessages() const;
 
    int maxUnconfirmedBytes() const;
 
    int consumerPriority() const;
 
    bool suspendsOnBadHostHealth() const;
 
    bool hasMaxUnconfirmedMessages() const;
 
    bool hasMaxUnconfirmedBytes() const;
 
    bool hasConsumerPriority() const;
 
    bool hasSuspendsOnBadHostHealth() const;
 
    bool loadSubscription(Subscription*             subscription,
                          const SubscriptionHandle& handle) const;
 
    void loadSubscriptions(SubscriptionsSnapshot* snapshot) const;
 
    bsl::ostream&
    print(bsl::ostream& stream, int level = 0, int spacesPerLevel = 4) const;
};
 
// FREE OPERATORS
 
bool operator==(const QueueOptions& lhs, const QueueOptions& rhs);
 
bool operator!=(const QueueOptions& lhs, const QueueOptions& rhs);
 
bsl::ostream& operator<<(bsl::ostream& stream, const QueueOptions& rhs);
 
// ============================================================================
//                             INLINE DEFINITIONS
// ============================================================================
 
// ------------------
// class QueueOptions
// ------------------
 
// MANIPULATORS
 
inline QueueOptions& QueueOptions::operator=(const QueueOptions& rhs)
{
    if (this != &rhs) {
        d_info                    = rhs.d_info;
        d_suspendsOnBadHostHealth = rhs.d_suspendsOnBadHostHealth;
        d_hadSubscriptions        = rhs.d_hadSubscriptions;
        d_subscriptions           = Subscriptions(rhs.d_subscriptions,
                                        this->d_allocator_p);
    }
 
    return *this;
}
 
inline QueueOptions& QueueOptions::setMaxUnconfirmedMessages(int value)
{
    d_info.setMaxUnconfirmedMessages(value);
    return *this;
}
 
inline QueueOptions& QueueOptions::setMaxUnconfirmedBytes(int value)
{
    d_info.setMaxUnconfirmedBytes(value);
    return *this;
}
 
inline QueueOptions& QueueOptions::setConsumerPriority(int value)
{
    d_info.setConsumerPriority(value);
    return *this;
}
 
inline QueueOptions& QueueOptions::setSuspendsOnBadHostHealth(bool value)
{
    d_suspendsOnBadHostHealth.emplace(value);
    return *this;
}
 
// ACCESSORS
inline int QueueOptions::maxUnconfirmedMessages() const
{
    return d_info.maxUnconfirmedMessages();
}
 
inline int QueueOptions::maxUnconfirmedBytes() const
{
    return d_info.maxUnconfirmedBytes();
}
 
inline int QueueOptions::consumerPriority() const
{
    return d_info.consumerPriority();
}
 
inline bool QueueOptions::suspendsOnBadHostHealth() const
{
    return d_suspendsOnBadHostHealth.value_or(
        k_DEFAULT_SUSPENDS_ON_BAD_HOST_HEALTH);
}
 
inline bool QueueOptions::hasMaxUnconfirmedMessages() const
{
    return d_info.hasMaxUnconfirmedMessages();
}
 
inline bool QueueOptions::hasMaxUnconfirmedBytes() const
{
    return d_info.hasMaxUnconfirmedBytes();
}
 
inline bool QueueOptions::hasConsumerPriority() const
{
    return d_info.hasConsumerPriority();
}
 
inline bool QueueOptions::hasSuspendsOnBadHostHealth() const
{
    return d_suspendsOnBadHostHealth.has_value();
}
 
}  // close package namespace
 
// ------------------
// class QueueOptions
// ------------------
 
inline bool bmqt::operator==(const bmqt::QueueOptions& lhs,
                             const bmqt::QueueOptions& rhs)
{
    return lhs.maxUnconfirmedMessages() == rhs.maxUnconfirmedMessages() &&
           lhs.maxUnconfirmedBytes() == rhs.maxUnconfirmedBytes() &&
           lhs.consumerPriority() == rhs.consumerPriority() &&
           lhs.suspendsOnBadHostHealth() == rhs.suspendsOnBadHostHealth();
}
 
inline bool bmqt::operator!=(const bmqt::QueueOptions& lhs,
                             const bmqt::QueueOptions& rhs)
{
    return lhs.maxUnconfirmedMessages() != rhs.maxUnconfirmedMessages() ||
           lhs.maxUnconfirmedBytes() != rhs.maxUnconfirmedBytes() ||
           lhs.consumerPriority() != rhs.consumerPriority() ||
           lhs.suspendsOnBadHostHealth() != rhs.suspendsOnBadHostHealth();
}
 
inline bsl::ostream& bmqt::operator<<(bsl::ostream&             stream,
                                      const bmqt::QueueOptions& rhs)
{
    return rhs.print(stream, 0, -1);
}
 
}  // close enterprise namespace
 
#endif