@@ -12,6 +12,7 @@ const {
1212 DataViewPrototypeGetByteLength,
1313 ErrorCaptureStackTrace,
1414 FunctionPrototypeBind,
15+ FunctionPrototypeCall,
1516 Number,
1617 ObjectDefineProperties,
1718 ObjectKeys,
@@ -210,6 +211,7 @@ const {
210211 kSendHeaders,
211212 kSessionApplication,
212213 kSessionTicket,
214+ kStopSending,
213215 kTrailers,
214216 kVersionNegotiation,
215217 kInspect,
@@ -991,6 +993,14 @@ setCallbacks({
991993 this[kOwner][kReset](error);
992994 },
993995
996+ onStreamStopSending(error) {
997+ if (error !== undefined) {
998+ error = convertQuicError(error);
999+ }
1000+ debug('stream stop sending callback', this[kOwner], error);
1001+ this[kOwner][kStopSending](error);
1002+ },
1003+
9941004 onStreamHeaders(headers, kind) {
9951005 // Called when the stream C++ handle has received a full block of headers.
9961006 debug(`stream ${this[kOwner].id} headers callback`, headers, kind);
@@ -1576,6 +1586,7 @@ class QuicStream {
15761586 onerror: undefined,
15771587 onblocked: undefined,
15781588 onreset: undefined,
1589+ onstopsending: undefined,
15791590 onheaders: undefined,
15801591 ontrailers: undefined,
15811592 oninfo: undefined,
@@ -1781,6 +1792,25 @@ class QuicStream {
17811792 }
17821793 }
17831794
1795+ /** @type {OnStreamErrorCallback} */
1796+ get onstopsending() {
1797+ assertIsQuicStream(this);
1798+ return this.#inner.onstopsending;
1799+ }
1800+
1801+ set onstopsending(fn) {
1802+ assertIsQuicStream(this);
1803+ const inner = this.#inner;
1804+ if (fn === undefined) {
1805+ inner.onstopsending = undefined;
1806+ inner.state.wantsStopSending = false;
1807+ } else {
1808+ validateFunction(fn, 'onstopsending');
1809+ inner.onstopsending = FunctionPrototypeBind(fn, this);
1810+ inner.state.wantsStopSending = true;
1811+ }
1812+ }
1813+
17841814 /** @type {OnHeadersCallback} */
17851815 get onheaders() {
17861816 assertIsQuicStream(this);
@@ -2143,6 +2173,19 @@ class QuicStream {
21432173 }
21442174 };
21452175
2176+ const onStopSending = stream[kStopSending];
2177+ stream[kStopSending] = (reason) => {
2178+ if (!closed && !errored) {
2179+ errored = true;
2180+ error = reason;
2181+ if (drainWakeup != null) {
2182+ drainWakeup.reject(error);
2183+ drainWakeup = null;
2184+ }
2185+ }
2186+ FunctionPrototypeCall(onStopSending, stream, reason);
2187+ };
2188+
21462189 // A note on backpressure handling: per the stream/iter spec, the default
21472190 // backpressure policy for writers is strict, meaning that if the stream
21482191 // signals backpressure additional writes are rejected until the buffer has
@@ -2543,6 +2586,7 @@ class QuicStream {
25432586 inner.pendingClose.resolve = undefined;
25442587 inner.onblocked = undefined;
25452588 inner.onreset = undefined;
2589+ inner.onstopsending = undefined;
25462590 inner.onheaders = undefined;
25472591 inner.onerror = undefined;
25482592 inner.ontrailers = undefined;
@@ -2596,6 +2640,12 @@ class QuicStream {
25962640 safeCallbackInvoke(inner.onreset, this, error);
25972641 }
25982642
2643+ [kStopSending](error) {
2644+ const inner = this.#inner;
2645+ assert(inner.onstopsending, 'Unexpected stop sending event');
2646+ safeCallbackInvoke(inner.onstopsending, this, error);
2647+ }
2648+
25992649 [kHeaders](headers, kind) {
26002650 const block = parseHeaderPairs(headers);
26012651 const kindName = kHeadersKindName[kind] ?? kind;
0 commit comments