processor_partition_client.hpp

// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
#pragma once
#include "checkpoint_store.hpp"
#include "consumer_client.hpp"
 
namespace Azure { namespace Messaging { namespace EventHubs {
 
  class ProcessorPartitionClient final {
    friend class Processor;
 
  public:
    ProcessorPartitionClient(ProcessorPartitionClient const& other) = delete;
 
    ProcessorPartitionClient(ProcessorPartitionClient&& other) = default;
 
    ProcessorPartitionClient& operator=(ProcessorPartitionClient const& other) = delete;
 
    ProcessorPartitionClient& operator=(ProcessorPartitionClient&& other) = default;
 
    ~ProcessorPartitionClient();
 
    std::vector<std::shared_ptr<const Models::ReceivedEventData>> ReceiveEvents(
        uint32_t maxBatchSize,
        Core::Context const& context = {})
    {
      return m_partitionClient->ReceiveEvents(maxBatchSize, context);
    }
 
    void UpdateCheckpoint(
        std::shared_ptr<const Models::ReceivedEventData> const& eventData,
        Core::Context const& context = {});
 
    std::string PartitionId() const { return m_partitionId; }
 
    void Close(Core::Context const& context = {})
    {
      if (m_cleanupFunc)
      {
        m_cleanupFunc();
      }
      m_partitionClient->Close(context);
    }
 
  private:
    std::string m_partitionId;
    std::unique_ptr<PartitionClient> m_partitionClient{};
    std::shared_ptr<CheckpointStore> m_checkpointStore;
    std::function<void()> m_cleanupFunc;
    Models::ConsumerClientDetails m_consumerClientDetails;
 
    ProcessorPartitionClient(
        std::string partitionId,
        std::shared_ptr<CheckpointStore> checkpointStore,
        Models::ConsumerClientDetails consumerClientDetails,
        std::function<void()> cleanupFunc)
        : m_partitionId(partitionId), m_checkpointStore(checkpointStore),
          m_cleanupFunc(cleanupFunc), m_consumerClientDetails(consumerClientDetails)
    {
    }
 
    void SetPartitionClient(std::unique_ptr<PartitionClient>& partitionClient)
    {
      m_partitionClient = std::move(partitionClient);
    }
 
    void UpdateCheckpoint(
        Azure::Core::Amqp::Models::AmqpMessage const& amqpMessage,
        Core::Context const& context = {});
    std::string GetPartitionId() { return m_partitionId; }
  };
}}} // namespace Azure::Messaging::EventHubs