upcastable_message_demuxer (2024.07.24)

[eao197]

Another iteration around issue #25.

A sketch

It may looks like:

#include <so_5_extra/upcastable_msg_demuxer.hpp>
...
namespace demuxer = so_5::extra::uncastable_msg_demuxer;

//
// Definitions of messages to be used.
//
// NOTE: the necessity of using root_t<T> and node_t<D, B> is
// because C++ (17, 20, 23) lacks reflection and there is no
// official, legal and cross platform way to enumerate base
// classes of a type. 

// Base type for messages.
struct image_base : public demuxer::root_t<image_base>
{
  ...
  image_base() = default;
};

// Derived messages.
struct camera_vendor_1
  : public image_base
  , public demuxer::node_t<camera_vendor_1, image_base>
{
  ...
  // A constructor has to be provided because the constructor
  // of node_t<D, B> requires a reference to an instance of type D.
  camera_vendor_1()
    : image_base{}
    , demuxer::node_t<camera_vendor_1, image_base>{ *this }
  {}
};

struct camera_vendor_2
  : public image_base
  , public demuxer::node_t<camera_vendor_2, image_base>
{
  ...
  // A constructor has to be provided.
  camera_vendor_2()
    : image_base{}
    , demuxer::node_t<camera_vendor_2, image_base>{ *this }
  {}
};

struct special_image
  : public camera_vendor_2
  , public demuxer::node_t<special_image, camera_vendor_2>
{
  ...
  // A constructor has to be provided.
  special_image()
    : image_base{}
    , demuxer::node_t<special_image, camera_vendor_2>{ *this }
  {}
};

//
// Type of the demuxer to be used.
//
// The root of the type hierarchy is specified as template parameter.
using image_demuxer = demuxer::demuxer_t<image_base>;

//
// Type of agent that wants to receive messages derived from image_base
//
class image_receiver : public so_5::agent_t
{
  // Demuxer for receiving messages.
  std::shared_ptr<image_demuxer> m_demuxer;

  // ID of demuxed messages consumer.
  demuxer::consumer_id_t m_consumer_id;

public:
  image_receiver(context_t ctx, std::shared_ptr<image_demuxer> demuxer)
    : so_5::agent_t{std::move(ctx)}
    , m_demuxer{std::move(demuxer)}
    , m_consumer_id{m_demuxer->allocate_consumer_id()}
  {}

  void so_define_agent() override
  {
    // A separate mbox will be used for every type of incoming messages.
    const auto special_image_mbox = m_demuxer->receiveing_mbox_for<special_image>(
        // Have to specify consumer_id to say demuxer that the same
        // receiver is behind every separate mbox.
        m_consumer_id);
    const auto vendor_2_mbox = m_demuxer->receiving_mbox_for<camera_vendor_2>(
        // The same receiver.
        m_consumer_id);
    const auto base_mbox = m_demuxer->receiving_mbox_for<image_base>(
        // The same receiver.
        m_consumer_id); 

    // Now we can define delivery_filters and subscriptions for messages.
    so_set_delivery_filter(
        special_image_mbox,
        [](const special_image & msg) { ... });
    so_set_delivery_filter(
        base_mbox,
        [](const image_base & msg) { ... });

    so_subscribe(special_image_mbox)
        .event([this](mhood_t<special_image> cmd) {...});
    so_subscribe(vendor_2_mbox)
        .event([this](mhood_t<camera_vendor_2> cmd) {...});
    so_subscribe(base_mbox)
        .event([this](mhood_t<image_base> cmd) {...});
  }
  ...
};

//
// Type of agent that collects images from a camera (manufactured by vendor-1)
// and publishes them to a demuxer.
//
class vendor_1_cam_image_acquier : public so_5::agent_t
{
  // Mbox for messages to be published.
  const so_5::mbox_t m_dest;

public:
  vendor_1_cam_image_acquier(context_t ctx, const image_demuxer & demuxer)
    : so_5::agent_t{std::move(ctx)}
    // Get a mbox for sending outgoing messages.
    , m_dest{demuxer.sending_mbox()}
  {}
  ...
  void evt_time_to_acquire(mhood_t<msg_timer_tick>)
  {
    so_5::message_holder_t<camera_vendor_1> msg = acquire_frame();
    // Sending the acquired image to the receivers.
    so_5::send(m_dest, msg);
  }
};

//
// Creation of agents with message demuxer.
//
auto demuxer_instance = std::make_shared<image_demuxer>();
so_5::environment_t & env = ...;
env.introduce_coop([&](so_5::coop_t & coop) {
    coop.make_agent<image_receiver>(demuxer_instance);
    coop.make_agent<vendor_1_cam_image_acquier>(*demuxer_instance);
    ...
  });

General Description

So the main idea is to have a special object, demuxer, instead of just one mbox.

This demuxer object owns a bunch of separate mboxes:

• one of them is a "sending mbox" (or input for the demuxer). All outgoing messages have to be sent to this mbox;
• several "receiving mboxes" (or output for the demuxer). They can be used for subscription only. There is no way to send a message to such a mbox.

A "receiving mbox" is created as a unique instance for two keys:

• type of messages to be received from it. As shown above it can be a specific message type without any derived classes (like special_image), or it can be a root of type hierarchy (like image_base or camera_vendor_2);
• ID of a consumer of messages.

The ID of a consumer seems to be mandatory because a demuxer should understand that a message is already sent to a receiver. For example, let's assume that we have receiver-1 and receiver-2 agents. Both of them make subscriptions for image_base, but receiver-1 also has a subscription for special_image. It means that receiver-1 has subscriptions from two separate mboxes -- one for image_base, another for special_image.

When an instance of special_image is being sent the demuxer should push it to the mbox for special_image for receiver-1 and to the mbox for image_base for receiver-2. But if there is a common mbox for image_base for receiver-1 and receiver-2, then receiver-1 will receive the message twice: first time from a mbox for special_image, another time from a mbox for image_base.

The presence of a unique consumer ID resolves this problem.

[eao197]

This is just a dirty sketch of how root_t<T> and node_t<D, B> can be used for type hierarchy traversal down up. Please note that this code is not production ready and has obvious issues (like missing of copy/move constructors/operators for root_base_t class).

[eao197]

And this is another dirty sketch that shows type hierarchy traversal with respect to mutability of messages.

[eao197]

The first working sketch: https://github.com/Stiffstream/so5extra/blob/d3e1d71c8a17427a4daf93d57d8d118ec4492ab7/dev/test/so_5_extra/msg_hierarchy/simple/main.cpp#L17-L84