Customize Underlying Layer

Customization Points

You can use async_mqtt with not only predefined underlying layers (mqtt, mqtts, ws, and wss) but also your own underlying layers. In order to use your own underlying layer with async_mqtt, you need to define a class template specilization.

Lets' say you want to use your_own_stream as the underlying layer of async_mqtt.

Four check points

If you need some custom processes for your_own_stream at least one of the following four points, you need to define struct layer_customize specilization. If you don’t define the specialization, or its static member functions, the default version is used.

1. initialize

Do nothing by default.
initialize is called from the bottom layer to the top layer synchronously.
initialize is called when the endpoint is created.

2. async_read_some

Call your_own_stream.async_read_some() by default.
- See https://www.boost.org/doc/html/boost_asio/reference/basic_stream_socket/async_read_some.html

3. async_read

Call Boost.Asio’s async_read() with your_own_stream by default. Boost.Asio’s async_read() calls your_own_stream.async_read_some() internally.
- See https://www.boost.org/doc/html/boost_asio/reference/async_read/overload1.html

4. async_write

Call Boost.Asio’s async_write() with your_own_stream by default.
- See https://www.boost.org/doc/html/boost_asio/reference/async_write/overload1.html

5. async_close

Do nothing by default.
async_close is called from the top layer to the bottom layer asynchronously.
initialize is called when the endpoint is created.

Only if you want to customize some of the four points, then define the member function as follows:

customized_your_own_stream.hpp

#include <async_mqtt/util/stream_traits.hpp>

// Partial or full specialization here
template <typename NextLayer>
struct layer_customize<your_own_stream<NextLayer>> {

    // define if you want to customize initialize process
    static void initialize(your_own_stream<NextLayer>& stream) {
        // Special initialization for your_own_stream
    }

    // define if you want to customize async_read_some process
    template <
        typename MutableBufferSequence,
        typename CompletionToken
    >
    static auto
    async_read_some(
        your_own_stream<NextLayer>& stream,
        MutableBufferSequence const& mbs,
        CompletionToken&& token
    ) {
        return as::async_compose<
            CompletionToken,
            void(error_code const& ec, std::size_t size)
        > (
            async_read_some_impl<MutableBufferSequence>{
                stream,
                mbs
            },
            token
        );
    }

    // define if you want to customize async_read process
    template <
        typename MutableBufferSequence,
        typename CompletionToken
    >
    static auto
    async_read(
        your_own_stream<NextLayer>& stream,
        MutableBufferSequence const& mbs,
        CompletionToken&& token
    ) {
        return as::async_compose<
            CompletionToken,
            void(error_code const& ec, std::size_t size)
        > (
            async_read_impl<MutableBufferSequence>{
                stream,
                mbs
            },
            token
        );
    }

    // define if you want to customize async_write process
    template <
        typename ConstBufferSequence,
        typename CompletionToken
    >
    static auto
    async_write(
        your_own_stream<NextLayer>& stream,
        ConstBufferSequence const& cbs,
        CompletionToken&& token
    ) {
        return as::async_compose<
            CompletionToken,
            void(error_code const& ec, std::size_t size)
        > (
            async_write_impl<ConstBufferSequence>{
                stream,
                cbs
            },
            token
        );
    }


    // define if you want to customize async_close process
    template <
        typename CompletionToken
    >
    static auto
    async_close(
        bs::websocket::stream<NextLayer>& stream,
        CompletionToken&& token
    ) {
        return as::async_compose<
            CompletionToken,
            void(error_code const& ec)
        > (
            async_close_impl{
                stream
            },
            token
        );
    }
};

// define async_read_impl, async_write_impl, and async_close_impl for your own layer here

You can replace your_***_function(). async_mqtt has some of predefined struct layer_customize specializations.

See https://github.com/redboltz/async_mqtt/blob/main/include/async_mqtt/predefined_layer/

customized_*_stream.hpp help defining your own underlaying layer.