diff --git a/connections/implementation/mediums/webrtc/connection_flow_test.cc b/connections/implementation/mediums/webrtc/connection_flow_test.cc index 8c066163..f0e1d6bf 100644 --- a/connections/implementation/mediums/webrtc/connection_flow_test.cc +++ b/connections/implementation/mediums/webrtc/connection_flow_test.cc @@ -317,7 +317,7 @@ TEST_F(ConnectionFlowTest, TerminateAnswerer) { offerer_socket.result().GetOutputStream().Write(ByteArray{message}); ExceptionOr received_message = answerer_socket.result().GetInputStream().Read(4); - EXPECT_FALSE(received_message.ok()); + EXPECT_TRUE(received_message.GetResult().Empty()); } TEST_F(ConnectionFlowTest, TerminateOfferer) { @@ -398,7 +398,7 @@ TEST_F(ConnectionFlowTest, TerminateOfferer) { offerer_socket.result().GetOutputStream().Write(ByteArray{message}); ExceptionOr received_message = answerer_socket.result().GetInputStream().Read(4); - EXPECT_FALSE(received_message.ok()); + EXPECT_TRUE(received_message.GetResult().Empty()); } } // namespace } // namespace mediums diff --git a/connections/implementation/mediums/webrtc/webrtc_socket_impl_test.cc b/connections/implementation/mediums/webrtc/webrtc_socket_impl_test.cc index f8bd7b52..af3b6d84 100644 --- a/connections/implementation/mediums/webrtc/webrtc_socket_impl_test.cc +++ b/connections/implementation/mediums/webrtc/webrtc_socket_impl_test.cc @@ -168,7 +168,7 @@ TEST(WebRtcSocketTest, ReadFromClosedChannel) { webrtc_socket.GetOutputStream().Write(kMessage); webrtc_socket.Close(); - EXPECT_EQ(webrtc_socket.GetInputStream().Read(7).exception(), Exception::kIo); + EXPECT_TRUE(webrtc_socket.GetInputStream().Read(7).GetResult().Empty()); } TEST(WebRtcSocketTest, DataChannelCloseEventCleansUp) { @@ -181,7 +181,7 @@ TEST(WebRtcSocketTest, DataChannelCloseEventCleansUp) { webrtc_socket.OnStateChange(); - EXPECT_EQ(webrtc_socket.GetInputStream().Read(7).exception(), Exception::kIo); + EXPECT_TRUE(webrtc_socket.GetInputStream().Read(7).GetResult().Empty()); // Calling Close again should be safe even if the channel is already shut // down. diff --git a/internal/platform/base_pipe.cc b/internal/platform/base_pipe.cc index 7d1ee35d..94c6b917 100644 --- a/internal/platform/base_pipe.cc +++ b/internal/platform/base_pipe.cc @@ -38,21 +38,17 @@ ExceptionOr BasePipe::Read(size_t size) { } } - if (input_stream_closed_) { - return ExceptionOr{Exception::kIo}; + // 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 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 (first_chunk.Empty()) { - read_all_chunks_ = true; - return ExceptionOr{ByteArray{}}; - } - // If first_chunk is small enough to not overshoot the requested 'size', just // return that. if (first_chunk.size() <= size) { diff --git a/internal/platform/bluetooth_classic_test.cc b/internal/platform/bluetooth_classic_test.cc index a68fe9ad..210b2a23 100644 --- a/internal/platform/bluetooth_classic_test.cc +++ b/internal/platform/bluetooth_classic_test.cc @@ -184,6 +184,53 @@ INSTANTIATE_TEST_SUITE_P(ParametrisedBluetoothClassicMediumTest, BluetoothClassicMediumTest, ::testing::ValuesIn(kTestCases)); +TEST_F(BluetoothClassicMediumTest, SendData) { + adapter_a_->SetScanMode(BluetoothAdapter::ScanMode::kConnectable); + CountDownLatch found_latch(1); + BluetoothDevice* discovered_device = nullptr; + bt_a_->StartDiscovery(DiscoveryCallback{ + .device_discovered_cb = + [this, &found_latch, &discovered_device](BluetoothDevice& device) { + NEARBY_LOG(INFO, "Device discovered: %s", device.GetName().c_str()); + EXPECT_EQ(device.GetName(), adapter_b_->GetName()); + discovered_device = &device; + found_latch.CountDown(); + }, + }); + adapter_b_->SetScanMode(BluetoothAdapter::ScanMode::kConnectableDiscoverable); + EXPECT_EQ(adapter_b_->GetScanMode(), + BluetoothAdapter::ScanMode::kConnectableDiscoverable); + ASSERT_TRUE(found_latch.Await().Ok()); + std::string service_name{"service"}; + std::string service_uuid("service-uuid"); + BluetoothServerSocket server_socket = + bt_b_->ListenForService(service_name, service_uuid); + ASSERT_TRUE(server_socket.IsValid()); + { + ByteArray data("data"); + CancellationFlag flag; + SingleThreadExecutor server_executor; + SingleThreadExecutor client_executor; + client_executor.Execute([&, this]() { + BluetoothSocket socket_a = + bt_a_->ConnectToService(*discovered_device, service_uuid, &flag); + ASSERT_TRUE(socket_a.IsValid()); + EXPECT_TRUE(socket_a.GetOutputStream().Write(data).Ok()); + EXPECT_TRUE(socket_a.GetOutputStream().Close().Ok()); + }); + server_executor.Execute([&]() { + BluetoothSocket socket_b = server_socket.Accept(); + ASSERT_TRUE(socket_b.IsValid()); + ExceptionOr result = + socket_b.GetInputStream().Read(data.size()); + ASSERT_EQ(result.exception(), Exception::kSuccess); + ASSERT_TRUE(result.ok()); + EXPECT_EQ(result.GetResult(), data); + }); + } + server_socket.Close(); +} + TEST_F(BluetoothClassicMediumTest, ConstructorDestructorWorks) { // Make sure we can create functional adapters. ASSERT_TRUE(adapter_a_->IsValid()); diff --git a/internal/platform/input_stream.cc b/internal/platform/input_stream.cc index 2aaf2d3a..d9043e7e 100644 --- a/internal/platform/input_stream.cc +++ b/internal/platform/input_stream.cc @@ -34,7 +34,11 @@ ExceptionOr InputStream::Skip(size_t offset) { if (!result.ok()) { return result.GetException(); } - bytes_left -= chunk_size; + size_t bytes_read = result.GetResult().size(); + if (bytes_read == 0) { + return ExceptionOr(offset - bytes_left); + } + bytes_left -= bytes_read; } return ExceptionOr(offset); } diff --git a/internal/platform/input_stream.h b/internal/platform/input_stream.h index 82c216ef..f0315300 100644 --- a/internal/platform/input_stream.h +++ b/internal/platform/input_stream.h @@ -29,10 +29,13 @@ class InputStream { public: virtual ~InputStream() = default; - // throws Exception::kIo + // Reads at most `size` bytes from the input stream. + // Returns an empty byte array on end of file, or Exception::kIo on error. virtual ExceptionOr Read(std::int64_t size) = 0; - // throws Exception::kIo + // Skips `offset` bytes from the stream. + // Returns the number of bytes skipped, which can be less than offset on EOF, + // or Exception::kIo on error. virtual ExceptionOr Skip(size_t offset); // throws Exception::kIo diff --git a/internal/platform/pipe_test.cc b/internal/platform/pipe_test.cc index bebb8531..08312836 100644 --- a/internal/platform/pipe_test.cc +++ b/internal/platform/pipe_test.cc @@ -126,8 +126,8 @@ TEST(PipeTest, ReadAfterInputStreamClosed) { input_stream.Close(); ExceptionOr read_data = input_stream.Read(Pipe::kChunkSize); - EXPECT_TRUE(!read_data.ok()); - EXPECT_TRUE(read_data.GetException().Raised(Exception::kIo)); + EXPECT_TRUE(read_data.ok()); + EXPECT_TRUE(read_data.GetResult().Empty()); } TEST(PipeTest, WriteAfterOutputStreamClosed) { diff --git a/internal/platform/wifi_direct_test.cc b/internal/platform/wifi_direct_test.cc index 2c39f4af..e7f045e7 100644 --- a/internal/platform/wifi_direct_test.cc +++ b/internal/platform/wifi_direct_test.cc @@ -157,7 +157,7 @@ TEST_P(WifiDirectMediumTest, CanStartDirectGOThatOtherCanConnect) { socket_b.Close(); EXPECT_FALSE(out_stream.Write(ByteArray(data)).Ok()); read_data = in_stream.Read(kChunkSize); - EXPECT_FALSE(read_data.ok()); + EXPECT_TRUE(read_data.GetResult().Empty()); server_socket.Close(); EXPECT_TRUE(wifi_direct_b.DisconnectWifiDirect()); diff --git a/internal/platform/wifi_hotspot_test.cc b/internal/platform/wifi_hotspot_test.cc index f9bd78d1..a21ee7a8 100644 --- a/internal/platform/wifi_hotspot_test.cc +++ b/internal/platform/wifi_hotspot_test.cc @@ -20,9 +20,9 @@ #include "protobuf-matchers/protocol-buffer-matchers.h" #include "gtest/gtest.h" #include "absl/time/clock.h" -#include "internal/platform/medium_environment.h" #include "internal/platform/count_down_latch.h" #include "internal/platform/logging.h" +#include "internal/platform/medium_environment.h" #include "internal/platform/wifi_credential.h" namespace nearby { @@ -68,16 +68,13 @@ class WifiHotspotMediumTest : public testing::TestWithParam { env_.Stop(); env_.Start(); } - ~WifiHotspotMediumTest() override{ - env_.Stop(); - } + ~WifiHotspotMediumTest() override { env_.Stop(); } MediumEnvironment& env_{MediumEnvironment::Instance()}; }; INSTANTIATE_TEST_SUITE_P(ParametrisedWifiHotspotMediumTest, - WifiHotspotMediumTest, - testing::ValuesIn(kTestCases)); + WifiHotspotMediumTest, testing::ValuesIn(kTestCases)); TEST_F(WifiHotspotMediumTest, ConstructorDestructorWorks) { auto wifi_hotspot_a = std::make_unique(); @@ -163,7 +160,7 @@ TEST_P(WifiHotspotMediumTest, CanStartHotspotThatOtherConnect) { EXPECT_TRUE(socket_a.IsValid()); EXPECT_TRUE(socket_b.IsValid()); InputStream& in_stream = socket_a.GetInputStream(); - OutputStream& out_stream = socket_b.GetOutputStream(); + OutputStream& out_stream = socket_b.GetOutputStream(); std::string data(kData); EXPECT_TRUE(out_stream.Write(ByteArray(data)).Ok()); ExceptionOr read_data = in_stream.Read(kChunkSize); @@ -174,7 +171,7 @@ TEST_P(WifiHotspotMediumTest, CanStartHotspotThatOtherConnect) { socket_b.Close(); EXPECT_FALSE(out_stream.Write(ByteArray(data)).Ok()); read_data = in_stream.Read(kChunkSize); - EXPECT_FALSE(read_data.ok()); + EXPECT_TRUE(read_data.GetResult().Empty()); server_socket.Close(); EXPECT_TRUE(wifi_hotspot_b->DisconnectWifiHotspot()); @@ -235,7 +232,6 @@ TEST_P(WifiHotspotMediumTest, CanStartHotspotThatOtherCanCancelConnect) { EXPECT_FALSE(socket_b.IsValid()); } - server_socket.Close(); EXPECT_TRUE(wifi_hotspot_b->DisconnectWifiHotspot()); EXPECT_TRUE(wifi_hotspot_a->StopWifiHotspot());