consumer_client.hpp

// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
 
// cspell: word myeventhub
 
#pragma once
#include "models/consumer_client_models.hpp"
#include "models/management_models.hpp"
#include "partition_client.hpp"
 
#include <azure/core/amqp.hpp>
#include <azure/core/amqp/internal/connection.hpp>
#include <azure/core/context.hpp>
#include <azure/core/credentials/credentials.hpp>
#include <azure/core/diagnostics/logger.hpp>
#include <azure/core/http/policies/policy.hpp>
#include <azure/core/internal/diagnostics/log.hpp>
namespace Azure { namespace Messaging { namespace EventHubs {
 
  constexpr const char* DefaultConsumerGroup = "$Default";
 
  struct ConsumerClientOptions final
  {
    std::string ApplicationID;
 
    Azure::Core::Http::Policies::RetryOptions RetryOptions{};
 
    std::string Name{};
  };
 
  class ConsumerClient final {
  public:
    ConsumerClient(ConsumerClient const& other) = delete;
 
    ConsumerClient(ConsumerClient&& other) = delete;
 
    ConsumerClient& operator=(ConsumerClient const& other) = delete;
 
    ConsumerClient& operator=(ConsumerClient&& other) = delete;
 
    ~ConsumerClient();
 
    std::string const& GetEventHubName() const { return m_eventHub; }
 
    std::string const& GetConsumerGroup() const { return m_consumerGroup; }
 
    Models::ConsumerClientDetails GetDetails() const
    {
      Models::ConsumerClientDetails details;
      details.ClientId = m_consumerClientOptions.ApplicationID;
      details.ConsumerGroup = m_consumerGroup;
      details.EventHubName = m_eventHub;
      details.FullyQualifiedNamespace = m_fullyQualifiedNamespace;
      return details;
    }
    Azure::Core::Http::Policies::RetryOptions const& GetRetryOptions() const
    {
      return m_consumerClientOptions.RetryOptions;
    }
 
    ConsumerClient(
        std::string const& connectionString,
        std::string const& eventHub = {},
        std::string const& consumerGroup = DefaultConsumerGroup,
        ConsumerClientOptions const& options = {});
 
    ConsumerClient(
        std::string const& fullyQualifiedNamespace,
        std::string const& eventHub,
        std::shared_ptr<Azure::Core::Credentials::TokenCredential> credential,
        std::string const& consumerGroup = DefaultConsumerGroup,
        ConsumerClientOptions const& options = {});
 
    PartitionClient CreatePartitionClient(
        std::string const& partitionId,
        PartitionClientOptions const& options = {},
        Azure::Core::Context const& context = {});
 
    Models::EventHubProperties GetEventHubProperties(Core::Context const& context = {});
 
    Models::EventHubPartitionProperties GetPartitionProperties(
        std::string const& partitionID,
        Core::Context const& context = {});
 
  private:
    std::string m_connectionString;
 
    std::string m_fullyQualifiedNamespace;
 
    std::string m_eventHub;
 
    std::string m_consumerGroup;
 
    std::shared_ptr<Core::Credentials::TokenCredential> m_credential;
 
    std::string m_hostUrl;
 
    std::mutex m_receiversLock;
    std::map<std::string, Azure::Core::Amqp::_internal::MessageReceiver> m_receivers;
 
    std::mutex m_sessionsLock;
    std::map<std::string, Azure::Core::Amqp::_internal::Session> m_sessions;
    std::map<std::string, Azure::Core::Amqp::_internal::Connection> m_connections;
 
    ConsumerClientOptions m_consumerClientOptions;
 
    void EnsureConnection(std::string const& partitionId);
    void EnsureSession(std::string const& partitionId);
    Azure::Core::Amqp::_internal::Connection CreateConnection(std::string const& partitionId) const;
    Azure::Core::Amqp::_internal::Session CreateSession(std::string const& partitionId);
    Azure::Core::Amqp::_internal::Session GetSession(std::string const& partitionId);
  };
}}} // namespace Azure::Messaging::EventHubs