[serial] Set high water mark for SerialPortUnderlyingSink to 1

The high water mark for the count queuing strategy used for a
SerialPort's WritableStream needs to be at least one or else the
desiredSize property of the writer will always be zero. This prevented
methods like pipeTo() from working.

New tests for piping readable and writable through a TransformStream
have been added.

Bug: 893334
Change-Id: Ie7dc63e3a3272aab369c510a82b6b1fb9cd65f64
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/1846074
Commit-Queue: Reilly Grant <reillyg@chromium.org>
Commit-Queue: Ovidio de Jesús Ruiz-Henríquez <odejesush@chromium.org>
Reviewed-by: Ovidio de Jesús Ruiz-Henríquez <odejesush@chromium.org>
Auto-Submit: Reilly Grant <reillyg@chromium.org>
Cr-Commit-Position: refs/heads/master@{#703931}

third_party/blink/renderer/modules/serial/serial_port.cc

@@ -430,8 +430,12 @@ void SerialPort::InitializeReadableStream(
   DCHECK(!readable_);
   underlying_source_ = MakeGarbageCollected<SerialPortUnderlyingSource>(
       script_state, this, std::move(readable_pipe));
+  // Ideally the stream would report the number of bytes that can be read from
+  // the underlying Mojo data pipe. As an approximation the high water mark is
+  // set to 0 so that data remains in the pipe rather than being queued in the
+  // stream and thus adding an extra layer of buffering.
   readable_ = ReadableStream::CreateWithCountQueueingStrategy(
-      script_state, underlying_source_, 0);
+      script_state, underlying_source_, /*high_water_mark=*/0);
 }
 
 void SerialPort::InitializeWritableStream(
@@ -441,8 +445,13 @@ void SerialPort::InitializeWritableStream(
   DCHECK(!writable_);
   underlying_sink_ = MakeGarbageCollected<SerialPortUnderlyingSink>(
       this, std::move(writable_pipe));
+  // Ideally the stream would report the number of bytes that could be written
+  // to the underlying Mojo data pipe. As an approximation the high water mark
+  // is set to 1 so that the stream appears ready but producers observing
+  // backpressure won't queue additional chunks in the stream and thus add an
+  // extra layer of buffering.
   writable_ = WritableStream::CreateWithCountQueueingStrategy(
-      script_state, underlying_sink_, 0);
+      script_state, underlying_sink_, /*high_water_mark=*/1);
 }
 
 void SerialPort::OnGetSignals(

third_party/blink/web_tests/serial/serialPort_readable.html

@@ -98,4 +98,23 @@ serial_test(async (t, fake) => {
   assert_equals(undefined, value);
 }, 'Parity error closes readable and replaces it with a new stream');
 
+serial_test(async (t, fake) => {
+  const { port, fakePort } = await getFakeSerialPort(fake);
+  // Select a buffer size larger than the amount of data transferred.
+  await port.open({ baudrate: 9600, buffersize: 64 });
+
+  fakePort.write(new TextEncoder().encode("Hello world!"));
+
+  const reader = port.readable.pipeThrough(new TextDecoderStream()).getReader();
+  let { value, done } = await reader.read();
+  assert_false(done);
+  assert_equals("Hello world!", value);
+
+  port.close();
+
+  ({ value, done } = await reader.read());
+  assert_true(done);
+  assert_equals(undefined, value);
+}, 'Can pipe readable through a transform stream.')
+
 </script>

third_party/blink/web_tests/serial/serialPort_writable.html

@@ -68,4 +68,22 @@ serial_test(async (t, fake) => {
   compareArrays(data, value);
 }, 'Can read a large amount of data');
 
+serial_test(async (t, fake) => {
+  const { port, fakePort } = await getFakeSerialPort(fake);
+
+  await port.open({ baudrate: 9600 });
+  assert_true(port.writable instanceof WritableStream);
+
+  const encoder = new TextEncoderStream();
+  encoder.readable.pipeTo(port.writable);
+  const writer = encoder.writable.getWriter();
+  let writePromise = writer.write("Hello world!");
+  let { value, done } = await fakePort.read();
+  await writePromise;
+  assert_equals("Hello world!", new TextDecoder().decode(value));
+
+  port.close();
+  assert_equals(null, port.writable);
+}, 'Can pipe a stream to writable');
+
 </script>