Reconnecting a Stream after creating it - possible?

[alvestrand]

Domenic said this might be a good place to ask...
I'm designing an API where I intend to use WHATWG Streams.
I have to make a choice between a model where I show the user a bunch of readable/writable/transferStreams and tell the user "connect them all", and a model where I present a connected network of streams, and tell the user "make the changes you want".

I experimented a little, and found that the code below didn't work - but I could find no obvious API to do what I wanted; once you have connected a stream, there seems to be no convenient API for reconnecting them, short of messing around with the readers and writers explicitly.

Is there something obvious I'm missing?

Thanks in advance!

var dest1 = new WritableStream({
  write(chunk) {
    console.log('Dest 1 got ' + chunk);
  }
});

var dest2 = new WritableStream({
  write(chunk) {
    console.log('Dest 2 got ' + chunk);
  }
});

var src = new ReadableStream({
  start(controller) {
    const producer = () => {
      controller.enqueue('chunk');
      setTimeout(producer, 1000);
    };
    setTimeout(producer, 1000);
  }
});

src.pipeTo(dest1);

const pipeTo1 = () => {
  console.log('Piping to 1');
  src.pipeTo(dest1);
}

const pipeTo2 = async () => {
  console.log('Piping to 2');
  // Here there's something missing
  src.pipeTo(dest2);
}

[MattiasBuelens]

You can pass an AbortSignal to pipeTo() to stop an ongoing pipe (*). You can use preventAbort and preventCancel to prevent the aborted pipe from also cancelling the input and/or aborting the output.

Note that the pipe still waits for any pending writes to complete, and both ends remain locked until everything has been written. That means you still have to wait for the pipeTo() promise to settle (either resolve or reject) before you can start a new pipe.

let currentPipePromise = Promise.resolve();
let currentPipeAbortController = new AbortController();

async function startNewPipe(dest) {
  // Abort the previous pipe
  currentPipeAbortController.abort();
  currentPipeAbortController = new AbortController();
  // Wait for previous pipe to complete
  await currentPipePromise.catch(() => {});
  // Start new pipe, but make sure we can abort it later
  currentPipePromise = src.pipeTo(dest, {
    preventAbort: true,
    preventCancel: true,
    signal: currentPipeAbortController.signal
  });
}

startNewPipe(dest1);

const pipeTo1 = () => {
  console.log('Piping to 1');
  startNewPipe(dest1);
}

const pipeTo2 = async () => {
  console.log('Piping to 2');
  startNewPipe(dest2);
}

(*) Not yet natively supported in any browser, but you could use a polyfill.

[alvestrand]

Thanks for the tips!

Unfortunately abort() didn't cause the currentPipePromise to be rejected - the promise remained unresolved. This was true for both the polyfill version and the native version (Chrome 75).

In the polyfill version, data stopped flowing, though - in native, it just continued. So something happened.

Attaching a zip with the current code (Github doesn't want .html or .js files)
streammove.zip

[alvestrand]

If I removed the "preventAbort" from the pipeTo operation, the first attempt to change the destination of the output stream worked with the shim, but an attempt to move the stream back resulted in a DOMError.

It seems that reattaching streams is an area that hasn't been explored much.

[domenic]

I agree something is going wrong here, either in the spec or polyfill. All the pieces (AbortController, preventAbort/preventCancel) are in place, but when we put them together, the desired effect isn't showing up.

My demo is at https://boom-bath.glitch.me/stream-reconnect.html. If @MattiasBuelens or @ricea could dig in to this and let us know where things are going wrong, I'd be very grateful. Otherwise, I'll try to make some time next week.

[MattiasBuelens]

Very interesting.

Since both preventAbort and preventCancel are true, the list of actions in ReadableStreamPipeTo's abortAlgorithm is empty. The action passed to "shutdown with an action" consists of "waiting for all of the actions in actions". This is defined here.

However, the steps in that algorithm do not account for the case where promises is an empty list! It attaches a fulfillmentHandler to every promise in promises, and checks whether we're done only inside that handler. But if no promises were passed in, then no handlers are attached and we never check whether we're done.

The fix is obvious: if the list of promises passed to the "wait for all" algorithm is empty, then it should immediately call successSteps with an empty list. I'll make a PR for the reference implementation and the polyfill, and I'll try to write a spec change for the promises guide.

EDIT: Oh, and I guess this would be a good case for a web platform test as well. Even more PRs! 😅

[MattiasBuelens]

One more thing: since we're aborting the pipe, the pipe promise will reject with an AbortError instead of resolving. So we need to use .catch() instead of .then():

document.querySelector("#switch").onclick = () => {
  controller.abort();
  pipePromise.catch(() => {
    rsA.pipeTo(wsB);
  });
};

(You should probably check whether the rejection reason was actually an AbortError, but this should be good enough for demo purposes. 😛)

[domenic]

This is now fixed in the reference implementation and polyfill, and web platform tests exist too!