Direct Sockets: Browser test for writing to TCP socket

We add a browser test that writes to the socket using script.

Bug: 905818
Change-Id: Ica84ba877eab844d3f9ae0b4cb76fe18b0046b59
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2624371
Commit-Queue: Eric Willigers <ericwilligers@chromium.org>
Commit-Queue: Glen Robertson <glenrob@chromium.org>
Reviewed-by: Glen Robertson <glenrob@chromium.org>
Auto-Submit: Eric Willigers <ericwilligers@chromium.org>
Cr-Commit-Position: refs/heads/master@{#843417}

content/browser/direct_sockets/direct_sockets_browsertest.cc

@@ -190,6 +190,87 @@ net::Error UnconditionallyPermitConnection(
   return net::OK;
 }
 
+class ReadWaiter {
+ public:
+  explicit ReadWaiter(uint32_t required_bytes)
+      : required_bytes_(required_bytes) {}
+
+  void Init(mojo::Remote<network::mojom::TCPServerSocket>& tcp_server_socket) {
+    tcp_server_socket->Accept(
+        /*observer=*/mojo::NullRemote(),
+        base::BindLambdaForTesting(
+            [this](int result,
+                   const base::Optional<net::IPEndPoint>& remote_addr,
+                   mojo::PendingRemote<network::mojom::TCPConnectedSocket>
+                       accepted_socket,
+                   mojo::ScopedDataPipeConsumerHandle consumer_handle,
+                   mojo::ScopedDataPipeProducerHandle producer_handle) {
+              DCHECK(!accepted_socket_);
+              DCHECK_EQ(result, net::OK);
+
+              accepted_socket_.Bind(std::move(accepted_socket));
+              receive_stream_ = std::move(consumer_handle);
+              read_watcher_ = std::make_unique<mojo::SimpleWatcher>(
+                  FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL);
+              read_watcher_->Watch(
+                  receive_stream_.get(),
+                  MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+                  MOJO_TRIGGER_CONDITION_SIGNALS_SATISFIED,
+                  base::BindRepeating(&ReadWaiter::OnReadReady,
+                                      base::Unretained(this)));
+
+              read_watcher_->ArmOrNotify();
+            }));
+  }
+
+  void Await() { run_loop_.Run(); }
+
+ private:
+  void OnReadReady(MojoResult result, const mojo::HandleSignalsState& state) {
+    ReadData();
+  }
+
+  void ReadData() {
+    while (true) {
+      DCHECK(receive_stream_.is_valid());
+      DCHECK_LT(bytes_read_, required_bytes_);
+      const void* buffer = nullptr;
+      uint32_t num_bytes = 0;
+      const MojoResult mojo_result = receive_stream_->BeginReadData(
+          &buffer, &num_bytes, MOJO_READ_DATA_FLAG_NONE);
+      if (mojo_result == MOJO_RESULT_SHOULD_WAIT) {
+        read_watcher_->ArmOrNotify();
+        return;
+      }
+      DCHECK_EQ(mojo_result, MOJO_RESULT_OK);
+
+      // This is guaranteed by Mojo.
+      DCHECK_GT(num_bytes, 0u);
+
+      const unsigned char* current = static_cast<const unsigned char*>(buffer);
+      const unsigned char* const end = current + num_bytes;
+      while (current < end) {
+        EXPECT_EQ(*current, bytes_read_ % 256);
+        ++current;
+        ++bytes_read_;
+      }
+
+      receive_stream_->EndReadData(num_bytes);
+      if (bytes_read_ == required_bytes_) {
+        run_loop_.Quit();
+        return;
+      }
+    }
+  }
+
+  base::RunLoop run_loop_;
+  const uint32_t required_bytes_;
+  mojo::Remote<network::mojom::TCPConnectedSocket> accepted_socket_;
+  mojo::ScopedDataPipeConsumerHandle receive_stream_;
+  std::unique_ptr<mojo::SimpleWatcher> read_watcher_;
+  uint32_t bytes_read_ = 0;
+};
+
 }  // anonymous namespace
 
 class DirectSocketsBrowserTest : public ContentBrowserTest {
@@ -250,6 +331,10 @@ class DirectSocketsBrowserTest : public ContentBrowserTest {
     return local_addr.port();
   }
 
+  mojo::Remote<network::mojom::TCPServerSocket>& tcp_server_socket() {
+    return tcp_server_socket_;
+  }
+
  protected:
   void SetUp() override {
     embedded_test_server()->AddDefaultHandlers(GetTestDataFilePath());
@@ -410,6 +495,21 @@ IN_PROC_BROWSER_TEST_F(DirectSocketsBrowserTest, OpenTcp_OptionsTwo) {
   EXPECT_EQ(true, call.no_delay);
 }
 
+IN_PROC_BROWSER_TEST_F(DirectSocketsBrowserTest, WriteTcp) {
+  const uint32_t kRequiredBytes = 10000;
+  EXPECT_TRUE(NavigateToURL(shell(), GetTestPageURL()));
+
+  ReadWaiter read_waiter(kRequiredBytes);
+  const uint16_t listening_port = StartTcpServer();
+  read_waiter.Init(tcp_server_socket());
+
+  const std::string script = base::StringPrintf(
+      "writeTcp({remoteAddress: '127.0.0.1', remotePort: %d}, %u)",
+      listening_port, kRequiredBytes);
+  EXPECT_EQ("write succeeded", EvalJs(shell(), script));
+  read_waiter.Await();
+}
+
 IN_PROC_BROWSER_TEST_F(DirectSocketsBrowserTest, CloseTcp) {
   EXPECT_TRUE(NavigateToURL(shell(), GetTestPageURL()));
 

content/test/data/direct_sockets/index.html

@@ -16,6 +16,35 @@
       }
     }
 
+    async function writeLoop(writer, requiredBytes) {
+      if (!(writer instanceof WritableStreamDefaultWriter))
+        return 'write failed: writer is not a WritableStreamDefaultWriter';
+      let bytesWritten = 0;
+      let chunkLength = 0;
+      while (bytesWritten < requiredBytes) {
+        chunkLength = Math.min(chunkLength + 1,
+                               requiredBytes - bytesWritten);
+        let chunk = new Uint8Array(chunkLength);
+        for (let index = 0; index < chunkLength; ++index) {
+          chunk[index] = bytesWritten % 256;
+          ++bytesWritten;
+        }
+        await writer.ready;
+        await writer.write(chunk);
+      }
+      return 'write succeeded';
+    }
+
+    async function writeTcp(options, requiredBytes) {
+      try {
+        let tcpSocket = await navigator.openTCPSocket(options);
+        let writer = tcpSocket.writable.getWriter();
+        return await writeLoop(writer, requiredBytes);
+      } catch(error) {
+        return ('writeTcp failed: ' + error);
+      }
+    }
+
     async function closeTcp(options, closeWriter) {
       try {
         let tcpSocket = await navigator.openTCPSocket(options);