Method readLoop

src/rpc.ts:929–1056

()

  }

  private async readLoop() {
    while (!this.abortReason) {
      // Each receive needs its own abort promise so Promise.race() doesn't keep old reads.
      let readCanceled = Promise.withResolvers<never>();
      this.cancelReadLoop = readCanceled.reject;

      let raw: unknown;

      try {
        raw = await Promise.race([this.transport.receive(), readCanceled.promise]);
      } finally {
        if (this.cancelReadLoop === readCanceled.reject) {
          this.cancelReadLoop = undefined;
        }
      }
      if (this.abortReason) break;  // check again before processing

      // Only parse JSON at "string" level; otherwise message is already an object
      let msg = this.encodingLevel === "string" ? JSON.parse(raw as string) : raw;

      if (msg instanceof Array) {
        switch (msg[0]) {
          case "push":  // ["push", Expression]
            if (msg.length > 1) {
              let payload = new Evaluator(this, this.encodingLevel).evaluate(msg[1]);
              let hook = new PayloadStubHook(payload);

              // It's possible for a rejection to occur before the client gets a chance to send
              // a "pull" message or to use the promise in a pipeline. We don't want that to be
              // treated as an unhandled rejection on our end.
              hook.ignoreUnhandledRejections();

              this.exports.push({ hook, refcount: 1 });
              continue;
            }
            break;

          case "stream": {  // ["stream", Expression]
            // Like "push", but:
            // - Promise pipelining on the result is not supported.
            // - The export is automatically considered "pulled".
            // - Once the "resolve" is sent, the export is implicitly released.
            if (msg.length > 1) {
              let payload = new Evaluator(this, this.encodingLevel).evaluate(msg[1]);
              let hook = new PayloadStubHook(payload);
              hook.ignoreUnhandledRejections();

              let exportId = this.exports.length;
              this.exports.push({ hook, refcount: 1, autoRelease: true });

              // Automatically pull since stream messages are always pulled.
              this.ensureResolvingExport(exportId);
              continue;
            }
            break;
          }

          case "pipe": {  // ["pipe"]