MqttConnectionCore.h

Go to the documentation of this file.

#pragma once

#include <aws/crt/Exports.h>
#include <aws/crt/StlAllocator.h>
#include <aws/crt/Types.h>
#include <aws/crt/http/HttpConnection.h>
#include <aws/crt/io/SocketOptions.h>
#include <aws/crt/io/TlsOptions.h>
#include <aws/crt/mqtt/MqttTypes.h>
#include <aws/crt/mqtt/private/MqttShared.h>

#include <aws/mqtt/client.h>
#include <aws/mqtt/v5/mqtt5_client.h>

#include <functional>
#include <memory>

namespace Aws
{
    namespace Crt
    {
        namespace Mqtt
        {
            class MqttConnection;

            class MqttConnectionCore final : public std::enable_shared_from_this<MqttConnectionCore>
            {
                friend MqttConnection;

              public:
                ~MqttConnectionCore();
                MqttConnectionCore(const MqttConnectionCore &) = delete;
                MqttConnectionCore(MqttConnectionCore &&) = delete;
                MqttConnectionCore &operator=(const MqttConnectionCore &) = delete;
                MqttConnectionCore &operator=(MqttConnectionCore &&) = delete;

                operator bool() const noexcept;

                void Destroy();

                int LastError() const noexcept;

                bool SetWill(const char *topic, QOS qos, bool retain, const ByteBuf &payload) noexcept;

                bool SetLogin(const char *username, const char *password) noexcept;

                bool SetHttpProxyOptions(const Http::HttpClientConnectionProxyOptions &proxyOptions) noexcept;

                bool SetReconnectTimeout(uint64_t min_seconds, uint64_t max_seconds) noexcept;

                bool Connect(
                    const char *clientId,
                    bool cleanSession,
                    uint16_t keepAliveTimeSecs,
                    uint32_t pingTimeoutMs,
                    uint32_t protocolOperationTimeoutMs,
                    bool setWebSocketInterceptor) noexcept;

                bool Disconnect() noexcept;

                aws_mqtt_client_connection *GetUnderlyingConnection() const noexcept;

                uint16_t Subscribe(
                    const char *topicFilter,
                    QOS qos,
                    OnMessageReceivedHandler &&onMessage,
                    OnSubAckHandler &&onSubAck) noexcept;

                uint16_t Subscribe(
                    const Vector<std::pair<const char *, OnMessageReceivedHandler>> &topicFilters,
                    QOS qos,
                    OnMultiSubAckHandler &&onOpComplete) noexcept;

                bool SetOnMessageHandler(OnMessageReceivedHandler &&onMessage) noexcept;

                uint16_t Unsubscribe(const char *topicFilter, OnOperationCompleteHandler &&onOpComplete) noexcept;

                uint16_t Publish(
                    const char *topic,
                    QOS qos,
                    bool retain,
                    const ByteBuf &payload,
                    OnOperationCompleteHandler &&onOpComplete) noexcept;

                const MqttConnectionOperationStatistics &GetOperationStatistics() noexcept;

              private:
                static std::shared_ptr<MqttConnectionCore> s_createMqttConnectionCore(
                    aws_mqtt_client *client,
                    std::shared_ptr<MqttConnection> connection,
                    MqttConnectionOptions options) noexcept;

                static std::shared_ptr<MqttConnectionCore> s_createMqttConnectionCore(
                    aws_mqtt5_client *mqtt5Client,
                    std::shared_ptr<MqttConnection> connection,
                    MqttConnectionOptions options) noexcept;

                static void s_onConnectionTermination(void *userData);

                static void s_onConnectionInterrupted(aws_mqtt_client_connection *, int errorCode, void *userData);
                static void s_onConnectionCompleted(
                    aws_mqtt_client_connection *,
                    int errorCode,
                    enum aws_mqtt_connect_return_code returnCode,
                    bool sessionPresent,
                    void *userData);

                static void s_onConnectionSuccess(
                    aws_mqtt_client_connection *,
                    ReturnCode returnCode,
                    bool sessionPresent,
                    void *userData);

                static void s_onConnectionFailure(aws_mqtt_client_connection *, int errorCode, void *userData);

                static void s_onConnectionResumed(
                    aws_mqtt_client_connection *,
                    ReturnCode returnCode,
                    bool sessionPresent,
                    void *userData);

                static void s_onConnectionClosed(
                    aws_mqtt_client_connection *,
                    on_connection_closed_data *data,
                    void *userData);

                static void s_onDisconnect(aws_mqtt_client_connection *connection, void *userData);
                static void s_onPublish(
                    aws_mqtt_client_connection *connection,
                    const aws_byte_cursor *topic,
                    const aws_byte_cursor *payload,
                    bool dup,
                    enum aws_mqtt_qos qos,
                    bool retain,
                    void *userData);

                static void s_onSubAck(
                    aws_mqtt_client_connection *connection,
                    uint16_t packetId,
                    const struct aws_byte_cursor *topic,
                    enum aws_mqtt_qos qos,
                    int errorCode,
                    void *userdata);
                static void s_onMultiSubAck(
                    aws_mqtt_client_connection *connection,
                    uint16_t packetId,
                    const struct aws_array_list *topicSubacks,
                    int errorCode,
                    void *userdata);
                static void s_onOpComplete(
                    aws_mqtt_client_connection *connection,
                    uint16_t packetId,
                    int errorCode,
                    void *userdata);

                static void s_onWebsocketHandshake(
                    struct aws_http_message *request,
                    void *userData,
                    aws_mqtt_transform_websocket_handshake_complete_fn *completeFn,
                    void *completeCtx);

                MqttConnectionCore(
                    aws_mqtt_client *client,
                    aws_mqtt5_client *mqtt5Client,
                    std::shared_ptr<MqttConnection> connection,
                    MqttConnectionOptions options) noexcept;

                void createUnderlyingConnection(aws_mqtt_client *mqttClient);
                void createUnderlyingConnection(aws_mqtt5_client *mqtt5Client);
                void connectionInit();

                std::shared_ptr<MqttConnection> obtainConnectionInstance();

                aws_mqtt_client_connection *m_underlyingConnection;
                String m_hostName;
                uint32_t m_port;
                Crt::Io::TlsContext m_tlsContext;
                Io::TlsConnectionOptions m_tlsOptions;
                Io::SocketOptions m_socketOptions;
                Crt::Optional<Http::HttpClientConnectionProxyOptions> m_proxyOptions;
                void *m_onAnyCbData;
                bool m_useTls;
                bool m_useWebsocket;
                MqttConnectionOperationStatistics m_operationStatistics;
                bool m_enableMetrics = true;
                IoTDeviceSDKMetrics m_sdkMetrics;
                Allocator *m_allocator;

                std::weak_ptr<MqttConnection> m_connection;

                std::shared_ptr<MqttConnectionCore> m_self;
            };
        } // namespace Mqtt
    } // namespace Crt
} // namespace Aws