implement trie based wildcard support

[redboltz]

See #684 #676 #685 #681

We have topic_name and topic_filter validatiton functions. They are already been meged.
I already merged tire supporting functions.

Now, it is a time to implement trie version of wildcard support to test_broker.

@kleunen , could you demonstrate how to use subscription_map and retained_topic map ?

Could you write a code example or pull request ?

I'm not sure sub_con_online and sub_con_offline need to be shared_ptr or not. I remember your previous approach uses shared_ptr for them. Please check it is still required in the current master.

I guess that you replace

mqtt_cpp/test/test_broker.hpp

Lines 1699 to 1700 in d035c5b

	mi_session_online sessions_online_; ///< Map of active client id and connections
	mi_session_offline sessions_offline_; ///< Storage for sessions not currently active. Indexed by client id.

to subscription_map.

I think that subscription_map needs to have search by the address of endpoint_t and client_id.
Consider the following scenario.

1. client1 connected to the broker with CleanStart:false SessionExpiryInterval:inifinity
2. client1 subscribed topic1. Broker updates subscription_map.
3. client1 disconnected. Broker need to search subscription_map by the address of endpoint_t and move it to offline one.

It is implemented using multi_index now. See tag_con.

mqtt_cpp/test/test_broker.hpp

Lines 1554 to 1557 in d035c5b

	mi::ordered_non_unique<
	mi::tag<tag_con>,
	BOOST_MULTI_INDEX_MEMBER(sub_con_online, con_sp_t, con)
	>,

There are some kind of events to be a trigger to access to subscription_map.

1. publish: access to subscription_map by topic_name and get a set of matched sub_con_online or sub_con_offline.
2. subscribe: insert to subscription_map.
3. unsubscribe: remove from subscription_map. The key is topic_filter.
4. disconnect: move or update sub_con_online element in subscription_map. The key is the address of endpoint_t.
5. connect: move or update sub_con_offine element in subscription_map. The key is client_id.
6. timeout: Session expire case. remove sub_con_offline from subscription_map. The key is client_id.

@kleunen , could you tell me how to implement them ?

[kleunen]

@kleunen , could you demonstrate how to use subscription_map and retained_topic map ?

Yes, i have an example broker and a small unit test setup which shows how the maps should be used. How can I send you the code example ?

[redboltz]

Please put the code here with cpp decoration as follows:

// your code example
int main() {
    subscription_map m;
    ...
}

[kleunen]

Tsumaranai monodesuga ( hope this is right thing to say :D )

It does say:
Remaining size should be 1 (root element only)
Remaining size: 4

So something is wrong somewhere, i think this is in the test actually, not the map itself.

//
// Created by wkl04 on 10-12-2019.
//
#include "precomp.h"

#include "subscription_map.h"
#include "retained_topic_map.h"

#include <iostream>
#include <chrono>

void TestSingleSubscription()
{
    std::string text = "example/test/A";

    single_subscription_map< std::function< void() > > map;
    map.insert("example/test/A", [] { std::cout << "example/test/A: Subscriber 1" << std::endl; } );

    try {
        map.insert("example/test/A", [] { std::cout << "example/test/A: Subscriber 2" << std::endl; } );
    } catch(std::exception &e)
    {
        // Should fail due to duplicate subscription
        std::cout << e.what() << std::endl;
    }

    auto handle = map.insert("example/test/B", [] { std::cout << "example/test/B" << std::endl; } );
    std::cout << "Handle of new subscription: " << map.handle_to_subscription(handle) << std::endl;
    map.update(handle, [] { std::cout << "example/test/B updated" << std::endl; } );
   // map.remove(handle);

    map.insert("example/+/A", [] { std::cout << "example/+/A: Plus" << std::endl; } );
    map.insert("example/#", [] { std::cout << "example/#" << std::endl; } );

    std::cout << "Test with hash, plus and full path" << std::endl;
    map.find("example/test/A", [](std::function< void() >  const &a) {
        a();
    });

    map.remove("example/+/A");

    std::cout << "Test with hash, full path" << std::endl;
    map.find("example/test/A", [](std::function< void() > const &a) {
        a();
    });

    map.remove("example/#");

    std::cout << "Test with full path only" << std::endl;
    map.find("example/test/A", [](std::function< void() >  const &a) {
        a();
    });

    map.remove("example/test/A");
    // map.remove("example/test/A");
    map.remove("example/test/B");

    std::cout << "No entries exist anymore" << std::endl;
    map.find("example/test/A", [](std::function< void() >  const &a) {
        a();
    });

    std::cout << "Remaining size should be 1 (root element only)" << std::endl;
    std::cout << "Remaining size: " << map.size() << std::endl;
}

void TestMultipleSubscription()
{
    std::string text = "example/test/A";

    multiple_subscription_map<std::string> map;

    try {
        map.insert("doubleslash//#", "Should fail");
    } catch(std::exception &e)
    {
        // Should fail due to empty path usage
        std::cout << e.what() << std::endl;
    }

    map.insert("example/test/A", "example/test/A: Subscriber 1");
    map.insert("example/test/A", "example/test/A: Subscriber 2");
    map.insert("example/test/B", "example/test/B");
    map.insert("example/+/A", "example/+/A: Plus");
    map.insert("example/#", "example/#");
   // map.insert("#", "Root hash");

    std::cout << "Lookup handle of an entry" << std::endl;
    std::cout << "Size: " << map.lookup("example/test/B").size() << std::endl;
    std::cout << "Path: " << map.handle_to_subscription(map.lookup("example/test/B")) << std::endl;

    map.remove(map.lookup("example/test/A"), "example/test/A: Subscriber 1");

    std::cout << "Test with hash, plus and full path" << std::endl;
    map.find("example/test/A", [](std::string const &a) {
        std::cout << a << std::endl;
    });

    /* map.remove("example/+/A", "example/+/A: Plus");

    std::cout << "Test with hash, full path" << std::endl;
    map.find("example/test/A", [](std::string const &a) {
        std::cout << a << std::endl;
    });

    map.remove("example/#", "example/#");

    std::cout << "Test with full path only" << std::endl;
    map.find("example/test/A", [](std::string const &a) {
        std::cout << a << std::endl;
    });

    map.remove("example/test/A", "example/test/A: Subscriber 1");
    map.remove("example/test/A", "example/test/A: Subscriber 2");
    map.remove("example/test/B", "example/test/B");

    std::cout << "No entries exist anymore" << std::endl;
    map.find("example/test/A", [](std::string const &a) {
        std::cout << a << std::endl;
    }); */

    std::cout << "Remaining size should be 1 (root element only)" << std::endl;
    std::cout << "Remaining size: " << map.size() << std::endl;
}

void TestRetainedTopics()
{
    retained_topic_map<std::string> map;
    map.insert_or_update("example/test/A", "example/test/A: Content 1");
    map.insert_or_update("example/test/B", "example/test/B: Content 1");
    map.insert_or_update("example/A/test", "example/A/test: Content 1");
    map.insert_or_update("example/B/test", "example/B/test: Content 1");

    /*map.find("example/test/A", [](std::string const &a) {
        std::cout << a << std::endl;
    });

    map.find("example/test/B", [](std::string const &a) {
        std::cout << a << std::endl;
    });

    map.find("example/test/+", [](std::string const &a) {
        std::cout << a << std::endl;
    });

    map.find("example/+/B", [](std::string const &a) {
        std::cout << a << std::endl;
    }); */

    std::cout << "Match: example/#" << std::endl;
    map.find("example/#", [](std::string const &a) {
        std::cout << a << " " << a.size() << std::endl;
    });
    std::cout << "Done" << std::endl;

    map.remove("example/test/A");
    map.find("example/+/A", [](std::string const &a) {
        std::cout << a << std::endl;
    });

}


int main(int, char**)
{
    try {
        TestSingleSubscription();
        TestMultipleSubscription();
        TestRetainedTopics();


    } catch(std::exception &e)
    {
        std::cout << e.what() << std::endl;
    }
}
``

[kleunen]

And the small broker I made:

#include "precomp.h"

#include <string>
#include <iostream>

#include "mqtt_server_cpp.hpp"
#include "subscription_map.h"
#include "retained_topic_map.h"

using con_t = MQTT_NS::server<>::endpoint_t;
using con_sp_t = std::shared_ptr<con_t>;

struct session_t;
using session_ptr_t = std::shared_ptr<session_t>;
using session_weak_ptr_t = std::weak_ptr<session_t>;

using session_map_t = std::map<MQTT_NS::buffer, session_ptr_t>;

struct session_subscription_t
{
    session_ptr_t session;
    MQTT_NS::qos qos;

    session_subscription_t(const session_ptr_t &session, MQTT_NS::qos qos = MQTT_NS::qos())
        : session(session), qos(qos)
    { }
};

inline bool operator<(session_subscription_t const &a, session_subscription_t const &b)
{
    return a.session < b.session;
}

using subscription_map_t = multiple_subscription_map< session_subscription_t >;

// Map of all subscription -> session
subscription_map_t subs_map;

struct session_t
{
    MQTT_NS::buffer client_id;
    std::weak_ptr<con_t> con;
    std::map<MQTT_NS::buffer, subscription_map_t::handle> subscriptions;

    session_t(const std::weak_ptr<con_t> &con)
        : con(con)
    { }

    ~session_t()
    {
        std::cout << "Release session: " << client_id << std::endl;
    }

    std::shared_ptr<con_t> get_connection()
    {
        auto sp = con.lock();
        BOOST_ASSERT(sp);
        return sp;
    }

    void add_subscription(MQTT_NS::buffer topic, subscription_map_t::handle const &handle)
    {
        subscriptions[topic] = handle;
    }

    subscription_map_t::handle get_subscription(MQTT_NS::buffer topic)
    {
        auto sub_iter = subscriptions.find(topic);
        if(sub_iter == subscriptions.end())
            return subscription_map_t::handle();
        else
            return sub_iter->second;
    }

    subscription_map_t::handle remove_subscription(MQTT_NS::buffer topic)
    {
        auto sub_iter = subscriptions.find(topic);
        if(sub_iter == subscriptions.end())
            return subscription_map_t::handle();

        auto result = sub_iter->second;
        subscriptions.erase(sub_iter);
        return result;
    }



    void publish(MQTT_NS::buffer topic_name, MQTT_NS::buffer contents, MQTT_NS::publish_options options) {
        auto sp = con.lock();
        if(sp) {
            sp->async_publish(
                    boost::asio::buffer(topic_name),
                    boost::asio::buffer(contents),
                    std::make_pair(topic_name, contents), options);
        }
    }
};


// All stored retained topics
retained_topic_map< std::pair<MQTT_NS::buffer, MQTT_NS::buffer> > retained_map;

// Map of all sessions
session_map_t sessions;

inline void close_session(session_ptr_t session) {
    // Remove all subscriptions for this session
    for(auto const &i: session->subscriptions) {
        subs_map.remove(i.second, session_subscription_t(session));
    }

    // Remove the client id from the list of active sessions
    auto i = sessions.find(session->client_id);
    if(i != sessions.end() && i->second == session)
        sessions.erase(i);

    std::cout << "Active sessions: " << sessions.size() << std::endl;
    std::cout << "Subs map size: " << subs_map.size() << std::endl;
}

int main(int argc, char** argv) {
    if (argc != 2) {
        std::cout << argv[0] << " port" << std::endl;
        return -1;
    }
    boost::asio::io_context ioc;

    auto server_endpoint = boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(),boost::lexical_cast<std::uint16_t>(argv[1]));
    auto s = MQTT_NS::server<>(server_endpoint, ioc);

    s.set_error_handler(
            [](MQTT_NS::error_code ec) {
                std::cout << "error: " << ec.message() << std::endl;
            }
    );


    s.set_accept_handler(
            [](con_sp_t spep) {
                auto& ep = *spep;

                session_ptr_t session = std::make_shared<session_t>(std::weak_ptr<con_t>(spep));

                using packet_id_t = typename std::remove_reference_t<decltype(ep)>::packet_id_t;
                std::cout << "accept" << std::endl;

                // Pass spep to keep lifetime.
                // It makes sure wp.lock() never return nullptr in the handlers below
                // including close_handler and error_handler.
                ep.start_session(session);

                // set connection (lower than MQTT) level handlers
                ep.set_close_handler(
                        [session]() {
                            std::cout << "[server] closed session: " << session->client_id << std::endl;
                            close_session(session);
                        });

                ep.set_error_handler(
                        [session](MQTT_NS::error_code ec) {
                            std::cout << "[server] error: " << ec.message() << " " << session->client_id << std::endl;
                            close_session(session);
                        });

                // set MQTT level handlers
                ep.set_connect_handler(
                        [session](MQTT_NS::buffer client_id, MQTT_NS::optional<MQTT_NS::buffer> username, MQTT_NS::optional<MQTT_NS::buffer> password, MQTT_NS::optional<MQTT_NS::will>, bool clean_session, std::uint16_t keep_alive) {
                            auto sp = session->get_connection();

                            using namespace MQTT_NS::literals;
                            std::cout << "[server] client_id    : " << client_id << std::endl;
                            std::cout << "[server] username     : " << (username ? username.value() : "none"_mb) << std::endl;
                            std::cout << "[server] password     : " << (password ? password.value() : "none"_mb) << std::endl;
                            std::cout << "[server] clean_session: " << std::boolalpha << clean_session << std::endl;
                            std::cout << "[server] keep_alive   : " << keep_alive << std::endl;

                            if(sessions.find(client_id) != sessions.end())
                            {
                                // A session with this client id already exists
                                sp->connack(false, MQTT_NS::connect_return_code::identifier_rejected);
                                return false;
                            }

                            session->client_id = client_id;
                            sessions[client_id] = session;

                            sp->connack(false, MQTT_NS::connect_return_code::accepted);
                            return true;
                        }
                );

                ep.set_pingreq_handler([session]() {
                    auto sp = session->get_connection();
                    sp->async_pingresp();
                    return true;
                });

                ep.set_disconnect_handler(
                        [session]() {
                            auto sp = session->get_connection();
                            std::cout << "[server] disconnect received." << std::endl;
                            close_session(session);
                        });

                ep.set_puback_handler(
                        [](packet_id_t packet_id){
                          //  std::cout << "[server] puback received. packet_id: " << packet_id << std::endl;
                            return true;
                        });

                ep.set_pubrec_handler(
                        [](packet_id_t packet_id){
                          //  std::cout << "[server] pubrec received. packet_id: " << packet_id << std::endl;
                            return true;
                        });

                ep.set_pubrel_handler(
                        [](packet_id_t packet_id){
//                            std::cout << "[server] pubrel received. packet_id: " << packet_id << std::endl;
                            return true;
                        });

                ep.set_pubcomp_handler(
                        [](packet_id_t packet_id){
                           // std::cout << "[server] pubcomp received. packet_id: " << packet_id << std::endl;
                            return true;
                        });

                ep.set_publish_handler(
                        [session]
                                (MQTT_NS::optional<packet_id_t> packet_id,
                                 MQTT_NS::publish_options pubopts,
                                 MQTT_NS::buffer topic_name,
                                 MQTT_NS::buffer contents){
                            /* std::cout << "[server] publish received."
                                      << " dup: "    << pubopts.get_dup()
                                      << " qos: "    << pubopts.get_qos()
                                      << " retain: " << pubopts.get_retain() << std::endl;
                            if (packet_id)
                                std::cout << "[server] packet_id: " << *packet_id << std::endl;
                            std::cout << "[server] topic_name: " << topic_name << std::endl;
                            std::cout << "[server] contents: " << contents << std::endl; */

                            if(!mqtt_valid_topic(topic_name))
                                return false;

                            if(pubopts.get_retain() == MQTT_NS::retain::yes)
                                retained_map.insert_or_update(topic_name, std::make_pair(topic_name, contents));

                            std::map< session_ptr_t, MQTT_NS::qos > subscribers;
                            subs_map.find(std::string(topic_name.data(), topic_name.size()), [&topic_name, &contents, &pubopts, &subscribers](session_subscription_t const &r){
                                subscribers[r.session] = std::max(r.qos, subscribers[r.session]);
                            });

                            for(auto const &r: subscribers)
                                r.first->publish(topic_name, contents, std::min(r.second, pubopts.get_qos()) |  pubopts.get_retain());

                            return true;
                        });

                ep.set_subscribe_handler(
                        [session] (packet_id_t packet_id, std::vector<std::tuple<MQTT_NS::buffer, MQTT_NS::subscribe_options>> entries) {
                            std::cout << "[server]subscribe received. packet_id: " << packet_id << std::endl;
                            std::vector<MQTT_NS::suback_return_code> res;
                            res.reserve(entries.size());

                            auto sp = session->get_connection();

                            for (auto const& e : entries) {
                                MQTT_NS::buffer topic = std::get<0>(e);
                                MQTT_NS::qos qos_value = std::get<1>(e).get_qos();

                                if(mqtt_valid_subscription(topic)) {

                                    auto handle = session->get_subscription(topic);
                                    if(!handle.empty())
                                        subs_map.update(handle, session_subscription_t(session, qos_value));
                                    else
                                        session->add_subscription(topic, subs_map.insert(topic, session_subscription_t(session, qos_value)));

                                    res.emplace_back(MQTT_NS::qos_to_suback_return_code(qos_value));

                                    retained_map.find(topic, [&session, &topic, qos_value](std::pair<MQTT_NS::buffer, MQTT_NS::buffer> const &retained_message) {
                                        session->publish(retained_message.first, retained_message.second, qos_value);
                                    });
                                } else {
                                    res.emplace_back(MQTT_NS::suback_return_code::failure);
                                }


                            }

                            sp->suback(packet_id, res);
                            return true;
                        }
                );

                ep.set_unsubscribe_handler([session](packet_id_t packet_id, std::vector<MQTT_NS::buffer> topics) {
                            auto sp = session->get_connection();
                            std::cout << "[server]unsubscribe received. packet_id: " << packet_id << ", client id: " << session->client_id << std::endl;

                            for (auto const& topic : topics) {
                                auto handle = session->remove_subscription(topic);

                                subs_map.remove(handle, session_subscription_t(session));
                            }

                            sp->unsuback(packet_id);
                            return true;
                        }
                );
            }
    );

    s.listen();

    ioc.run();
}

[kleunen]

If you really consider serializing the retained and published messages to disk, i would recommend using an embedded database such as: LevelDB orKyotoCabinet. And perform the actual writing on different thread then the network/mqtt handling.

Not sure how easy it would be to add this to the design now, that is something to consider for the future ?

[redboltz]

Thanks you! I will take a look your codes.

If you really consider serializing the retained and published messages to disk, i would recommend using an embedded database such as: LevelDB orKyotoCabinet. And perform the actual writing on different thread then the network/mqtt handling.

Yes, I wrote examples on wiki https://github.com/redboltz/mqtt_cpp/wiki/Persistence
And my proprietary broker (not mqtt_cpp one) uses redis to store message.

Not sure how easy it would be to add this to the design now, that is something to consider for the future ?

test_broker.hpp is originally designed for testing for client code.
Now, it is becoming one of reference implementation of MQTT v5 broker.
Perhaps test_broker.hpp is moved to something like utility/simple_broker.hpp .

[redboltz]

Tsumaranai monodesuga ( hope this is right thing to say :D )

It is perfect! つまらないものですが

Thank you for interested in Japanese :)

[redboltz]

mqtt_cpp/test/subscription_map.hpp

Lines 390 to 393 in a897c81

	// Insert a value at the specified subscription path
	handle insert(MQTT_NS::string_view subscription, Value const& value) {
	return insert_or_update(subscription, value);
	}

It seems that the function insert() calls insert_or_update(). Is this right?
I think that insert() should be failed if the same element exists.
The return value is similar to https://en.cppreference.com/w/cpp/container/set/insert
maybe std::pair<handle, bool>.

[redboltz]

I need to know newly inserted or updated.

If subscription option retain_handling is send_only_new_subscription, then send retained message to the subscriber only if the subscription is new one.

See

mqtt_cpp/test/test_broker.hpp

Lines 981 to 1012 in a897c81

	std::vector<bool> new_sub;
	new_sub.reserve(entries.size());
	auto add_or_overwrite_sub =
	[&](MQTT_NS::buffer topic,
	MQTT_NS::subscribe_options subopts,
	MQTT_NS::optional<std::size_t> sid = MQTT_NS::nullopt) {
	auto& idx = subs_online_.get<tag_con_topic>();
	auto r = idx.equal_range(std::make_tuple(spep, topic));
	if (r.first == r.second) {
	// add new subscription
	MQTT_LOG("mqtt_broker", trace)
	<< MQTT_ADD_VALUE(address, spep.get())
	<< "subs_online_.emplace() " << "topic:" << topic << " qos:" << subopts.get_qos();
	idx.emplace_hint(r.first, MQTT_NS::force_move(topic), spep, subopts, sid);
	new_sub.emplace_back(true);
	}
	else {
	// update subscription
	MQTT_LOG("mqtt_broker", trace)
	<< MQTT_ADD_VALUE(address, spep.get())
	<< "subs_online_.update " << "topic:" << topic << " qos:" << subopts.get_qos();
	idx.modify(
	r.first,
	[&](sub_con_online& e) {
	e.subopts = subopts;
	e.sid = sid;
	}
	);
	new_sub.emplace_back(false);
	}
	};

mqtt_cpp/test/test_broker.hpp

Lines 1073 to 1098 in a897c81

	auto new_sub_it = new_sub.begin();
	// Publish any retained messages that match the newly subscribed topic.
	for (auto const& e : entries) {
	for( auto const& retain : retains_) {
	MQTT_NS::buffer const& topic = std::get<0>(e);
	MQTT_NS::subscribe_options options = std::get<1>(e);
	if (compare_topic_filter(topic, retain.topic)) {
	auto rh = options.get_retain_handling();
	if (rh == MQTT_NS::retain_handling::send ||
	(rh == MQTT_NS::retain_handling::send_only_new_subscription && *new_sub_it)) {
	auto props = retain.props;
	if (sid) {
	props.push_back(MQTT_NS::v5::property::subscription_identifier(*sid));
	}
	ep.publish(
	as::buffer(retain.topic),
	as::buffer(retain.contents),
	std::min(retain.qos_value, options.get_qos()) | MQTT_NS::retain::yes,
	props,
	std::make_pair(retain.topic, retain.contents)
	);
	}
	}
	}
	++new_sub_it;
	}

This algorithm will be refined using subscription_map.

[redboltz]

@kleunen ,

I wrote tests code for subscription map. See #687
It is also the code of PoC how to use it in the test_broker.

So far, I don't want to mix offline/online subscription. So I create two subscription_map s.
At first, I thought that I could replace mi_sub_con_online with subscription_map. But I noticed that it is bad idea.
The biggest reason is disconnect/connect process.
I need a efficient way to traverse all topic filters corresponding to one sesson(connection).
multi_index does it well by tag_con or tag_cid.

So I decided keep mi_sub_con_online, and use subscription_map for deliver publish message.
Hence subscription_map 's Value is reference_wrapper.
There is one a little inefficient point.
During disconnect/connect process, I can get target subscriptions from multi_index efficiently. And then, I need to call subscription_map::remove() for each target subscription.
But each subscription_map::remove() call is efficient enough. So I accept it.

The ideal scenario is adapting your subscription_map structure to multi_index. I mean somehow integrate subscription_map to multi_index as a custom index. And use it tag_subscription_map. But I'm not sure it is possible and how difficult.

I've read you broker code. If you add close process, then you would experience the same issue.

Could you check #687 ?
Note: Offline part is not implemented but almost the same as online part.

[kleunen]

maybe std::pair<handle, bool>

I agree, i will make this a pair like the insert on a set. Also, the insert in the single subscription map throws an exception if a duplicate store is detected. I will also change that to returning a pair.

[kleunen]

mqtt_cpp/test/subscription_map.hpp

Lines 390 to 393 in a897c81

	// Insert a value at the specified subscription path
	handle insert(MQTT_NS::string_view subscription, Value const& value) {
	return insert_or_update(subscription, value);
	}

It seems that the function insert() calls insert_or_update(). Is this right?
I think that insert() should be failed if the same element exists.
The return value is similar to https://en.cppreference.com/w/cpp/container/set/insert
maybe std::pair<handle, bool>.

I don't think it should fail when the element already exists. The c++ container set also does not really fail, it just states whether it inserted the element or it is returning the existing element. So actually the set insert is insert_or_update. In case of a single subscription map, it should fail.

[redboltz]

Ok, I come up with a new question. What is the difference between single_subscription_map and multiple_subscription_map?
I think that it is similar to map and multi_map relationship.

I think that multiple_subscription_map::insert_or_update() is quite misleading.Because it always insert never update. Is that right?
I think that multiple_subscription_map::insert_or_update() should be removed.
How about multiple_subscription_map::update() ? I'm not sure the behavior.

single_subscription_map::insert_or_update() doesn't exist. I think that it should exist.

What do you think?

[kleunen]

I think that multiple_subscription_map::insert_or_update() is quite misleading.Because it always insert never update. Is that right?

yes, exactly, insert_or_update should be renamed to insert, same as in a set.