Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions js/qmux/src/error.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,9 @@
* socket error, decode failure, ...).
*/
export class SessionError extends Error implements Pick<WebTransportError, "source" | "streamErrorCode"> {
/** Always `"session"`: the whole session failed, not one stream. */
readonly source = "session" as const;
/** Always `null`: a session failure carries no stream error code. */
readonly streamErrorCode = null;

constructor(message: string, options?: { cause?: unknown }) {
Expand All @@ -28,7 +30,9 @@ export class SessionError extends Error implements Pick<WebTransportError, "sour
* than being formatted into the message and lost.
*/
export class StreamError extends Error implements Pick<WebTransportError, "source" | "streamErrorCode"> {
/** Always `"stream"`: one stream was reset, not the whole session. */
readonly source = "stream" as const;
/** The application error code the reset carried. */
readonly streamErrorCode: number;

constructor(code: number, message = "stream reset") {
Expand Down
12 changes: 11 additions & 1 deletion js/qmux/src/session.ts
Original file line number Diff line number Diff line change
Expand Up @@ -500,6 +500,7 @@ interface StreamFlowState {
recvConsumed: bigint;
}

/** A `WebTransport` session carried over a WebSocket, for hosts without native WebTransport. */
export default class Session implements WebTransport {
// The transport: a native `WebSocketStream` when the platform has one (real
// backpressure), otherwise the `WebSocketStream` ponyfill over a plain
Expand Down Expand Up @@ -531,27 +532,32 @@ export default class Session implements WebTransport {
// with the actual version derived from `ws.protocol`.
#version: WireFormat = "webtransport";

#protocol = "";
/** The negotiated application-level subprotocol, or empty string if none.
*
* The prefix is stripped; this returns only the application protocol name.
*/
#protocol = "";
get protocol(): string {
return this.#protocol;
}

/** Resolves once the WebSocket opens and the session can carry streams. */
readonly ready: Promise<void>;
#readyResolve: () => void;
#readyReject: (err: Error) => void;
/** Resolves with the close code and reason on a clean close, or rejects with a {@link SessionError}. */
readonly closed: Promise<WebTransportCloseInfo>;
#closedResolve: (info: WebTransportCloseInfo) => void;
#closedReject: (err: Error) => void;

/** Bidirectional streams the peer opens, in arrival order. */
readonly incomingBidirectionalStreams: ReadableStream<WebTransportBidirectionalStream>;
#incomingBidirectionalQueue: IncomingStreamQueue<WebTransportBidirectionalStream>;
/** Unidirectional streams the peer opens, in arrival order. */
readonly incomingUnidirectionalStreams: ReadableStream<ReadableStream<Uint8Array>>;
#incomingUnidirectionalQueue: IncomingStreamQueue<ReadableStream<Uint8Array>>;

/** Unreliable datagrams sent and received as QMux DATAGRAM frames. */
readonly datagrams = new Datagrams((data) => this.#sendDatagram(data));

// Flow control state
Expand Down Expand Up @@ -1755,6 +1761,7 @@ export default class Session implements WebTransport {
});
}

/** Open a bidirectional stream, waiting for the peer to grant stream credit. */
async createBidirectionalStream(options?: WebTransportSendStreamOptions): Promise<WebTransportBidirectionalStream> {
await this.ready;

Expand Down Expand Up @@ -1834,6 +1841,7 @@ export default class Session implements WebTransport {
return { readable: recvStream.readable, writable: writer };
}

/** Open a send-only stream, waiting for the peer to grant stream credit. */
async createUnidirectionalStream(options?: WebTransportSendStreamOptions): Promise<WritableStream<Uint8Array>> {
await this.ready;

Expand Down Expand Up @@ -2012,6 +2020,7 @@ export default class Session implements WebTransport {
void sleep(Nano.fromMilli(Milli(100))).then(() => this.#transportClose());
}

/** Close the session with an application code and reason; a no-op once closed. */
close(info?: { closeCode?: number; reason?: string }) {
if (this.#closed) return;
const code = info?.closeCode ?? 0;
Expand All @@ -2033,6 +2042,7 @@ export default class Session implements WebTransport {
this.#wss?.setHighWaterMark?.(this.#sendBufferSize);
}

/** Always `"default"`: the underlying TCP connection does its own congestion control. */
get congestionControl(): string {
return "default";
}
Expand Down
Loading