Added shared subscription support preparation code.

example/broker.cpp

@@ -14,7 +14,7 @@
 int main() {
     MQTT_NS::setup_log();
     boost::asio::io_context ioc;
-    MQTT_NS::broker b(ioc);
+    MQTT_NS::broker::broker_t b(ioc);
     test_server_no_tls s(ioc, b);
     ioc.run();
 }

include/mqtt/broker/broker_namespace.hpp

@@ -0,0 +1,26 @@
+// Copyright Takatoshi Kondo 2020
+//
+// Distributed under the Boost Software License, Version 1.0.
+// (See accompanying file LICENSE_1_0.txt or copy at
+// http://www.boost.org/LICENSE_1_0.txt)
+
+#if !defined(MQTT_BROKER_BROKER_NAMESPACE_HPP)
+#define MQTT_BROKER_BROKER_NAMESPACE_HPP
+
+#include <mqtt/config.hpp>
+
+#include <mqtt/broker/broker_namespace.hpp>
+
+#if __cplusplus >= 201703L
+
+#define MQTT_BROKER_NS_BEGIN namespace MQTT_NS::broker {
+#define MQTT_BROKER_NS_END }
+
+#else  // __cplusplus >= 201703L
+
+#define MQTT_BROKER_NS_BEGIN namespace MQTT_NS { namespace broker {
+#define MQTT_BROKER_NS_END } }
+
+#endif // __cplusplus >= 201703L
+
+#endif // MQTT_BROKER_BROKER_NAMESPACE_HPP

include/mqtt/broker/common_type.hpp

@@ -0,0 +1,24 @@
+// Copyright Takatoshi Kondo 2020
+//
+// Distributed under the Boost Software License, Version 1.0.
+// (See accompanying file LICENSE_1_0.txt or copy at
+// http://www.boost.org/LICENSE_1_0.txt)
+
+#if !defined(MQTT_BROKER_COMMON_TYPE_HPP)
+#define MQTT_BROKER_COMMON_TYPE_HPP
+
+#include <mqtt/config.hpp>
+
+#include <mqtt/broker/broker_namespace.hpp>
+#include <mqtt/server.hpp>
+
+MQTT_BROKER_NS_BEGIN
+
+using endpoint_t = server<>::endpoint_t;
+using con_sp_t = std::shared_ptr<endpoint_t>;
+using con_wp_t = std::weak_ptr<endpoint_t>;
+using packet_id_t = endpoint_t::packet_id_t;
+
+MQTT_BROKER_NS_END
+
+#endif // MQTT_BROKER_COMMON_TYPE_HPP

include/mqtt/broker/inflight_message.hpp

@@ -0,0 +1,148 @@
+// Copyright Takatoshi Kondo 2020
+//
+// Distributed under the Boost Software License, Version 1.0.
+// (See accompanying file LICENSE_1_0.txt or copy at
+// http://www.boost.org/LICENSE_1_0.txt)
+
+#if !defined(MQTT_BROKER_INFLIGHT_MESSAGE_HPP)
+#define MQTT_BROKER_INFLIGHT_MESSAGE_HPP
+
+#include <mqtt/config.hpp>
+
+#include <chrono>
+
+#include <boost/asio/steady_timer.hpp>
+
+#include <mqtt/broker/broker_namespace.hpp>
+#include <mqtt/message_variant.hpp>
+#include <mqtt/any.hpp>
+#include <mqtt/visitor_util.hpp>
+
+#include <mqtt/broker/common_type.hpp>
+#include <mqtt/broker/tags.hpp>
+#include <mqtt/broker/property_util.hpp>
+
+MQTT_BROKER_NS_BEGIN
+
+class inflight_messages;
+
+class inflight_message {
+public:
+    inflight_message(
+        store_message_variant msg,
+        any life_keeper,
+        std::shared_ptr<as::steady_timer> tim_message_expiry)
+        :msg_ { force_move(msg) },
+         life_keeper_ { force_move(life_keeper) },
+         tim_message_expiry_ { force_move(tim_message_expiry) }
+    {}
+
+    packet_id_t packet_id() const {
+        return
+            MQTT_NS::visit(
+                make_lambda_visitor(
+                    [](auto const& m) {
+                        return m.packet_id();
+                    }
+                ),
+                msg_
+            );
+    }
+
+    void send(endpoint_t& ep) const {
+        optional<store_message_variant> msg_opt;
+        if (tim_message_expiry_) {
+            MQTT_NS::visit(
+                make_lambda_visitor(
+                    [&](v5::basic_publish_message<sizeof(packet_id_t)> const& m) {
+                        auto updated_msg = m;
+                        auto& props = updated_msg.props();
+
+                        auto d =
+                            std::chrono::duration_cast<std::chrono::seconds>(
+                                tim_message_expiry_->expiry() - std::chrono::steady_clock::now()
+                            ).count();
+                        if (d < 0) d = 0;
+                        set_property<v5::property::message_expiry_interval>(
+                            props,
+                            v5::property::message_expiry_interval(
+                                static_cast<uint32_t>(d)
+                            )
+                        );
+                        msg_opt.emplace(force_move(updated_msg));
+                    },
+                    [](auto const&) {
+                    }
+                ),
+                msg_
+            );
+        }
+        ep.send_store_message(msg_opt ? msg_opt.value() : msg_, life_keeper_);
+    }
+
+private:
+    friend class inflight_messages;
+
+    store_message_variant msg_;
+    any life_keeper_;
+    std::shared_ptr<as::steady_timer> tim_message_expiry_;
+};
+
+class inflight_messages {
+public:
+    void insert(
+        store_message_variant msg,
+        any life_keeper,
+        std::shared_ptr<as::steady_timer> tim_message_expiry
+    ) {
+        messages_.emplace_back(
+            force_move(msg),
+            force_move(life_keeper),
+            force_move(tim_message_expiry)
+        );
+    }
+
+    void send_all_messages(endpoint_t& ep) {
+        for (auto const& ifm : messages_) {
+            ifm.send(ep);
+        }
+    }
+
+    void clear() {
+        messages_.clear();
+    }
+
+    template <typename Tag>
+    decltype(auto) get() {
+        return messages_.get<Tag>();
+    }
+
+    template <typename Tag>
+    decltype(auto) get() const {
+        return messages_.get<Tag>();
+    }
+
+private:
+    using mi_inflight_message = mi::multi_index_container<
+        inflight_message,
+        mi::indexed_by<
+            mi::sequenced<
+                mi::tag<tag_seq>
+            >,
+            mi::ordered_unique<
+                mi::tag<tag_pid>,
+                BOOST_MULTI_INDEX_CONST_MEM_FUN(inflight_message, packet_id_t, packet_id)
+            >,
+            mi::ordered_non_unique<
+                mi::tag<tag_tim>,
+                BOOST_MULTI_INDEX_MEMBER(inflight_message, std::shared_ptr<as::steady_timer>, tim_message_expiry_)
+            >
+        >
+    >;
+
+    mi_inflight_message messages_;
+};
+
+MQTT_BROKER_NS_END
+
+#endif // MQTT_BROKER_INFLIGHT_MESSAGE_HPP

include/mqtt/broker/offline_message.hpp

@@ -0,0 +1,139 @@
+// Copyright Takatoshi Kondo 2020
+//
+// Distributed under the Boost Software License, Version 1.0.
+// (See accompanying file LICENSE_1_0.txt or copy at
+// http://www.boost.org/LICENSE_1_0.txt)
+
+#if !defined(MQTT_BROKER_OFFLINE_MESSAGE_HPP)
+#define MQTT_BROKER_OFFLINE_MESSAGE_HPP
+
+#include <mqtt/config.hpp>
+
+#include <boost/asio/steady_timer.hpp>
+#include <boost/multi_index_container.hpp>
+#include <boost/multi_index/ordered_index.hpp>
+#include <boost/multi_index/sequenced_index.hpp>
+#include <boost/multi_index/member.hpp>
+
+#include <mqtt/buffer.hpp>
+#include <mqtt/property_variant.hpp>
+#include <mqtt/publish.hpp>
+
+#include <mqtt/broker/broker_namespace.hpp>
+#include <mqtt/broker/common_type.hpp>
+#include <mqtt/broker/tags.hpp>
+#include <mqtt/broker/property_util.hpp>
+
+MQTT_BROKER_NS_BEGIN
+
+namespace mi = boost::multi_index;
+
+class offline_messages;
+
+// The offline_message structure holds messages that have been published on a
+// topic that a not-currently-connected client is subscribed to.
+// When a new connection is made with the client id for this saved data,
+// these messages will be published to that client, and only that client.
+class offline_message {
+public:
+    offline_message(
+        buffer topic,
+        buffer contents,
+        v5::properties props,
+        publish_options pubopts,
+        std::shared_ptr<as::steady_timer> tim_message_expiry)
+        : topic_(force_move(topic)),
+          contents_(force_move(contents)),
+          props_(force_move(props)),
+          pubopts_(pubopts),
+          tim_message_expiry_(force_move(tim_message_expiry))
+    { }
+
+    void send(endpoint_t& ep) const {
+        auto props = props_;
+        if (tim_message_expiry_) {
+            auto d =
+                std::chrono::duration_cast<std::chrono::seconds>(
+                    tim_message_expiry_->expiry() - std::chrono::steady_clock::now()
+                ).count();
+            if (d < 0) d = 0;
+            set_property<v5::property::message_expiry_interval>(
+                props,
+                v5::property::message_expiry_interval(
+                    static_cast<uint32_t>(d)
+                )
+            );
+        }
+
+        ep.publish(topic_, contents_, pubopts_, force_move(props));
+    }
+
+private:
+    friend class offline_messages;
+
+    buffer topic_;
+    buffer contents_;
+    v5::properties props_;
+    publish_options pubopts_;
+    std::shared_ptr<as::steady_timer> tim_message_expiry_;
+};
+
+class offline_messages {
+public:
+    void send_all_messages(endpoint_t& ep) {
+        try {
+            auto& idx = messages_.get<tag_seq>();
+            while (!idx.empty()) {
+                idx.modify(
+                    idx.begin(),
+                    [&](auto& e) {
+                        e.send(ep);
+                    }
+                );
+                idx.pop_front();
+            }
+        }
+        catch (packet_id_exhausted_error const& e) {
+            MQTT_LOG("mqtt_broker", warning)
+                << MQTT_ADD_VALUE(address, &ep)
+                << e.what();
+        }
+        for (auto const& oflm : messages_) {
+            oflm.send(ep);
+        }
+    }
+
+    void clear() {
+        messages_.clear();
+    }
+
+    template <typename Tag>
+    decltype(auto) get() {
+        return messages_.get<Tag>();
+    }
+
+    template <typename Tag>
+    decltype(auto) get() const {
+        return messages_.get<Tag>();
+    }
+
+private:
+    using mi_offline_message = mi::multi_index_container<
+        offline_message,
+        mi::indexed_by<
+            mi::sequenced<
+                mi::tag<tag_seq>
+            >,
+            mi::ordered_non_unique<
+                mi::tag<tag_tim>,
+                BOOST_MULTI_INDEX_MEMBER(offline_message, std::shared_ptr<as::steady_timer>, tim_message_expiry_)
+            >
+        >
+    >;
+
+    mi_offline_message messages_;
+};
+
+MQTT_BROKER_NS_END
+
+#endif // MQTT_BROKER_OFFLINE_MESSAGE_HPP

include/mqtt/broker/property_util.hpp

@@ -0,0 +1,49 @@
+// Copyright Takatoshi Kondo 2020
+//
+// Distributed under the Boost Software License, Version 1.0.
+// (See accompanying file LICENSE_1_0.txt or copy at
+// http://www.boost.org/LICENSE_1_0.txt)
+
+#if !defined(MQTT_BROKER_PROPERTY_UTIL_HPP)
+#define MQTT_BROKER_PROPERTY_UTIL_HPP
+
+#include <mqtt/config.hpp>
+
+#include <mqtt/broker/broker_namespace.hpp>
+#include <mqtt/optional.hpp>
+#include <mqtt/property_variant.hpp>
+#include <mqtt/visitor_util.hpp>
+
+MQTT_BROKER_NS_BEGIN
+
+template <typename T>
+inline optional<T> get_property(v5::properties const& props) {
+    optional<T> result;
+
+    auto visitor = make_lambda_visitor(
+        [&result](T const& t) { result = t; },
+        [](auto const&) { }
+    );
+
+    for (auto const& p : props) {
+        MQTT_NS::visit(visitor, p);
+    }
+
+    return result;
+}
+
+template <typename T>
+inline void set_property(v5::properties& props, T&& v) {
+    auto visitor = make_lambda_visitor(
+        [&v](T& t) mutable { t = std::forward<T>(v); },
+        [](auto&) { }
+    );
+
+    for (auto& p : props) {
+        MQTT_NS::visit(visitor, p);
+    }
+}
+
+MQTT_BROKER_NS_END
+
+#endif // MQTT_BROKER_PROPERTY_UTIL_HPP