[Nearby] Handle peer reset events in BT Streams

Turns out we were not handling the peer reset events on the SendMore and
ReceiveMore functions which could cause an infinite looping when the
other side resets.

Fixed: 1132118

Change-Id: I87113518b01f56497f09e361931cc64647819df8
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2430294
Auto-Submit: James Vecore <vecore@google.com>
Commit-Queue: Josh Nohle <nohle@chromium.org>
Reviewed-by: Josh Nohle <nohle@chromium.org>
Cr-Commit-Position: refs/heads/master@{#810685}

chrome/services/sharing/nearby/platform_v2/bluetooth_socket.cc

@@ -91,6 +91,13 @@ class InputStreamImpl : public InputStream {
     DCHECK_LT(pending_read_buffer_pos_, pending_read_buffer_->size());
     DCHECK(task_run_);
 
+    if (state.peer_closed()) {
+      exception_or_received_byte_array_ =
+          ExceptionOr<ByteArray>(Exception::kIo);
+      task_run_->Signal();
+      return;
+    }
+
     if (result == MOJO_RESULT_OK) {
       uint32_t num_bytes = static_cast<uint32_t>(pending_read_buffer_->size() -
                                                  pending_read_buffer_pos_);
@@ -205,6 +212,12 @@ class OutputStreamImpl : public OutputStream {
     DCHECK_LT(pending_write_buffer_pos_, pending_write_buffer_->size());
     DCHECK(task_run_);
 
+    if (state.peer_closed()) {
+      write_success_ = false;
+      task_run_->Signal();
+      return;
+    }
+
     if (result == MOJO_RESULT_OK) {
       uint32_t num_bytes = static_cast<uint32_t>(pending_write_buffer_->size() -
                                                  pending_write_buffer_pos_);

chrome/services/sharing/nearby/platform_v2/bluetooth_socket_unittest.cc

@@ -281,6 +281,37 @@ TEST_F(BluetoothSocketTest, TestOutputStream_MultipleChunks) {
   run_loop.Run();
 }
 
+TEST_F(BluetoothSocketTest, TestInputStreamResetHandler) {
+  InputStream& input_stream = bluetooth_socket_->GetInputStream();
+
+  // Setup a message to receive that would work if the connection was not reset.
+  std::string message = "ReceivedMessage";
+  uint32_t message_size = message.size();
+  EXPECT_EQ(MOJO_RESULT_OK,
+            receive_stream_->WriteData(message.data(), &message_size,
+                                       MOJO_WRITE_DATA_FLAG_NONE));
+  EXPECT_EQ(message.size(), message_size);
+
+  // Reset the pipe on the other side to trigger a peer_reset state.
+  receive_stream_.reset();
+
+  ExceptionOr<ByteArray> exception_or_byte_array =
+      input_stream.Read(message_size);
+  ASSERT_FALSE(exception_or_byte_array.ok());
+  EXPECT_EQ(Exception::kIo, exception_or_byte_array.exception());
+}
+
+TEST_F(BluetoothSocketTest, TestOutputStreamResetHandling) {
+  OutputStream& output_stream = bluetooth_socket_->GetOutputStream();
+
+  // Reset the pipe on the other side to trigger a peer_reset state.
+  send_stream_.reset();
+
+  std::string message = "SentMessage";
+  ByteArray byte_array(message);
+  EXPECT_EQ(Exception::kIo, output_stream.Write(byte_array).value);
+}
+
 }  // namespace chrome
 }  // namespace nearby
 }  // namespace location