corrected send flow for client/server routers

Author: serges147

src/common/ipc/client_router.cpp

@@ -262,7 +262,7 @@ class ClientRouterImpl final : public ClientRouter
     ///
     /// Called on the gateway disposal (correspondingly on its channel destruction).
     /// The "dying" gateway might wish to notify the remote router about its disposal.
-    /// The router fulfills the wish if the gateway was registered and the router is connected.
+    /// This local router fulfills the wish if the gateway was registered and the router is connected.
     ///
     void onGatewayDisposal(const Endpoint& endpoint, const bool send_ch_end)
     {

src/common/ipc/server_router.cpp

@@ -147,8 +147,9 @@ class ServerRouterImpl final : public ServerRouter
 
         ~GatewayImpl()
         {
-            router_.unregisterGateway(endpoint_, true);
             ::syslog(LOG_DEBUG, "~Gateway(cl=%zu, tag=%zu).", endpoint_.getClientId(), endpoint_.getTag());
+
+            router_.onGatewayDisposal(endpoint_);
         }
 
         // detail::Gateway
@@ -159,6 +160,10 @@ class ServerRouterImpl final : public ServerRouter
             {
                 return static_cast<int>(ErrorCode::NotConnected);
             }
+            if (!router_.isRegisteredGateway(endpoint_))
+            {
+                return static_cast<int>(ErrorCode::Disconnected);
+            }
 
             Route_1_0 route{&router_.memory_};
 
@@ -183,16 +188,8 @@ class ServerRouterImpl final : public ServerRouter
 
         void subscribe(EventHandler event_handler) override
         {
-            if (event_handler)
-            {
-                event_handler_ = std::move(event_handler);
-                router_.registerGateway(endpoint_, *this);
-            }
-            else
-            {
-                event_handler_ = nullptr;
-                router_.unregisterGateway(endpoint_, false);
-            }
+            event_handler_ = std::move(event_handler);
+            router_.onGatewaySubscription(endpoint_);
         }
 
     private:
@@ -211,24 +208,62 @@ class ServerRouterImpl final : public ServerRouter
         return connected_client_ids_.find(endpoint.getClientId()) != connected_client_ids_.end();
     }
 
-    void registerGateway(const Endpoint& endpoint, GatewayImpl& gateway)
+    CETL_NODISCARD bool isRegisteredGateway(const Endpoint& endpoint) const noexcept
+    {
+        return endpoint_to_gateway_.find(endpoint) != endpoint_to_gateway_.end();
+    }
+
+    template <typename Action>
+    void findAndActOnRegisteredGateway(const Endpoint endpoint, Action&& action)
+    {
+        const auto ep_to_gw = endpoint_to_gateway_.find(endpoint);
+        if (ep_to_gw != endpoint_to_gateway_.end())
+        {
+            const auto gateway = ep_to_gw->second.lock();
+            if (gateway)
+            {
+                std::forward<Action>(action)(*gateway, ep_to_gw);
+            }
+        }
+    }
+
+    void onGatewaySubscription(const Endpoint endpoint)
     {
-        endpoint_to_gateway_[endpoint] = gateway.shared_from_this();
         if (isConnected(endpoint))
         {
-            gateway.event(detail::Gateway::Event::Connected{});
+            findAndActOnRegisteredGateway(endpoint, [](auto& gateway, auto) {
+                //
+                gateway.event(detail::Gateway::Event::Connected{});
+            });
         }
     }
 
-    void unregisterGateway(const Endpoint& endpoint, const bool is_disposed = false)
+    /// Unregisters the gateway associated with the given endpoint.
+    ///
+    /// Called on the gateway disposal (correspondingly on its channel destruction).
+    /// The "dying" gateway wishes to notify the remote client router about its disposal.
+    /// This local router fulfills the wish if the gateway was registered and the client router is connected.
+    ///
+    void onGatewayDisposal(const Endpoint& endpoint)
     {
-        endpoint_to_gateway_.erase(endpoint);
+        const bool was_registered = (endpoint_to_gateway_.erase(endpoint) > 0);
 
-        // Notify "remote" router about the gateway disposal.
-        // The router will deliver "disconnected" event to the counterpart gateway (if it exists).
+        // Notify remote client router about the gateway disposal (aka channel completion).
+        // The router will propagate "ChEnd" event to the counterpart gateway (if it's registered).
         //
-        if (is_disposed && isConnected(endpoint))
+        if (was_registered && isConnected(endpoint))
         {
+            Route_1_0 route{&memory_};
+            auto&     channel_end  = route.set_channel_end();
+            channel_end.tag        = endpoint.getTag();
+            channel_end.error_code = 0;  // No error b/c it's a normal channel completion.
+
+            const int result = tryPerformOnSerialized(route, [this, &endpoint](const auto payload) {
+                //
+                return server_pipe_->send(endpoint.getClientId(), {{payload}});
+            });
+            // Best efforts strategy - gateway anyway is gone, so nowhere to report.
+            (void) result;
         }
     }
 
@@ -318,7 +353,8 @@ class ServerRouterImpl final : public ServerRouter
             const auto si_to_cf = service_id_to_channel_factory_.find(route_ch_msg.service_id);
             if (si_to_cf != service_id_to_channel_factory_.end())
             {
-                auto gateway = GatewayImpl::create(*this, endpoint);
+                auto gateway                   = GatewayImpl::create(*this, endpoint);
+                endpoint_to_gateway_[endpoint] = gateway;
                 si_to_cf->second(gateway, msg_payload);
             }
         }

test/common/ipc/ipc_gtest_helpers.hpp

@@ -164,7 +164,7 @@ testing::PolymorphicMatcher<PayloadMatcher<T>> PayloadWith(
     return testing::MakePolymorphicMatcher(PayloadMatcher<T>(matcher, memory));
 }
 
-inline auto PayloadRouteConnectEq(cetl::pmr::memory_resource& mr,
+inline auto PayloadOfRouteConnect(cetl::pmr::memory_resource& mr,
                                   const std::uint8_t          ver_major = VERSION_MAJOR,
                                   const std::uint8_t          ver_minor = VERSION_MINOR)
 {
@@ -182,6 +182,14 @@ auto PayloadOfRouteChannel(cetl::pmr::memory_resource& mr,
     return PayloadWith<Route_1_0>(testing::VariantWith<RouteChannelMsg_1_0>(msg), mr);
 }
 
+inline auto PayloadOfRouteChannelEnd(cetl::pmr::memory_resource& mr,  //
+                                     const std::uint64_t         tag,
+                                     const ErrorCode             error_code)
+{
+    const RouteChannelEnd_1_0 ch_end{{tag, static_cast<std::int32_t>(error_code), &mr}, &mr};
+    return PayloadWith<Route_1_0>(testing::VariantWith<RouteChannelEnd_1_0>(ch_end), mr);
+}
+
 }  // namespace ipc
 }  // namespace common
 }  // namespace ocvsmd

test/common/ipc/test_client_router.cpp

@@ -13,7 +13,6 @@
 #include "pipe/client_pipe_mock.hpp"
 #include "tracking_memory_resource.hpp"
 
-#include "ocvsmd/common/ipc/RouteChannelEnd_1_0.hpp"
 #include "ocvsmd/common/ipc/RouteChannelMsg_1_0.hpp"
 #include "ocvsmd/common/ipc/RouteConnect_1_0.hpp"
 #include "ocvsmd/common/ipc/Route_1_0.hpp"
@@ -82,7 +81,8 @@ class TestClientRouter : public testing::Test
     void emulateRouteConnect(pipe::ClientPipeMock& client_pipe_mock)
     {
         // client RouteConnect -> server
-        EXPECT_CALL(client_pipe_mock, send(PayloadRouteConnectEq(mr_))).WillOnce(Return(0));
+        EXPECT_CALL(client_pipe_mock, send(PayloadOfRouteConnect(mr_)))  //
+            .WillOnce(Return(0));
         client_pipe_mock.event_handler_(pipe::ClientPipe::Event::Connected{});
 
         // Server -> client RouteConnect
@@ -205,13 +205,15 @@ TEST_F(TestClientRouter, makeChannel_send)
 
     const std::uint64_t tag = 0;
     std::uint64_t       seq = 0;
-    EXPECT_CALL(client_pipe_mock, send(PayloadOfRouteChannel<Msg>(mr_, tag, seq++))).WillOnce(Return(0));
+    EXPECT_CALL(client_pipe_mock, send(PayloadOfRouteChannel<Msg>(mr_, tag, seq++)))  //
+        .WillOnce(Return(0));
     EXPECT_THAT(channel.send(msg), 0);
 
-    EXPECT_CALL(client_pipe_mock, send(PayloadOfRouteChannel<Msg>(mr_, tag, seq++))).WillOnce(Return(0));
+    EXPECT_CALL(client_pipe_mock, send(PayloadOfRouteChannel<Msg>(mr_, tag, seq++)))  //
+        .WillOnce(Return(0));
     EXPECT_THAT(channel.send(msg), 0);
 
-    EXPECT_CALL(client_pipe_mock, send(PayloadWith<Route_1_0>(VariantWith<RouteChannelEnd_1_0>(_), mr_)))
+    EXPECT_CALL(client_pipe_mock, send(PayloadOfRouteChannelEnd(mr_, tag, ErrorCode::Success)))  //
         .WillOnce(Return(0));
 }
 

test/common/ipc/test_server_router.cpp

@@ -12,7 +12,6 @@
 #include "pipe/server_pipe_mock.hpp"
 #include "tracking_memory_resource.hpp"
 
-#include "ocvsmd/common/ipc/RouteConnect_1_0.hpp"
 #include "ocvsmd/common/ipc/Route_1_0.hpp"
 #include "ocvsmd/common/node_command/ExecCmd_1_0.hpp"
 
@@ -169,40 +168,45 @@ TEST_F(TestServerRouter, channel_send)
     EXPECT_THAT(server_router->start(), 0);
     EXPECT_THAT(server_pipe_mock.event_handler_, IsTrue());
 
-    StrictMock<MockFunction<void(const Channel::EventVar&)>> ch1_event_mock;
+    StrictMock<MockFunction<void(const Channel::EventVar&)>> ch_event_mock;
 
     cetl::optional<Channel> maybe_channel;
     server_router->registerChannel<Channel>("", [&](Channel&& ch, const auto& input) {
         //
-        ch.subscribe(ch1_event_mock.AsStdFunction());
+        ch.subscribe(ch_event_mock.AsStdFunction());
         maybe_channel = std::move(ch);
-        ch1_event_mock.Call(input);
+        ch_event_mock.Call(input);
     });
     EXPECT_THAT(maybe_channel.has_value(), IsFalse());
 
     // Emulate that client #42 is connected.
     //
-    EXPECT_CALL(ch1_event_mock, Call(VariantWith<Channel::Connected>(_))).Times(1);
-    emulatePipeConnect(42, server_pipe_mock);
+    constexpr std::uint64_t cl_id = 42;
+    EXPECT_CALL(ch_event_mock, Call(VariantWith<Channel::Connected>(_))).Times(1);
+    emulatePipeConnect(cl_id, server_pipe_mock);
 
-    // Emulate that client posted initial `RouteChannelMsg` on 42/1 client/tag pair.
+    // Emulate that client posted initial `RouteChannelMsg` on 42/7 client/tag pair.
     //
-    const std::uint64_t tag = 1;
+    const std::uint64_t tag = 7;
     std::uint64_t       seq = 0;
-    EXPECT_CALL(ch1_event_mock, Call(VariantWith<Channel::Input>(_))).Times(1);
-    emulateRouteChannelMsg(42, server_pipe_mock, tag, Channel::Input{&mr_}, seq);
+    EXPECT_CALL(ch_event_mock, Call(VariantWith<Channel::Input>(_))).Times(1);
+    emulateRouteChannelMsg(cl_id, server_pipe_mock, tag, Channel::Input{&mr_}, seq);
     ASSERT_THAT(maybe_channel.has_value(), IsTrue());
+    EXPECT_CALL(server_pipe_mock, send(cl_id, PayloadOfRouteChannelEnd(mr_, tag, ErrorCode::Success)))  //
+        .WillOnce(Return(0));
 
-    // Emulate that client posted one more `RouteChannelMsg` on the same 42/1 client/tag pair.
+    // Emulate that client posted one more `RouteChannelMsg` on the same 42/7 client/tag pair.
     //
-    EXPECT_CALL(ch1_event_mock, Call(VariantWith<Channel::Input>(_))).Times(1);
-    emulateRouteChannelMsg(42, server_pipe_mock, tag, Channel::Input{&mr_}, seq);
+    EXPECT_CALL(ch_event_mock, Call(VariantWith<Channel::Input>(_))).Times(1);
+    emulateRouteChannelMsg(cl_id, server_pipe_mock, tag, Channel::Input{&mr_}, seq);
 
     seq = 0;
-    EXPECT_CALL(server_pipe_mock, send(42, PayloadOfRouteChannel<Msg>(mr_, tag, seq++))).WillOnce(Return(0));
+    EXPECT_CALL(server_pipe_mock, send(cl_id, PayloadOfRouteChannel<Msg>(mr_, tag, seq++)))  //
+        .WillOnce(Return(0));
     EXPECT_THAT(maybe_channel->send(Channel::Output{&mr_}), 0);
 
-    EXPECT_CALL(server_pipe_mock, send(42, PayloadOfRouteChannel<Msg>(mr_, tag, seq++))).WillOnce(Return(0));
+    EXPECT_CALL(server_pipe_mock, send(cl_id, PayloadOfRouteChannel<Msg>(mr_, tag, seq++)))  //
+        .WillOnce(Return(0));
     EXPECT_THAT(maybe_channel->send(Channel::Output{&mr_}), 0);
 }