coffee/etl
1
0
Fork 0
mirror of https://github.com/ETLCPP/etl.git synced 2026-10-28 18:07:10 +08:00
Code Issues Packages Projects Releases Wiki Activity

Added destination_router_id parameter overrides to receive() virtuaL functions

Browse Source
This commit is contained in:
John Wellbelove 2023-01-29 20:46:41 +01:00
parent e67811065d
commit df6f6d9d96
2 changed files with 51 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
include/etl/message_broker.h
View File

@ -184,6 +184,18 @@ namespace etl
//*******************************************
virtual void receive(const etl::imessage& msg) ETL_OVERRIDE
{
receive(etl::imessage_router::ALL_MESSAGE_ROUTERS, msg);
}
virtual void receive(etl::shared_message shared_msg) ETL_OVERRIDE
{
receive(etl::imessage_router::ALL_MESSAGE_ROUTERS, shared_msg);
}
//*******************************************
virtual void receive(etl::message_router_id_t destination_router_id,
const etl::imessage& msg) ETL_OVERRIDE
{
const etl::message_id_t id = msg.get_message_id();
@ -200,7 +212,13 @@ namespace etl
if (itr != message_ids.end())
{
sub->get_router()->receive(msg);
etl::imessage_router* router = sub->get_router();
if (destination_router_id == etl::imessage_router::ALL_MESSAGE_ROUTERS ||
destination_router_id == router->get_message_router_id())
{
sub->get_router()->receive(msg);
}
}
sub = sub->next_subscription();
@ -212,12 +230,13 @@ namespace etl
{
etl::imessage_router& successor = get_successor();
successor.receive(msg);
successor.receive(destination_router_id, msg);
}
}
//*******************************************
virtual void receive(etl::shared_message shared_msg) ETL_OVERRIDE
virtual void receive(etl::message_router_id_t destination_router_id,
etl::shared_message shared_msg) ETL_OVERRIDE
{
const etl::message_id_t id = shared_msg.get_message().get_message_id();
@ -234,7 +253,13 @@ namespace etl
if (itr != message_ids.end())
{
sub->get_router()->receive(shared_msg);
etl::imessage_router* router = sub->get_router();
if (destination_router_id == etl::imessage_router::ALL_MESSAGE_ROUTERS ||
destination_router_id == router->get_message_router_id())
{
sub->get_router()->receive(shared_msg);
}
}
sub = sub->next_subscription();
@ -244,7 +269,7 @@ namespace etl
// Always pass the message on to a successor.
if (has_successor())
{
get_successor().receive(shared_msg);
get_successor().receive(destination_router_id, shared_msg);
}
}

21
test/test_message_broker.cpp
View File

@ -438,6 +438,27 @@ namespace
CHECK_EQUAL(1, router3.message_unknown_count);
}
//*************************************************************************
TEST(message_broker_send_messages_to_specific_subscribers)
{
Broker broker;
Router router1(1);
Router router2(2);
Router router3(3);
Subscription subscription1{ router1, { Message1::ID, Message2::ID, Message3::ID, Message4::ID } };
Subscription subscription2{ router2, { Message1::ID, Message2::ID } };
Subscription subscription3{ router2, { Message1::ID, Message3::ID } };
broker.subscribe(subscription1);
broker.subscribe(subscription2);
broker.set_successor(router3);
broker.receive(Message1());
broker.receive(1, Message1());
broker.receive(2, Message2());
}
//*************************************************************************
TEST(message_broker_send_messages_to_subscribers_after_unsubscribe)
{