1 // Copyright 2024 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_channel/stream_channel.h"
16
17 #include <algorithm>
18 #include <array>
19
20 #include "pw_async2/pend_func_task.h"
21 #include "pw_bytes/suffix.h"
22 #include "pw_multibuf/simple_allocator_for_test.h"
23 #include "pw_status/status.h"
24 #include "pw_stream/mpsc_stream.h"
25 #include "pw_thread/test_thread_context.h"
26 #include "pw_toolchain/globals.h"
27 #include "pw_toolchain/no_destructor.h"
28 #include "pw_unit_test/framework.h"
29
30 namespace {
31
32 using ::pw::async2::Context;
33 using ::pw::async2::PendFuncTask;
34 using ::pw::async2::Pending;
35 using ::pw::async2::Poll;
36 using ::pw::async2::Ready;
37 using ::pw::multibuf::MultiBuf;
38 using ::pw::multibuf::test::SimpleAllocatorForTest;
39 using ::pw::operator""_b;
40
41 template <typename ActualIterable, typename ExpectedIterable>
ExpectElementsEqual(const ActualIterable & actual,const ExpectedIterable & expected)42 void ExpectElementsEqual(const ActualIterable& actual,
43 const ExpectedIterable& expected) {
44 auto actual_iter = actual.begin();
45 auto expected_iter = expected.begin();
46 for (; expected_iter != expected.end(); ++actual_iter, ++expected_iter) {
47 ASSERT_NE(actual_iter, actual.end());
48 EXPECT_EQ(*actual_iter, *expected_iter);
49 }
50 }
51
52 template <typename ActualIterable, typename T>
ExpectElementsEqual(const ActualIterable & actual,std::initializer_list<T> expected)53 void ExpectElementsEqual(const ActualIterable& actual,
54 std::initializer_list<T> expected) {
55 ExpectElementsEqual<ActualIterable, std::initializer_list<T>>(actual,
56 expected);
57 }
58
59 struct LiveForeverTestData {
LiveForeverTestData__anonb11e6e200111::LiveForeverTestData60 LiveForeverTestData() {
61 pw::stream::CreateMpscStream(channel_input_reader, channel_input_writer);
62 pw::stream::CreateMpscStream(channel_output_reader, channel_output_writer);
63 }
64 pw::stream::BufferedMpscReader<512> channel_input_reader;
65 pw::stream::MpscWriter channel_input_writer;
66 pw::stream::BufferedMpscReader<512> channel_output_reader;
67 pw::stream::MpscWriter channel_output_writer;
68 SimpleAllocatorForTest<> allocator;
69 pw::thread::test::TestThreadContext read_thread_cx;
70 pw::thread::test::TestThreadContext write_thread_cx;
71 };
72
TEST(StreamChannel,ReadsAndWritesData)73 TEST(StreamChannel, ReadsAndWritesData) {
74 static pw::NoDestructor<LiveForeverTestData> test_data;
75 static pw::RuntimeInitGlobal<pw::channel::StreamChannel> stream_channel(
76 test_data->allocator,
77 test_data->channel_input_reader,
78 test_data->read_thread_cx.options(),
79 test_data->channel_output_writer,
80 test_data->write_thread_cx.options());
81
82 PendFuncTask read_task([&](Context& cx) -> Poll<> {
83 auto read = stream_channel->PendRead(cx);
84 if (read.IsPending()) {
85 return Pending();
86 }
87 EXPECT_EQ(read->status(), pw::OkStatus());
88 if (read->ok()) {
89 ExpectElementsEqual(**read, {1_b, 2_b, 3_b});
90 }
91 return Ready();
92 });
93
94 MultiBuf to_send = test_data->allocator.BufWith({4_b, 5_b, 6_b});
95 PendFuncTask write_task([&](Context& cx) -> Poll<> {
96 if (stream_channel->PendReadyToWrite(cx).IsPending()) {
97 return Pending();
98 }
99 PW_TEST_EXPECT_OK(stream_channel->StageWrite(std::move(to_send)));
100 return Ready();
101 });
102
103 pw::async2::Dispatcher dispatcher;
104 dispatcher.Post(write_task);
105 dispatcher.Post(read_task);
106
107 EXPECT_EQ(Pending(), dispatcher.RunUntilStalled());
108 std::array<const std::byte, 3> data_to_send({1_b, 2_b, 3_b});
109 ASSERT_EQ(pw::OkStatus(),
110 test_data->channel_input_writer.Write(data_to_send));
111 dispatcher.RunToCompletion();
112 }
113
114 } // namespace
115