processor.hpp

// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
#pragma once
 
#include "checkpoint_store.hpp"
#include "consumer_client.hpp"
#include "models/processor_load_balancer_models.hpp"
#include "models/processor_models.hpp"
#include "processor_partition_client.hpp"
 
#include <azure/core/amqp/internal/common/async_operation_queue.hpp>
#include <azure/core/context.hpp>
 
#include <chrono>
#include <thread>
 
#ifdef _azure_TESTING_BUILD_AMQP
namespace Azure { namespace Messaging { namespace EventHubs { namespace Test {
  class ProcessorTest_LoadBalancing_Test;
}}}} // namespace Azure::Messaging::EventHubs::Test
#endif
namespace Azure { namespace Messaging { namespace EventHubs {
  struct ProcessorOptions final
  {
    Models::ProcessorStrategy LoadBalancingStrategy{
        Models::ProcessorStrategy::ProcessorStrategyBalanced};
 
    Azure::DateTime::duration UpdateInterval{std::chrono::seconds(10)};
 
    Azure::DateTime::duration PartitionExpirationDuration{std::chrono::seconds(60)};
 
    Models::StartPositions StartPositions;
 
    int32_t Prefetch{300};
 
    int32_t MaximumNumberOfPartitions{0};
  };
 
  namespace _detail {
    class ProcessorLoadBalancer;
  }
 
  class Processor final {
#ifdef _azure_TESTING_BUILD_AMQP
    friend class Test::ProcessorTest_LoadBalancing_Test;
#endif
 
  public:
    Processor(
        std::shared_ptr<ConsumerClient> consumerClient,
        std::shared_ptr<CheckpointStore> checkpointStore,
        ProcessorOptions const& options = {});
 
    ~Processor();
 
    Processor(Processor const& other) = delete;
 
    Processor& operator=(Processor const& other) = delete;
 
    std::shared_ptr<ProcessorPartitionClient> NextPartitionClient(
        Azure::Core::Context const& context = {});
 
    void Run(Core::Context const& context);
 
    void Start(Azure::Core::Context const& context = {});
 
    void Stop();
 
    void Close(Core::Context const& context = {})
    {
      if (m_isRunning)
      {
        throw std::runtime_error("cannot close a processor that is running");
      }
 
      // Drain the partition clients queue.
      for (;;)
      {
        auto client = m_nextPartitionClients.TryRemove();
        if (client)
        {
          client->Close(context);
        }
        else
        {
          break;
        }
      }
      (void)context;
    }
 
  private:
    template <class T> class Channel {
    public:
      Channel() : m_maximumDepth{0} {}
 
      ~Channel()
      {
        Azure::Core::Diagnostics::_internal::Log::Stream(
            Azure::Core::Diagnostics::Logger::Level::Verbose)
            << "~Channel. Currently depth is " << m_channelDepth << " and maximum depth is "
            << m_maximumDepth;
        Azure::Core::Diagnostics::_internal::Log::Stream(
            Azure::Core::Diagnostics::Logger::Level::Verbose)
            << "Clear channel queue.";
        while (m_channelDepth > 0)
        {
          auto value = m_channelQueue.TryWaitForResult();
          if (value)
          {
            m_channelDepth -= 1;
          }
        }
      }
 
      // Insert an item into the channel, returning true if successful, false if the channel is
      // full.
      bool Insert(T item)
      {
        std::lock_guard<std::mutex> lock{m_channelLock};
        if ((m_maximumDepth != 0) && (m_channelDepth >= m_maximumDepth))
        {
          return false;
        }
        m_channelQueue.CompleteOperation(item);
        m_channelDepth += 1;
        return true;
      }
 
      // Remove an item from the channel.
      T Remove(Azure::Core::Context const& context)
      {
        auto value = m_channelQueue.WaitForResult(context);
        if (!value)
        {
          throw Azure::Core::OperationCancelledException("Operation was cancelled.");
        }
        std::lock_guard<std::mutex> lock{m_channelLock};
        m_channelDepth -= 1;
        return std::get<0>(*value);
      }
 
      T TryRemove()
      {
        std::lock_guard<std::mutex> lock{m_channelLock};
        auto value = m_channelQueue.TryWaitForResult();
        if (value)
        {
          m_channelDepth -= 1;
          return std::get<0>(*value);
        }
        return T{};
      }
 
      void SetMaximumDepth(size_t maximumDepth)
      {
        std::lock_guard<std::mutex> lock{m_channelLock};
        m_maximumDepth = maximumDepth;
      }
 
    private:
      std::mutex m_channelLock;
      size_t m_channelDepth{};
      size_t m_maximumDepth{};
      Core::Amqp::Common::_internal::AsyncOperationQueue<T> m_channelQueue;
    };
 
    Azure::DateTime::duration m_ownershipUpdateInterval;
    Models::StartPositions m_defaultStartPositions;
    int32_t m_maximumNumberOfPartitions;
    std::shared_ptr<CheckpointStore> m_checkpointStore;
    std::shared_ptr<ConsumerClient> m_consumerClient;
    int32_t m_prefetch;
    Channel<std::shared_ptr<ProcessorPartitionClient>> m_nextPartitionClients;
    Models::ConsumerClientDetails m_consumerClientDetails;
    std::shared_ptr<_detail::ProcessorLoadBalancer> m_loadBalancer;
    int64_t m_processorOwnerLevel{0};
    bool m_isRunning{false};
    std::thread m_processorThread;
 
    typedef std::map<std::string, std::shared_ptr<ProcessorPartitionClient>> ConsumersType;
 
    void Dispatch(
        Models::EventHubProperties const& eventHubProperties,
        std::shared_ptr<ConsumersType> consumers,
        Core::Context const& context);
 
    void AddPartitionClient(
        Models::Ownership const& ownership,
        std::map<std::string, Models::Checkpoint>& checkpoints,
        std::weak_ptr<ConsumersType> consumers,
        Core::Context const& context);
 
    void RunInternal(Core::Context const& context, bool manualRun);
 
    Models::StartPosition GetStartPosition(
        Models::Ownership const& ownership,
        std::map<std::string, Models::Checkpoint> const& checkpoints)
    {
      Models::StartPosition startPosition = m_defaultStartPositions.Default;
 
      if (checkpoints.find(ownership.PartitionId) != checkpoints.end())
      {
        Models::Checkpoint checkpoint = checkpoints.at(ownership.PartitionId);
 
        if (checkpoint.Offset.HasValue())
        {
          startPosition.Offset = checkpoint.Offset;
        }
        else if (checkpoint.SequenceNumber.HasValue())
        {
          startPosition.SequenceNumber = checkpoint.SequenceNumber;
        }
        else
        {
          throw std::runtime_error(
              "invalid checkpoint" + ownership.PartitionId + "no offset or sequence number");
        }
      }
      else if (
          m_defaultStartPositions.PerPartition.find(ownership.PartitionId)
          != m_defaultStartPositions.PerPartition.end())
      {
        startPosition = m_defaultStartPositions.PerPartition.at(ownership.PartitionId);
      }
      return startPosition;
    }
 
    std::map<std::string, Models::Checkpoint> GetCheckpointsMap(Core::Context const& context);
  };
}}} // namespace Azure::Messaging::EventHubs