`ReadableStreams` and `Response` objects are not supported

[dario-piotrowicz]

Passing ReadableStreams (and as a result Response objects as well) is currently not supported

This is a known limitation as described in the README however I am creating this issue to hopefully track this and see if there is any progress on potentially supporting them.

Also I am opening this off of cloudflare/workers-sdk#11114 (since that is not a wrangler limitation but a capnweb one).

[pliashkou]

Hey @kentonv , thanks for providing more information in the PR. I agree with the response. Would you mind, please, sharing your vision of how it should work when you have a bit of time? Thanks again!

[benfoxall]

I'm interested in ReadableStream support.

But I was also wondering if this functionality might work as an async generator/iterator?

class MyApiServer extends RpcTarget {
  hello(name) {
    return `Hello, ${name}!`
  }
  async * countUp() {
     let i = 0;
     while(true) {
        await new Promise(r => setTimeout(r, 1000))
        yield i++;
     }
  }
}


// on the client
for await(const i of api.countUp()) {
   console.log(`counter: ${i}`)

   if(i > 10) break;
}

[nakasyou]

I found how to respond AsyncIterator, so I share it.

// server
export class MyRpc extends RpcTarget {
  stream() {
    const stream = new ReadableStream({
      start(controller) {
        controller.enqueue('h')
        controller.enqueue('e')
        controller.enqueue('l')
        controller.enqueue('l')
        controller.enqueue('o')
        controller.enqueue('!')
        controller.close()
      },
    })[Symbol.asyncIterator]()
    return {
      next: stream.next.bind(stream),
      return: stream.return?.bind(stream),
      throw: stream.throw?.bind(stream),
    }
  }
}

// client
const sess = newWebSocketRpcSession<MyRpc>('..')

const res = await sess.stream()

for await (const chunk of {
  [Symbol.asyncIterator]() {
    return res
  }
}) {
  console.log('Received chunk:', chunk)
}

Result:

[ericclemmons]

Ah, so this could be workaround for the error I'm seeing with my AsyncGenerator in my Agent:

Currently, I have:

export class Sandbox extends Base {
  async *streamTestRun() {
    for await (const event of parseSSEStream<ExecEvent>(
      await this.execStream("npm run test")
    )) {
      yield event;
    }
  }
}

But, based on @nakasyou 's comment, I should be able to wrap the *streamTestRun with an RpcTarget-friendly version?

[ericclemmons]

Ok, here's how I was able to get it working without too much extra code thanks to @nakasyou!

In my server-side oRPC code that was running into this bug, I'm able to call a "regular" async function on my Cloudflare Sandbox:

const iterator = sandbox.doSomething();
const iterable: AsyncIterable<ExecEvent> = {
  [Symbol.asyncIterator]: () => iterator,
};

for await (const event of iterable) {
  // This is streamed via https://orpc.dev/docs/integrations/tanstack-query#streamed-query-options
  yield event;
}

Inside my sandbox.ts, I have:

doSomething(): AsyncIterator<ExecEvent> {
  const iterator = this.doSomethingGenerator(testContent);

  return {
    next: iterator.next.bind(iterator),
    return: iterator.return?.bind(iterator),
    throw: iterator.throw?.bind(iterator),
  };
}

private async *doSomethingGenerator() {
  for await (const event of parseSSEStream<ExecEvent>(
    await this.execStream("npm install")
  )) {
    yield event;
  }
}

And now it's working! I get SSE and RPC functionality!