1 // Copyright 2023 The Pigweed Authors
2 //
3 // Licensed under the Apache License, Version 2.0 (the "License"); you may not
4 // use this file except in compliance with the License. You may obtain a copy of
5 // the License at
6 //
7 // https://www.apache.org/licenses/LICENSE-2.0
8 //
9 // Unless required by applicable law or agreed to in writing, software
10 // distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11 // WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12 // License for the specific language governing permissions and limitations under
13 // the License.
14
15 #include "pw_chrono/system_clock.h"
16 #include "pw_rpc/client_server.h"
17 #include "pw_rpc/synchronous_call.h"
18 #include "pw_rpc_transport/egress_ingress.h"
19 #include "pw_rpc_transport/internal/test.rpc.pwpb.h"
20 #include "pw_rpc_transport/local_rpc_egress.h"
21 #include "pw_rpc_transport/service_registry.h"
22 #include "pw_rpc_transport/socket_rpc_transport.h"
23 #include "pw_status/status.h"
24 #include "pw_string/string.h"
25 #include "pw_thread/thread.h"
26 #include "pw_thread/thread_core.h"
27 #include "pw_thread_stl/options.h"
28 #include "pw_unit_test/framework.h"
29
30 namespace pw::rpc {
31 namespace {
32
33 using namespace std::chrono_literals;
34
35 constexpr size_t kMaxTestMessageSize = 1024;
36 constexpr uint32_t kTestChannelId = 1;
37
38 class TestService final
39 : public pw_rpc_transport::testing::pw_rpc::pwpb::TestService::Service<
40 TestService> {
41 public:
Echo(const pw_rpc_transport::testing::pwpb::EchoMessage::Message & request,pw_rpc_transport::testing::pwpb::EchoMessage::Message & response)42 Status Echo(
43 const pw_rpc_transport::testing::pwpb::EchoMessage::Message& request,
44 pw_rpc_transport::testing::pwpb::EchoMessage::Message& response) {
45 response.msg = request.msg;
46 return OkStatus();
47 }
48 };
49
50 template <size_t kMaxPacketSize, size_t kLocalEgressQueueSize>
51 struct SocketRpcEndpoint {
SocketRpcEndpointpw::rpc::__anon921fc77a0111::SocketRpcEndpoint52 explicit SocketRpcEndpoint(SocketRpcTransport<kMaxPacketSize>& transport)
53 : transport(transport),
54 rpc_egress("tx", transport),
55 tx_channels({rpc::Channel::Create<kTestChannelId>(&rpc_egress)}),
56 rx_channels({ChannelEgress{kTestChannelId, local_egress}}),
57 rpc_ingress(rx_channels),
58 service_registry(tx_channels) {
59 local_egress.set_packet_processor(service_registry);
60 transport.set_ingress(rpc_ingress);
61 }
62
63 LocalRpcEgress<kLocalEgressQueueSize, kMaxPacketSize> local_egress;
64 SocketRpcTransport<kMaxPacketSize>& transport;
65 SimpleRpcEgress<kMaxPacketSize> rpc_egress;
66 std::array<rpc::Channel, 1> tx_channels;
67 std::array<ChannelEgress, 1> rx_channels;
68 SimpleRpcIngress<kMaxPacketSize> rpc_ingress;
69 ServiceRegistry service_registry;
70 };
71
TEST(RpcIntegrationTest,SocketTransport)72 TEST(RpcIntegrationTest, SocketTransport) {
73 constexpr size_t kMaxPacketSize = 512;
74 constexpr size_t kLocalEgressQueueSize = 20;
75 constexpr size_t kMessageSize = 50;
76
77 SocketRpcTransport<kMaxPacketSize> a_to_b_transport(
78 SocketRpcTransport<kMaxPacketSize>::kAsServer, /*port=*/0);
79 auto a = SocketRpcEndpoint<kMaxPacketSize, kLocalEgressQueueSize>(
80 a_to_b_transport);
81 auto a_local_egress_thread =
82 thread::Thread(thread::stl::Options(), a.local_egress);
83 auto a_transport_thread = thread::Thread(thread::stl::Options(), a.transport);
84
85 a_to_b_transport.WaitUntilReady();
86
87 SocketRpcTransport<kMaxPacketSize> b_to_a_transport(
88 SocketRpcTransport<kMaxPacketSize>::kAsClient,
89 "localhost",
90 a_to_b_transport.port());
91
92 auto b = SocketRpcEndpoint<kMaxPacketSize, kLocalEgressQueueSize>(
93 b_to_a_transport);
94 auto b_local_egress_thread =
95 thread::Thread(thread::stl::Options(), b.local_egress);
96 auto b_transport_thread = thread::Thread(thread::stl::Options(), b.transport);
97
98 TestService b_test_service;
99 b.service_registry.RegisterService(b_test_service);
100 a_to_b_transport.WaitUntilConnected();
101 b_to_a_transport.WaitUntilConnected();
102
103 for (int i = 0; i < 10; ++i) {
104 InlineString<kMaxTestMessageSize> test_message;
105 test_message.append(kMessageSize, '*');
106 auto echo_request = pw_rpc_transport::testing::pwpb::EchoMessage::Message{
107 .msg = test_message};
108 const auto echo_response = rpc::SynchronousCall<
109 pw_rpc_transport::testing::pw_rpc::pwpb::TestService::Echo>(
110 a.service_registry.client_server().client(),
111 kTestChannelId,
112 echo_request);
113 EXPECT_EQ(echo_response.status(), OkStatus());
114 EXPECT_EQ(echo_response.response().msg, test_message);
115 }
116
117 // Shut everything down.
118 a.local_egress.Stop();
119 b.local_egress.Stop();
120 a.transport.Stop();
121 b.transport.Stop();
122
123 a_local_egress_thread.join();
124 b_local_egress_thread.join();
125 a_transport_thread.join();
126 b_transport_thread.join();
127 }
128
129 } // namespace
130 } // namespace pw::rpc
131