Wrapping RPC Client with Async Generator

[harrysolovay]

What is the best approach for getting an async iterator over incoming messages with an RPC client? Ideally, when the iteration is ended, we can trigger a client-specific callback (to close the connection, for example).

for await (const message of rpc.iter(client)) {
  // utilize
  if (isLastMessage(message)) {
    break;
  }
}

// `client` is no longer needed, so the connection is closed

[tjjfvi]

export function rpcIter(client: RpcClient): AsyncIterableIterator<IngressMessage> {
  const valueQueue: IngressMessage[] = [];
  const callbackQueue: Array<(value: IteratorYieldResult<IngressMessage>) => void> = [];
  let finished = false;

  const endListen = client.listen((value) => {
    const callback = callbackQueue.shift();
    if (callback) {
      callback({ done: false, value });
    } else {
      valueQueue.push(value);
    }
  });

  return {
    next() {
      if (valueQueue.length) {
        return Promise.resolve({
          done: false,
          value: valueQueue.shift()!,
        });
      }

      if (finished) {
        return Promise.resolve({ done: true, value: undefined });
      }

      return new Promise((resolve) => callbackQueue.push(resolve));
    },

    return() {
      endListen();
      // client.close()
      finished = true;
      return Promise.resolve({ done: true, value: undefined });
    },

    [Symbol.asyncIterator]() {
      return this;
    },
  };
}

[harrysolovay]

This is implemented. See util here and example here. Not sure if this is a DX we want to further pursue.