Use consistent return value in Pipe read.

Previously reading from an empty, closed Pipe returned:
* IO exception if there were no writes to the pipe,
* Empty byte array if there were writes to the pipe.
With this change, we return an empty byte array on both paths.

This also fixes a subtle race condition in the following flow:
1. Pipe is empty
2. Thread A is reading from the pipe (blocking call).
3. Thread B writes to the pipe and immediately closes it.
4. Thread A is unblocked and returns IO error. The data written by thread B is
   ignored!

The new test in bluetooth_classic_test triggered that defect consistently.

PiperOrigin-RevId: 518107700

connections/implementation/mediums/webrtc/connection_flow_test.cc

@@ -317,7 +317,7 @@ TEST_F(ConnectionFlowTest, TerminateAnswerer) {
   offerer_socket.result().GetOutputStream().Write(ByteArray{message});
   ExceptionOr<ByteArray> 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<ByteArray> received_message =
       answerer_socket.result().GetInputStream().Read(4);
-  EXPECT_FALSE(received_message.ok());
+  EXPECT_TRUE(received_message.GetResult().Empty());
 }
 }  // namespace
 }  // namespace mediums

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.

internal/platform/base_pipe.cc

@@ -38,21 +38,17 @@ ExceptionOr<ByteArray> BasePipe::Read(size_t size) {
     }
   }
 
-  if (input_stream_closed_) {
-    return ExceptionOr<ByteArray>{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{}};
   }
 
   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>{ByteArray{}};
-  }
-
   // If first_chunk is small enough to not overshoot the requested 'size', just
   // return that.
   if (first_chunk.size() <= size) {

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<ByteArray> 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());

internal/platform/input_stream.cc

@@ -34,7 +34,11 @@ ExceptionOr<size_t> 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<size_t>(offset - bytes_left);
+    }
+    bytes_left -= bytes_read;
   }
   return ExceptionOr<size_t>(offset);
 }

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<ByteArray> 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<size_t> Skip(size_t offset);
 
   // throws Exception::kIo

internal/platform/pipe_test.cc

@@ -126,8 +126,8 @@ TEST(PipeTest, ReadAfterInputStreamClosed) {
   input_stream.Close();
 
   ExceptionOr<ByteArray> 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) {

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());

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<FeatureFlags> {
     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<WifiHotspotMedium>();
@@ -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<ByteArray> 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());