tor-browser

RTCDataChannel-worker-shim.js (7270B)

---

// This can be used to transfer an RTCDataChannel to a worker, and expose an
// interface that acts as a passthrough to the channel on the worker. There are
// some caveats though:
// * certain kinds of error won't propagate back to the callsite, and will
//  manifest as an unhandled error (eg; webidl errors on attribute setters)
// * the event handler/GC interactions won't translate exactly (because the
//  worker code registers all handler types up front, instead of waiting for
//  registrations to happen on the wrapper)
// * RTCDataChannel.label must be unique on the worker
class WorkerBackedDataChannel extends EventTarget {
 #worker;
 #dcAttrs;
 #eventHandlers;
 #errorPromise;
 #label;

 // If you want to make multiple of these with the same worker, create first
 // with no args, and the others with first.worker
 constructor(worker = null) {
   super();
   this.#worker = worker || WorkerBackedDataChannel.makeWorker();

   // Cache of the RTTCDataChannel's state attributes, filled on init. Some are
   // updated by state updates later.
   this.#dcAttrs = null;

   // For tracking the onxxxx-style event callbacks
   // TODO: Maybe there's a simpler way to do this?
   this.#eventHandlers = new Map();

   this.#listenForEventMessages();

   // Ejection seat that we put in our promises, for cases where we've misused
   // the worker (or its code), or encountered some sort of unhandled error
   this.#errorPromise = new Promise((_, reject) => {
     // the Worker 'error' and 'messageerror' events
     const onErrorEvent = (e) => {
       switch (e.type) {
         case 'error':
         case 'messageerror':
           reject(new Error(`Worker sent ${e.type} event: ${e.message}`));
           break;
       }
     };
     this.#worker.addEventListener('error', onErrorEvent);
     this.#worker.addEventListener('messageerror', onErrorEvent);

     // Unhandled exceptions thrown by *our* worker code; not Worker error
     // events (those are handled above), and not errors thrown by
     // RTCDataChannel (those are handled in #sendRequestToWorker)
     this.#worker.addEventListener('message', ({data}) => {
       const {type, label, result} = data;
       if (type == 'workerError' &&
           (label === undefined || label == this.#label)) {
         reject(new Error(
           `Worker code sent error message: ${result}`));
       }
     });
   });
 }

 async init(channel) {
   this.#label = channel.label;

   // DO NOT GO ASYNC BEFORE THIS! Doing so will render the channel
   // untransferable.
   const initPromise = this.#sendRequestToWorker('init', channel, [channel]);
   this.#dcAttrs = await Promise.race([initPromise, this.#errorPromise]);
   return this.#dcAttrs;
 }

 static makeWorker() {
   return new Worker('/webrtc/RTCDataChannel-worker.js');
 }

 // Make it easy to put more channels on this worker
 get worker() { return this.#worker; }

 // Read-only attributes
 get label() { return this.#dcAttrs.label; }
 get ordered() { return this.#dcAttrs.ordered; }
 get maxPacketLifeTime() { return this.#dcAttrs.maxPacketLifeTime; }
 get maxRetransmits() { return this.#dcAttrs.maxRetransmits; }
 get protocol() { return this.#dcAttrs.protocol; }
 get negotiated() { return this.#dcAttrs.negotiated; }
 get id() { return this.#dcAttrs.id; }
 get readyState() { return this.#dcAttrs.readyState; }
 get bufferedAmount() { return this.#dcAttrs.bufferedAmount; }

 // Writable attributes
 set bufferedAmountLowThreshold(val) {
   this.#dcAttrs.bufferedAmountLowThreshold = val;
   this.#sendRequestToWorker('setBufferedAmountLowThreshold', val);
 }
 get bufferedAmountLowThreshold() {
   return this.#dcAttrs.bufferedAmountLowThreshold;
 }

 set binaryType(val) {
   this.#dcAttrs.binaryType = val;
   this.#sendRequestToWorker('setBinaryType', val);
 }
 get binaryType() {
   return this.#dcAttrs.binaryType;
 }

 // Note: these do not try to match the way the handler is registered on the
 // other end (eg; dc.onopen = handler is performed on the worker as an
 // addEventListener call, not as workerDc.onopen = func). This means that
 // this wrapper is not suitable for testing GC logic based on event handlers.
 set onopen(fn) { this.#setEventHandler('open', fn); }
 set onbufferedamountlow(fn) { this.#setEventHandler('bufferedamountlow', fn); }
 set onerror(fn) { this.#setEventHandler('error', fn); }
 set onclosing(fn) { this.#setEventHandler('closing', fn); }
 set onclose(fn) { this.#setEventHandler('close', fn); }
 set onmessage(fn) { this.#setEventHandler('message', fn); }

 async send(data) {
   return this.#sendRequestToWorker('send', data);
 }

 async close() {
   return this.#sendRequestToWorker('close');
 }

 // Used to refresh readyState, bufferedAmount, and id
 async updateState() {
   const resp = await Promise.race([this.#sendRequestToWorker('queryState'), this.#errorPromise]);
   this.#dcAttrs.readyState = resp.readyState;
   this.#dcAttrs.bufferedAmount = resp.bufferedAmount;
   this.#dcAttrs.id = resp.id;
   return resp;
 }

 #setEventHandler(type, handler) {
   // Listener might not exist, removeEventListener doesn't care
   this.removeEventListener(type, this.#eventHandlers.get(type));
   this.#eventHandlers.delete(type);
   if (handler) {
     this.addEventListener(type, handler);
     this.#eventHandlers.set(type, handler);
   }
 }

 #listenForEventMessages() {
   this.#worker.addEventListener('message', ({data}) => {
     const { type, label, result } = data;

     const eventTypes =
       ['open', 'bufferedamountlow', 'error', 'closing', 'close', 'message'];

     if (label == this.#label && eventTypes.includes(type)) {
       let e;
       if (type == 'message') {
         const {data, origin} = result;
         e = new MessageEvent(type, {data, origin});
       } else {
         e = new Event(type);
       }
       this.dispatchEvent(e);
     }
   });
 }


 #sendRequestToWorker(type, arg, transferOrOptions) {
   if (!this.#label) {
     throw new Error('RTCDataChannel worker shim not initialized!');
   }

   return new Promise((resolve, reject) => {
     // We currently assume that if multiple requests of the same type are
     // sent, they get responses in the same order. That probably won't
     // change, but if it does we'll need a transaction id.
     const msg = { type, label: this.#label, arg };
     const responseType = `${type}Response`

     const onResponse = ({data}) => {
       const {type, label, result} = data;
       if (type == responseType && label == this.#label) {
         this.#worker.removeEventListener('message', onResponse);
         if (result?.error) {
           // Error thrown by RTCDataChannel, other error cases are handled by
           // the code in this.#errorPromise
           // TODO: Maybe re-synthesize the specific error thrown by the
           // RTCDataChannel?
           reject(new Error(`RTCDataChannel error: ${result.error.message}`));
         } else {
           resolve(result);
         }
       }
     };

     this.#worker.addEventListener('message', onResponse);
     this.#worker.postMessage(msg, transferOrOptions);
   });
 }
}