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/management.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 {
 
  struct ProducerClientOptions final
  {
    std::string ApplicationID = "";
 
    Azure::Core::Http::Policies::RetryOptions RetryOptions{};
 
    std::string Name{};
 
    Azure::Nullable<std::uint64_t> MaxMessageSize{};
  };
 
  class ProducerClient final {
    std::string m_connectionString;
 
    std::string m_fullyQualifiedNamespace;
 
    std::string m_eventHub{};
 
    std::string m_targetUrl{};
 
    std::shared_ptr<Core::Credentials::TokenCredential> m_credential{};
 
    ProducerClientOptions m_producerClientOptions{};
    std::map<std::string, Azure::Core::Amqp::_internal::MessageSender> m_senders{};
    std::map<std::string, Azure::Core::Amqp::_internal::Session> m_sessions{};
 
  public:
    std::string const& GetEventHubName() { return m_eventHub; }
 
    Azure::Core::Http::Policies::RetryOptions const& GetRetryOptions()
    {
      return m_producerClientOptions.RetryOptions;
    }
 
    ProducerClient(ProducerClient const& other) = default;
 
    ProducerClient& operator=(ProducerClient const& other) = default;
 
    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()
    {
      for (auto& sender : m_senders)
      {
        sender.second.Close();
      }
      m_senders.clear();
    }
 
    bool SendEventDataBatch(
        EventDataBatch const& eventDataBatch,
        Core::Context const& context = {});
 
    Models::EventHubProperties GetEventHubProperties(Core::Context const& context = {});
 
    Models::EventHubPartitionProperties GetPartitionProperties(
        std::string const& partitionID,
        Core::Context const& context = {});
 
  private:
    Azure::Core::Amqp::_internal::MessageSender GetSender(std::string const& partitionId = "");
    void CreateSender(std::string const& partitionId = "");
  };
}}} // namespace Azure::Messaging::EventHubs