message_receiver.hpp

// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
 
#pragma once
 
#include "azure/core/amqp/internal/amqp_settle_mode.hpp"
#include "azure/core/amqp/internal/models/amqp_error.hpp"
#include "azure/core/amqp/models/amqp_message.hpp"
#include "azure/core/amqp/models/amqp_value.hpp"
#include "claims_based_security.hpp"
#include "common/async_operation_queue.hpp"
#include "connection_string_credential.hpp"
#include "link.hpp"
#include "session.hpp"
 
#include <azure/core/credentials/credentials.hpp>
#include <azure/core/nullable.hpp>
 
#include <vector>
 
namespace Azure { namespace Core { namespace Amqp { namespace _detail {
  class MessageReceiverImpl;
  class MessageReceiverFactory;
}}}} // namespace Azure::Core::Amqp::_detail
 
#if defined(_azure_TESTING_BUILD)
namespace Azure { namespace Core { namespace Amqp { namespace Tests { namespace MessageTests {
  class MockServiceEndpoint;
}}}}} // namespace Azure::Core::Amqp::Tests::MessageTests
#endif
 
namespace Azure { namespace Core { namespace Amqp { namespace _internal {
  enum class MessageReceiverState
  {
    Invalid,
    Idle,
    Opening,
    Open,
    Closing,
    Error,
  };
  std::ostream& operator<<(std::ostream& stream, _internal::MessageReceiverState state);
 
  class MessageReceiver;
 
  struct MessageReceiverOptions final
  {
    std::string Name;
 
    ReceiverSettleMode SettleMode{ReceiverSettleMode::First};
 
    Models::_internal::MessageTarget MessageTarget;
 
    Nullable<uint32_t> InitialDeliveryCount;
 
    Nullable<uint64_t> MaxMessageSize;
 
    uint32_t MaxLinkCredit{};
 
    Models::AmqpMap Properties;
 
    bool EnableTrace{false};
 
    bool AuthenticationRequired{true};
  };
 
  class MessageReceiverEvents {
  protected:
    ~MessageReceiverEvents() = default;
 
  public:
    virtual void OnMessageReceiverStateChanged(
        MessageReceiver const& receiver,
        MessageReceiverState newState,
        MessageReceiverState oldState)
        = 0;
    virtual Models::AmqpValue OnMessageReceived(
        MessageReceiver const& receiver,
        std::shared_ptr<Models::AmqpMessage> const& message)
        = 0;
    virtual void OnMessageReceiverDisconnected(
        MessageReceiver const& receiver,
        Models::_internal::AmqpError const& error)
        = 0;
  };
 
  class MessageReceiver final {
  public:
    ~MessageReceiver() noexcept;
 
    MessageReceiver(MessageReceiver const&) = default;
    MessageReceiver& operator=(MessageReceiver const&) = default;
    MessageReceiver(MessageReceiver&&) = default;
    MessageReceiver& operator=(MessageReceiver&&) = default;
 
    void Open(Context const& context = {});
 
    void Close(Context const& context = {});
 
    std::string GetLinkName() const;
 
    std::string GetSourceName() const;
 
    std::pair<std::shared_ptr<const Models::AmqpMessage>, Models::_internal::AmqpError>
    WaitForIncomingMessage(Context const& context = {});
 
    std::pair<std::shared_ptr<const Models::AmqpMessage>, Models::_internal::AmqpError>
    TryWaitForIncomingMessage();
 
  private:
    MessageReceiver(std::shared_ptr<_detail::MessageReceiverImpl> impl) : m_impl{impl} {}
    friend class _detail::MessageReceiverFactory;
    std::shared_ptr<_detail::MessageReceiverImpl> m_impl;
 
#if _azure_TESTING_BUILD
    friend class Azure::Core::Amqp::Tests::MessageTests::MockServiceEndpoint;
 
    // There is a deadlock associated with the link polling if it is enabled from an AMQP event
    // callback. To work around this, link polling is disabled when creating a message receiver from
    // an existing link endpoint. This method should be called to enable it at a time when it is
    // safer to enable link polling.
 
    // This is a test hook and should not be used outside of test code.
    void EnableLinkPolling();
#endif
  };
}}}} // namespace Azure::Core::Amqp::_internal