diff --git a/include/etl/message_broker.h b/include/etl/message_broker.h index abedc8e1..08a0d0b6 100644 --- a/include/etl/message_broker.h +++ b/include/etl/message_broker.h @@ -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); } } diff --git a/test/test_message_broker.cpp b/test/test_message_broker.cpp index 8dbdf05e..3a414d8e 100644 --- a/test/test_message_broker.cpp +++ b/test/test_message_broker.cpp @@ -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) {