如何关闭通过 TransformStream 传输的 Web 串行端口？

Question

我有一个 Web 串行端口，我想从中读取一些数据。我想通过使用pipeThrough 转换数据，使用TransformStream 进行一些处理（例如，将字节解码为字符串、分离出逻辑消息等）。然而，一旦我这样做了，我就无法再通过调用reader.releaseLock()来释放端口上的锁了。我在这里做错了什么？

该代码按照我的预期工作（在安全上下文中在浏览器中运行，无需依赖项）：

async serialTestWorking() {
  const port = await navigator.serial.requestPort();

  await port.open({baudRate: 9600});
  console.log("Is locked after open?", port.readable.locked);
  // Prints "Is locked after open? false"

  let reader = port.readable.getReader();
  console.log("Is locked after getReader?", port.readable.locked);
  // Prints "Is locked after getReader? true"

  reader.releaseLock();
  console.log("Is locked after releaseLock?", port.readable.locked);
  // Prints "Is locked after releaseLock? false"

  await port.close();
  console.log("Port closed");
  // Prints "Port closed"
}

但是，如果我使用 PipeThrough 通过无操作的 TransformStream 发送输出，一切都崩溃了。锁在 releaseLock 时并未释放，最终的close 无法工作。

async serialTestBroken() {
  const port = await navigator.serial.requestPort();

  await port.open({baudRate: 9600});
  console.log("Is locked after open?", port.readable.locked);
  // Prints "Is locked after open? false"

  let reader = port.readable.pipeThrough(new TransformStream()).getReader();
  console.log("Is locked after getReader?", port.readable.locked);
  // Prints "Is locked after getReader? true"

  reader.releaseLock();
  console.log("Is locked after releaseLock?", port.readable.locked);
  // Prints "Is locked after releaseLock? true"

  await port.close();
  console.log("Port closed");
  // Doesn't make it to the log line
  //throws "TypeError: Failed to execute 'close' on 'SerialPort': Cannot cancel a locked stream"
}

我在这里做错了什么？释放 TransformStream 上的锁真的不会传播到上游吗？我是否必须跟踪管道中每个变压器的实例，以便确保将它们全部解锁？

流规范表示管道在管道持续时间内锁定可读和可写流手术。

管道锁定可读和可写流，防止它们在管道操作期间被操纵。这允许实现执行重要的优化，例如直接将数据从底层源传输到底层接收器，同时绕过许多中间队列。

是否有其他方法可以表明“管道操作”已完成？

Answer 1

正如 https://web.dev/serial/#close-port 中所述，port.close() 如果串行端口的 readable 和 writable 成员已解锁，则关闭串行端口，这意味着 releaseLock() 已被解锁呼吁各自的读者和作者。

不过，使用转换流时关闭串行端口有点复杂。调用 reader.cancel()，然后调用 writer.close() 和 port.close()。这会将错误通过转换流传播到底层串行端口。由于错误传播不会立即发生，因此您需要使用之前创建的 readableStreamClosed 和 writableStreamClosed Promise 来检测 port.ready 和 何时发生>port.writable 已解锁。取消读取器会导致流中止；这就是为什么您必须捕获并忽略由此产生的错误。

const textDecoder = new TextDecoderStream();
const readableStreamClosed = port.readable.pipeTo(textDecoder.writable);
const reader = textDecoder.readable.getReader();

// Listen to data coming from the serial device.
while (true) {
  const { value, done } = await reader.read();
  if (done) {
    reader.releaseLock();
    break;
  }
  // value is a string.
  console.log(value);
}

// Later...

const textEncoder = new TextEncoderStream();
const writer = textEncoder.writable.getWriter();
const writableStreamClosed = textEncoder.readable.pipeTo(port.writable);

reader.cancel();
await readableStreamClosed.catch(() => { /* Ignore the error */ });

writer.close();
await writableStreamClosed;

await port.close();

Answer 2

对于可能关心的人来说，我还发现了这个解决方案，它一直有效，直到不再有效。 await thing.catch(Object) as no-op 是准确的（同时也很尴尬），但总的来说，经过 1 小时的混乱，我得出的结论是关闭端口并释放所有内容的最简单方法是传递 AbortController.signal。

pipelineTo 确实接受第二个 options 参数，该参数又接受一个 signal 属性，因此这就是我最终所做的，并最终正确释放并关闭了端口：

const writable = new WritableStream({ ... });

// pass { signal: aborter.signal } or just aborter
const aborter = new AbortController;
const readerClosed = port.readable.pipeTo(writable, aborter);

// later on ...
aborter.abort(); // that's it ...
// the rest of the dance
writer.close();
await writerClosed;
await readerClosed.catch(Object);
await port.close();

我希望这个提示将帮助那些像我今天早上一样疯狂的人。