connection.hpp

// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
 
#pragma once
 
#include "azure/core/amqp/internal/models/amqp_protocol.hpp"
#include "azure/core/amqp/models/amqp_value.hpp"
#include "common/async_operation_queue.hpp"
#include "connection_string_credential.hpp"
#include "session.hpp"
 
#include <azure/core/credentials/credentials.hpp>
 
#include <chrono>
#include <limits>
#include <memory>
#include <string>
 
namespace Azure { namespace Core { namespace Amqp { namespace _detail {
  class ConnectionImpl;
  class ConnectionFactory;
}}}} // namespace Azure::Core::Amqp::_detail
 
#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 MessageListenerEvents;
  } // namespace MessageTests
 
  class TestConnections_ConnectionAttributes_Test;
  class TestConnections_ConnectionOpenClose_Test;
  class TestConnections_ConnectionListenClose_Test;
  class TestSocketListenerEvents;
  class LinkSocketListenerEvents;
  class TestLinks_LinkAttachDetach_Test;
  class TestSessions_MultipleSessionBeginEnd_Test;
  class TestMessages_SenderOpenClose_Test;
  class TestMessages_TestLocalhostVsTls_Test;
  class TestMessages_SenderSendAsync_Test;
  class TestMessages_SenderOpenClose_Test;
  class TestMessages_ReceiverOpenClose_Test;
  class TestMessages_ReceiverReceiveAsync_Test;
 
}}}} // namespace Azure::Core::Amqp::Tests
#endif // _azure_TESTING_BUILD
#if defined(SAMPLES_BUILD)
namespace LocalServerSample {
int LocalServerSampleMain();
} // namespace LocalServerSample
#endif // SAMPLES_BUILD
 
namespace Azure { namespace Core { namespace Amqp { namespace _internal {
  class Session;
 
  class Error;
 
  constexpr uint16_t AmqpPort = 5672;
 
  constexpr uint16_t AmqpTlsPort = 5671;
 
  enum class ConnectionState
  {
    Start,
    HeaderReceived,
    HeaderSent,
    HeaderExchanged,
    OpenPipe,
    OcPipe,
    OpenReceived,
    OpenSent,
    ClosePipe,
    Opened,
    CloseReceived,
    CloseSent,
    Discarding,
    End,
 
    Error,
  };
 
  std::ostream& operator<<(std::ostream& stream, ConnectionState value);
 
  class Connection;
 
  class ConnectionEvents {
  protected:
    virtual ~ConnectionEvents() = default;
 
  public:
    virtual void OnConnectionStateChanged(
        Connection const& connection,
        ConnectionState newState,
        ConnectionState oldState)
        = 0;
 
    virtual void OnIOError(Connection const& connection) = 0;
  };
 
  class ConnectionEndpointEvents {
  protected:
    virtual ~ConnectionEndpointEvents() = default;
 
  public:
    virtual bool OnNewEndpoint(Connection const& connection, Endpoint& endpoint) = 0;
  };
 
  struct ConnectionOptions final
  {
    std::vector<std::string> AuthenticationScopes;
 
    std::chrono::milliseconds IdleTimeout{std::chrono::minutes(1)};
 
    uint32_t MaxFrameSize{(std::numeric_limits<uint32_t>::max)()};
 
    uint16_t MaxChannelCount{65535};
 
    Models::AmqpMap Properties;
 
    uint16_t Port{AmqpTlsPort};
 
    std::string ContainerId;
 
    bool EnableTrace{false};
  };
 
  class Connection final {
  public:
    Connection(
        std::string const& hostName,
        std::shared_ptr<Credentials::TokenCredential> credential,
        ConnectionOptions const& options,
        ConnectionEvents* eventHandler = nullptr);
 
    Connection(
        Network::_internal::Transport const& transport,
        ConnectionOptions const& options,
        ConnectionEvents* eventHandler,
        ConnectionEndpointEvents* endpointEvents);
 
    ~Connection();
 
    Session CreateSession(SessionOptions const& options = {}, SessionEvents* eventHandler = nullptr)
        const;
 
    Session CreateSession(
        Endpoint& newEndpoint,
        SessionOptions const& options = {},
        SessionEvents* eventHandler = nullptr) const;
 
    void Poll();
 
  private:
    void Open();
 
    void Listen();
 
    void Close(
        std::string const& condition = {},
        std::string const& description = {},
        Models::AmqpValue info = {});
 
    std::string GetHost() const;
 
    uint16_t GetPort() const;
 
    uint32_t GetMaxFrameSize() const;
 
    uint32_t GetRemoteMaxFrameSize() const;
 
    uint16_t GetMaxChannel() const;
 
    std::chrono::milliseconds GetIdleTimeout() const;
 
    Models::AmqpMap GetProperties() const;
 
    void SetIdleEmptyFrameSendPercentage(double idleTimeoutEmptyFrameSendRatio);
 
  private:
    Connection(std::shared_ptr<_detail::ConnectionImpl> impl) : m_impl{impl} {}
 
    std::shared_ptr<_detail::ConnectionImpl> m_impl;
    friend class _detail::ConnectionFactory;
#if _azure_TESTING_BUILD
    friend class Azure::Core::Amqp::Tests::MessageTests::AmqpServerMock;
    friend class Azure::Core::Amqp::Tests::MessageTests::MessageListenerEvents;
    friend class Azure::Core::Amqp::Tests::TestSocketListenerEvents;
    friend class Azure::Core::Amqp::Tests::LinkSocketListenerEvents;
    friend class Azure::Core::Amqp::Tests::TestConnections_ConnectionAttributes_Test;
    friend class Azure::Core::Amqp::Tests::TestConnections_ConnectionOpenClose_Test;
    friend class Azure::Core::Amqp::Tests::TestConnections_ConnectionListenClose_Test;
    friend class Azure::Core::Amqp::Tests::TestSessions_MultipleSessionBeginEnd_Test;
    friend class Azure::Core::Amqp::Tests::TestLinks_LinkAttachDetach_Test;
    friend class Azure::Core::Amqp::Tests::TestMessages_SenderOpenClose_Test;
    friend class Azure::Core::Amqp::Tests::TestMessages_TestLocalhostVsTls_Test;
    friend class Azure::Core::Amqp::Tests::TestMessages_SenderSendAsync_Test;
    friend class Azure::Core::Amqp::Tests::TestMessages_SenderOpenClose_Test;
 
#endif // _azure_TESTING_BUILD
#if SAMPLES_BUILD
    friend int LocalServerSample::LocalServerSampleMain();
#endif // SAMPLES_BUILD
  };
}}}} // namespace Azure::Core::Amqp::_internal