lib/streams.js
function transformWithCancel(cancel) {
  let pulled = false;
  let backpressureChangePromiseResolve;
  let outputController;
  return {
    readable: new ReadableStream({
      start(controller) {
        outputController = controller;
      },
      pull() {
        if (backpressureChangePromiseResolve) {
          backpressureChangePromiseResolve();
        } else {
          pulled = true;
        }
      },
      cancel
    }, {highWaterMark: 0}),
    writable: new WritableStream({
      write: async function(chunk) {
        outputController.enqueue(chunk);
        if (!pulled) {
          await new Promise(resolve => {
            backpressureChangePromiseResolve = resolve;
          });
          backpressureChangePromiseResolve = null;
        } else {
          pulled = false;
        }
      },
      close: outputController.close.bind(outputController),
      abort: outputController.error.bind(outputController)
    })
  };
}
function transformWithCancel(cancel) {
  let pulled = false;
  let backpressureChangePromiseResolve;
  let outputController;
  return {
    readable: new ReadableStream({
      start(controller) {
        outputController = controller;
      },
      pull() {
        if (backpressureChangePromiseResolve) {
          backpressureChangePromiseResolve();
        } else {
          pulled = true;
        }
      },
      cancel
    }, {highWaterMark: 0}),
    writable: new WritableStream({
      write: async function(chunk) {
        outputController.enqueue(chunk);
        if (!pulled) {
          await new Promise(resolve => {
            backpressureChangePromiseResolve = resolve;
          });
          backpressureChangePromiseResolve = null;
        } else {
          pulled = false;
        }
      },
      close: outputController.close.bind(outputController),
      abort: outputController.error.bind(outputController)
    })
  };
}
function transformWithCancel ( cancel ) { let pulled = false ; let backpressureChangePromiseResolve ; let outputController ; return { readable : new ReadableStream ( { start ( controller ) { outputController = controller ; } , pull ( ) { if ( backpressureChangePromiseResolve ) { backpressureChangePromiseResolve ( ) ; } else { pulled = true ; } } , cancel } , { highWaterMark : 0 } ) , writable : new WritableStream ( { write : async function ( chunk ) { outputController . enqueue ( chunk ) ; if ( ! pulled ) { await new Promise ( resolve => { backpressureChangePromiseResolve = resolve ; } ) ; backpressureChangePromiseResolve = null ; } else { pulled = false ; } } , close : outputController . close . bind ( outputController ) , abort : outputController . error . bind ( outputController ) } ) } ; }
Create a cancelable TransformStream.
@param {Function} cancel
@returns {TransformStream}
Create a cancelable TransformStream .