producer_client.hpp

// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
 
// cspell: words myeventhub
 
#pragma once
#include "event_data_batch.hpp"
#include "models/management_models.hpp"
 
#include <azure/core/amqp.hpp>
#include <azure/core/amqp/internal/message_sender.hpp>
#include <azure/core/context.hpp>
#include <azure/core/credentials/credentials.hpp>
#include <azure/core/http/policies/policy.hpp>
 
#include <iostream>
 
namespace Azure { namespace Messaging { namespace EventHubs {
  namespace _detail {
    class EventHubsPropertiesClient;
  } // namespace _detail
 
  struct ProducerClientOptions final
  {
    std::string ApplicationID = "";
 
    Azure::Core::Http::Policies::RetryOptions RetryOptions{};
 
    std::string Name{};
 
    Azure::Nullable<std::uint64_t> MaxMessageSize{};
  };
 
  class ProducerClient final {
 
  public:
    std::string const& GetEventHubName() { return m_eventHub; }
 
    Azure::Core::Http::Policies::RetryOptions const& GetRetryOptions() const
    {
      return m_producerClientOptions.RetryOptions;
    }
 
    ProducerClient(ProducerClient const& other) = delete;
 
    ProducerClient& operator=(ProducerClient const& other) = delete;
 
    ProducerClient(ProducerClient&& other) = delete;
    ProducerClient& operator=(ProducerClient&& other) = delete;
 
    ProducerClient() = default;
 
    ProducerClient(
        std::string const& connectionString,
        std::string const& eventHub,
        ProducerClientOptions options = {});
 
    ProducerClient(
        std::string const& fullyQualifiedNamespace,
        std::string const& eventHub,
        std::shared_ptr<Azure::Core::Credentials::TokenCredential> credential,
        ProducerClientOptions options = {});
 
    ~ProducerClient() { Close(); }
 
    void Close(Azure::Core::Context const& context = {})
    {
      for (auto& sender : m_senders)
      {
        sender.second.Close(context);
      }
      m_senders.clear();
 
      // Close needs to tear down all outstanding sessions and connections, but the functionality to
      // tear these down isn't complete yet.
      //    for (auto& session : m_sessions)
      //    {
      // session.second.Close(context);
      // }
      //    for (auto& connection : m_connections)
      //    {
      // connection.second.Close(context);
      // }
    }
 
    EventDataBatch CreateBatch(
        EventDataBatchOptions const& options = {},
        Azure::Core::Context const& context = {});
 
    void Send(EventDataBatch const& eventDataBatch, Core::Context const& context = {});
 
    void Send(Models::EventData const& eventData, Core::Context const& context = {});
 
    void Send(std::vector<Models::EventData> const& eventData, 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_targetUrl{};
 
    uint16_t m_targetPort = Azure::Core::Amqp::_internal::AmqpTlsPort;
 
    std::shared_ptr<Core::Credentials::TokenCredential> m_credential{};
 
    ProducerClientOptions m_producerClientOptions{};
 
    // Protects m_senders and m_connection.
    std::mutex m_sendersLock;
    std::map<std::string, Azure::Core::Amqp::_internal::Connection> m_connections{};
    std::map<std::string, Azure::Core::Amqp::_internal::MessageSender> m_senders{};
 
    std::recursive_mutex m_sessionsLock;
    std::map<std::string, Azure::Core::Amqp::_internal::Session> m_sessions{};
 
    std::mutex m_propertiesClientLock;
    std::shared_ptr<_detail::EventHubsPropertiesClient> m_propertiesClient;
 
    Azure::Core::Amqp::_internal::Connection CreateConnection() const;
    Azure::Core::Amqp::_internal::Session CreateSession(std::string const& partitionId);
 
    // Ensure that the connection for this producer has been established.
    void EnsureConnection(const std::string& partitionId);
 
    // Ensure that a session for the specified partition ID has been established.
    void EnsureSession(std::string const& partitionId);
 
    // Ensure that a message sender for the specified partition has been created.
    void EnsureSender(std::string const& partitionId, Azure::Core::Context const& context = {});
 
    std::shared_ptr<_detail::EventHubsPropertiesClient> GetPropertiesClient();
 
    Azure::Core::Amqp::_internal::MessageSender GetSender(std::string const& partitionId);
    Azure::Core::Amqp::_internal::Session GetSession(std::string const& partitionId);
  };
}}} // namespace Azure::Messaging::EventHubs