BiDi: Support Send/Receive for "conversational" APIs

[bhollis]

For APIs that stream in only one direction (stream request, single response or single request, stream response), the AsyncIterator abstraction works really well. I can feed a generator into the request, or consume the result with for await, and it's quite convenient.

However, for bidirectional streaming APIs (stream request, stream response), this is not as convenient, since you need to provide an AsyncIterator for the request, and then consume the AsyncIterator that's returned. This might be OK for simple "streaming-pipeline" cases where the request and response have nothing to do with each other. However, we make use of "conversational" bidi streaming APIs, where the client will stream a message to the server, then expect a corresponding response, and so on back and forth until both sides hang up. Effectively, each request and response message are paired up to form a little mini-RPC within the broader context of a single API call. Our case is supporting an interactive database transaction API, but you can imagine lots of use cases for this sort of API.

We've managed to get things working by building a mildly complicated custom AsyncIterator class that can be passed as the request, which will then block on next() and allow request messages to be asynchronously enqueued. It allows us to effectively tie together the request and response streams so they can interact with each other, but it's a lot of tricky coordination code.

For this type of API, something closer to how connect-go manages bidi streams would be much easier to deal with. In connect-go, bidi APIs return a BidiStreamForClient, which has Send and Receive methods that can be used to send or receive a single message (along with CloseRequest and CloseResponse to hang up each end). This is much easier to use for these conversational APIs, since we now have a single handle that can be used to send a request, then await a response.

Would folks be open to changing connect-es to use a similar API for bidi streams? Or to have this sort of API as an alternative option?

[srikrsna-buf]

Hey! Thank you for the detailed write up. The connect library also has a custom AsyncIterable that can be used as the request to a bidi rpc. Here's an example demonstrating its usage:

connect-es/packages/connect-node/src/node-readme.spec.ts

Lines 269 to 279 in dc7d75a

	const req = createWritableIterable<PartialMessage<ConverseRequest>>();
	try {
	const res = client.converse(req);
	for await (const next of res) {
	expect(next.sentence).toBe("ping");
	await req.write({ sentence: "pong" });
	break;
	}
	} finally {
	req.close();
	}

Can you give this a try and let us know if it works for you?

[bhollis]

Thanks, I don't think I've seen that documented anywhere. It's roughly the same as what I built on my own, but I'll definitely switch to this implementation. Mine has suffered from what might be a bug in my implementation or may be a bug in connect-es v2 where I cannot drain out responses once I'm done without hitting an error. I already owe you a minimal repro for that.

My use case is a bit different from how that example works - in my case I'm doing what I believe should be a common pattern, where we send a request and then wait for the response, rather than waiting for the response first and then sending a request. That just means we can't drive things with a for await loop.

Regardless of all this, I still think using AsyncIterable for bidi APIs isn't a great fit vs. the send/receive APIs that I have in Go, Python, etc. It's just far easier to handle two queues directly than having to go through these iterables.

[srikrsna-buf]

You don't need a for await for AsyncIterable, that is just syntactic sugar around them. You can use them like this:

const res = client.converse(req);
const it = res[Symbol.asyncIterator]();
await req.write({ sentence: "pong" });
const next = await it.next();
if (next.done !== true) {
  expect(next.value.sentence).toBe("ping");
}

[bhollis]

Of course, and that's how I'm using it. I'm just pointing out that because you can't really make effective use of for await, using AsyncIterable isn't that valuable.