...

nodejs · Apr 26, 2021 · ebf932c · ebf932c
1 parent 6e8df73
commit ebf932c
Show file tree

Hide file tree

Showing 3 changed files with 341 additions and 55 deletions.
diff --git a/lib/internal/quic/binding.js b/lib/internal/quic/binding.js
@@ -67,8 +67,12 @@ function onStreamHeaders() {}
 
 function onStreamBlocked() {}
 
+let initialized = false;
+
 module.exports = {
   initializeBinding() {
+    if (initialized) return;
+    initialized = true;
     initializeCallbacks({
       onEndpointClose,
       onEndpointDone,

diff --git a/lib/internal/quic/config.js b/lib/internal/quic/config.js
@@ -118,7 +118,6 @@ const kRandomConnectionIdStrategy = new RandomConnectionIDStrategy();
  * @property {string} [ccAlgorithm]
  * @property {UDPOptions} [udp]
  * @property {ArrayBuffer|TypedArray|DataView} [resetTokenSecret]
- * @property {AbortSignal} [signal]
  *
  * @typedef {Object} PreferredAddress
  * @property {SocketAddressOrOptions} [ipv4]
@@ -176,6 +175,22 @@ const kRandomConnectionIdStrategy = new RandomConnectionIDStrategy();
  * @property {AbortSignal} [signal]
  * @typedef {EndpointConfig|EndpointConfigOptions} EndpointConfigOrOptions
  * @typedef {SessionConfig|SessionConfigOptions} SessionConfigOrOptions
+ *
+ * @typedef {import('../blob.js').Blob} Blob
+ * @typedef {import('stream').Readable} Readable
+ * @typedef {ArrayBuffer|TypedArray|DataView|Blob|Readable|string} StreamPayload
+ * @typedef {Object} StreamOptionsInit
+ * @property {boolean} [unidirectional]
+ * @property {Object|Map<string,string>} [headers]
+ * @property {Object|Map<string,string>} [trailers]
+ * @property {StreamPayload|Promise<StreamPayload>} [body]
+ * @property {string} [encoding]
+ *
+ * @typedef {Object} ResponseOptionsInit
+ * @property {Object|Map<string,string>} [headers]
+ * @property {Object|Map<string,string>} [trailers]
+ * @property {StreamPayload|Promise<StreamPayload>} [body]
+ * @property {string} [encoding]
  */
 class EndpointConfig {
   [kType] = 'endpoint-config';
@@ -214,7 +229,6 @@ class EndpointConfig {
       ccAlgorithm,
       udp,
       resetTokenSecret,
-      signal,
     } = options;
 
     if (!SocketAddress.isSocketAddress(address)) {
@@ -377,13 +391,6 @@ class EndpointConfig {
       }
     }
 
-    if (signal !== undefined) {
-      validateAbortSignal(signal, 'options.signal');
-      if (signal.aborted)
-        throw new AbortError();
-    }
-
-
     this[kOptions] = {
       address,
       retryTokenExpiration,
@@ -408,7 +415,6 @@ class EndpointConfig {
       sendBufferSize,
       ttl,
       resetTokenSecret: resetTokenSecret || '(generated)',
-      signal,
     };
 
     this[kHandle] = new ConfigObject(
@@ -455,11 +461,6 @@ class EndpointConfig {
 
     return `EndpointConfig ${inspect(this[kOptions], opts)}`;
   }
-
-  /** @type {AbortSignal} */
-  get signal() {
-    return this[kOptions].signal;
-  }
 }
 
 class SessionConfig {
@@ -863,6 +864,124 @@ class SessionConfig {
   }
 }
 
+class StreamOptions {
+  [kType] = 'stream-options';
+
+  /**
+   * @param {*} val
+   * @returns
+   */
+  static isStreamOptions(val) {
+    return val?.[kType] === 'stream-options';
+  }
+
+  /**
+   * @param {StreamOptionsInit} [options]
+   */
+  constructor(options = {}) {
+    validateObject(options, 'options');
+    const {
+      unidirectional = false,
+      headers,
+      trailers,
+      body,
+      encoding,
+    } = options;
+
+    this[kOptions] = {
+      unidirectional,
+      headers,
+      trailers,
+      body,
+      encoding,
+    };
+  }
+
+  get unidirectional() {
+    return this[kOptions].unidirectional;
+  }
+
+  get headers() {
+    return this[kOptions].headers;
+  }
+
+  get body() {
+    return this[kOptions].body;
+  }
+
+  get encoding() {
+    return this[kOptions].encoding;
+  }
+
+  [kInspect](depth, options) {
+    if (depth < 0)
+      return this;
+
+    const opts = {
+      ...options,
+      depth: options.depth == null ? null : options.depth - 1
+    };
+
+    return `StreamOptions ${inspect(this[kOptions], opts)}`;
+  }
+}
+
+class ResponseOptions {
+  [kType] = 'response-options';
+
+  /**
+   * @param {*} val
+   * @returns
+   */
+  static isResponseOptions(val) {
+    return val?.[kType] === 'response-options';
+  }
+
+  /**
+   * @param {ResponseOptionsInit} [options]
+   */
+  constructor(options = {}) {
+    validateObject(options, 'options');
+    const {
+      headers,
+      trailers,
+      body,
+      encoding,
+    } = options;
+
+    this[kOptions] = {
+      headers,
+      trailers,
+      body,
+      encoding,
+    };
+  }
+
+  get headers() {
+    return this[kOptions].headers;
+  }
+
+  get body() {
+    return this[kOptions].body;
+  }
+
+  get encoding() {
+    return this[kOptions].encoding;
+  }
+
+  [kInspect](depth, options) {
+    if (depth < 0)
+      return this;
+
+    const opts = {
+      ...options,
+      depth: options.depth == null ? null : options.depth - 1
+    };
+
+    return `ResponseOptions ${inspect(this[kOptions], opts)}`;
+  }
+}
+
 /**
  * @param {ArrayBuffer|TypedArray|DataView} sessionTicket
  * @param {ArrayBuffer|TypedArray|DataView} transportParams
@@ -900,6 +1019,8 @@ function validateResumeOptions(sessionTicket, transportParams) {
 module.exports = {
   EndpointConfig,
   SessionConfig,
+  StreamOptions,
+  ResponseOptions,
   validateResumeOptions,
   kSecureContext,
   kHandle,