request-response-stream

RequestResponseStream lets you take a single duplex connection (e.g. a WebSocket, a TCP socket, or anything else you can represent as a TransformStream) and use it to send a sequence of requests, each of which gets a streamed response back. It takes care of:

• reusing the same connection across requests, and reconnecting when needed
• optionally pipelining requests (writing the next request before the previous response has finished being read)
• draining or discarding in-flight responses correctly when a request is cancelled

Installation

npm install request-response-stream

Example

The library doesn't know how to open a connection itself - you provide a getConnection function that returns a TransformStream, where writing to the connection sends data and reading from it receives data. The example below imagines a connection to a line-delimited JSON RPC server, where each response ends with a {"type": "end"} chunk:

import RequestResponseStream from "request-response-stream";

// Imagined: opens a WebSocket to the server and exposes it as a
// TransformStream, writing/reading one JSON message at a time.
declare function connectToServer(): Promise<TransformStream<string, string>>;

const rrs = new RequestResponseStream<string, string>({
  getConnection: connectToServer,
  pipelineRequests: true,
  reconnectOnCancel: true,
  isOutputClose: (chunk) => JSON.parse(chunk).type === "end",
});

const response = rrs.request([
  JSON.stringify({ type: "greet", name: "World" }),
]);

for await (const chunk of response) {
  console.log(JSON.parse(chunk));
}

// Once you're done making requests, close the underlying connection.
await rrs.close();

API

new RequestResponseStream<In, Out, RequestContext = void>(options)

Creates a new RequestResponseStream. No connection is made until the first call to request().

The optional RequestContext type parameter lets each call to request() carry a value that is passed through to isOutputStart and isOutputClose, so they can make decisions specific to the request a chunk belongs to (e.g. matching a request ID embedded in the response). If it's left as its default of void, request() takes no context argument.

options has the following properties:

• getConnection: () => TransformStream<In, Out> | Promise<TransformStream<In, Out>> Called to establish a new connection whenever one is needed. The writable side is used to send requests and the readable side is used to receive responses.
• pipelineRequests: boolean Set this to true to allow the next request to be sent before the previous response is fully read. This improves latency for many small requests, providing that the server being connected to supports this feature.
• reconnectOnCancel: boolean Reconnect if there is a downstream error or if a response stream is cancelled. If this is false, we will instead just consume the rest of the response in order to get to the next response.
• isOutputStart?: (chunk: Out, context: RequestContext) => boolean Test a chunk and return true if it should be treated as the start of a response, false otherwise. Chunks read before the start of the response are discarded rather than being emitted from the response stream. If this function throws, the response stream will error and the readable stream from the connection will be cancelled with the error as the reason. If this option is not specified, the response starts immediately after the previous response.
• isOutputClose: (chunk: Out, context: RequestContext) => boolean Test a chunk and return true if it should be treated as the end of a response, false otherwise. If this function throws, the response stream will error and the readable stream from the connection will be cancelled with the error as the reason.
• makeError?: (code: ErrorCode, message: string) => Error Optionally customize the errors this library generates (see Error Codes).

rrs.request(input, context?)

Makes a single request on the underlying connection.

• input: readonly In[] | ReadableStream<In> - the chunks to write to the connection for this request, either as an array or a readable stream.
• context: RequestContext - passed through as the second argument to isOutputStart and isOutputClose for this request. Required if you specified a RequestContext type parameter, otherwise omitted.
• Returns a ReadableStream<Out> of the response chunks. If isOutputStart is set, this starts with the chunk it identified as the start of the response (any earlier chunks are discarded); otherwise it starts immediately after the previous response. It always ends with the chunk that isOutputClose identified as the end of the response.

rrs.close()

Waits for any requests that are already in flight, then closes the connection by closing the writable stream. Returns a Promise<void>.

Safe to call more than once - subsequent calls just wait for the same close to complete.

Error Handling

If getConnection throws the error is propagated to the current response, the input is cancelled (if it's a readable stream), and the next request will make a fresh connection.

If the input stream errors, the writable stream for the connection is aborted with that error, that error is also propagated to the current response, and the next request will make a fresh connection.

If an error is thrown by the connection streams, the error is propagated to the current response, and the next request will make a fresh connection.

If a downstream error is thrown or the readable stream for a response is cancelled (this can happen at any point, including before the connection has even been established):

• If reconnectOnCancel === false: the connection continues to be read until the end of that response and the next request proceeds as normal.
• If reconnectOnCancel === true: the readable stream for the connection is cancelled with that error and the next request will make a fresh connection.

If an error is thrown by the isOutputStart or isOutputClose method, the error is propagated to the current response, the readable stream for the connection is cancelled with that error and the next request will make a fresh connection.

If the readable side of the connection ends, an error is thrown for the current request, and the next request will make a fresh connection.

If pipelineRequests === true, a request may already have started writing to a connection by the time an earlier request (reading its response concurrently) discovers the connection is no longer usable (e.g. because of a downstream cancellation with reconnectOnCancel, or a read error). In that case the later request's response errors with "A previous request/response errored so this connection is no longer usable", and the next request after that will make a fresh connection. The input for that later request is not cancelled in this case, since it may already be locked by an in-progress write.

If close() has already been called, any further call to request() immediately errors with "Cannot make a request after closing the connection" and cancels the input (if it's a readable stream). Calling close() itself is safe to call more than once.

Error Codes

request-response-stream synthesizes its own errors for a handful of situations. By default these are Error instances with a code property set to one of the ErrorCode values below; you can customize how they're constructed with the makeError option. Errors that come from elsewhere (the connection itself, or a cancellation reason you provided) are propagated unchanged and won't have one of these codes unless you added it yourself.

• CONNECTION_CLOSED - "Cannot make a request after closing the connection". Thrown from request() when it's called after close() has already been called (or is in progress).
• UNEXPECTED_END - "Response stream ended unexpectedly". Thrown when the readable side of the connection ends (closes) before an end-of-response chunk (as identified by isOutputClose) was received.
• PREVIOUS_ERROR - "A previous request/response errored so this connection is no longer usable". Only possible with pipelineRequests: true: thrown for a request that already started writing to a connection by the time an earlier, concurrently-read request discovered that the connection had become unusable.