PDF.js 源码解读: MessageHandler

// 信封通用部分
{
  sourceName:   "main" | "worker" | "d1" | "d1_worker", // 消息来源
  targetName:   "main" | "worker" | "d1" | "d1_worker", // 消息目的
  // ... 具体形态的字段
}

if (data.stream) { /* 流式控制或数据 */ }
if (data.callback) { /* callbackId 的回包 */ }
// 剩下的才是 action

constructor(sourceName, targetName, comObj) {
  this.sourceName = sourceName;
  this.targetName = targetName;
  this.comObj = comObj;
  this.callbackId = 1;
  this.streamId = 1;
  this.streamSinks = Object.create(null);       // streamId → sink（我这边是数据接收端）
  this.streamControllers = Object.create(null); // streamId → controller（我这边是数据生产端）
  this.callbackCapabilities = Object.create(null); // callbackId → Promise.withResolvers()
  this.actionHandler = Object.create(null);     // actionName → handler
  comObj.addEventListener("message", this.#onMessage.bind(this), {
    signal: this.#messageAC.signal,
  });
}

on(actionName, handler) {
  if (typeof PDFJSDev === "undefined" || PDFJSDev.test("TESTING")) {
    assert(typeof handler === "function",
      'MessageHandler.on: Expected "handler" to be a function.');
  }
  const ah = this.actionHandler;
  if (ah[actionName]) {
    throw new Error(`There is already an actionName called "${actionName}"`);
  }
  ah[actionName] = handler;
}

send(actionName, data, transfers) {
  this.comObj.postMessage({
    sourceName: this.sourceName,
    targetName: this.targetName,
    action: actionName,
    data,
  }, transfers);
}

sendWithPromise(actionName, data, transfers) {
  const callbackId = this.callbackId++;
  const capability = Promise.withResolvers();
  this.callbackCapabilities[callbackId] = capability;
  try {
    this.comObj.postMessage({
      sourceName: this.sourceName,
      targetName: this.targetName,
      action: actionName,
      callbackId,
      data,
    }, transfers);
  } catch (ex) {
    capability.reject(ex);
  }
  return capability.promise;
}

sendWithStream(actionName, data, queueingStrategy, transfers) {
  const streamId = this.streamId++,
    sourceName = this.sourceName,
    targetName = this.targetName,
    comObj = this.comObj;
 
  return new ReadableStream({
    start: controller => {
      const startCapability = Promise.withResolvers();
      this.streamControllers[streamId] = {
        controller, startCall: startCapability,
        pullCall: null, cancelCall: null, isClosed: false,
      };
      comObj.postMessage({
        sourceName, targetName,
        action: actionName, streamId, data,
        desiredSize: controller.desiredSize, // 首次推送的背压信息
      }, transfers);
      return startCapability.promise; // 告诉 ReadableStream：等我这个 promise 才开始 pull
    },
    pull: controller => {
      const pullCapability = Promise.withResolvers();
      this.streamControllers[streamId].pullCall = pullCapability;
      comObj.postMessage({ sourceName, targetName,
        stream: StreamKind.PULL, streamId,
        desiredSize: controller.desiredSize });
      return pullCapability.promise; // 告诉 ReadableStream：等我回 PULL_COMPLETE 再 pull 下一次
    },
    cancel: reason => {
      assert(reason instanceof Error, "cancel must have a valid reason");
      const cancelCapability = Promise.withResolvers();
      this.streamControllers[streamId].cancelCall = cancelCapability;
      this.streamControllers[streamId].isClosed = true;
      comObj.postMessage({ sourceName, targetName,
        stream: StreamKind.CANCEL, streamId,
        reason: wrapReason(reason) });
      return cancelCapability.promise; // 告诉 ReadableStream：等我 CANCEL_COMPLETE
    },
  }, queueingStrategy);
}

#createStreamSink(data) {
  const streamSink = {
    enqueue(chunk, size = 1, transfers) { /* 见下文 */ },
    close()  { /* 发 CLOSE */ },
    error(reason) { /* 发 ERROR */ },
    sinkCapability: Promise.withResolvers(),
    onPull: null,   // 被 PULL 消息触发
    onCancel: null, // 被 CANCEL 消息触发
    isCancelled: false,
    desiredSize: data.desiredSize, // 初始值来自 start 消息
    ready: null,
  };
  streamSink.sinkCapability.resolve();
  streamSink.ready = streamSink.sinkCapability.promise;
  this.streamSinks[streamId] = streamSink;
 
  Promise.try(action, data.data, streamSink).then(
    () => comObj.postMessage({ stream: StreamKind.START_COMPLETE, streamId, success: true }),
    reason => comObj.postMessage({ stream: StreamKind.START_COMPLETE, streamId, reason: wrapReason(reason) })
  );
}

case StreamKind.PULL:          // ReadableStream 端需要更多数据 → 触发 sink.onPull
case StreamKind.PULL_COMPLETE: // sink.onPull 执行完毕 → resolve ReadableStream 的 pull promise
case StreamKind.ENQUEUE:       // sink 发了数据 → controller.enqueue
case StreamKind.CLOSE:         // sink 结束 → controller.close
case StreamKind.ERROR:         // sink 报错 → controller.error
case StreamKind.CANCEL:        // ReadableStream 被取消 → 触发 sink.onCancel
case StreamKind.CANCEL_COMPLETE: // sink.onCancel 执行完毕 → resolve ReadableStream 的 cancel promise
case StreamKind.START_COMPLETE: // sink 创建 + action 执行完毕 → resolve ReadableStream 的 start promise

const lastDesiredSize = this.desiredSize;
this.desiredSize -= size;
if (lastDesiredSize > 0 && this.desiredSize <= 0) {
  this.sinkCapability = Promise.withResolvers();
  this.ready = this.sinkCapability.promise;
}

if (streamSink.desiredSize <= 0 && data.desiredSize > 0) {
  streamSink.sinkCapability.resolve();
}
streamSink.desiredSize = data.desiredSize;

async #deleteStreamController(streamController, streamId) {
  await Promise.allSettled([
    streamController.startCall?.promise,
    streamController.pullCall?.promise,
    streamController.cancelCall?.promise,
  ]);
  delete this.streamControllers[streamId];
}

#onMessage({ data }) {
  if (data.targetName !== this.sourceName) return; // 软路由
 
  if (data.stream) {
    this.#processStreamMessage(data);   // StreamKind 控制或数据
    return;
  }
  if (data.callback) {
    // 纯回包：从 callbackCapabilities 取出 capability，按 DATA/ERROR resolve 或 reject
    return;
  }
 
  // 剩下的是 action
  const action = this.actionHandler[data.action];
  if (!action) throw new Error(`Unknown action from worker: ${data.action}`);
 
  if (data.callbackId) {
    // action + callbackId：执行后把结果/异常回成 callback
    Promise.try(action, data.data).then(
      result => comObj.postMessage({ callback: DATA, callbackId, data: result }),
      reason => comObj.postMessage({ callback: ERROR, callbackId, reason: wrapReason(reason) })
    );
    return;
  }
  if (data.streamId) {
    // action + streamId：创建 sink，不回 action 形态的答复
    this.#createStreamSink(data);
    return;
  }
  action(data.data); // 纯 action，fire-and-forget
}

function wrapReason(ex) {
  // 已经是正确类型 → 直接返回（避免二次构造）
  if (ex instanceof AbortException || ex instanceof InvalidPDFException ||
      ex instanceof PasswordException || ex instanceof ResponseException ||
      ex instanceof UnknownErrorException) {
    return ex;
  }
  if (!(ex instanceof Error || (typeof ex === "object" && ex !== null))) {
    unreachable('wrapReason: Expected "reason" to be a (possibly cloned) Error.');
  }
  switch (ex.name) {
    case "AbortException":     return new AbortException(ex.message);
    case "InvalidPDFException": return new InvalidPDFException(ex.message);
    case "PasswordException":   return new PasswordException(ex.message, ex.code);
    case "ResponseException":   return new ResponseException(ex.message, ex.status, ex.missing);
    case "UnknownErrorException": return new UnknownErrorException(ex.message, ex.details);
  }
  return new UnknownErrorException(ex.message, ex.toString());
}

destroy() {
  this.#messageAC?.abort();
  this.#messageAC = null;
}

const port = new LoopbackPort();
const id = `fake${PDFWorker.#fakeWorkerId++}`;
const workerHandler = new MessageHandler(id + "_worker", id, port);
WorkerMessageHandler.setup(workerHandler, port);
this.#messageHandler = new MessageHandler(id, id + "_worker", port);