diff --git a/connections/c/payload_w.cc b/connections/c/payload_w.cc index 6205ffda..5faf69c0 100644 --- a/connections/c/payload_w.cc +++ b/connections/c/payload_w.cc @@ -13,9 +13,16 @@ // limitations under the License. #include "connections/c/payload_w.h" +#include +#include +#include +#include + #include "connections/c/file_w.h" #include "connections/payload.h" +#include "connections/payload_type.h" #include "internal/platform/byte_array.h" +#include "internal/platform/input_stream.h" #include "internal/platform/payload_id.h" namespace nearby { @@ -32,7 +39,7 @@ PayloadW::PayloadW() : impl_(std::unique_ptr( new connections::Payload())) {} -PayloadW::~PayloadW() {} +PayloadW::~PayloadW() = default; PayloadW::PayloadW(PayloadW &&other) noexcept : impl_(std::move(other.impl_)) {} PayloadW &PayloadW::operator=(PayloadW &&other) noexcept { @@ -49,10 +56,9 @@ PayloadW::PayloadW(InputFileW &file) : impl_(std::unique_ptr( new connections::Payload(InputFile(std::move(*file.GetImpl()))))) {} -// TODO(jfcarroll): Convert std::function to function pointer -PayloadW::PayloadW(std::function stream) +PayloadW::PayloadW(std::unique_ptr stream) : impl_(std::unique_ptr( - new connections::Payload(stream))) {} + new connections::Payload(std::move(stream)))) {} // Constructors for incoming payloads. PayloadW::PayloadW(PayloadId id, const char *bytes, const size_t bytes_size) @@ -69,9 +75,9 @@ PayloadW::PayloadW(const char *parent_folder, const char *file_name, new connections::Payload(parent_folder, file_name, std::move(*file.GetImpl())))) {} -PayloadW::PayloadW(PayloadId id, std::function stream) +PayloadW::PayloadW(PayloadId id, std::unique_ptr stream) : impl_(std::unique_ptr( - new connections::Payload(id, stream))) {} + new connections::Payload(id, std::move(stream)))) {} // Returns ByteArray payload, if it has been defined, or empty ByteArray. bool PayloadW::AsBytes(const char *&bytes, size_t &bytes_size) const & { diff --git a/connections/c/payload_w.h b/connections/c/payload_w.h index d4baeaf1..17972e6e 100644 --- a/connections/c/payload_w.h +++ b/connections/c/payload_w.h @@ -60,19 +60,18 @@ class DLL_API PayloadW { ~PayloadW(); // Constructors for outgoing payloads. - explicit PayloadW(const char* bytes, const size_t size); + explicit PayloadW(const char* bytes, size_t size); explicit PayloadW(InputFileW& file); - explicit PayloadW(std::function stream); + explicit PayloadW(std::unique_ptr stream); // Constructors for incoming payloads. - PayloadW(PayloadId id, const char* bytes, const size_t size); + PayloadW(PayloadId id, const char* bytes, size_t size); PayloadW(PayloadId id, InputFileW file); explicit PayloadW(const char* parent_folder, const char* file_name, InputFileW file); - // TODO(jfcarroll): Convert std::function to function pointer - PayloadW(PayloadId id, std::function stream); + PayloadW(PayloadId id, std::unique_ptr stream); // Returns ByteArray payload, if it has // been defined, or empty ByteArray. bool AsBytes(const char*& bytes, size_t& bytes_size) const&; diff --git a/connections/implementation/BUILD b/connections/implementation/BUILD index 2179906b..fe357f98 100644 --- a/connections/implementation/BUILD +++ b/connections/implementation/BUILD @@ -226,6 +226,7 @@ cc_test( ":internal", ":internal_test", "//connections:core_types", + "//connections/implementation/analytics", "//connections/implementation/flags:connections_flags", "//connections/implementation/mediums", "//connections/implementation/proto:offline_wire_formats_cc_proto", @@ -241,6 +242,7 @@ cc_test( "//internal/test", "//proto:connections_enums_cc_proto", "@com_github_protobuf_matchers//protobuf-matchers", + "@com_google_absl//absl/base:core_headers", "@com_google_absl//absl/container:flat_hash_set", "@com_google_absl//absl/strings", "@com_google_absl//absl/strings:str_format", diff --git a/connections/implementation/base_endpoint_channel_test.cc b/connections/implementation/base_endpoint_channel_test.cc index 03ca6898..00646455 100644 --- a/connections/implementation/base_endpoint_channel_test.cc +++ b/connections/implementation/base_endpoint_channel_test.cc @@ -14,7 +14,9 @@ #include "connections/implementation/base_endpoint_channel.h" +#include #include +#include #include #include @@ -22,9 +24,13 @@ #include "gmock/gmock.h" #include "protobuf-matchers/protocol-buffer-matchers.h" #include "gtest/gtest.h" +#include "absl/strings/string_view.h" #include "absl/synchronization/mutex.h" +#include "absl/time/clock.h" #include "absl/time/time.h" +#include "connections/implementation/client_proxy.h" #include "connections/implementation/encryption_runner.h" +#include "connections/implementation/endpoint_channel.h" #include "connections/implementation/offline_frames.h" #include "internal/platform/byte_array.h" #include "internal/platform/count_down_latch.h" @@ -34,7 +40,6 @@ #include "internal/platform/multi_thread_executor.h" #include "internal/platform/output_stream.h" #include "internal/platform/pipe.h" -#include "internal/platform/single_thread_executor.h" #include "proto/connections_enums.pb.h" namespace nearby { @@ -44,6 +49,7 @@ namespace { using ::location::nearby::proto::connections::DisconnectionReason; using ::location::nearby::proto::connections::Medium; using EncryptionContext = BaseEndpointChannel::EncryptionContext; +constexpr size_t kChunkSize = 64 * 1024; class TestEndpointChannel : public BaseEndpointChannel { public: @@ -62,7 +68,7 @@ std::function MakeDataPump( return [label, input, output, monitor]() { NEARBY_LOGS(INFO) << "streaming data through '" << label << "'"; while (true) { - auto read_response = input->Read(Pipe::kChunkSize); + auto read_response = input->Read(kChunkSize); if (!read_response.ok()) { NEARBY_LOGS(INFO) << "Peer reader closed on '" << label << "'"; output->Close(); @@ -158,21 +164,17 @@ DoDhKeyExchange(BaseEndpointChannel* channel_a, } TEST(BaseEndpointChannelTest, ConstructorDestructorWorks) { - Pipe pipe; - InputStream& input_stream = pipe.GetInputStream(); - OutputStream& output_stream = pipe.GetOutputStream(); + auto [input, output] = CreatePipe(); - TestEndpointChannel test_channel(&input_stream, &output_stream); + TestEndpointChannel test_channel(input.get(), output.get()); } TEST(BaseEndpointChannelTest, ReadWrite) { // Direct not-encrypted IO. - Pipe pipe_a; // channel_a writes to pipe_a, reads from pipe_b. - Pipe pipe_b; // channel_b writes to pipe_b, reads from pipe_a. - TestEndpointChannel channel_a(&pipe_b.GetInputStream(), - &pipe_a.GetOutputStream()); - TestEndpointChannel channel_b(&pipe_a.GetInputStream(), - &pipe_b.GetOutputStream()); + auto pipe_a = CreatePipe(); // channel_a writes to pipe_a, reads from pipe_b. + auto pipe_b = CreatePipe(); // channel_b writes to pipe_b, reads from pipe_a. + TestEndpointChannel channel_a(pipe_b.first.get(), pipe_a.second.get()); + TestEndpointChannel channel_b(pipe_a.first.get(), pipe_b.second.get()); ByteArray tx_message{"data message"}; channel_a.Write(tx_message); ByteArray rx_message = std::move(channel_b.Read().result()); @@ -180,8 +182,8 @@ TEST(BaseEndpointChannelTest, ReadWrite) { } TEST(BaseEndpointChannelTest, ChannelUnencryptedByDefault) { - Pipe pipe; - TestEndpointChannel channel(&pipe.GetInputStream(), &pipe.GetOutputStream()); + auto pipe = CreatePipe(); + TestEndpointChannel channel(pipe.first.get(), pipe.second.get()); ExceptionOr result = channel.TryDecrypt(ByteArray("message")); @@ -192,12 +194,10 @@ TEST(BaseEndpointChannelTest, ChannelUnencryptedByDefault) { TEST(BaseEndpointChannelTest, TryDecrypt) { absl::string_view kMessage = "message"; - Pipe pipe_a; // channel_a writes to pipe_a, reads from pipe_b. - Pipe pipe_b; // channel_b writes to pipe_b, reads from pipe_a. - TestEndpointChannel channel_a(&pipe_b.GetInputStream(), - &pipe_a.GetOutputStream()); - TestEndpointChannel channel_b(&pipe_a.GetInputStream(), - &pipe_b.GetOutputStream()); + auto pipe_a = CreatePipe(); // channel_a writes to pipe_a, reads from pipe_b. + auto pipe_b = CreatePipe(); // channel_b writes to pipe_b, reads from pipe_a. + TestEndpointChannel channel_a(pipe_b.first.get(), pipe_a.second.get()); + TestEndpointChannel channel_b(pipe_a.first.get(), pipe_b.second.get()); auto [context_a, context_b] = DoDhKeyExchange(&channel_a, &channel_b); ASSERT_NE(context_a, nullptr); ASSERT_NE(context_b, nullptr); @@ -215,12 +215,10 @@ TEST(BaseEndpointChannelTest, TryDecrypt) { } TEST(BaseEndpointChannelTest, TryDecryptFailsWhenDecryptionFails) { - Pipe pipe_a; // channel_a writes to pipe_a, reads from pipe_b. - Pipe pipe_b; // channel_b writes to pipe_b, reads from pipe_a. - TestEndpointChannel channel_a(&pipe_b.GetInputStream(), - &pipe_a.GetOutputStream()); - TestEndpointChannel channel_b(&pipe_a.GetInputStream(), - &pipe_b.GetOutputStream()); + auto pipe_a = CreatePipe(); // channel_a writes to pipe_a, reads from pipe_b. + auto pipe_b = CreatePipe(); // channel_b writes to pipe_b, reads from pipe_a. + TestEndpointChannel channel_a(pipe_b.first.get(), pipe_a.second.get()); + TestEndpointChannel channel_b(pipe_a.first.get(), pipe_b.second.get()); auto [context_a, context_b] = DoDhKeyExchange(&channel_a, &channel_b); ASSERT_NE(context_a, nullptr); channel_a.EnableEncryption(context_a); @@ -240,25 +238,27 @@ TEST(BaseEndpointChannelTest, NotEncryptedReadWriteCanBeIntercepted) { absl::Mutex mutex; std::string capture_a; std::string capture_b; - Pipe client_a; // Channel "a" writes to client "a", reads from server "a". - Pipe client_b; // Channel "b" writes to client "b", reads from server "b". - Pipe server_a; // Data pump "a" reads from client "a", writes to server "b". - Pipe server_b; // Data pump "b" reads from client "b", writes to server "a". - TestEndpointChannel channel_a(&server_a.GetInputStream(), - &client_a.GetOutputStream()); - TestEndpointChannel channel_b(&server_b.GetInputStream(), - &client_b.GetOutputStream()); + auto client_a = + CreatePipe(); // Channel "a" writes to client "a", reads from server "a". + auto client_b = + CreatePipe(); // Channel "b" writes to client "b", reads from server "b". + auto server_a = CreatePipe(); // Data pump "a" reads from client "a", writes + // to server "b". + auto server_b = CreatePipe(); // Data pump "b" reads from client "b", writes + // to server "a". + TestEndpointChannel channel_a(server_a.first.get(), client_a.second.get()); + TestEndpointChannel channel_b(server_b.first.get(), client_b.second.get()); ON_CALL(channel_a, GetMedium).WillByDefault([]() { return Medium::BLE; }); ON_CALL(channel_b, GetMedium).WillByDefault([]() { return Medium::BLE; }); MultiThreadExecutor executor(2); - executor.Execute(MakeDataPump( - "pump_a", &client_a.GetInputStream(), &server_b.GetOutputStream(), - MakeDataMonitor("monitor_a", &capture_a, &mutex))); - executor.Execute(MakeDataPump( - "pump_b", &client_b.GetInputStream(), &server_a.GetOutputStream(), - MakeDataMonitor("monitor_b", &capture_b, &mutex))); + executor.Execute( + MakeDataPump("pump_a", client_a.first.get(), server_b.second.get(), + MakeDataMonitor("monitor_a", &capture_a, &mutex))); + executor.Execute( + MakeDataPump("pump_b", client_b.first.get(), server_a.second.get(), + MakeDataMonitor("monitor_b", &capture_b, &mutex))); EXPECT_EQ(channel_a.GetType(), "BLE"); EXPECT_EQ(channel_b.GetType(), "BLE"); @@ -289,14 +289,16 @@ TEST(BaseEndpointChannelTest, EncryptedReadWriteCanNotBeIntercepted) { absl::Mutex mutex; std::string capture_a; std::string capture_b; - Pipe client_a; // Channel "a" writes to client "a", reads from server "a". - Pipe client_b; // Channel "b" writes to client "b", reads from server "b". - Pipe server_a; // Data pump "a" reads from client "a", writes to server "b". - Pipe server_b; // Data pump "b" reads from client "b", writes to server "a". - TestEndpointChannel channel_a(&server_a.GetInputStream(), - &client_a.GetOutputStream()); - TestEndpointChannel channel_b(&server_b.GetInputStream(), - &client_b.GetOutputStream()); + auto client_a = + CreatePipe(); // Channel "a" writes to client "a", reads from server "a". + auto client_b = + CreatePipe(); // Channel "b" writes to client "b", reads from server "b". + auto server_a = CreatePipe(); // Data pump "a" reads from client "a", writes + // to server "b". + auto server_b = CreatePipe(); // Data pump "b" reads from client "b", writes + // to server "a". + TestEndpointChannel channel_a(server_a.first.get(), client_a.second.get()); + TestEndpointChannel channel_b(server_b.first.get(), client_b.second.get()); ON_CALL(channel_a, GetMedium).WillByDefault([]() { return Medium::BLUETOOTH; @@ -306,12 +308,12 @@ TEST(BaseEndpointChannelTest, EncryptedReadWriteCanNotBeIntercepted) { }); MultiThreadExecutor executor(2); - executor.Execute(MakeDataPump( - "pump_a", &client_a.GetInputStream(), &server_b.GetOutputStream(), - MakeDataMonitor("monitor_a", &capture_a, &mutex))); - executor.Execute(MakeDataPump( - "pump_b", &client_b.GetInputStream(), &server_a.GetOutputStream(), - MakeDataMonitor("monitor_b", &capture_b, &mutex))); + executor.Execute( + MakeDataPump("pump_a", client_a.first.get(), server_b.second.get(), + MakeDataMonitor("monitor_a", &capture_a, &mutex))); + executor.Execute( + MakeDataPump("pump_b", client_b.first.get(), server_a.second.get(), + MakeDataMonitor("monitor_b", &capture_b, &mutex))); // Run DH key exchange; setup encryption contexts for channels. auto [context_a, context_b] = DoDhKeyExchange(&channel_a, &channel_b); @@ -346,12 +348,10 @@ TEST(BaseEndpointChannelTest, EncryptedReadWriteCanNotBeIntercepted) { TEST(BaseEndpointChannelTest, CanBesuspendedAndResumed) { // Setup test communication environment. - Pipe pipe_a; // channel_a writes to pipe_a, reads from pipe_b. - Pipe pipe_b; // channel_b writes to pipe_b, reads from pipe_a. - TestEndpointChannel channel_a(&pipe_b.GetInputStream(), - &pipe_a.GetOutputStream()); - TestEndpointChannel channel_b(&pipe_a.GetInputStream(), - &pipe_b.GetOutputStream()); + auto pipe_a = CreatePipe(); // channel_a writes to pipe_a, reads from pipe_b. + auto pipe_b = CreatePipe(); // channel_b writes to pipe_b, reads from pipe_a. + TestEndpointChannel channel_a(pipe_b.first.get(), pipe_a.second.get()); + TestEndpointChannel channel_b(pipe_a.first.get(), pipe_b.second.get()); ON_CALL(channel_a, GetMedium).WillByDefault([]() { return Medium::WIFI_LAN; @@ -399,14 +399,12 @@ TEST(BaseEndpointChannelTest, CanBesuspendedAndResumed) { } TEST(BaseEndpointChannelTest, ReadAfterInputStreamClosed) { - Pipe pipe; - InputStream& input_stream = pipe.GetInputStream(); - OutputStream& output_stream = pipe.GetOutputStream(); + auto [input, output] = CreatePipe(); - TestEndpointChannel test_channel(&input_stream, &output_stream); + TestEndpointChannel test_channel(input.get(), output.get()); // Close the output stream before trying to read from the input. - output_stream.Close(); + output->Close(); // Trying to read should fail gracefully with an IO error. ExceptionOr read_data = test_channel.Read(); @@ -417,12 +415,10 @@ TEST(BaseEndpointChannelTest, ReadAfterInputStreamClosed) { TEST(BaseEndpointChannelTest, ReadUnencryptedFrameOnEncryptedChannel) { // Setup test communication environment. - Pipe pipe_a; // channel_a writes to pipe_a, reads from pipe_b. - Pipe pipe_b; // channel_b writes to pipe_b, reads from pipe_a. - TestEndpointChannel channel_a(&pipe_b.GetInputStream(), - &pipe_a.GetOutputStream()); - TestEndpointChannel channel_b(&pipe_a.GetInputStream(), - &pipe_b.GetOutputStream()); + auto pipe_a = CreatePipe(); // channel_a writes to pipe_a, reads from pipe_b. + auto pipe_b = CreatePipe(); // channel_b writes to pipe_b, reads from pipe_a. + TestEndpointChannel channel_a(pipe_b.first.get(), pipe_a.second.get()); + TestEndpointChannel channel_b(pipe_a.first.get(), pipe_b.second.get()); ON_CALL(channel_a, GetMedium).WillByDefault([]() { return Medium::BLUETOOTH; diff --git a/connections/implementation/base_pcp_handler_test.cc b/connections/implementation/base_pcp_handler_test.cc index 9e849bef..ddb38e7e 100644 --- a/connections/implementation/base_pcp_handler_test.cc +++ b/connections/implementation/base_pcp_handler_test.cc @@ -18,21 +18,32 @@ #include #include #include +#include +#include #include "gmock/gmock.h" #include "protobuf-matchers/protocol-buffer-matchers.h" #include "gtest/gtest.h" +#include "absl/base/thread_annotations.h" #include "absl/strings/string_view.h" +#include "absl/time/clock.h" #include "absl/time/time.h" +#include "connections/advertising_options.h" +#include "connections/connection_options.h" #include "connections/discovery_options.h" +#include "connections/implementation/analytics/packet_meta_data.h" #include "connections/implementation/base_endpoint_channel.h" #include "connections/implementation/bwu_manager.h" #include "connections/implementation/client_proxy.h" #include "connections/implementation/encryption_runner.h" +#include "connections/implementation/endpoint_manager.h" +#include "connections/implementation/mediums/mediums.h" #include "connections/implementation/offline_frames.h" +#include "connections/implementation/pcp.h" #include "connections/implementation/proto/offline_wire_formats.pb.h" #include "connections/listeners.h" #include "connections/medium_selector.h" +#include "connections/out_of_band_connection_metadata.h" #include "connections/params.h" #include "connections/status.h" #include "connections/strategy.h" @@ -41,7 +52,12 @@ #include "internal/interop/device_provider.h" #include "internal/platform/byte_array.h" #include "internal/platform/exception.h" +#include "internal/platform/feature_flags.h" +#include "internal/platform/future.h" +#include "internal/platform/input_stream.h" +#include "internal/platform/logging.h" #include "internal/platform/medium_environment.h" +#include "internal/platform/output_stream.h" #include "internal/platform/pipe.h" #include "proto/connections_enums.pb.h" #include "proto/connections_enums.proto.h" @@ -108,9 +124,12 @@ class FakePresenceDeviceProvider : public NearbyDeviceProvider { class MockEndpointChannel : public BaseEndpointChannel { public: - explicit MockEndpointChannel(Pipe* reader, Pipe* writer) - : BaseEndpointChannel("service_id", "channel", &reader->GetInputStream(), - &writer->GetOutputStream()) {} + explicit MockEndpointChannel(std::unique_ptr reader, + std::unique_ptr writer) + : BaseEndpointChannel("service_id", "channel", reader.get(), + writer.get()), + input_stream_(std::move(reader)), + output_stream_(std::move(writer)) {} ExceptionOr DoRead() { return BaseEndpointChannel::Read(); } Exception DoWrite(const ByteArray& data) { @@ -136,6 +155,10 @@ class MockEndpointChannel : public BaseEndpointChannel { MOCK_METHOD(absl::Time, GetLastReadTimestamp, (), (const override)); bool broken_write_{false}; + + private: + std::unique_ptr input_stream_; + std::unique_ptr output_stream_; }; class MockPcpHandler : public BasePcpHandler { @@ -459,10 +482,13 @@ class BasePcpHandlerTest std::pair, std::unique_ptr> SetupConnection( - Pipe& pipe_a, Pipe& pipe_b, location::nearby::proto::connections::Medium medium) { // NOLINT - auto channel_a = std::make_unique(&pipe_b, &pipe_a); - auto channel_b = std::make_unique(&pipe_a, &pipe_b); + auto [input_a, output_a] = CreatePipe(); + auto [input_b, output_b] = CreatePipe(); + auto channel_a = std::make_unique(std::move(input_a), + std::move(output_b)); + auto channel_b = std::make_unique(std::move(input_b), + std::move(output_a)); // On initiator (A) side, we drop the first write, since this is a // connection establishment packet, and we don't have the peer entity, just // the peer channel. The rest of the exchange must happen for the benefit of @@ -648,9 +674,6 @@ class BasePcpHandlerTest expected_result); NEARBY_LOG(INFO, "Stopping Encryption Runner"); } - - Pipe pipe_a_; - Pipe pipe_b_; MockConnectionListener mock_connection_listener_; MockDiscoveryListener mock_discovery_listener_; ConnectionListener connection_listener_{ @@ -839,7 +862,7 @@ TEST_F(BasePcpHandlerTest, WifiMediumFailFallBackToBT) { auto mediums = pcp_handler.GetDiscoveryMediums(&client); auto connect_medium = mediums[mediums.size() - 1]; - auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium); + auto channel_pair = SetupConnection(connect_medium); auto& channel_a = channel_pair.first; auto& channel_b = channel_pair.second; EXPECT_CALL(*channel_a, CloseImpl).Times(1); @@ -865,7 +888,7 @@ TEST_P(BasePcpHandlerTest, RequestConnectionChangesState) { StartDiscovery(&client, &pcp_handler); auto mediums = pcp_handler.GetDiscoveryMediums(&client); auto connect_medium = mediums[mediums.size() - 1]; - auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium); + auto channel_pair = SetupConnection(connect_medium); auto& channel_a = channel_pair.first; auto& channel_b = channel_pair.second; EXPECT_CALL(*channel_a, CloseImpl).Times(1); @@ -910,7 +933,7 @@ TEST_P(BasePcpHandlerTest, CanRequestConnectionPresence) { StartDiscovery(&client, &pcp_handler); auto mediums = pcp_handler.GetDiscoveryMediums(&client); auto connect_medium = mediums[mediums.size() - 1]; - auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium); + auto channel_pair = SetupConnection(connect_medium); auto& channel_a = channel_pair.first; auto& channel_b = channel_pair.second; EXPECT_CALL(*channel_a, CloseImpl).Times(1); @@ -941,7 +964,7 @@ TEST_P(BasePcpHandlerTest, CanRequestConnectionLegacy) { StartDiscovery(&client, &pcp_handler); auto mediums = pcp_handler.GetDiscoveryMediums(&client); auto connect_medium = mediums[mediums.size() - 1]; - auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium); + auto channel_pair = SetupConnection(connect_medium); auto& channel_a = channel_pair.first; auto& channel_b = channel_pair.second; EXPECT_CALL(*channel_a, CloseImpl).Times(1); @@ -968,7 +991,7 @@ TEST_P(BasePcpHandlerTest, IoError_RequestConnectionFails) { StartDiscovery(&client, &pcp_handler); auto mediums = pcp_handler.GetDiscoveryMediums(&client); auto connect_medium = mediums[mediums.size() - 1]; - auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium); + auto channel_pair = SetupConnection(connect_medium); auto& channel_a = channel_pair.first; auto& channel_b = channel_pair.second; EXPECT_CALL(*channel_a, CloseImpl).Times(AtLeast(1)); @@ -997,7 +1020,7 @@ TEST_P(BasePcpHandlerTest, AcceptConnectionChangesState) { StartDiscovery(&client, &pcp_handler); auto mediums = pcp_handler.GetDiscoveryMediums(&client); auto connect_medium = mediums[mediums.size() - 1]; - auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium); + auto channel_pair = SetupConnection(connect_medium); auto& channel_a = channel_pair.first; auto& channel_b = channel_pair.second; EXPECT_CALL(*channel_a, CloseImpl).Times(1); @@ -1028,7 +1051,7 @@ TEST_P(BasePcpHandlerTest, RejectConnectionChangesState) { StartDiscovery(&client, &pcp_handler); auto mediums = pcp_handler.GetDiscoveryMediums(&client); auto connect_medium = mediums[mediums.size() - 1]; - auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium); + auto channel_pair = SetupConnection(connect_medium); auto& channel_b = channel_pair.second; EXPECT_CALL(mock_connection_listener_.rejected_cb, Call).Times(1); RequestConnection(endpoint_id, std::move(channel_pair.first), channel_b.get(), @@ -1056,7 +1079,7 @@ TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) { StartDiscovery(&client, &pcp_handler); auto mediums = pcp_handler.GetDiscoveryMediums(&client); auto connect_medium = mediums[mediums.size() - 1]; - auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium); + auto channel_pair = SetupConnection(connect_medium); auto& channel_a = channel_pair.first; auto& channel_b = channel_pair.second; EXPECT_CALL(*channel_a, CloseImpl).Times(1); @@ -1098,7 +1121,7 @@ TEST_P(BasePcpHandlerTest, DestructorIsCalledOnProtocolEndpoint) { StartDiscovery(&client, &pcp_handler); auto mediums = pcp_handler.GetDiscoveryMediums(&client); auto connect_medium = mediums[mediums.size() - 1]; - auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium); + auto channel_pair = SetupConnection(connect_medium); auto& channel_a = channel_pair.first; auto& channel_b = channel_pair.second; EXPECT_CALL(*channel_a, CloseImpl).Times(1); @@ -1141,7 +1164,7 @@ TEST_P(BasePcpHandlerTest, MultipleMediumsProduceSingleEndpointLostEvent) { StartDiscovery(&client, &pcp_handler); auto mediums = pcp_handler.GetDiscoveryMediums(&client); auto connect_medium = mediums[mediums.size() - 1]; - auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium); + auto channel_pair = SetupConnection(connect_medium); auto& channel_a = channel_pair.first; auto& channel_b = channel_pair.second; EXPECT_CALL(*channel_a, CloseImpl).Times(1); @@ -1582,7 +1605,7 @@ TEST_F(BasePcpHandlerTest, TestDeviceFilterForConnectionsWithUnknown) { .first.Ok()); ASSERT_TRUE(client.IsListeningForIncomingConnections()); ASSERT_TRUE(pcp_handler.CanReceiveIncomingConnection(&client)); - auto channel_pair = SetupConnection(pipe_a_, pipe_b_, Medium::BLUETOOTH); + auto channel_pair = SetupConnection(Medium::BLUETOOTH); ByteArray serialized_frame = parser::ForConnectionRequestConnections( {}, { .local_endpoint_id = "ABCD", @@ -1631,7 +1654,7 @@ TEST_F(BasePcpHandlerTest, TestDeviceFilterForPresenceWithUnknown) { .first.Ok()); ASSERT_TRUE(client.IsListeningForIncomingConnections()); ASSERT_TRUE(pcp_handler.CanReceiveIncomingConnection(&client)); - auto channel_pair = SetupConnection(pipe_a_, pipe_b_, Medium::BLUETOOTH); + auto channel_pair = SetupConnection(Medium::BLUETOOTH); ByteArray serialized_frame = parser::ForConnectionRequestConnections( {}, { .local_endpoint_id = "ABCD", @@ -1681,7 +1704,7 @@ TEST_F(BasePcpHandlerTest, TestDeviceFilterForPresenceWithConnections) { .first.Ok()); ASSERT_TRUE(client.IsListeningForIncomingConnections()); ASSERT_TRUE(pcp_handler.CanReceiveIncomingConnection(&client)); - auto channel_pair = SetupConnection(pipe_a_, pipe_b_, Medium::BLUETOOTH); + auto channel_pair = SetupConnection(Medium::BLUETOOTH); ByteArray serialized_frame = parser::ForConnectionRequestConnections( {}, { .local_endpoint_id = "ABCD", @@ -1732,7 +1755,7 @@ TEST_F(BasePcpHandlerTest, TestDeviceFilterForPresenceWithPresence) { .first.Ok()); ASSERT_TRUE(client.IsListeningForIncomingConnections()); ASSERT_TRUE(pcp_handler.CanReceiveIncomingConnection(&client)); - auto channel_pair = SetupConnection(pipe_a_, pipe_b_, Medium::BLUETOOTH); + auto channel_pair = SetupConnection(Medium::BLUETOOTH); ByteArray serialized_frame = parser::ForConnectionRequestConnections( {}, { .local_endpoint_id = "ABCD", @@ -1782,7 +1805,7 @@ TEST_F(BasePcpHandlerTest, TestDeviceFilterForConnectionsWithConnections) { .first.Ok()); ASSERT_TRUE(client.IsListeningForIncomingConnections()); ASSERT_TRUE(pcp_handler.CanReceiveIncomingConnection(&client)); - auto channel_pair = SetupConnection(pipe_a_, pipe_b_, Medium::BLUETOOTH); + auto channel_pair = SetupConnection(Medium::BLUETOOTH); ByteArray serialized_frame = parser::ForConnectionRequestConnections( {}, { .local_endpoint_id = "ABCD", @@ -1832,7 +1855,7 @@ TEST_F(BasePcpHandlerTest, TestDeviceFilterForConnectionsWithPresence) { .first.Ok()); ASSERT_TRUE(client.IsListeningForIncomingConnections()); ASSERT_TRUE(pcp_handler.CanReceiveIncomingConnection(&client)); - auto channel_pair = SetupConnection(pipe_a_, pipe_b_, Medium::BLUETOOTH); + auto channel_pair = SetupConnection(Medium::BLUETOOTH); ByteArray serialized_frame = parser::ForConnectionRequestConnections( {}, { .local_endpoint_id = "ABCD", diff --git a/connections/implementation/encryption_runner_test.cc b/connections/implementation/encryption_runner_test.cc index 0bc91960..4dbd624e 100644 --- a/connections/implementation/encryption_runner_test.cc +++ b/connections/implementation/encryption_runner_test.cc @@ -14,23 +14,30 @@ #include "connections/implementation/encryption_runner.h" -#include "gmock/gmock.h" -#include "protobuf-matchers/protocol-buffer-matchers.h" +#include +#include + #include "gtest/gtest.h" -#include "absl/time/clock.h" +#include "absl/time/time.h" +#include "connections/implementation/analytics/analytics_recorder.h" #include "connections/implementation/client_proxy.h" #include "connections/implementation/endpoint_channel.h" #include "internal/platform/byte_array.h" #include "internal/platform/count_down_latch.h" +#include "internal/platform/exception.h" +#include "internal/platform/input_stream.h" +#include "internal/platform/output_stream.h" #include "internal/platform/pipe.h" #include "internal/platform/system_clock.h" #include "proto/connections_enums.pb.h" +#include "third_party/ukey2/src/main/cpp/include/securegcm/ukey2_handshake.h" namespace nearby { namespace connections { namespace { using ::location::nearby::proto::connections::Medium; +constexpr size_t kChunkSize = 64 * 1024; class FakeEndpointChannel : public EndpointChannel { public: @@ -38,13 +45,11 @@ class FakeEndpointChannel : public EndpointChannel { : in_(in), out_(out) {} ExceptionOr Read() override { read_timestamp_ = SystemClock::ElapsedRealtime(); - return in_ ? in_->Read(Pipe::kChunkSize) - : ExceptionOr{Exception::kIo}; + return in_ ? in_->Read(kChunkSize) : ExceptionOr{Exception::kIo}; } ExceptionOr Read(PacketMetaData& packet_meta_data) override { read_timestamp_ = SystemClock::ElapsedRealtime(); - return in_ ? in_->Read(Pipe::kChunkSize) - : ExceptionOr{Exception::kIo}; + return in_ ? in_->Read(kChunkSize) : ExceptionOr{Exception::kIo}; } Exception Write(const ByteArray& data) override { write_timestamp_ = SystemClock::ElapsedRealtime(); @@ -103,8 +108,7 @@ class FakeEndpointChannel : public EndpointChannel { }; struct User { - User(Pipe* reader, Pipe* writer) - : channel(&reader->GetInputStream(), &writer->GetOutputStream()) {} + User(InputStream* reader, OutputStream* writer) : channel(reader, writer) {} FakeEndpointChannel channel; EncryptionRunner crypto; @@ -126,10 +130,12 @@ struct Response { TEST(EncryptionRunnerTest, ConstructorDestructorWorks) { EncryptionRunner enc; } TEST(EncryptionRunnerTest, ReadWrite) { - Pipe from_a_to_b; - Pipe from_b_to_a; - User user_a(/*reader=*/&from_b_to_a, /*writer=*/&from_a_to_b); - User user_b(/*reader=*/&from_a_to_b, /*writer=*/&from_b_to_a); + auto from_a_to_b = CreatePipe(); + auto from_b_to_a = CreatePipe(); + User user_a(/*reader=*/from_b_to_a.first.get(), + /*writer=*/from_a_to_b.second.get()); + User user_b(/*reader=*/from_a_to_b.first.get(), + /*writer=*/from_b_to_a.second.get()); Response response; user_a.crypto.StartServer( diff --git a/connections/implementation/endpoint_channel_manager_test.cc b/connections/implementation/endpoint_channel_manager_test.cc index fe34336b..466f67c0 100644 --- a/connections/implementation/endpoint_channel_manager_test.cc +++ b/connections/implementation/endpoint_channel_manager_test.cc @@ -14,20 +14,24 @@ #include "connections/implementation/endpoint_channel_manager.h" +#include #include #include #include #include -#include "securegcm/d2d_connection_context_v1.h" #include "securegcm/ukey2_handshake.h" #include "gmock/gmock.h" #include "protobuf-matchers/protocol-buffer-matchers.h" #include "gtest/gtest.h" +#include "absl/strings/string_view.h" #include "absl/synchronization/mutex.h" #include "absl/time/time.h" #include "connections/implementation/base_endpoint_channel.h" +#include "connections/implementation/client_proxy.h" #include "connections/implementation/encryption_runner.h" +#include "connections/implementation/endpoint_channel.h" +#include "internal/platform/byte_array.h" #include "internal/platform/count_down_latch.h" #include "internal/platform/exception.h" #include "internal/platform/input_stream.h" @@ -45,6 +49,7 @@ using ::location::nearby::proto::connections::DisconnectionReason; using ::location::nearby::proto::connections::Medium; using EncryptionContext = BaseEndpointChannel::EncryptionContext; +constexpr size_t kChunkSize = 64 * 1024; constexpr absl::string_view kEndpointId = "EndpointId"; constexpr absl::string_view kMonitorA = "MonitorA"; constexpr absl::string_view kMonitorB = "MonitorB"; @@ -66,7 +71,7 @@ std::function MakeDataPump( return [label, input, output, monitor]() { NEARBY_LOGS(INFO) << "streaming data through '" << label << "'"; while (true) { - auto read_response = input->Read(Pipe::kChunkSize); + auto read_response = input->Read(kChunkSize); if (!read_response.ok()) { NEARBY_LOGS(INFO) << "Peer reader closed on '" << label << "'"; output->Close(); @@ -169,14 +174,18 @@ TEST(BaseEndpointChannelManagerTest, RegisterChannelEncryptedReadwrite) { std::string capture_b; ClientProxy proxy_a; ClientProxy proxy_b; - Pipe client_a; // Channel "a" writes to client "a", reads from server "a". - Pipe client_b; // Channel "b" writes to client "b", reads from server "b". - Pipe server_a; // Data pump "a" reads from client "a", writes to server "b". - Pipe server_b; // Data pump "b" reads from client "b", writes to server "a". - auto channel_a = std::make_unique( - &server_a.GetInputStream(), &client_a.GetOutputStream()); - auto channel_b = std::make_unique( - &server_b.GetInputStream(), &client_b.GetOutputStream()); + auto client_a = + CreatePipe(); // Channel "a" writes to client "a", reads from server "a". + auto client_b = + CreatePipe(); // Channel "b" writes to client "b", reads from server "b". + auto server_a = CreatePipe(); // Data pump "a" reads from client "a", writes + // to server "b". + auto server_b = CreatePipe(); // Data pump "b" reads from client "b", writes + // to server "a". + auto channel_a = std::make_unique(server_a.first.get(), + client_a.second.get()); + auto channel_b = std::make_unique(server_b.first.get(), + client_b.second.get()); auto channel_a_raw = channel_a.get(); auto channel_b_raw = channel_b.get(); @@ -188,12 +197,12 @@ TEST(BaseEndpointChannelManagerTest, RegisterChannelEncryptedReadwrite) { }); MultiThreadExecutor executor(2); - executor.Execute(MakeDataPump( - kPumpA, &client_a.GetInputStream(), &server_b.GetOutputStream(), - MakeDataMonitor(kMonitorA, &capture_a, &mutex))); - executor.Execute(MakeDataPump( - kPumpB, &client_b.GetInputStream(), &server_a.GetOutputStream(), - MakeDataMonitor(kMonitorB, &capture_b, &mutex))); + executor.Execute( + MakeDataPump(kPumpA, client_a.first.get(), server_b.second.get(), + MakeDataMonitor(kMonitorA, &capture_a, &mutex))); + executor.Execute( + MakeDataPump(kPumpB, client_b.first.get(), server_a.second.get(), + MakeDataMonitor(kMonitorB, &capture_b, &mutex))); // Run DH key exchange; setup encryption contexts for channels. auto context = DoDhKeyExchange(channel_a.get(), channel_b.get()); @@ -240,14 +249,18 @@ TEST(BaseEndpointChannelManagerTest, ReplaceChannelNoEncrypted) { std::string capture_b; ClientProxy proxy_a; ClientProxy proxy_b; - Pipe client_a; // Channel "a" writes to client "a", reads from server "a". - Pipe client_b; // Channel "b" writes to client "b", reads from server "b". - Pipe server_a; // Data pump "a" reads from client "a", writes to server "b". - Pipe server_b; // Data pump "b" reads from client "b", writes to server "a". - auto channel_a = std::make_unique( - &server_a.GetInputStream(), &client_a.GetOutputStream()); - auto channel_b = std::make_unique( - &server_b.GetInputStream(), &client_b.GetOutputStream()); + auto client_a = + CreatePipe(); // Channel "a" writes to client "a", reads from server "a". + auto client_b = + CreatePipe(); // Channel "b" writes to client "b", reads from server "b". + auto server_a = CreatePipe(); // Data pump "a" reads from client "a", writes + // to server "b". + auto server_b = CreatePipe(); // Data pump "b" reads from client "b", writes + // to server "a". + auto channel_a = std::make_unique(server_a.first.get(), + client_a.second.get()); + auto channel_b = std::make_unique(server_b.first.get(), + client_b.second.get()); auto channel_a_raw = channel_a.get(); auto channel_b_raw = channel_b.get(); @@ -259,12 +272,12 @@ TEST(BaseEndpointChannelManagerTest, ReplaceChannelNoEncrypted) { }); MultiThreadExecutor executor(2); - executor.Execute(MakeDataPump( - kPumpA, &client_a.GetInputStream(), &server_b.GetOutputStream(), - MakeDataMonitor(kMonitorA, &capture_a, &mutex))); - executor.Execute(MakeDataPump( - kPumpB, &client_b.GetInputStream(), &server_a.GetOutputStream(), - MakeDataMonitor(kMonitorB, &capture_b, &mutex))); + executor.Execute( + MakeDataPump(kPumpA, client_a.first.get(), server_b.second.get(), + MakeDataMonitor(kMonitorA, &capture_a, &mutex))); + executor.Execute( + MakeDataPump(kPumpB, client_b.first.get(), server_a.second.get(), + MakeDataMonitor(kMonitorB, &capture_b, &mutex))); // Run DH key exchange; setup encryption contexts for channels. auto context = DoDhKeyExchange(channel_a.get(), channel_b.get()); diff --git a/connections/implementation/internal_payload_factory.cc b/connections/implementation/internal_payload_factory.cc index 59d7de9d..38268fc9 100644 --- a/connections/implementation/internal_payload_factory.cc +++ b/connections/implementation/internal_payload_factory.cc @@ -14,24 +14,25 @@ #include "connections/implementation/internal_payload_factory.h" +#include #include #include #include #include -#include "absl/memory/memory.h" -#include "connections/implementation/offline_frames_validator.h" +#include "absl/strings/str_cat.h" +#include "connections/implementation/internal_payload.h" +#include "connections/implementation/proto/offline_wire_formats.pb.h" #include "connections/payload.h" +#include "connections/payload_type.h" #include "internal/platform/byte_array.h" -#include "internal/platform/condition_variable.h" #include "internal/platform/exception.h" -#include "internal/platform/feature_flags.h" #include "internal/platform/file.h" #include "internal/platform/implementation/platform.h" -#include "internal/platform/implementation/shared/file.h" +#include "internal/platform/input_stream.h" #include "internal/platform/logging.h" -#include "internal/platform/mutex.h" #include "internal/platform/os_name.h" +#include "internal/platform/output_stream.h" #include "internal/platform/pipe.h" namespace nearby { @@ -155,8 +156,9 @@ class OutgoingStreamInternalPayload : public InternalPayload { class IncomingStreamInternalPayload : public InternalPayload { public: - IncomingStreamInternalPayload(Payload payload, std::shared_ptr pipe) - : InternalPayload(std::move(payload)), pipe_(pipe) {} + IncomingStreamInternalPayload(Payload payload, + std::unique_ptr output) + : InternalPayload(std::move(payload)), output_(std::move(output)) {} PayloadTransferFrame::PayloadHeader::PayloadType GetType() const override { return PayloadTransferFrame::PayloadHeader::STREAM; @@ -174,7 +176,7 @@ class IncomingStreamInternalPayload : public InternalPayload { return {Exception::kSuccess}; } - return pipe_->GetOutputStream().Write(chunk); + return output_->Write(chunk); } ExceptionOr SkipToOffset(size_t offset) override { @@ -183,10 +185,10 @@ class IncomingStreamInternalPayload : public InternalPayload { return {Exception::kIo}; } - void Close() override { pipe_->GetOutputStream().Close(); } + void Close() override { output_->Close(); } private: - std::shared_ptr pipe_; + std::unique_ptr output_; }; class OutgoingFileInternalPayload : public InternalPayload { @@ -311,14 +313,14 @@ std::unique_ptr CreateOutgoingInternalPayload( Payload payload) { switch (payload.GetType()) { case PayloadType::kBytes: - return absl::make_unique(std::move(payload)); + return std::make_unique(std::move(payload)); case PayloadType::kFile: { - return absl::make_unique(std::move(payload)); + return std::make_unique(std::move(payload)); } case PayloadType::kStream: - return absl::make_unique( + return std::make_unique( std::move(payload)); default: @@ -359,19 +361,15 @@ std::unique_ptr CreateIncomingInternalPayload( const Payload::Id payload_id = frame.payload_header().id(); switch (frame.payload_header().type()) { case PayloadTransferFrame::PayloadHeader::BYTES: { - return absl::make_unique( + return std::make_unique( Payload(payload_id, ByteArray(frame.payload_chunk().body()))); } case PayloadTransferFrame::PayloadHeader::STREAM: { - auto pipe = std::make_shared(); + auto [input, output] = CreatePipe(); - return absl::make_unique( - Payload(payload_id, - [pipe]() -> InputStream& { - return pipe->GetInputStream(); // NOLINT - }), - pipe); + return std::make_unique( + Payload(payload_id, std::move(input)), std::move(output)); } case PayloadTransferFrame::PayloadHeader::FILE: { @@ -411,11 +409,11 @@ std::unique_ptr CreateIncomingInternalPayload( // there will be no input file to open. // On Chrome the file path should be empty, so use the payload id. if (ImplementationPlatform::GetCurrentOS() == OSName::kChromeOS) { - return absl::make_unique( + return std::make_unique( Payload(payload_id, InputFile(payload_id, total_size)), OutputFile(payload_id), total_size); } else { - return absl::make_unique( + return std::make_unique( Payload(payload_id, parent_folder, file_name, InputFile(file_path, total_size)), OutputFile(file_path), total_size); diff --git a/connections/implementation/internal_payload_factory_test.cc b/connections/implementation/internal_payload_factory_test.cc index d17556e1..9f6979ae 100644 --- a/connections/implementation/internal_payload_factory_test.cc +++ b/connections/implementation/internal_payload_factory_test.cc @@ -14,15 +14,19 @@ #include "connections/implementation/internal_payload_factory.h" +#include +#include #include #include -#include "gmock/gmock.h" -#include "protobuf-matchers/protocol-buffer-matchers.h" #include "gtest/gtest.h" -#include "connections/implementation/offline_frames.h" +#include "connections/implementation/internal_payload.h" #include "connections/implementation/proto/offline_wire_formats.pb.h" +#include "connections/payload.h" +#include "connections/payload_type.h" #include "internal/platform/byte_array.h" +#include "internal/platform/exception.h" +#include "internal/platform/file.h" #include "internal/platform/pipe.h" namespace nearby { @@ -44,11 +48,9 @@ TEST(InternalPayloadFactoryTest, CanCreateInternalPayloadFromBytePayload) { } TEST(InternalPayloadFactoryTest, CanCreateInternalPayloadFromStreamPayload) { - auto pipe = std::make_shared(); + auto [input, output] = CreatePipe(); std::unique_ptr internal_payload = - CreateOutgoingInternalPayload(Payload{[pipe]() -> InputStream& { - return pipe->GetInputStream(); // NOLINT - }}); + CreateOutgoingInternalPayload(Payload(std::move(input))); EXPECT_NE(internal_payload, nullptr); Payload payload = internal_payload->ReleasePayload(); EXPECT_EQ(payload.AsFile(), nullptr); @@ -212,13 +214,11 @@ TEST(InternalPayloadFactoryTest, SkipToOffset_StreamPayloadValidOffset_SkipsOffset) { ByteArray contents("0123456789"); constexpr size_t kOffset = 6; - auto pipe = std::make_shared(); + auto [input, output] = CreatePipe(); std::unique_ptr internal_payload = - CreateOutgoingInternalPayload(Payload{[pipe]() -> InputStream& { - return pipe->GetInputStream(); // NOLINT - }}); + CreateOutgoingInternalPayload(Payload(std::move(input))); EXPECT_NE(internal_payload, nullptr); - pipe->GetOutputStream().Write(contents); + output->Write(contents); ExceptionOr result = internal_payload->SkipToOffset(kOffset); diff --git a/connections/implementation/mediums/webrtc/webrtc_socket_impl.cc b/connections/implementation/mediums/webrtc/webrtc_socket_impl.cc index 63e66752..96d1b9bb 100644 --- a/connections/implementation/mediums/webrtc/webrtc_socket_impl.cc +++ b/connections/implementation/mediums/webrtc/webrtc_socket_impl.cc @@ -12,10 +12,18 @@ // See the License for the specific language governing permissions and // limitations under the License. +#include +#include +#include +#include + +#include "internal/platform/byte_array.h" +#include "internal/platform/exception.h" +#include "internal/platform/input_stream.h" +#include "internal/platform/pipe.h" #ifndef NO_WEBRTC #include "connections/implementation/mediums/webrtc/webrtc_socket_impl.h" - #include "internal/platform/logging.h" #include "internal/platform/mutex_lock.h" @@ -61,6 +69,7 @@ WebRtcSocket::WebRtcSocket( : name_(name), data_channel_(std::move(data_channel)) { NEARBY_LOGS(INFO) << "WebRtcSocket::WebRtcSocket(" << name_ << ") this: " << this; + std::tie(pipe_input_, pipe_output_) = CreatePipe(); data_channel_->RegisterObserver(this); } @@ -77,7 +86,7 @@ WebRtcSocket::~WebRtcSocket() { << ") this: " << this << " done"; } -InputStream& WebRtcSocket::GetInputStream() { return pipe_.GetInputStream(); } +InputStream& WebRtcSocket::GetInputStream() { return *pipe_input_; } OutputStream& WebRtcSocket::GetOutputStream() { return output_stream_; } @@ -129,12 +138,12 @@ void WebRtcSocket::OnMessage(const webrtc::DataBuffer& buffer) { // we don't block signaling. OffloadFromSignalingThread( [this, buffer = ByteArray(buffer.data.data(), buffer.size())] { - if (!pipe_.GetOutputStream().Write(buffer).Ok()) { + if (!pipe_output_->Write(buffer).Ok()) { Close(); return; } - if (!pipe_.GetOutputStream().Flush().Ok()) { + if (!pipe_output_->Flush().Ok()) { Close(); } }); @@ -159,8 +168,8 @@ void WebRtcSocket::ClosePipe() { // This is thread-safe to close these sockets even if a read or write is in // process on another thread, Close will wait for the exclusive mutex before // setting state. - pipe_.GetInputStream().Close(); - pipe_.GetOutputStream().Close(); + pipe_input_->Close(); + pipe_output_->Close(); WakeUpWriter(); NEARBY_LOGS(INFO) << "WebRtcSocket::ClosePipe(" << name_ << ") this: " << this << " done"; diff --git a/connections/implementation/mediums/webrtc/webrtc_socket_impl.h b/connections/implementation/mediums/webrtc/webrtc_socket_impl.h index beb1000d..723280b8 100644 --- a/connections/implementation/mediums/webrtc/webrtc_socket_impl.h +++ b/connections/implementation/mediums/webrtc/webrtc_socket_impl.h @@ -100,8 +100,8 @@ class WebRtcSocket : public Socket, public webrtc::DataChannelObserver { std::string name_; rtc::scoped_refptr data_channel_; - Pipe pipe_; - + std::unique_ptr pipe_input_; + std::unique_ptr pipe_output_; OutputStreamImpl output_stream_{this}; AtomicBoolean closed_{false}; diff --git a/connections/implementation/offline_service_controller_test.cc b/connections/implementation/offline_service_controller_test.cc index 43daafea..5b4e0989 100644 --- a/connections/implementation/offline_service_controller_test.cc +++ b/connections/implementation/offline_service_controller_test.cc @@ -12,31 +12,42 @@ // See the License for the specific language governing permissions and // limitations under the License. -#include "connections/implementation/offline_service_controller.h" - #include +#include +#include +#include #include "gmock/gmock.h" #include "protobuf-matchers/protocol-buffer-matchers.h" #include "gtest/gtest.h" +#include "absl/strings/string_view.h" #include "absl/time/time.h" +#include "connections/advertising_options.h" +#include "connections/discovery_options.h" #include "connections/implementation/flags/nearby_connections_feature_flags.h" #include "connections/implementation/offline_simulation_user.h" +#include "connections/listeners.h" +#include "connections/medium_selector.h" +#include "connections/out_of_band_connection_metadata.h" +#include "connections/payload.h" #include "connections/status.h" +#include "connections/strategy.h" #include "internal/flags/nearby_flags.h" +#include "internal/platform/byte_array.h" #include "internal/platform/count_down_latch.h" +#include "internal/platform/input_stream.h" #include "internal/platform/logging.h" #include "internal/platform/medium_environment.h" #include "internal/platform/output_stream.h" #include "internal/platform/pipe.h" -#include "internal/platform/system_clock.h" +#include "proto/connections_enums.proto.h" namespace nearby { namespace connections { namespace { using ::testing::Eq; - +constexpr size_t kChunkSize = 64 * 1024; constexpr std::array kFakeMacAddress = {'a', 'b', 'c', 'd', 'e', 'f'}; constexpr absl::string_view kServiceId = "service-id"; constexpr absl::string_view kDeviceA = "device-a"; @@ -302,12 +313,9 @@ TEST_P(OfflineServiceControllerTest, CanSendStreamPayload) { user_b.ExpectPayload(payload_latch_); ASSERT_TRUE(SetupConnection(user_a, user_b)); ByteArray message(std::string{kMessage}); - auto pipe = std::make_shared(); - OutputStream& tx = pipe->GetOutputStream(); - user_a.SendPayload(Payload([pipe]() -> InputStream& { - return pipe->GetInputStream(); // NOLINT - })); - tx.Write(message); + auto [input, tx] = CreatePipe(); + user_a.SendPayload(Payload(std::move(input))); + tx->Write(message); EXPECT_TRUE(payload_latch_.Await(kLongTimeout)); ASSERT_NE(user_b.GetPayload().AsStream(), nullptr); InputStream& rx = *user_b.GetPayload().AsStream(); @@ -316,7 +324,7 @@ TEST_P(OfflineServiceControllerTest, CanSendStreamPayload) { return info.bytes_transferred >= size; }, kLongTimeout)); - EXPECT_EQ(rx.Read(Pipe::kChunkSize).result(), message); + EXPECT_EQ(rx.Read(kChunkSize).result(), message); user_a.Stop(); user_b.Stop(); env_.Stop(); @@ -329,12 +337,9 @@ TEST_P(OfflineServiceControllerTest, CanCancelStreamPayload) { user_b.ExpectPayload(payload_latch_); ASSERT_TRUE(SetupConnection(user_a, user_b)); ByteArray message(std::string{kMessage}); - auto pipe = std::make_shared(); - OutputStream& tx = pipe->GetOutputStream(); - user_a.SendPayload(Payload([pipe]() -> InputStream& { - return pipe->GetInputStream(); // NOLINT - })); - tx.Write(message); + auto [input, tx] = CreatePipe(); + user_a.SendPayload(Payload(std::move(input))); + tx->Write(message); EXPECT_TRUE(payload_latch_.Await(kLongTimeout)); ASSERT_NE(user_b.GetPayload().AsStream(), nullptr); InputStream& rx = *user_b.GetPayload().AsStream(); @@ -343,11 +348,11 @@ TEST_P(OfflineServiceControllerTest, CanCancelStreamPayload) { return info.bytes_transferred >= size; }, kLongTimeout)); - EXPECT_EQ(rx.Read(Pipe::kChunkSize).result(), message); + EXPECT_EQ(rx.Read(kChunkSize).result(), message); user_b.CancelPayload(); absl::Time start_time = SystemClock::ElapsedRealtime(); while (true) { - if (!tx.Write(message).Ok()) break; + if (!tx->Write(message).Ok()) break; absl::Duration run_time = SystemClock::ElapsedRealtime() - start_time; if (run_time >= kLongTimeout) { EXPECT_LT(run_time, kLongTimeout); diff --git a/connections/implementation/payload_manager_test.cc b/connections/implementation/payload_manager_test.cc index 2c2c5f9f..f1984264 100644 --- a/connections/implementation/payload_manager_test.cc +++ b/connections/implementation/payload_manager_test.cc @@ -14,19 +14,29 @@ #include "connections/implementation/payload_manager.h" -#include "gmock/gmock.h" -#include "protobuf-matchers/protocol-buffer-matchers.h" +#include +#include +#include + #include "gtest/gtest.h" #include "absl/strings/string_view.h" +#include "absl/time/time.h" #include "connections/implementation/simulation_user.h" +#include "connections/listeners.h" +#include "connections/medium_selector.h" +#include "connections/payload.h" +#include "connections/status.h" #include "internal/platform/byte_array.h" +#include "internal/platform/count_down_latch.h" +#include "internal/platform/logging.h" +#include "internal/platform/medium_environment.h" #include "internal/platform/pipe.h" -#include "internal/platform/system_clock.h" namespace nearby { namespace connections { namespace { +constexpr size_t kChunkSize = 64 * 1024; constexpr absl::string_view kServiceId = "service-id"; constexpr absl::string_view kDeviceA = "device-a"; constexpr absl::string_view kDeviceB = "device-b"; @@ -165,18 +175,14 @@ TEST_P(PayloadManagerTest, CanSendStreamPayload) { PayloadSimulationUser user_b(kDeviceB, GetParam()); ASSERT_TRUE(SetupConnection(user_a, user_b)); - auto pipe = std::make_shared(); - OutputStream& tx = pipe->GetOutputStream(); - + auto [input, tx] = CreatePipe(); user_a.ExpectPayload(payload_latch_); const ByteArray message{std::string(kMessage)}; // The first write to the output stream will send the first PAYLOAD_TRANSFER // packet with payload info and message data. - tx.Write(message); + tx->Write(message); - user_b.SendPayload(Payload([pipe]() -> InputStream& { - return pipe->GetInputStream(); // NOLINT - })); + user_b.SendPayload(Payload(std::move(input))); ASSERT_TRUE(payload_latch_.Await(kDefaultTimeout).result()); ASSERT_NE(user_a.GetPayload().AsStream(), nullptr); InputStream& rx = *user_a.GetPayload().AsStream(); @@ -187,22 +193,22 @@ TEST_P(PayloadManagerTest, CanSendStreamPayload) { return info.bytes_transferred >= message.size(); }, kProgressTimeout)); - ByteArray result = rx.Read(Pipe::kChunkSize).result(); + ByteArray result = rx.Read(kChunkSize).result(); EXPECT_EQ(result, message); NEARBY_LOG(INFO, "Packet 1 handled."); - tx.Write(message); + tx->Write(message); EXPECT_TRUE(user_a.WaitForProgress( [&message](const PayloadProgressInfo& info) { return info.bytes_transferred >= 2 * message.size(); }, kProgressTimeout)); - ByteArray result2 = rx.Read(Pipe::kChunkSize).result(); + ByteArray result2 = rx.Read(kChunkSize).result(); EXPECT_EQ(result2, message); NEARBY_LOG(INFO, "Packet 2 handled."); rx.Close(); - tx.Close(); + tx->Close(); NEARBY_LOG(INFO, "Test completed."); user_a.Stop(); user_b.Stop(); @@ -214,17 +220,12 @@ TEST_P(PayloadManagerTest, CanCancelPayloadOnReceiverSide) { PayloadSimulationUser user_a(kDeviceA, GetParam()); PayloadSimulationUser user_b(kDeviceB, GetParam()); ASSERT_TRUE(SetupConnection(user_a, user_b)); - - auto pipe = std::make_shared(); - OutputStream& tx = pipe->GetOutputStream(); - + auto [input, tx] = CreatePipe(); user_a.ExpectPayload(payload_latch_); const ByteArray message{std::string(kMessage)}; - tx.Write(message); + tx->Write(message); - user_b.SendPayload(Payload([pipe]() -> InputStream& { - return pipe->GetInputStream(); // NOLINT - })); + user_b.SendPayload(Payload(std::move(input))); ASSERT_TRUE(payload_latch_.Await(kDefaultTimeout).result()); ASSERT_NE(user_a.GetPayload().AsStream(), nullptr); InputStream& rx = *user_a.GetPayload().AsStream(); @@ -235,7 +236,7 @@ TEST_P(PayloadManagerTest, CanCancelPayloadOnReceiverSide) { return info.bytes_transferred >= message.size(); }, kProgressTimeout)); - ByteArray result = rx.Read(Pipe::kChunkSize).result(); + ByteArray result = rx.Read(kChunkSize).result(); EXPECT_EQ(result, message); NEARBY_LOG(INFO, "Packet 1 handled."); @@ -246,7 +247,7 @@ TEST_P(PayloadManagerTest, CanCancelPayloadOnReceiverSide) { // Once cancel is handled, write will fail. int count = 0; while (true) { - if (!tx.Write(message).Ok()) break; + if (!tx->Write(message).Ok()) break; SystemClock::Sleep(kDefaultTimeout); count++; } @@ -258,7 +259,7 @@ TEST_P(PayloadManagerTest, CanCancelPayloadOnReceiverSide) { kProgressTimeout)); NEARBY_LOG(INFO, "Stream cancelation received."); - tx.Close(); + tx->Close(); rx.Close(); NEARBY_LOG(INFO, "Test completed."); @@ -272,17 +273,12 @@ TEST_P(PayloadManagerTest, CanCancelPayloadOnSenderSide) { PayloadSimulationUser user_a(kDeviceA, GetParam()); PayloadSimulationUser user_b(kDeviceB, GetParam()); ASSERT_TRUE(SetupConnection(user_a, user_b)); - - auto pipe = std::make_shared(); - OutputStream& tx = pipe->GetOutputStream(); - + auto [input, tx] = CreatePipe(); user_a.ExpectPayload(payload_latch_); const ByteArray message{std::string(kMessage)}; - tx.Write(message); + tx->Write(message); - user_b.SendPayload(Payload([pipe]() -> InputStream& { - return pipe->GetInputStream(); // NOLINT - })); + user_b.SendPayload(Payload(std::move(input))); ASSERT_TRUE(payload_latch_.Await(kDefaultTimeout).result()); ASSERT_NE(user_a.GetPayload().AsStream(), nullptr); InputStream& rx = *user_a.GetPayload().AsStream(); @@ -293,7 +289,7 @@ TEST_P(PayloadManagerTest, CanCancelPayloadOnSenderSide) { return info.bytes_transferred >= message.size(); }, kProgressTimeout)); - ByteArray result = rx.Read(Pipe::kChunkSize).result(); + ByteArray result = rx.Read(kChunkSize).result(); EXPECT_EQ(result, message); NEARBY_LOG(INFO, "Packet 1 handled."); @@ -304,7 +300,7 @@ TEST_P(PayloadManagerTest, CanCancelPayloadOnSenderSide) { // Once cancel is handled, write will fail. int count = 0; while (true) { - if (!tx.Write(message).Ok()) break; + if (!tx->Write(message).Ok()) break; SystemClock::Sleep(kDefaultTimeout); count++; } @@ -316,7 +312,7 @@ TEST_P(PayloadManagerTest, CanCancelPayloadOnSenderSide) { kProgressTimeout)); NEARBY_LOG(INFO, "Stream cancelation received."); - tx.Close(); + tx->Close(); rx.Close(); NEARBY_LOG(INFO, "Test completed."); @@ -331,19 +327,14 @@ TEST_P(PayloadManagerTest, SendPayloadWithSkip_StreamPayload) { PayloadSimulationUser user_a(kDeviceA, GetParam()); PayloadSimulationUser user_b(kDeviceB, GetParam()); ASSERT_TRUE(SetupConnection(user_a, user_b)); - - auto pipe = std::make_shared(); - OutputStream& tx = pipe->GetOutputStream(); - + auto [input, tx] = CreatePipe(); user_a.ExpectPayload(payload_latch_); const ByteArray message{std::string(kMessage)}; // The first write to the output stream will send the first PAYLOAD_TRANSFER // packet with payload info and message data. - tx.Write(message); + tx->Write(message); - Payload payload([pipe]() -> InputStream& { - return pipe->GetInputStream(); // NOLINT - }); + Payload payload(std::move(input)); payload.SetOffset(kOffset); user_b.SendPayload(std::move(payload)); ASSERT_TRUE(payload_latch_.Await(kDefaultTimeout).result()); @@ -356,22 +347,22 @@ TEST_P(PayloadManagerTest, SendPayloadWithSkip_StreamPayload) { return info.bytes_transferred >= message.size() - kOffset; }, kProgressTimeout)); - ByteArray result = rx.Read(Pipe::kChunkSize).result(); + ByteArray result = rx.Read(kChunkSize).result(); EXPECT_EQ(result, ByteArray("sage")); NEARBY_LOG(INFO, "Packet 1 handled."); - tx.Write(message); + tx->Write(message); EXPECT_TRUE(user_a.WaitForProgress( [&message](const PayloadProgressInfo& info) { return info.bytes_transferred >= 2 * message.size() - kOffset; }, kProgressTimeout)); - ByteArray result2 = rx.Read(Pipe::kChunkSize).result(); + ByteArray result2 = rx.Read(kChunkSize).result(); EXPECT_EQ(result2, message); NEARBY_LOG(INFO, "Packet 2 handled."); rx.Close(); - tx.Close(); + tx->Close(); NEARBY_LOG(INFO, "Test completed."); user_a.Stop(); user_b.Stop(); diff --git a/connections/payload.cc b/connections/payload.cc index 81cd9015..8b43133f 100644 --- a/connections/payload.cc +++ b/connections/payload.cc @@ -15,7 +15,18 @@ #include "connections/payload.h" #include +#include +#include +#include #include +#include +#include + +#include "connections/payload_type.h" +#include "internal/platform/byte_array.h" +#include "internal/platform/file.h" +#include "internal/platform/input_stream.h" +#include "internal/platform/prng.h" namespace nearby { namespace connections { @@ -43,8 +54,7 @@ Payload::~Payload() = default; Payload& Payload::operator=(Payload&& other) noexcept = default; // Default (invalid) payload. -Payload::Payload() - : type_(PayloadType::kUnknown), content_(absl::monostate()) {} +Payload::Payload() : type_(PayloadType::kUnknown), content_(std::monostate()) {} // Constructors for outgoing payloads. Payload::Payload(ByteArray&& bytes) @@ -73,7 +83,7 @@ Payload::Payload(std::string parent_folder, std::string file_name, type_(PayloadType::kFile), content_(std::move(input_file)) {} -Payload::Payload(std::function stream) +Payload::Payload(std::unique_ptr stream) : type_(PayloadType::kStream), content_(std::move(stream)) {} // Constructors for incoming payloads. @@ -91,22 +101,23 @@ Payload::Payload(Id id, std::string parent_folder, std::string file_name, type_(PayloadType::kFile), content_(std::move(input_file)) {} -Payload::Payload(Id id, std::function stream) +Payload::Payload(Id id, std::unique_ptr stream) : id_(id), type_(PayloadType::kStream), content_(std::move(stream)) {} // Returns ByteArray payload, if it has been defined, or empty ByteArray. const ByteArray& Payload::AsBytes() const& { static const ByteArray empty; // NOLINT: function-level static is OK. - auto* result = absl::get_if(&content_); + auto* result = std::get_if(&content_); return result ? *result : empty; } // Returns InputStream* payload, if it has been defined, or nullptr. InputStream* Payload::AsStream() { - auto* result = absl::get_if>(&content_); - return result ? &(*result)() : nullptr; + auto* result = std::get_if>(&content_); + return result ? result->get() : nullptr; } + // Returns InputFile* payload, if it has been defined, or nullptr. -InputFile* Payload::AsFile() { return absl::get_if(&content_); } +InputFile* Payload::AsFile() { return std::get_if(&content_); } // Returns Payload unique ID. Payload::Id Payload::GetId() const { return id_; } diff --git a/connections/payload.h b/connections/payload.h index f5a01d42..4c554828 100644 --- a/connections/payload.h +++ b/connections/payload.h @@ -19,6 +19,7 @@ #include #include #include +#include #include "absl/types/variant.h" #include "connections/payload_type.h" @@ -40,8 +41,8 @@ class Payload { using Id = PayloadId; // Order of types in variant, and values in Type enum is important. // Enum values must match respective variant types. - using Content = absl::variant, InputFile>; + using Content = std::variant, InputFile>; Payload(Payload&& other) noexcept; ~Payload(); @@ -71,7 +72,7 @@ class Payload { explicit Payload(std::string parent_folder, std::string file_name, InputFile file); - explicit Payload(std::function stream); + explicit Payload(std::unique_ptr stream); // Constructors for incoming payloads. Payload(Id id, ByteArray&& bytes); @@ -79,7 +80,7 @@ class Payload { Payload(Id id, InputFile file); Payload(Id id, std::string parent_folder, std::string file_name, InputFile input_file); - Payload(Id id, std::function stream); + Payload(Id id, std::unique_ptr stream); // Returns ByteArray payload, if it has been defined, or empty ByteArray. const ByteArray& AsBytes() const&; diff --git a/connections/payload_test.cc b/connections/payload_test.cc index 06d126cd..7a3aad02 100644 --- a/connections/payload_test.cc +++ b/connections/payload_test.cc @@ -16,6 +16,7 @@ #include #include +#include #include "gmock/gmock.h" #include "protobuf-matchers/protocol-buffer-matchers.h" @@ -95,19 +96,14 @@ TEST(PayloadTest, TEST(PayloadTest, SupportsStreamType) { constexpr size_t kOffset = 1234456; - auto pipe = std::make_shared(); + auto [input, output] = CreatePipe(); + InputStream* input_stream = input.get(); - Payload payload([streamable = pipe]() -> InputStream& { - // For some reason, linter warns us that we return a dangling reference. - // This is not true: we return a reference to internal variable of a - // shared_ptr which remains valid while Payload is valid, since - // shared_ptr is captured by value. - return streamable->GetInputStream(); // NOLINT - }); + Payload payload(std::move(input)); payload.SetOffset(kOffset); EXPECT_EQ(payload.GetType(), PayloadType::kStream); - EXPECT_EQ(payload.AsStream(), &pipe->GetInputStream()); + EXPECT_EQ(payload.AsStream(), input_stream); EXPECT_EQ(payload.AsFile(), nullptr); EXPECT_EQ(payload.AsBytes(), ByteArray{}); EXPECT_EQ(payload.GetOffset(), kOffset); diff --git a/connections/swift/NearbyCoreAdapter/BUILD b/connections/swift/NearbyCoreAdapter/BUILD index d92ec9c2..e6103c51 100644 --- a/connections/swift/NearbyCoreAdapter/BUILD +++ b/connections/swift/NearbyCoreAdapter/BUILD @@ -36,7 +36,6 @@ objc_library( "//internal/platform:base", "//internal/platform/implementation/apple", # buildcleaner: keep "//third_party/apple_frameworks:Foundation", - "//third_party/apple_frameworks:ObjectiveC", "//third_party/objective_c/google_toolbox_for_mac:GTM_Logger", ], ) diff --git a/connections/swift/NearbyCoreAdapter/Sources/CPPInputStream.h b/connections/swift/NearbyCoreAdapter/Sources/CPPInputStream.h new file mode 100644 index 00000000..67be0b85 --- /dev/null +++ b/connections/swift/NearbyCoreAdapter/Sources/CPPInputStream.h @@ -0,0 +1,45 @@ +// Copyright 2022 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Note: File language is detected using heuristics. Many Objective-C++ headers +// are incorrectly classified as C++ resulting in invalid linter errors. The use +// of "NSArray" and other Foundation classes like "NSData", "NSDictionary" and +// "NSUUID" are highly weighted for Objective-C and Objective-C++ scores. Oddly, +// "#import " does not contribute any points. This +// comment alone should be enough to trick the IDE in to believing this is +// actually some sort of Objective-C file. See: +// cs/google3/devtools/search/lang/recognize_language_classifiers_data + +#import + +#ifdef __cplusplus + +// TODO(b/239758418): Change this to the non-internal version when available. +#include "internal/platform/input_stream.h" + +class CPPInputStream : public nearby::InputStream { + public: + explicit CPPInputStream(NSInputStream *iStream); + + ~CPPInputStream() override; + + nearby::ExceptionOr Read(std::int64_t size) override; + + nearby::Exception Close() override; + + private: + NSInputStream *iStream_; +}; + +#endif diff --git a/connections/swift/NearbyCoreAdapter/Sources/CPPInputStream.mm b/connections/swift/NearbyCoreAdapter/Sources/CPPInputStream.mm new file mode 100644 index 00000000..a3fe9728 --- /dev/null +++ b/connections/swift/NearbyCoreAdapter/Sources/CPPInputStream.mm @@ -0,0 +1,46 @@ +// Copyright 2022 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#import "connections/swift/NearbyCoreAdapter/Sources/CPPInputStream.h" + +#import + +#include +#include + +using ::nearby::ByteArray; +using ::nearby::Exception; +using ::nearby::ExceptionOr; + +CPPInputStream::CPPInputStream(NSInputStream *iStream) : iStream_(iStream) { [iStream_ open]; } + +CPPInputStream::~CPPInputStream() { Close(); } + +ExceptionOr CPPInputStream::Read(std::int64_t size) { + std::vector buffer; + buffer.reserve(size); + NSInteger numberOfBytesRead = [iStream_ read:buffer.data() maxLength:size]; + if (numberOfBytesRead == 0) { + return ExceptionOr(); + } + if (numberOfBytesRead < 0) { + return ExceptionOr(Exception::kIo); + } + return ExceptionOr(ByteArray((const char *)buffer.data(), numberOfBytesRead)); +} + +Exception CPPInputStream::Close() { + [iStream_ close]; + return Exception{Exception::kSuccess}; +} diff --git a/connections/swift/NearbyCoreAdapter/Sources/CPPInputStreamBinding.h b/connections/swift/NearbyCoreAdapter/Sources/CPPInputStreamBinding.h deleted file mode 100644 index 06aac0d9..00000000 --- a/connections/swift/NearbyCoreAdapter/Sources/CPPInputStreamBinding.h +++ /dev/null @@ -1,50 +0,0 @@ -// Copyright 2022 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// https://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#import - -#ifdef __cplusplus - -namespace nearby { -class InputStream; -} - -#endif - -@interface CPPInputStreamBinding : NSObject - -/** - * Creates and attaches a @c CPPInputStreamBinding object to the provided stream as an associated - * object. - * - * This association makes it possible for a c++ pointer to live as long as the original user - * provided @c NSInputStream object. - * - * @param stream The stream to become associated with. - */ -+ (void)bindToStream:(NSInputStream *)stream; - -#ifdef __cplusplus -/** - * Retreives a reference to the c++ pointer associated to the provided stream. - * - * CPPInputStreamBinding::bindToStream: must be called on the this stream before calling this - * function. - * - * @param stream The stream that has a c++ pointer associated with it. - */ -+ (nearby::InputStream &)getRefFromStream:(NSInputStream *)stream; -#endif - -@end diff --git a/connections/swift/NearbyCoreAdapter/Sources/CPPInputStreamBinding.mm b/connections/swift/NearbyCoreAdapter/Sources/CPPInputStreamBinding.mm deleted file mode 100644 index d546ba5c..00000000 --- a/connections/swift/NearbyCoreAdapter/Sources/CPPInputStreamBinding.mm +++ /dev/null @@ -1,90 +0,0 @@ -// Copyright 2022 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// https://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#import "connections/swift/NearbyCoreAdapter/Sources/CPPInputStreamBinding.h" - -#import -#import - -#include -#include - -// TODO(b/239758418): Change this to the non-internal version when available. -#include "internal/platform/input_stream.h" - -using ::nearby::ByteArray; -using ::nearby::Exception; -using ::nearby::ExceptionOr; -using ::nearby::InputStream; - -class CPPInputStream : public InputStream { - public: - explicit CPPInputStream(NSInputStream *iStream) : iStream_(iStream) { [iStream_ open]; } - - ~CPPInputStream() override { Close(); } - - ExceptionOr Read(std::int64_t size) override { - std::vector buffer; - buffer.reserve(size); - NSInteger numberOfBytesRead = [iStream_ read:buffer.data() maxLength:size]; - if (numberOfBytesRead == 0) { - return ExceptionOr(); - } - if (numberOfBytesRead < 0) { - return ExceptionOr(Exception::kIo); - } - return ExceptionOr(ByteArray((const char *)buffer.data(), numberOfBytesRead)); - } - - Exception Close() override { - [iStream_ close]; - return Exception{Exception::kSuccess}; - } - - private: - // Prevent a retain cycle since NSInputStream will have a strong reference to this object. - __weak NSInputStream *iStream_; -}; - -// Wrap a c++ InputStream subclass in objective-c so it can be associated with the original -// NSInputStream object. The association makes it possible for the unique_ptr to live as long as -// the original user provided NSInputStream. -@implementation CPPInputStreamBinding { - @public - std::unique_ptr _cppStream; -} - -// This field's address is used as a unique identifier. -static char gAssociatedStreamKey; - -- (instancetype)initWithStream:(NSInputStream *)stream { - self = [super init]; - if (self) { - _cppStream = std::make_unique(stream); - } - return self; -} - -+ (void)bindToStream:(NSInputStream *)stream { - CPPInputStreamBinding *binding = [[CPPInputStreamBinding alloc] initWithStream:stream]; - objc_setAssociatedObject(stream, &gAssociatedStreamKey, binding, - OBJC_ASSOCIATION_RETAIN_NONATOMIC); -} - -+ (InputStream &)getRefFromStream:(NSInputStream *)stream { - CPPInputStreamBinding *binding = objc_getAssociatedObject(stream, &gAssociatedStreamKey); - return *binding->_cppStream; -} - -@end diff --git a/connections/swift/NearbyCoreAdapter/Sources/GNCInputStream.h b/connections/swift/NearbyCoreAdapter/Sources/GNCInputStream.h index bdd698e1..151a1a52 100644 --- a/connections/swift/NearbyCoreAdapter/Sources/GNCInputStream.h +++ b/connections/swift/NearbyCoreAdapter/Sources/GNCInputStream.h @@ -17,8 +17,10 @@ #ifdef __cplusplus namespace nearby { -class InputStream; +namespace connections { +class Payload; } +} // namespace nearby #endif @@ -29,7 +31,7 @@ class InputStream; - (nonnull instancetype)initWithURL:(nonnull NSURL *)url NS_UNAVAILABLE; #ifdef __cplusplus -- (nonnull instancetype)initWithCppInputStream:(nonnull nearby::InputStream *)stream +- (nonnull instancetype)initWithPayload:(nearby::connections::Payload)payload NS_DESIGNATED_INITIALIZER; #endif diff --git a/connections/swift/NearbyCoreAdapter/Sources/GNCInputStream.mm b/connections/swift/NearbyCoreAdapter/Sources/GNCInputStream.mm index d4c3096f..2b697a5f 100644 --- a/connections/swift/NearbyCoreAdapter/Sources/GNCInputStream.mm +++ b/connections/swift/NearbyCoreAdapter/Sources/GNCInputStream.mm @@ -18,31 +18,30 @@ #include -// TODO(b/239758418): Change this to the non-internal version when available. -#include "internal/platform/input_stream.h" +#include "connections/payload.h" #import "connections/swift/NearbyCoreAdapter/Sources/GNCException+Internal.h" #import "connections/swift/NearbyCoreAdapter/Sources/Public/NearbyCoreAdapter/GNCException.h" using ::nearby::ByteArray; using ::nearby::ExceptionOr; -using ::nearby::InputStream; using ::nearby::connections::NSErrorFromCppException; +using ::nearby::connections::Payload; @implementation GNCInputStream { NSStreamStatus _streamStatus; NSError *_streamError; id _delegate; - InputStream *_stream; + Payload _payload; } -- (instancetype)initWithCppInputStream:(InputStream *)stream { +- (instancetype)initWithPayload:(Payload)payload { // Init with empty data because init is not a designated initializer. self = [super initWithData:[[NSData alloc] init]]; if (self) { _streamStatus = NSStreamStatusNotOpen; _delegate = self; - _stream = stream; + _payload = std::move(payload); } return self; @@ -53,7 +52,7 @@ using ::nearby::connections::NSErrorFromCppException; } - (NSInteger)read:(uint8_t *)buffer maxLength:(NSUInteger)maxLen { - ExceptionOr readResult = _stream->Read(maxLen); + ExceptionOr readResult = _payload.AsStream()->Read(maxLen); if (!readResult.ok()) { _streamError = NSErrorFromCppException(readResult.GetException()); @@ -87,7 +86,7 @@ using ::nearby::connections::NSErrorFromCppException; - (void)close { _streamStatus = NSStreamStatusClosed; - _stream->Close(); + _payload.AsStream()->Close(); } - (NSStreamStatus)streamStatus { diff --git a/connections/swift/NearbyCoreAdapter/Sources/GNCPayload+CppConversions.h b/connections/swift/NearbyCoreAdapter/Sources/GNCPayload+CppConversions.h index 7d265e97..97e54e1f 100644 --- a/connections/swift/NearbyCoreAdapter/Sources/GNCPayload+CppConversions.h +++ b/connections/swift/NearbyCoreAdapter/Sources/GNCPayload+CppConversions.h @@ -29,10 +29,20 @@ class Payload; @interface GNCPayload (CppConversions) #ifdef __cplusplus +/** + * @note @c fromCpp should not be used to convert a @c Payload created from @c toCpp. For some + * payload types, each conversion creates a new object holding a reference to the previous, + * resulting in potentially endless nesting of objects. + */ + (nonnull GNCPayload *)fromCpp:(nearby::connections::Payload)payload; #endif #ifdef __cplusplus +/** + * @note @c toCPP should not be used to convert a @c GNCPayload created from @c fromCpp. For some + * payload types, each conversion creates a new object holding a reference to the previous, + * resulting in potentially endless nesting of objects. + */ - (nearby::connections::Payload)toCpp; #endif diff --git a/connections/swift/NearbyCoreAdapter/Sources/GNCPayload+CppConversions.mm b/connections/swift/NearbyCoreAdapter/Sources/GNCPayload+CppConversions.mm index 71956a9b..6cc8aa4b 100644 --- a/connections/swift/NearbyCoreAdapter/Sources/GNCPayload+CppConversions.mm +++ b/connections/swift/NearbyCoreAdapter/Sources/GNCPayload+CppConversions.mm @@ -20,13 +20,12 @@ #include "connections/payload.h" -#import "connections/swift/NearbyCoreAdapter/Sources/CPPInputStreamBinding.h" +#import "connections/swift/NearbyCoreAdapter/Sources/CPPInputStream.h" #import "connections/swift/NearbyCoreAdapter/Sources/GNCInputStream.h" #import "connections/swift/NearbyCoreAdapter/Sources/GNCPayload+CppConversions.h" using ::nearby::ByteArray; using ::nearby::InputFile; -using ::nearby::InputStream; using ::nearby::connections::Payload; @implementation GNCPayload (CppConversions) @@ -51,7 +50,7 @@ using ::nearby::connections::Payload; identifier:payloadId]; } case nearby::connections::PayloadType::kStream: { - GNCInputStream *stream = [[GNCInputStream alloc] initWithCppInputStream:payload.AsStream()]; + GNCInputStream *stream = [[GNCInputStream alloc] initWithPayload:std::move(payload)]; return [[GNCStreamPayload alloc] initWithStream:stream identifier:payloadId]; } case nearby::connections::PayloadType::kUnknown: @@ -76,14 +75,7 @@ using ::nearby::connections::Payload; @implementation GNCStreamPayload (CppConversions) - (Payload)toCpp { - // GNCStreamPayload will most likely be destroyed almost immediately, so a weak self would be - // useless and a strong self will cause a retain cycle. This is why we are keeping a weak - // reference of the stream instead. The input stream should be kept alive by a strong reference - // on the user end. - __weak NSInputStream *stream = self.stream; - return Payload(self.identifier, [stream]() -> InputStream & { - return [CPPInputStreamBinding getRefFromStream:stream]; - }); + return Payload(self.identifier, std::make_unique(self.stream)); } @end diff --git a/connections/swift/NearbyCoreAdapter/Sources/GNCPayload.mm b/connections/swift/NearbyCoreAdapter/Sources/GNCPayload.mm index 7c8415a2..58d6c032 100644 --- a/connections/swift/NearbyCoreAdapter/Sources/GNCPayload.mm +++ b/connections/swift/NearbyCoreAdapter/Sources/GNCPayload.mm @@ -20,8 +20,6 @@ #include "connections/payload.h" -#import "connections/swift/NearbyCoreAdapter/Sources/CPPInputStreamBinding.h" - using ::nearby::connections::Payload; @implementation GNCPayload @@ -62,7 +60,6 @@ using ::nearby::connections::Payload; self = [super initWithIdentifier:identifier]; if (self) { _stream = stream; - [CPPInputStreamBinding bindToStream:stream]; } return self; } diff --git a/internal/platform/BUILD b/internal/platform/BUILD index e839fdbd..20c15bfd 100644 --- a/internal/platform/BUILD +++ b/internal/platform/BUILD @@ -73,13 +73,11 @@ cc_library( name = "util", srcs = [ "base_input_stream.cc", - "base_pipe.cc", "byte_utils.cc", ], hdrs = [ "base_input_stream.h", "base_mutex_lock.h", - "base_pipe.h", "byte_utils.h", ], visibility = [ @@ -465,6 +463,7 @@ cc_test( shard_count = 16, deps = [ ":base", + ":cancellation_flag", ":comm", ":connection_info", ":test_util", diff --git a/internal/platform/base_pipe.cc b/internal/platform/base_pipe.cc deleted file mode 100644 index 94c6b917..00000000 --- a/internal/platform/base_pipe.cc +++ /dev/null @@ -1,103 +0,0 @@ -// Copyright 2020 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// https://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#include "internal/platform/base_pipe.h" - -#include "internal/platform/base_mutex_lock.h" -#include "internal/platform/input_stream.h" -#include "internal/platform/output_stream.h" - -namespace nearby { - -ExceptionOr BasePipe::Read(size_t size) { - BaseMutexLock lock(mutex_.get()); - - // We're done reading all the chunks that were written before the OutputStream - // was closed, so there's nothing to do here other than return an empty chunk - // to serve as an EOF indication to callers. - if (read_all_chunks_) { - return ExceptionOr{ByteArray{}}; - } - - while (buffer_.empty() && !input_stream_closed_) { - Exception wait_exception = cond_->Wait(); - - if (wait_exception.Raised()) { - return ExceptionOr{wait_exception}; - } - } - - // If we received our sentinel chunk, mark the fact that there cannot - // possibly be any more chunks to read here on in, and return an empty chunk - // to serve as an EOF indication to callers. - if (buffer_.empty() || buffer_.front().Empty()) { - read_all_chunks_ = true; - return ExceptionOr{ByteArray{}}; - } - - ByteArray first_chunk{buffer_.front()}; - buffer_.pop_front(); - - // If first_chunk is small enough to not overshoot the requested 'size', just - // return that. - if (first_chunk.size() <= size) { - return ExceptionOr{first_chunk}; - } else { - // Break first_chunk into 2 parts -- the first one of which (next_chunk) - // will be 'size' bytes long, and will be returned, and the second one of - // which (overflow_chunk) will be re-inserted into buffer_, at the head of - // the queue, to be served up in the next call to read(). - ByteArray next_chunk(first_chunk.data(), size); - buffer_.push_front( - ByteArray(first_chunk.data() + size, first_chunk.size() - size)); - return ExceptionOr{next_chunk}; - } -} - -Exception BasePipe::Write(const ByteArray& data) { - BaseMutexLock lock(mutex_.get()); - - return WriteLocked(data); -} - -void BasePipe::MarkInputStreamClosed() { - BaseMutexLock lock(mutex_.get()); - - input_stream_closed_ = true; - // Trigger cond_ to unblock a potentially-blocked call to read(), and to let - // it know to return Exception::IO. - cond_->Notify(); -} - -void BasePipe::MarkOutputStreamClosed() { - BaseMutexLock lock(mutex_.get()); - - // Write a sentinel null chunk before marking output_stream_closed as true. - WriteLocked(ByteArray{}); - output_stream_closed_ = true; -} - -Exception BasePipe::WriteLocked(const ByteArray& data) { - if (input_stream_closed_ || output_stream_closed_) { - return {Exception::kIo}; - } - - buffer_.push_back(data); - // Trigger cond_ to unblock a potentially-blocked call to read(), now that - // there's more data for it to consume. - cond_->Notify(); - return {Exception::kSuccess}; -} - -} // namespace nearby diff --git a/internal/platform/base_pipe.h b/internal/platform/base_pipe.h deleted file mode 100644 index 3b8987de..00000000 --- a/internal/platform/base_pipe.h +++ /dev/null @@ -1,136 +0,0 @@ -// Copyright 2020 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// https://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#ifndef PLATFORM_BASE_BASE_PIPE_H_ -#define PLATFORM_BASE_BASE_PIPE_H_ - -#include -#include -#include - -#include "absl/base/thread_annotations.h" -#include "internal/platform/implementation/condition_variable.h" -#include "internal/platform/implementation/mutex.h" -#include "internal/platform/byte_array.h" -#include "internal/platform/exception.h" -#include "internal/platform/input_stream.h" -#include "internal/platform/output_stream.h" - -namespace nearby { - -// Common Pipe implementation. -// It does not depend on platform implementation, and this allows it to -// be used in the platform implementation itself. -// Concrete class must be derived from it, as follows: -// -// class DerivedPipe : public BasePipe { -// public: -// DerivedPipe() { -// auto mutex = /* construct platform-dependent mutex */; -// auto cond = /* construct platform-dependent condition variable */; -// Setup(std::move(mutex), std::move(cond)); -// } -// ~DerivedPipe() override = default; -// DerivedPipe(DerivedPipe&&) = default; -// DerivedPipe& operator=(DerivedPipe&&) = default; -// }; -class BasePipe { - public: - static constexpr const size_t kChunkSize = 64 * 1024; - virtual ~BasePipe() = default; - - // Pipe is not copyable or movable, because copy/move will invalidate - // references to input and output streams. - // If move is required, Pipe could be wrapped with std::unique_ptr<>. - BasePipe(BasePipe&&) = delete; - BasePipe& operator=(BasePipe&&) = delete; - - // Get...() methods return references to input and output steam facades. - // It is safe to call Get...() methods multiple times. - InputStream& GetInputStream() { return input_stream_; } - OutputStream& GetOutputStream() { return output_stream_; } - - protected: - BasePipe() = default; - - void Setup(std::unique_ptr mutex, - std::unique_ptr cond) { - mutex_ = std::move(mutex); - cond_ = std::move(cond); - } - - private: - class BasePipeInputStream : public InputStream { - public: - explicit BasePipeInputStream(BasePipe* pipe) : pipe_(pipe) {} - ~BasePipeInputStream() override { DoClose(); } - - ExceptionOr Read(std::int64_t size) override { - return pipe_->Read(size); - } - Exception Close() override { return DoClose(); } - - private: - Exception DoClose() { - pipe_->MarkInputStreamClosed(); - return {Exception::kSuccess}; - } - BasePipe* pipe_; - }; - class BasePipeOutputStream : public OutputStream { - public: - explicit BasePipeOutputStream(BasePipe* pipe) : pipe_(pipe) {} - ~BasePipeOutputStream() override { DoClose(); } - - Exception Write(const ByteArray& data) override { - return pipe_->Write(data); - } - Exception Flush() override { return {Exception::kSuccess}; } - Exception Close() override { return DoClose(); } - - private: - Exception DoClose() { - pipe_->MarkOutputStreamClosed(); - return {Exception::kSuccess}; - } - BasePipe* pipe_; - }; - - ExceptionOr Read(size_t size) ABSL_LOCKS_EXCLUDED(mutex_); - Exception Write(const ByteArray& data) ABSL_LOCKS_EXCLUDED(mutex_); - - void MarkInputStreamClosed() ABSL_LOCKS_EXCLUDED(mutex_); - void MarkOutputStreamClosed() ABSL_LOCKS_EXCLUDED(mutex_); - - Exception WriteLocked(const ByteArray& data) - ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - - // Order of declaration matters: - // - mutex must be defined before condvar; - // - input & output streams must be after both mutex and condvar. - bool input_stream_closed_ ABSL_GUARDED_BY(mutex_) = false; - bool output_stream_closed_ ABSL_GUARDED_BY(mutex_) = false; - bool read_all_chunks_ ABSL_GUARDED_BY(mutex_) = false; - - std::deque ABSL_GUARDED_BY(mutex_) buffer_; - std::unique_ptr mutex_; - std::unique_ptr cond_; - - BasePipeInputStream input_stream_{this}; - BasePipeOutputStream output_stream_{this}; -}; - -} // namespace nearby - -#endif // PLATFORM_BASE_BASE_PIPE_H_ diff --git a/internal/platform/bluetooth_classic_test.cc b/internal/platform/bluetooth_classic_test.cc index 12d97989..3eabdb03 100644 --- a/internal/platform/bluetooth_classic_test.cc +++ b/internal/platform/bluetooth_classic_test.cc @@ -18,12 +18,16 @@ #include #include -#include "gmock/gmock.h" -#include "protobuf-matchers/protocol-buffer-matchers.h" #include "gtest/gtest.h" +#include "absl/strings/string_view.h" #include "absl/time/time.h" #include "internal/platform/bluetooth_adapter.h" +#include "internal/platform/byte_array.h" +#include "internal/platform/cancellation_flag.h" #include "internal/platform/count_down_latch.h" +#include "internal/platform/exception.h" +#include "internal/platform/feature_flags.h" +#include "internal/platform/implementation/bluetooth_adapter.h" #include "internal/platform/implementation/bluetooth_classic.h" #include "internal/platform/logging.h" #include "internal/platform/medium_environment.h" @@ -270,7 +274,7 @@ TEST_F(BluetoothClassicMediumTest, SendData) { server_socket.Close(); } -TEST_F(BluetoothClassicMediumTest, IoOnClosedSocketReturnsError) { +TEST_F(BluetoothClassicMediumTest, IoOnClosedSocketReturnsEmpty) { adapter_a_->SetScanMode(BluetoothAdapter::ScanMode::kConnectable); CountDownLatch found_latch(1); BluetoothDevice* discovered_device = nullptr; @@ -308,7 +312,7 @@ TEST_F(BluetoothClassicMediumTest, IoOnClosedSocketReturnsError) { BluetoothSocket socket_b = server_socket.Accept(); ASSERT_TRUE(socket_b.IsValid()); socket_b.Close(); - EXPECT_FALSE(socket_b.GetInputStream().Read(data.size()).ok()); + EXPECT_TRUE(socket_b.GetInputStream().Read(data.size()).result().Empty()); }); } server_socket.Close(); diff --git a/internal/platform/implementation/g3/BUILD b/internal/platform/implementation/g3/BUILD index 02f45c6c..9a4359d6 100644 --- a/internal/platform/implementation/g3/BUILD +++ b/internal/platform/implementation/g3/BUILD @@ -30,7 +30,6 @@ cc_library( "log_message.h", "multi_thread_executor.h", "mutex.h", - "pipe.h", "preferences_manager.h", "scheduled_executor.h", "single_thread_executor.h", @@ -98,7 +97,10 @@ cc_library( "@com_google_absl//absl/base:core_headers", "@com_google_absl//absl/container:flat_hash_map", "@com_google_absl//absl/container:flat_hash_set", + "@com_google_absl//absl/functional:any_invocable", + "@com_google_absl//absl/log:check", "@com_google_absl//absl/status", + "@com_google_absl//absl/status:statusor", "@com_google_absl//absl/strings", "@com_google_absl//absl/strings:str_format", "@com_google_absl//absl/synchronization", diff --git a/internal/platform/implementation/g3/ble.cc b/internal/platform/implementation/g3/ble.cc index 310d19be..f60b2391 100644 --- a/internal/platform/implementation/g3/ble.cc +++ b/internal/platform/implementation/g3/ble.cc @@ -17,10 +17,18 @@ #include #include #include +#include +#include "absl/functional/any_invocable.h" +#include "absl/log/check.h" #include "absl/synchronization/mutex.h" +#include "internal/platform/byte_array.h" +#include "internal/platform/cancellation_flag.h" #include "internal/platform/cancellation_flag_listener.h" +#include "internal/platform/exception.h" #include "internal/platform/implementation/ble.h" +#include "internal/platform/implementation/bluetooth_adapter.h" +#include "internal/platform/implementation/g3/bluetooth_adapter.h" #include "internal/platform/implementation/shared/count_down_latch.h" #include "internal/platform/logging.h" #include "internal/platform/medium_environment.h" diff --git a/internal/platform/implementation/g3/ble.h b/internal/platform/implementation/g3/ble.h index 78736ef8..243468ff 100644 --- a/internal/platform/implementation/g3/ble.h +++ b/internal/platform/implementation/g3/ble.h @@ -27,7 +27,6 @@ #include "internal/platform/implementation/g3/bluetooth_adapter.h" #include "internal/platform/implementation/g3/bluetooth_classic.h" #include "internal/platform/implementation/g3/multi_thread_executor.h" -#include "internal/platform/implementation/g3/pipe.h" #include "internal/platform/implementation/g3/socket_base.h" #include "internal/platform/input_stream.h" #include "internal/platform/output_stream.h" diff --git a/internal/platform/implementation/g3/ble_v2.cc b/internal/platform/implementation/g3/ble_v2.cc index f0fcf339..dce7d554 100644 --- a/internal/platform/implementation/g3/ble_v2.cc +++ b/internal/platform/implementation/g3/ble_v2.cc @@ -15,7 +15,6 @@ #include "internal/platform/implementation/g3/ble_v2.h" #include -#include #include #include #include @@ -23,15 +22,26 @@ #include #include +#include "absl/functional/any_invocable.h" +#include "absl/log/check.h" #include "absl/status/status.h" +#include "absl/status/statusor.h" #include "absl/strings/escaping.h" +#include "absl/strings/str_cat.h" +#include "absl/strings/string_view.h" #include "absl/synchronization/mutex.h" +#include "internal/platform/borrowable.h" +#include "internal/platform/byte_array.h" #include "internal/platform/cancellation_flag_listener.h" #include "internal/platform/count_down_latch.h" +#include "internal/platform/exception.h" #include "internal/platform/implementation/ble_v2.h" +#include "internal/platform/implementation/bluetooth_adapter.h" #include "internal/platform/implementation/g3/bluetooth_adapter.h" #include "internal/platform/logging.h" #include "internal/platform/medium_environment.h" +#include "internal/platform/prng.h" +#include "internal/platform/uuid.h" namespace nearby { namespace g3 { diff --git a/internal/platform/implementation/g3/ble_v2.h b/internal/platform/implementation/g3/ble_v2.h index 4b273a46..2165c014 100644 --- a/internal/platform/implementation/g3/ble_v2.h +++ b/internal/platform/implementation/g3/ble_v2.h @@ -29,7 +29,6 @@ #include "internal/platform/byte_array.h" #include "internal/platform/implementation/ble_v2.h" #include "internal/platform/implementation/g3/bluetooth_adapter.h" -#include "internal/platform/implementation/g3/pipe.h" #include "internal/platform/implementation/g3/socket_base.h" #include "internal/platform/medium_environment.h" #include "internal/platform/prng.h" diff --git a/internal/platform/implementation/g3/bluetooth_classic.cc b/internal/platform/implementation/g3/bluetooth_classic.cc index d7a1a832..a4d7b121 100644 --- a/internal/platform/implementation/g3/bluetooth_classic.cc +++ b/internal/platform/implementation/g3/bluetooth_classic.cc @@ -17,12 +17,18 @@ #include #include #include +#include +#include "absl/functional/any_invocable.h" +#include "absl/log/check.h" +#include "absl/strings/string_view.h" #include "absl/synchronization/mutex.h" +#include "internal/platform/cancellation_flag.h" #include "internal/platform/cancellation_flag_listener.h" +#include "internal/platform/exception.h" +#include "internal/platform/implementation/bluetooth_adapter.h" #include "internal/platform/implementation/bluetooth_classic.h" #include "internal/platform/implementation/g3/bluetooth_adapter.h" -#include "internal/platform/input_stream.h" #include "internal/platform/logging.h" #include "internal/platform/medium_environment.h" diff --git a/internal/platform/implementation/g3/bluetooth_classic.h b/internal/platform/implementation/g3/bluetooth_classic.h index 6bae1fa3..9c4d4257 100644 --- a/internal/platform/implementation/g3/bluetooth_classic.h +++ b/internal/platform/implementation/g3/bluetooth_classic.h @@ -26,7 +26,6 @@ #include "internal/platform/exception.h" #include "internal/platform/implementation/bluetooth_classic.h" #include "internal/platform/implementation/g3/bluetooth_adapter.h" -#include "internal/platform/implementation/g3/pipe.h" #include "internal/platform/implementation/g3/socket_base.h" #include "internal/platform/input_stream.h" #include "internal/platform/listeners.h" diff --git a/internal/platform/implementation/g3/pipe.h b/internal/platform/implementation/g3/pipe.h deleted file mode 100644 index 3c2f7f24..00000000 --- a/internal/platform/implementation/g3/pipe.h +++ /dev/null @@ -1,42 +0,0 @@ -// Copyright 2020 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// https://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#ifndef PLATFORM_IMPL_G3_PIPE_H_ -#define PLATFORM_IMPL_G3_PIPE_H_ - -#include - -#include "internal/platform/base_pipe.h" -#include "internal/platform/implementation/g3/condition_variable.h" -#include "internal/platform/implementation/g3/mutex.h" - -namespace nearby { -namespace g3 { - -class Pipe : public BasePipe { - public: - Pipe() { - auto mutex = std::make_unique(/*check=*/true); - auto cond = std::make_unique(mutex.get()); - Setup(std::move(mutex), std::move(cond)); - } - ~Pipe() override = default; - Pipe(Pipe&&) = delete; - Pipe& operator=(Pipe&&) = delete; -}; - -} // namespace g3 -} // namespace nearby - -#endif // PLATFORM_IMPL_G3_PIPE_H_ diff --git a/internal/platform/implementation/g3/socket_base.h b/internal/platform/implementation/g3/socket_base.h index 4da305b9..7678a9e6 100644 --- a/internal/platform/implementation/g3/socket_base.h +++ b/internal/platform/implementation/g3/socket_base.h @@ -18,14 +18,16 @@ #include #include #include +#include +#include #include "absl/base/thread_annotations.h" #include "absl/synchronization/mutex.h" #include "internal/platform/byte_array.h" #include "internal/platform/exception.h" -#include "internal/platform/implementation/g3/pipe.h" #include "internal/platform/input_stream.h" #include "internal/platform/output_stream.h" +#include "internal/platform/pipe.h" namespace nearby { namespace g3 { @@ -33,6 +35,7 @@ namespace g3 { // Common base for BT, BLE and Wifi socket implementations. class SocketBase { public: + SocketBase() { std::tie(input_for_remote_, output_) = CreatePipe(); } virtual ~SocketBase() { absl::MutexLock lock(&mutex_); DoClose(); @@ -43,23 +46,17 @@ class SocketBase { void Connect(SocketBase& other) ABSL_LOCKS_EXCLUDED(mutex_) { absl::MutexLock lock(&mutex_); remote_socket_ = &other; - input_ = other.output_; + input_ = std::move(other.input_for_remote_); } // Returns the InputStream of this connected socket. - InputStream& GetInputStream() ABSL_LOCKS_EXCLUDED(mutex_) { - absl::MutexLock lock(&mutex_); - if (IsConnectedLocked()) { - return input_->GetInputStream(); - } - return invalid_input_stream_; - } + InputStream& GetInputStream() { return input_proxy_; } // Returns the OutputStream of this connected socket. // This stream is for local side to write. OutputStream& GetOutputStream() ABSL_LOCKS_EXCLUDED(mutex_) { absl::MutexLock lock(&mutex_); - return output_->GetOutputStream(); + return *output_; } // Returns true if connection exists to the (possibly closed) remote socket. @@ -95,13 +92,15 @@ class SocketBase { void DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_) { if (!closed_) { remote_socket_ = nullptr; - output_->GetOutputStream().Close(); - output_->GetInputStream().Close(); - if (IsConnectedLocked()) { - input_->GetOutputStream().Close(); - input_->GetInputStream().Close(); - input_.reset(); + // The client can hold references to `output_` and `input_` streams. We + // can close them but we cannot destroy them. + output_->Close(); + if (input_) { + input_->Close(); } + // The client does not hold a reference to `input_for_remote_`, so we can + // destroy it. Connecting to this socket will fail after that. + input_for_remote_.reset(); closed_ = true; } } @@ -111,24 +110,42 @@ class SocketBase { return input_ != nullptr; } - class InvalidInputStream : public InputStream { + class InputProxyStream : public InputStream { public: + explicit InputProxyStream(SocketBase* socket) : socket_(socket) {} ExceptionOr Read(std::int64_t size) override { - return ExceptionOr(Exception::kIo); + if (!socket_->IsConnected()) { + return ExceptionOr(Exception::kIo); + } + return socket_->input_->Read(size); } ExceptionOr Skip(size_t offset) override { - return ExceptionOr(Exception::kIo); + if (!socket_->IsConnected()) { + return ExceptionOr(Exception::kIo); + } + return socket_->input_->Skip(offset); + } + Exception Close() override { + if (!socket_->IsConnected()) { + return {Exception::kIo}; + } + return socket_->input_->Close(); } - Exception Close() override { return {Exception::kIo}; } - }; - // Returned to the caller if the remote socket is destroyed. - InvalidInputStream invalid_input_stream_; - // Output pipe is initialized by constructor, it remains always valid, until - // it is closed. it represents output part of a local socket. Input part of a - // local socket comes from the peer socket, after connection. - std::shared_ptr output_{new Pipe}; - std::shared_ptr input_; + private: + SocketBase* socket_; + }; + InputProxyStream input_proxy_{this}; + + // Output stream is initialized by constructor, it remains always valid. It + // represents output part of a local socket. Input stream of a local socket + // comes from the peer socket, after connection. + std::unique_ptr output_; + std::unique_ptr input_; + // `input_for_remote_` is the other end of the pipe formed with `output_`. We + // give this stream to the remote socket when they connect to us, and it + // becomes their `input_` stream. + std::unique_ptr input_for_remote_; SocketBase* remote_socket_ ABSL_GUARDED_BY(mutex_) = nullptr; bool closed_ ABSL_GUARDED_BY(mutex_) = false; }; diff --git a/internal/platform/implementation/g3/wifi_direct.cc b/internal/platform/implementation/g3/wifi_direct.cc index d4d97261..8e49f7f4 100644 --- a/internal/platform/implementation/g3/wifi_direct.cc +++ b/internal/platform/implementation/g3/wifi_direct.cc @@ -16,15 +16,21 @@ #include #include -#include #include #include +#include "absl/log/check.h" +#include "absl/strings/str_cat.h" #include "absl/strings/str_format.h" +#include "absl/strings/string_view.h" #include "absl/synchronization/mutex.h" +#include "internal/platform/cancellation_flag.h" +#include "internal/platform/exception.h" #include "internal/platform/implementation/wifi_direct.h" #include "internal/platform/logging.h" #include "internal/platform/medium_environment.h" +#include "internal/platform/prng.h" +#include "internal/platform/wifi_credential.h" namespace nearby { namespace g3 { diff --git a/internal/platform/implementation/g3/wifi_direct.h b/internal/platform/implementation/g3/wifi_direct.h index fba77915..d4b07504 100644 --- a/internal/platform/implementation/g3/wifi_direct.h +++ b/internal/platform/implementation/g3/wifi_direct.h @@ -22,7 +22,6 @@ #include "absl/synchronization/mutex.h" #include "internal/platform/implementation/g3/multi_thread_executor.h" -#include "internal/platform/implementation/g3/pipe.h" #include "internal/platform/implementation/g3/socket_base.h" #include "internal/platform/implementation/wifi_direct.h" #include "internal/platform/input_stream.h" diff --git a/internal/platform/implementation/g3/wifi_hotspot.cc b/internal/platform/implementation/g3/wifi_hotspot.cc index d4ab753d..b8da8859 100644 --- a/internal/platform/implementation/g3/wifi_hotspot.cc +++ b/internal/platform/implementation/g3/wifi_hotspot.cc @@ -16,16 +16,22 @@ #include #include -#include #include #include +#include "absl/log/check.h" +#include "absl/strings/str_cat.h" #include "absl/strings/str_format.h" +#include "absl/strings/string_view.h" #include "absl/synchronization/mutex.h" +#include "internal/platform/cancellation_flag.h" #include "internal/platform/cancellation_flag_listener.h" +#include "internal/platform/exception.h" #include "internal/platform/implementation/wifi_hotspot.h" #include "internal/platform/logging.h" #include "internal/platform/medium_environment.h" +#include "internal/platform/prng.h" +#include "internal/platform/wifi_credential.h" namespace nearby { namespace g3 { diff --git a/internal/platform/implementation/g3/wifi_hotspot.h b/internal/platform/implementation/g3/wifi_hotspot.h index 6c248dfd..a3b3e69f 100644 --- a/internal/platform/implementation/g3/wifi_hotspot.h +++ b/internal/platform/implementation/g3/wifi_hotspot.h @@ -23,7 +23,6 @@ #include "absl/synchronization/mutex.h" #include "internal/platform/byte_array.h" #include "internal/platform/implementation/g3/multi_thread_executor.h" -#include "internal/platform/implementation/g3/pipe.h" #include "internal/platform/implementation/g3/socket_base.h" #include "internal/platform/implementation/wifi_hotspot.h" #include "internal/platform/input_stream.h" diff --git a/internal/platform/implementation/g3/wifi_lan.cc b/internal/platform/implementation/g3/wifi_lan.cc index 078c67b3..8d4626e3 100644 --- a/internal/platform/implementation/g3/wifi_lan.cc +++ b/internal/platform/implementation/g3/wifi_lan.cc @@ -19,10 +19,13 @@ #include #include -#include "absl/strings/escaping.h" +#include "absl/log/check.h" +#include "absl/strings/str_cat.h" #include "absl/strings/str_format.h" #include "absl/synchronization/mutex.h" +#include "internal/platform/cancellation_flag.h" #include "internal/platform/cancellation_flag_listener.h" +#include "internal/platform/exception.h" #include "internal/platform/implementation/wifi_lan.h" #include "internal/platform/logging.h" #include "internal/platform/medium_environment.h" diff --git a/internal/platform/implementation/g3/wifi_lan.h b/internal/platform/implementation/g3/wifi_lan.h index 19cb18d3..b51f05fd 100644 --- a/internal/platform/implementation/g3/wifi_lan.h +++ b/internal/platform/implementation/g3/wifi_lan.h @@ -24,7 +24,6 @@ #include "absl/synchronization/mutex.h" #include "internal/platform/byte_array.h" #include "internal/platform/implementation/g3/multi_thread_executor.h" -#include "internal/platform/implementation/g3/pipe.h" #include "internal/platform/implementation/g3/socket_base.h" #include "internal/platform/implementation/wifi_lan.h" #include "internal/platform/input_stream.h" diff --git a/internal/platform/pipe.cc b/internal/platform/pipe.cc index 25a0b9e9..8249686d 100644 --- a/internal/platform/pipe.cc +++ b/internal/platform/pipe.cc @@ -14,25 +14,181 @@ #include "internal/platform/pipe.h" +#include +#include +#include +#include +#include + +#include "absl/base/thread_annotations.h" +#include "internal/platform/base_mutex_lock.h" +#include "internal/platform/byte_array.h" +#include "internal/platform/exception.h" #include "internal/platform/implementation/condition_variable.h" #include "internal/platform/implementation/mutex.h" #include "internal/platform/implementation/platform.h" +#include "internal/platform/input_stream.h" +#include "internal/platform/output_stream.h" namespace nearby { namespace { using Platform = api::ImplementationPlatform; -} +class Pipe { + public: + Pipe() { #pragma push_macro("CreateMutex") #undef CreateMutex + mutex_ = Platform::CreateMutex(api::Mutex::Mode::kRegular); +#pragma pop_macro("CreateMutex") + cond_ = Platform::CreateConditionVariable(mutex_.get()); + } -Pipe::Pipe() { - auto mutex = Platform::CreateMutex(api::Mutex::Mode::kRegular); - auto cond = Platform::CreateConditionVariable(mutex.get()); - Setup(std::move(mutex), std::move(cond)); + class PipeInputStream : public InputStream { + public: + explicit PipeInputStream(std::shared_ptr pipe) : pipe_(pipe) {} + ~PipeInputStream() override { DoClose(); } + + ExceptionOr Read(std::int64_t size) override { + return pipe_->Read(size); + } + Exception Close() override { return DoClose(); } + + private: + Exception DoClose() { + pipe_->MarkInputStreamClosed(); + return {Exception::kSuccess}; + } + std::shared_ptr pipe_; + }; + + class PipeOutputStream : public OutputStream { + public: + explicit PipeOutputStream(std::shared_ptr pipe) : pipe_(pipe) {} + ~PipeOutputStream() override { DoClose(); } + + Exception Write(const ByteArray& data) override { + return pipe_->Write(data); + } + Exception Flush() override { return {Exception::kSuccess}; } + Exception Close() override { return DoClose(); } + + private: + Exception DoClose() { + pipe_->MarkOutputStreamClosed(); + return {Exception::kSuccess}; + } + std::shared_ptr pipe_; + }; + + private: + ExceptionOr Read(size_t size) ABSL_LOCKS_EXCLUDED(mutex_); + Exception Write(const ByteArray& data) ABSL_LOCKS_EXCLUDED(mutex_); + + void MarkInputStreamClosed() ABSL_LOCKS_EXCLUDED(mutex_); + void MarkOutputStreamClosed() ABSL_LOCKS_EXCLUDED(mutex_); + + Exception WriteLocked(const ByteArray& data) + ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + + bool input_stream_closed_ ABSL_GUARDED_BY(mutex_) = false; + bool output_stream_closed_ ABSL_GUARDED_BY(mutex_) = false; + bool read_all_chunks_ ABSL_GUARDED_BY(mutex_) = false; + + std::deque ABSL_GUARDED_BY(mutex_) buffer_; + // Order of declaration matters: + // - mutex must be defined before condvar; + std::unique_ptr mutex_; + std::unique_ptr cond_; +}; + +ExceptionOr Pipe::Read(size_t size) { + BaseMutexLock lock(mutex_.get()); + + // We're done reading all the chunks that were written before the OutputStream + // was closed, so there's nothing to do here other than return an empty chunk + // to serve as an EOF indication to callers. + if (read_all_chunks_) { + return ExceptionOr{ByteArray{}}; + } + + while (buffer_.empty() && !input_stream_closed_) { + Exception wait_exception = cond_->Wait(); + + if (wait_exception.Raised()) { + return ExceptionOr{wait_exception}; + } + } + + // If we received our sentinel chunk, mark the fact that there cannot + // possibly be any more chunks to read here on in, and return an empty chunk + // to serve as an EOF indication to callers. + if (buffer_.empty() || buffer_.front().Empty()) { + read_all_chunks_ = true; + return ExceptionOr{ByteArray{}}; + } + + ByteArray first_chunk{buffer_.front()}; + buffer_.pop_front(); + + // If first_chunk is small enough to not overshoot the requested 'size', just + // return that. + if (first_chunk.size() <= size) { + return ExceptionOr{first_chunk}; + } else { + // Break first_chunk into 2 parts -- the first one of which (next_chunk) + // will be 'size' bytes long, and will be returned, and the second one of + // which (overflow_chunk) will be re-inserted into buffer_, at the head of + // the queue, to be served up in the next call to read(). + ByteArray next_chunk(first_chunk.data(), size); + buffer_.push_front( + ByteArray(first_chunk.data() + size, first_chunk.size() - size)); + return ExceptionOr{next_chunk}; + } } -#pragma pop_macro("CreateMutex") +Exception Pipe::Write(const ByteArray& data) { + BaseMutexLock lock(mutex_.get()); + return WriteLocked(data); +} + +void Pipe::MarkInputStreamClosed() { + BaseMutexLock lock(mutex_.get()); + if (input_stream_closed_) return; + input_stream_closed_ = true; + // Trigger cond_ to unblock a potentially-blocked call to read(), and to let + // it know to return Exception::IO. + cond_->Notify(); +} + +void Pipe::MarkOutputStreamClosed() { + BaseMutexLock lock(mutex_.get()); + if (output_stream_closed_) return; + // Write a sentinel null chunk before marking output_stream_closed as true. + WriteLocked(ByteArray{}); + output_stream_closed_ = true; +} + +Exception Pipe::WriteLocked(const ByteArray& data) { + if (input_stream_closed_ || output_stream_closed_) { + return {Exception::kIo}; + } + + buffer_.push_back(data); + // Trigger cond_ to unblock a potentially-blocked call to read(), now that + // there's more data for it to consume. + cond_->Notify(); + return {Exception::kSuccess}; +} + +} // namespace + +std::pair, std::unique_ptr> +CreatePipe() { + auto pipe = std::make_shared(); + return std::make_pair(std::make_unique(pipe), + std::make_unique(pipe)); +} } // namespace nearby diff --git a/internal/platform/pipe.h b/internal/platform/pipe.h index a96bfbbc..4d0af630 100644 --- a/internal/platform/pipe.h +++ b/internal/platform/pipe.h @@ -15,19 +15,23 @@ #ifndef PLATFORM_PUBLIC_PIPE_H_ #define PLATFORM_PUBLIC_PIPE_H_ -#include "internal/platform/base_pipe.h" +#include +#include + +#include "internal/platform/input_stream.h" +#include "internal/platform/output_stream.h" namespace nearby { -// See for details: -// http://google3/platform/base/base_pipe.h -class Pipe final : public BasePipe { - public: - Pipe(); - ~Pipe() override = default; - Pipe(Pipe&&) = delete; - Pipe& operator=(Pipe&&) = delete; -}; +// Creates a pipe for streaming data between threads. +// ``` +// auto [input, output] = CreatePipe(); +// ReaderThread(std::move(input)); +// WriterThread(std::move(output)); +// ``` +// Pipe stays valid as long as either `input` or `output` exist. +std::pair, std::unique_ptr> +CreatePipe(); } // namespace nearby diff --git a/internal/platform/pipe_test.cc b/internal/platform/pipe_test.cc index 08312836..23ee4da0 100644 --- a/internal/platform/pipe_test.cc +++ b/internal/platform/pipe_test.cc @@ -17,141 +17,137 @@ #include #include +#include #include +#include #include #include +#include #include "gtest/gtest.h" +#include "absl/strings/string_view.h" +#include "internal/platform/byte_array.h" +#include "internal/platform/exception.h" +#include "internal/platform/input_stream.h" +#include "internal/platform/output_stream.h" #include "internal/platform/prng.h" #include "internal/platform/runnable.h" namespace nearby { +namespace { +constexpr size_t kChunkSize = 64 * 1024; +} + TEST(PipeTest, ConstructorDestructorWorks) { - Pipe pipe; + auto [input_stream, output_stream] = CreatePipe(); SUCCEED(); } TEST(PipeTest, SimpleWriteRead) { - Pipe pipe; - InputStream& input_stream{pipe.GetInputStream()}; - OutputStream& output_stream{pipe.GetOutputStream()}; - + auto [input_stream, output_stream] = CreatePipe(); std::string data("ABCD"); - EXPECT_TRUE(output_stream.Write(ByteArray(data)).Ok()); + EXPECT_TRUE(output_stream->Write(ByteArray(data)).Ok()); - ExceptionOr read_data = input_stream.Read(Pipe::kChunkSize); + ExceptionOr read_data = input_stream->Read(kChunkSize); EXPECT_TRUE(read_data.ok()); EXPECT_EQ(data, std::string(read_data.result())); } TEST(PipeTest, WriteEndClosedBeforeRead) { - Pipe pipe; - InputStream& input_stream{pipe.GetInputStream()}; - OutputStream& output_stream{pipe.GetOutputStream()}; + auto [input_stream, output_stream] = CreatePipe(); std::string data("ABCD"); - EXPECT_TRUE(output_stream.Write(ByteArray(data)).Ok()); + EXPECT_TRUE(output_stream->Write(ByteArray(data)).Ok()); // Close the write end before the read end has even begun reading. - EXPECT_TRUE(output_stream.Close().Ok()); + EXPECT_TRUE(output_stream->Close().Ok()); // We should still be able to read what was written. - ExceptionOr read_data = input_stream.Read(Pipe::kChunkSize); + ExceptionOr read_data = input_stream->Read(kChunkSize); EXPECT_TRUE(read_data.ok()); EXPECT_EQ(data, std::string(read_data.result())); // And after that, we should get our indication that all the data that could // ever be read, has already been read. - read_data = input_stream.Read(Pipe::kChunkSize); + read_data = input_stream->Read(kChunkSize); EXPECT_TRUE(read_data.ok()); EXPECT_TRUE(read_data.result().Empty()); } TEST(PipeTest, ReadEndClosedBeforeWrite) { - Pipe pipe; - InputStream& input_stream{pipe.GetInputStream()}; - OutputStream& output_stream{pipe.GetOutputStream()}; + auto [input_stream, output_stream] = CreatePipe(); // Close the read end before the write end has even begun writing. - EXPECT_TRUE(input_stream.Close().Ok()); + EXPECT_TRUE(input_stream->Close().Ok()); std::string data("ABCD"); - EXPECT_TRUE(output_stream.Write(ByteArray(data)).Raised(Exception::kIo)); + EXPECT_TRUE(output_stream->Write(ByteArray(data)).Raised(Exception::kIo)); } TEST(PipeTest, SizedReadMoreThanFirstChunkSize) { - Pipe pipe; - InputStream& input_stream{pipe.GetInputStream()}; - OutputStream& output_stream{pipe.GetOutputStream()}; + auto [input_stream, output_stream] = CreatePipe(); std::string data("ABCD"); - EXPECT_TRUE(output_stream.Write(ByteArray(data)).Ok()); + EXPECT_TRUE(output_stream->Write(ByteArray(data)).Ok()); // Even though we ask for double of what's there in the first chunk, we should // get back only what's there in that first chunk, and that's alright. - ExceptionOr read_data = input_stream.Read(data.size() * 2); + ExceptionOr read_data = input_stream->Read(data.size() * 2); EXPECT_TRUE(read_data.ok()); EXPECT_EQ(data, std::string(read_data.result())); } TEST(PipeTest, SizedReadLessThanFirstChunkSize) { - Pipe pipe; - InputStream& input_stream{pipe.GetInputStream()}; - OutputStream& output_stream{pipe.GetOutputStream()}; - + auto [input_stream, output_stream] = CreatePipe(); std::string data_first_part("ABCD"); std::string data_second_part("EFGHIJ"); std::string data = data_first_part + data_second_part; - EXPECT_TRUE(output_stream.Write(ByteArray(data)).Ok()); + EXPECT_TRUE(output_stream->Write(ByteArray(data)).Ok()); // When we ask for less than what's there in the first chunk, we should get // back exactly what we asked for, with the remainder still being available // for the next read. std::int64_t desired_size = data_first_part.size(); - ExceptionOr first_read_data = input_stream.Read(desired_size); + ExceptionOr first_read_data = input_stream->Read(desired_size); EXPECT_TRUE(first_read_data.ok()); EXPECT_EQ(data_first_part, std::string(first_read_data.result())); // Now read the remainder, and get everything that ought to have been left. - ExceptionOr second_read_data = input_stream.Read(Pipe::kChunkSize); + ExceptionOr second_read_data = input_stream->Read(kChunkSize); EXPECT_TRUE(second_read_data.ok()); EXPECT_EQ(data_second_part, std::string(second_read_data.result())); } TEST(PipeTest, ReadAfterInputStreamClosed) { - Pipe pipe; - InputStream& input_stream{pipe.GetInputStream()}; + auto [input_stream, output_stream] = CreatePipe(); - input_stream.Close(); + input_stream->Close(); - ExceptionOr read_data = input_stream.Read(Pipe::kChunkSize); + ExceptionOr read_data = input_stream->Read(kChunkSize); EXPECT_TRUE(read_data.ok()); EXPECT_TRUE(read_data.GetResult().Empty()); } TEST(PipeTest, WriteAfterOutputStreamClosed) { - Pipe pipe; - OutputStream& output_stream{pipe.GetOutputStream()}; + auto [input_stream, output_stream] = CreatePipe(); - output_stream.Close(); + output_stream->Close(); std::string data("ABCD"); - EXPECT_TRUE(output_stream.Write(ByteArray(data)).Raised(Exception::kIo)); + EXPECT_TRUE(output_stream->Write(ByteArray(data)).Raised(Exception::kIo)); } TEST(PipeTest, RepeatedClose) { - Pipe pipe; - InputStream& input_stream{pipe.GetInputStream()}; - OutputStream& output_stream{pipe.GetOutputStream()}; + auto [input_stream, output_stream] = CreatePipe(); - EXPECT_TRUE(output_stream.Close().Ok()); - EXPECT_TRUE(output_stream.Close().Ok()); - EXPECT_TRUE(output_stream.Close().Ok()); + EXPECT_TRUE(output_stream->Close().Ok()); + EXPECT_TRUE(output_stream->Close().Ok()); + EXPECT_TRUE(output_stream->Close().Ok()); - EXPECT_TRUE(input_stream.Close().Ok()); - EXPECT_TRUE(input_stream.Close().Ok()); - EXPECT_TRUE(input_stream.Close().Ok()); + EXPECT_TRUE(input_stream->Close().Ok()); + EXPECT_TRUE(input_stream->Close().Ok()); + EXPECT_TRUE(input_stream->Close().Ok()); } class Thread { @@ -186,17 +182,18 @@ TEST(PipeTest, ReadBlockedUntilWrite) { class ReaderRunnable { public: - ReaderRunnable(InputStream* input_stream, + ReaderRunnable(std::unique_ptr input_stream, absl::string_view expected_read_data, CrossThreadBool* ok_for_read_to_unblock) - : input_stream_(input_stream), + : input_stream_(std::move(input_stream)), expected_read_data_(expected_read_data), ok_for_read_to_unblock_(ok_for_read_to_unblock) {} + ReaderRunnable(ReaderRunnable&&) = default; ~ReaderRunnable() = default; // Signature "void()" satisfies Runnable. void operator()() { - ExceptionOr read_data = input_stream_->Read(Pipe::kChunkSize); + ExceptionOr read_data = input_stream_->Read(kChunkSize); // Make sure read() doesn't return before it's appropriate. if (!*ok_for_read_to_unblock_) { @@ -210,13 +207,12 @@ TEST(PipeTest, ReadBlockedUntilWrite) { } private: - InputStream* input_stream_; + std::unique_ptr input_stream_; const std::string expected_read_data_; CrossThreadBool* ok_for_read_to_unblock_; }; - Pipe pipe; - OutputStream& output_stream{pipe.GetOutputStream()}; + auto [input_stream, output_stream] = CreatePipe(); // State shared between this thread (the writer) and reader_thread. CrossThreadBool ok_for_read_to_unblock = false; @@ -225,7 +221,7 @@ TEST(PipeTest, ReadBlockedUntilWrite) { // Kick off reader_thread. Thread reader_thread; reader_thread.Start( - ReaderRunnable(&pipe.GetInputStream(), data, &ok_for_read_to_unblock)); + ReaderRunnable(std::move(input_stream), data, &ok_for_read_to_unblock)); // Introduce a delay before we actually write anything. absl::SleepFor(absl::Seconds(5)); @@ -236,7 +232,7 @@ TEST(PipeTest, ReadBlockedUntilWrite) { ok_for_read_to_unblock = true; // Perform the actual write. - EXPECT_TRUE(output_stream.Write(ByteArray(data)).Ok()); + EXPECT_TRUE(output_stream->Write(ByteArray(data)).Ok()); // And wait for reader_thread to finish. reader_thread.Join(); @@ -247,6 +243,7 @@ TEST(PipeTest, ConcurrentWriteAndRead) { protected: explicit BaseRunnable(const std::vector& chunks) : chunks_(chunks) {} + BaseRunnable(BaseRunnable&&) = default; virtual ~BaseRunnable() = default; void RandomSleep() { @@ -267,9 +264,10 @@ TEST(PipeTest, ConcurrentWriteAndRead) { class WriterRunnable : public BaseRunnable { public: - WriterRunnable(OutputStream* output_stream, + WriterRunnable(std::unique_ptr output_stream, const std::vector& chunks) - : BaseRunnable(chunks), output_stream_(output_stream) {} + : BaseRunnable(chunks), output_stream_(std::move(output_stream)) {} + WriterRunnable(WriterRunnable&&) = default; ~WriterRunnable() override = default; void operator()() { @@ -283,14 +281,15 @@ TEST(PipeTest, ConcurrentWriteAndRead) { } private: - OutputStream* output_stream_; + std::unique_ptr output_stream_; }; class ReaderRunnable : public BaseRunnable { public: - ReaderRunnable(InputStream* input_stream, + ReaderRunnable(std::unique_ptr input_stream, const std::vector& chunks) - : BaseRunnable(chunks), input_stream_(input_stream) {} + : BaseRunnable(chunks), input_stream_(std::move(input_stream)) {} + ReaderRunnable(ReaderRunnable&&) = default; ~ReaderRunnable() override = default; void operator()() { @@ -304,8 +303,7 @@ TEST(PipeTest, ConcurrentWriteAndRead) { std::string actual_data; while (true) { RandomSleep(); // Random pauses before each read. - ExceptionOr read_data = - input_stream_->Read(Pipe::kChunkSize); + ExceptionOr read_data = input_stream_->Read(kChunkSize); if (read_data.ok()) { ByteArray result = read_data.result(); if (result.Empty()) { @@ -322,11 +320,10 @@ TEST(PipeTest, ConcurrentWriteAndRead) { } private: - InputStream* input_stream_; + std::unique_ptr input_stream_; }; - Pipe pipe; - + auto [input_stream, output_stream] = CreatePipe(); std::vector chunks; chunks.push_back("ABCD"); chunks.push_back("EFGH"); @@ -334,8 +331,8 @@ TEST(PipeTest, ConcurrentWriteAndRead) { Thread writer_thread; Thread reader_thread; - writer_thread.Start(WriterRunnable(&pipe.GetOutputStream(), chunks)); - reader_thread.Start(ReaderRunnable(&pipe.GetInputStream(), chunks)); + writer_thread.Start(WriterRunnable(std::move(output_stream), chunks)); + reader_thread.Start(ReaderRunnable(std::move(input_stream), chunks)); writer_thread.Join(); reader_thread.Join(); }