message_sender.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/models/amqp_message.hpp"
#include "azure/core/amqp/models/amqp_value.hpp"
#include "cancellable.hpp"
#include "claims_based_security.hpp"
#include "common/async_operation_queue.hpp"
#include "connection.hpp"
#include "link.hpp"
#include "models/amqp_error.hpp"
 
#include <azure/core/nullable.hpp>
 
#include <tuple>
 
#if defined(_azure_TESTING_BUILD)
// Define the test classes dependant on this class here.
namespace Azure { namespace Core { namespace Amqp { namespace Tests { namespace MessageTests {
  class AmqpServerMock;
  class MockServiceEndpoint;
}}}}} // namespace Azure::Core::Amqp::Tests::MessageTests
#endif // _azure_TESTING_BUILDs
 
namespace Azure { namespace Core { namespace Amqp { namespace _detail {
  class MessageSenderImpl;
  class MessageSenderFactory;
}}}} // namespace Azure::Core::Amqp::_detail
 
namespace Azure { namespace Core { namespace Amqp { namespace _internal {
  enum class MessageSendStatus
  {
    Invalid,
    Ok,
    Error,
    Timeout,
    Cancelled,
  };
  std::ostream& operator<<(std::ostream& stream, MessageSendStatus status);
 
  enum class MessageSenderState
  {
    Invalid,
    Idle,
    Opening,
    Open,
    Closing,
    Error,
  };
  std::ostream& operator<<(std::ostream& stream, MessageSenderState state);
 
  class MessageSender;
  class MessageSenderEvents {
  protected:
    ~MessageSenderEvents() = default;
 
  public:
    virtual void OnMessageSenderStateChanged(
        MessageSender const& sender,
        MessageSenderState newState,
        MessageSenderState oldState)
        = 0;
    virtual void OnMessageSenderDisconnected(
        MessageSender const& sender,
        Models::_internal::AmqpError const& error)
        = 0;
  };
 
  struct MessageSenderOptions final
  {
    std::string Name;
 
    SenderSettleMode SettleMode{};
 
    Models::_internal::MessageSource MessageSource;
 
    Nullable<uint64_t> MaxMessageSize;
 
    uint32_t MaxLinkCredits{};
 
    Nullable<uint32_t> InitialDeliveryCount;
 
    bool EnableTrace{false};
 
    bool AuthenticationRequired{true};
  };
 
  class MessageSender final {
  public:
    using MessageSendCompleteCallback
        = std::function<void(MessageSendStatus sendResult, Models::AmqpValue const& deliveryState)>;
 
    ~MessageSender() noexcept;
 
    MessageSender(MessageSender const&) = default;
    MessageSender& operator=(MessageSender const&) = default;
    MessageSender(MessageSender&&) noexcept = default;
    MessageSender& operator=(MessageSender&&) noexcept = default;
 
    _azure_NODISCARD Models::_internal::AmqpError Open(Context const& context = {});
 
    void Close(Context const& context = {});
 
    std::string GetLinkName() const;
 
    std::uint64_t GetMaxMessageSize() const;
 
    _azure_NODISCARD std::tuple<MessageSendStatus, Models::_internal::AmqpError> Send(
        Models::AmqpMessage const& message,
        Context const& context = {});
 
  private:
    // Half-open the message sender (does not block waiting on the Open to complete).
    _azure_NODISCARD Models::_internal::AmqpError HalfOpen(Context const& context = {});
    MessageSender(std::shared_ptr<_detail::MessageSenderImpl> sender) : m_impl{sender} {}
 
    friend class _detail::MessageSenderFactory;
    std::shared_ptr<_detail::MessageSenderImpl> m_impl;
#if _azure_TESTING_BUILD
    friend class Azure::Core::Amqp::Tests::MessageTests::AmqpServerMock;
    friend class Azure::Core::Amqp::Tests::MessageTests::MockServiceEndpoint;
    friend class Azure::Core::Amqp::Tests::MessageTests::MessageListenerEvents;
#endif // _azure_TESTING_BUILD
  };
}}}} // namespace Azure::Core::Amqp::_internal