diff --git a/packages/transport-http/src/transport.ts b/packages/transport-http/src/transport.ts index d13b7b26..dae3baaa 100644 --- a/packages/transport-http/src/transport.ts +++ b/packages/transport-http/src/transport.ts @@ -24,7 +24,7 @@ function toArrayBuffer(uint8array: Uint8Array): ArrayBuffer { export class TransportHTTP implements Types.Transport { private _toDevice: WritableStream; private _fromDevice: ReadableStream; - private _fromDeviceController?: ReadableStreamDefaultController; + private fromDeviceController?: ReadableStreamDefaultController; private url: string; private receiveBatchRequests: boolean; @@ -32,12 +32,11 @@ export class TransportHTTP implements Types.Transport { private fetching: boolean; private interval: ReturnType | undefined; - private _inflightReadController?: AbortController; - private _inflightReadStartedAt = 0; + private inflightReadController?: AbortController; - private _lastStatus: Types.DeviceStatusEnum = + private lastStatus: Types.DeviceStatusEnum = Types.DeviceStatusEnum.DeviceDisconnected; - private _closingByUser = false; + private closingByUser = false; /** * Probe the device and return a connected HTTP transport. @@ -72,13 +71,10 @@ export class TransportHTTP implements Types.Transport { try { await this.writeToRadio(chunk); } catch (error) { - if (!this._closingByUser) { + if (!this.closingByUser) { this.emitStatus( Types.DeviceStatusEnum.DeviceDisconnected, - error instanceof DOMException && - (error.name === "AbortError" || error.name === "TimeoutError") - ? "write-timeout" - : "write-error", + this.isTimeoutOrAbort(error) ? "write-timeout" : "write-error", ); } throw error; @@ -88,8 +84,15 @@ export class TransportHTTP implements Types.Transport { this._fromDevice = new ReadableStream({ start: (ctrl) => { - this._fromDeviceController = ctrl; + this.fromDeviceController = ctrl; this.emitStatus(Types.DeviceStatusEnum.DeviceConnecting); + + // Start polling immediately + void this.safePoll(); + this.interval = setInterval( + () => void this.safePoll(), + this.fetchInterval, + ); }, cancel: () => { if (this.interval) { @@ -98,29 +101,6 @@ export class TransportHTTP implements Types.Transport { this.interval = undefined; }, }); - - this.interval = setInterval(async () => { - if (this.fetching) { - return; - } - - this.fetching = true; - try { - await this.readFromRadio(); - } catch (error) { - if (!this._closingByUser) { - this.emitStatus( - Types.DeviceStatusEnum.DeviceDisconnected, - error instanceof DOMException && - (error.name === "AbortError" || error.name === "TimeoutError") - ? "read-timeout" - : "read-error", - ); - } - } finally { - this.fetching = false; - } - }, this.fetchInterval); } /** Poll `/api/v1/fromradio` and enqueue incoming packets. */ @@ -129,8 +109,7 @@ export class TransportHTTP implements Types.Transport { while (readBuffer.byteLength > 0) { const inflight = new AbortController(); - this._inflightReadController = inflight; - this._inflightReadStartedAt = Date.now(); + this.inflightReadController = inflight; const signal = AbortSignal.any([ inflight.signal, @@ -152,20 +131,18 @@ export class TransportHTTP implements Types.Transport { ); } - if (this._lastStatus === Types.DeviceStatusEnum.DeviceDisconnected) { - this.emitStatus(Types.DeviceStatusEnum.DeviceConnected); - } + this.emitStatus(Types.DeviceStatusEnum.DeviceConnected); readBuffer = await response.arrayBuffer(); if (readBuffer.byteLength > 0) { - this._fromDeviceController?.enqueue({ + this.fromDeviceController?.enqueue({ type: "packet", data: new Uint8Array(readBuffer), }); } } finally { - this._inflightReadController = undefined; + this.inflightReadController = undefined; } } } @@ -183,13 +160,10 @@ export class TransportHTTP implements Types.Transport { throw new Error(`toradio ${response.status} ${response.statusText}`); } } catch (error) { - if (!this._closingByUser) { + if (!this.closingByUser) { this.emitStatus( Types.DeviceStatusEnum.DeviceDisconnected, - error instanceof DOMException && - (error.name === "AbortError" || error.name === "TimeoutError") - ? "write-timeout" - : "write-error", + this.isTimeoutOrAbort(error) ? "write-timeout" : "write-error", ); } throw error; @@ -210,7 +184,7 @@ export class TransportHTTP implements Types.Transport { * Stop polling and emit `DeviceDisconnected("user")`. */ disconnect(): Promise { - this._closingByUser = true; + this.closingByUser = true; if (this.interval) { clearInterval(this.interval); @@ -219,9 +193,9 @@ export class TransportHTTP implements Types.Transport { this.fetching = false; try { - this._inflightReadController?.abort(); + this.inflightReadController?.abort(); } catch {} - this._inflightReadController = undefined; + this.inflightReadController = undefined; this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "user"); @@ -229,13 +203,41 @@ export class TransportHTTP implements Types.Transport { } private emitStatus(next: Types.DeviceStatusEnum, reason?: string): void { - if (next === this._lastStatus) { + if (next === this.lastStatus) { return; } - this._lastStatus = next; - this._fromDeviceController?.enqueue({ + this.lastStatus = next; + this.fromDeviceController?.enqueue({ type: "status", data: { status: next, reason }, }); } + + private isTimeoutOrAbort(err: unknown): boolean { + return ( + (err instanceof DOMException && + (err.name === "AbortError" || err.name === "TimeoutError")) || + (err instanceof Error && + (err.name === "AbortError" || err.name === "TimeoutError")) + ); + } + + private async safePoll(): Promise { + if (this.fetching) { + return; + } + this.fetching = true; + try { + await this.readFromRadio(); + } catch (error) { + if (!this.closingByUser) { + this.emitStatus( + Types.DeviceStatusEnum.DeviceDisconnected, + this.isTimeoutOrAbort(error) ? "read-timeout" : "read-error", + ); + } + } finally { + this.fetching = false; + } + } } diff --git a/packages/transport-node-serial/src/transport.ts b/packages/transport-node-serial/src/transport.ts index 2c60933b..fc5df1a7 100644 --- a/packages/transport-node-serial/src/transport.ts +++ b/packages/transport-node-serial/src/transport.ts @@ -12,17 +12,18 @@ import { SerialPort } from "serialport"; export class TransportNodeSerial implements Types.Transport { private readonly _toDevice: WritableStream; private readonly _fromDevice: ReadableStream; - private _fromDeviceController?: ReadableStreamDefaultController; + private fromDeviceController?: ReadableStreamDefaultController; private port: SerialPort | undefined; - - private _lastStatus: Types.DeviceStatusEnum = + private pipePromise?: Promise; + private abortController: AbortController; + private lastStatus: Types.DeviceStatusEnum = Types.DeviceStatusEnum.DeviceDisconnected; - private _closingByUser = false; + private closingByUser = false; /** * Creates and connects a new TransportNode instance. * @param path - Path to the serial device - * @param baudRate - The port number for the TCP connection (defaults to 4403). + * @param baudRate - Baud rate for the serial connection (default is 115200). * @returns A promise that resolves with a connected TransportNode instance. */ public static create( @@ -60,7 +61,7 @@ export class TransportNodeSerial implements Types.Transport { this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "port-error"); }); this.port.on("close", () => { - if (this._closingByUser) { + if (this.closingByUser) { return; } this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "port-closed"); @@ -69,9 +70,12 @@ export class TransportNodeSerial implements Types.Transport { const fromDeviceSource = Readable.toWeb(port) as ReadableStream; const transformed = fromDeviceSource.pipeThrough(Utils.fromDeviceStream()); + this.abortController = new AbortController(); + const controller = this.abortController; + this._fromDevice = new ReadableStream({ start: async (ctrl) => { - this._fromDeviceController = ctrl; + this.fromDeviceController = ctrl; this.emitStatus(Types.DeviceStatusEnum.DeviceConnecting); this.emitStatus(Types.DeviceStatusEnum.DeviceConnected); @@ -83,15 +87,24 @@ export class TransportNodeSerial implements Types.Transport { if (done) { break; } - if (value) { - ctrl.enqueue(value); - } + ctrl.enqueue(value); } - } catch { - this.emitStatus( - Types.DeviceStatusEnum.DeviceDisconnected, - "read-error", - ); + ctrl.close(); + } catch (error) { + if (this.closingByUser) { + ctrl.close(); // graceful EOF on user + } else { + this.emitStatus( + Types.DeviceStatusEnum.DeviceDisconnected, + "read-error", + ); + ctrl.error( + error instanceof Error ? error : new Error(String(error)), + ); + } + try { + await transformed.cancel(); + } catch {} } finally { reader.releaseLock(); } @@ -99,13 +112,24 @@ export class TransportNodeSerial implements Types.Transport { }); // Stream for data going FROM the application TO the Meshtastic device. - const toDeviceTransform = Utils.toDeviceStream; - this._toDevice = toDeviceTransform.writable; - - toDeviceTransform.readable - .pipeTo(Writable.toWeb(port) as WritableStream) - .catch((err) => { - console.error("Error piping data to serial port:", err); + this._toDevice = Utils.toDeviceStream.writable; + + this.pipePromise = Utils.toDeviceStream.readable + .pipeTo(Writable.toWeb(port) as WritableStream, { + signal: controller.signal, + }) + .catch((error) => { + if (controller.signal.aborted || this.closingByUser) { + return; + } + console.error("Error piping data to serial port:", error); + this.emitStatus( + Types.DeviceStatusEnum.DeviceDisconnected, + "write-error", + ); + try { + this.port?.close(); + } catch {} }); } @@ -127,24 +151,29 @@ export class TransportNodeSerial implements Types.Transport { * Disconnect from the serial port and emit `DeviceDisconnected("user")`. * Safe to call multiple times. */ - disconnect() { + async disconnect() { try { - this._closingByUser = true; - this.port?.close(); - } finally { + this.closingByUser = true; this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "user"); + + this.abortController?.abort(); + await this.pipePromise?.catch(() => {}); + + try { + this.port?.close(); + } catch {} + } finally { this.port = undefined; - this._closingByUser = false; + this.closingByUser = false; } - return Promise.resolve(); } private emitStatus(next: Types.DeviceStatusEnum, reason?: string): void { - if (next === this._lastStatus) { + if (next === this.lastStatus) { return; } - this._lastStatus = next; - this._fromDeviceController?.enqueue({ + this.lastStatus = next; + this.fromDeviceController?.enqueue({ type: "status", data: { status: next, reason }, }); diff --git a/packages/transport-node/src/transport.ts b/packages/transport-node/src/transport.ts index 4b011bdb..03713d0b 100644 --- a/packages/transport-node/src/transport.ts +++ b/packages/transport-node/src/transport.ts @@ -12,13 +12,14 @@ import { Types, Utils } from "@meshtastic/core"; export class TransportNode implements Types.Transport { private readonly _toDevice: WritableStream; private readonly _fromDevice: ReadableStream; - private _fromDeviceController?: ReadableStreamDefaultController; + private fromDeviceController?: ReadableStreamDefaultController; private socket: Socket | undefined; - - private _lastStatus: Types.DeviceStatusEnum = + private pipePromise?: Promise; + private abortController: AbortController; + private lastStatus: Types.DeviceStatusEnum = Types.DeviceStatusEnum.DeviceDisconnected; - private _closingByUser = false; + private closingByUser = false; /** * Creates and connects a new TransportNode instance. @@ -53,7 +54,7 @@ export class TransportNode implements Types.Transport { this.socket.on("error", (err) => { console.error("Socket connection error:", err); - if (!this._closingByUser) { + if (!this.closingByUser) { this.emitStatus( Types.DeviceStatusEnum.DeviceDisconnected, "socket-error", @@ -62,7 +63,7 @@ export class TransportNode implements Types.Transport { }); this.socket.on("close", () => { - if (this._closingByUser) { + if (this.closingByUser) { return; // suppress close-derived disconnect in user flow } this.emitStatus( @@ -71,6 +72,9 @@ export class TransportNode implements Types.Transport { ); }); + this.abortController = new AbortController(); + const abortController = this.abortController; + const fromDeviceSource = Readable.toWeb( connection, ) as ReadableStream; @@ -78,7 +82,7 @@ export class TransportNode implements Types.Transport { this._fromDevice = new ReadableStream({ start: async (ctrl) => { - this._fromDeviceController = ctrl; + this.fromDeviceController = ctrl; this.emitStatus(Types.DeviceStatusEnum.DeviceConnecting); this.emitStatus(Types.DeviceStatusEnum.DeviceConnected); @@ -90,19 +94,22 @@ export class TransportNode implements Types.Transport { if (done) { break; } - if (value) { - ctrl.enqueue(value); - } + ctrl.enqueue(value); } ctrl.close(); } catch (error) { - if (!this._closingByUser) { + if (this.closingByUser) { + ctrl.close(); + } else { this.emitStatus( Types.DeviceStatusEnum.DeviceDisconnected, "read-error", ); + + ctrl.error( + error instanceof Error ? error : new Error(String(error)), + ); } - ctrl.error(error instanceof Error ? error : new Error(String(error))); try { await transformed.cancel(); } catch {} @@ -116,9 +123,14 @@ export class TransportNode implements Types.Transport { const toDeviceTransform = Utils.toDeviceStream; this._toDevice = toDeviceTransform.writable; - toDeviceTransform.readable - .pipeTo(Writable.toWeb(connection) as WritableStream) + this.pipePromise = toDeviceTransform.readable + .pipeTo(Writable.toWeb(connection) as WritableStream, { + signal: abortController.signal, + }) .catch((err) => { + if (abortController.signal.aborted) { + return; + } console.error("Error piping data to socket:", err); const error = err instanceof Error ? err : new Error(String(err)); this.socket?.destroy(error); @@ -139,24 +151,29 @@ export class TransportNode implements Types.Transport { * Disconnect from the TCP socket and emit `DeviceDisconnected("user")`. * Safe to call multiple times. */ - disconnect(): Promise { + async disconnect(): Promise { try { - this._closingByUser = true; + this.closingByUser = true; + this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "user"); + + this.abortController.abort(); + if (this.pipePromise) { + await this.pipePromise; + } + this.socket?.destroy(); } finally { - this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "user"); this.socket = undefined; - this._closingByUser = false; + this.closingByUser = false; } - return Promise.resolve(); } private emitStatus(next: Types.DeviceStatusEnum, reason?: string): void { - if (next === this._lastStatus) { + if (next === this.lastStatus) { return; } - this._lastStatus = next; - this._fromDeviceController?.enqueue({ + this.lastStatus = next; + this.fromDeviceController?.enqueue({ type: "status", data: { status: next, reason }, }); diff --git a/packages/transport-web-bluetooth/src/transport.ts b/packages/transport-web-bluetooth/src/transport.ts index c9d02c5c..00c0e0c4 100644 --- a/packages/transport-web-bluetooth/src/transport.ts +++ b/packages/transport-web-bluetooth/src/transport.ts @@ -21,19 +21,18 @@ function toArrayBuffer(uint8array: Uint8Array): ArrayBuffer { export class TransportWebBluetooth implements Types.Transport { private _toDevice: WritableStream; private _fromDevice: ReadableStream; - private _fromDeviceController?: ReadableStreamDefaultController; - private _isFirstWrite = true; + private fromDeviceController?: ReadableStreamDefaultController; private toRadioCharacteristic: BluetoothRemoteGATTCharacteristic; private fromRadioCharacteristic: BluetoothRemoteGATTCharacteristic; private fromNumCharacteristic: BluetoothRemoteGATTCharacteristic; private gattServer: BluetoothRemoteGATTServer; - private _lastStatus: Types.DeviceStatusEnum = + private lastStatus: Types.DeviceStatusEnum = Types.DeviceStatusEnum.DeviceDisconnected; - private _closingByUser = false; - + private closingByUser = false; + private reading = false; /** UUID for the "toRadio" write characteristic. */ static ToRadioUuid = "f75c76d2-129e-4dad-a1dd-7866124401e7"; /** UUID for the "fromRadio" read characteristic. */ @@ -43,6 +42,19 @@ export class TransportWebBluetooth implements Types.Transport { /** UUID for the Meshtastic GATT service. */ static ServiceUuid = "6ba1b218-15a8-461f-9fa8-5dcae273eafd"; + private onGattDisconnected = () => { + if (this.closingByUser) { + return; + } + this.emitStatus( + Types.DeviceStatusEnum.DeviceDisconnected, + "gatt-disconnected", + ); + }; + private onFromNumChanged = () => { + void this.readFromRadio(); + }; + /** * Prompts the user to select a Bluetooth device, connects it, and returns a transport. */ @@ -121,22 +133,34 @@ export class TransportWebBluetooth implements Types.Transport { this.gattServer = gattServer; this._fromDevice = new ReadableStream({ - start: (ctrl) => { - this._fromDeviceController = ctrl; + start: async (ctrl) => { + this.fromDeviceController = ctrl; this.emitStatus(Types.DeviceStatusEnum.DeviceConnecting); this.gattServer.device.addEventListener( "gattserverdisconnected", - () => { - if (this._closingByUser) { - return; - } - this.emitStatus( - Types.DeviceStatusEnum.DeviceDisconnected, - "gatt-disconnected", - ); - }, + this.onGattDisconnected, ); + + try { + await this.fromNumCharacteristic.startNotifications(); + this.fromNumCharacteristic.addEventListener( + "characteristicvaluechanged", + this.onFromNumChanged, + ); + this.emitStatus(Types.DeviceStatusEnum.DeviceConnected); + // prime once in case data already queued + void this.readFromRadio(); + } catch { + this.emitStatus( + Types.DeviceStatusEnum.DeviceDisconnected, + "notify-failed", + ); + this.gattServer.device.removeEventListener( + "gattserverdisconnected", + this.onGattDisconnected, + ); + } }, }); @@ -145,11 +169,7 @@ export class TransportWebBluetooth implements Types.Transport { try { const ab = toArrayBuffer(chunk); await this.toRadioCharacteristic.writeValue(ab); - - if (this._isFirstWrite) { - this._isFirstWrite = false; - setTimeout(() => this.readFromRadio(), 50); - } + void this.readFromRadio(); // ensure we read any response } catch (error) { this.emitStatus( Types.DeviceStatusEnum.DeviceDisconnected, @@ -159,25 +179,6 @@ export class TransportWebBluetooth implements Types.Transport { } }, }); - - this.fromNumCharacteristic.addEventListener( - "characteristicvaluechanged", - () => { - this.readFromRadio(); - }, - ); - - this.fromNumCharacteristic - .startNotifications() - .then(() => { - this.emitStatus(Types.DeviceStatusEnum.DeviceConnected); - }) - .catch(() => { - this.emitStatus( - Types.DeviceStatusEnum.DeviceDisconnected, - "notify-failed", - ); - }); } /** Writable stream of bytes to the device. */ @@ -195,20 +196,32 @@ export class TransportWebBluetooth implements Types.Transport { */ disconnect(): Promise { try { - this._closingByUser = true; + this.closingByUser = true; + this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "user"); + try { + this.fromNumCharacteristic.stopNotifications?.(); + } catch {} + this.fromNumCharacteristic.removeEventListener( + "characteristicvaluechanged", + this.onFromNumChanged, + ); + this.gattServer.device.removeEventListener( + "gattserverdisconnected", + this.onGattDisconnected, + ); + this.gattServer.disconnect(); } finally { - this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "user"); - this._closingByUser = false; + this.closingByUser = false; } return Promise.resolve(); } private async readFromRadio(): Promise { - const controller = this._fromDeviceController; - if (!controller) { + if (this.reading) { return; } + this.reading = true; try { let hasMoreData = true; @@ -224,23 +237,30 @@ export class TransportWebBluetooth implements Types.Transport { }); } } catch (error) { - this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "read-error"); + if (!this.closingByUser) { + this.emitStatus( + Types.DeviceStatusEnum.DeviceDisconnected, + "read-error", + ); + } throw error; + } finally { + this.reading = false; } } private emitStatus(next: Types.DeviceStatusEnum, reason?: string): void { - if (next === this._lastStatus) { + if (next === this.lastStatus) { return; } - this._lastStatus = next; - this._fromDeviceController?.enqueue({ + this.lastStatus = next; + this.fromDeviceController?.enqueue({ type: "status", data: { status: next, reason }, }); } private enqueue(output: Types.DeviceOutput): void { - this._fromDeviceController?.enqueue(output); + this.fromDeviceController?.enqueue(output); } } diff --git a/packages/transport-web-serial/src/transport.ts b/packages/transport-web-serial/src/transport.ts index eb088b3b..07c8822e 100644 --- a/packages/transport-web-serial/src/transport.ts +++ b/packages/transport-web-serial/src/transport.ts @@ -10,18 +10,18 @@ import { Types, Utils } from "@meshtastic/core"; export class TransportWebSerial implements Types.Transport { private _toDevice: WritableStream; private _fromDevice: ReadableStream; - private _fromDeviceController?: ReadableStreamDefaultController; - + private fromDeviceController?: ReadableStreamDefaultController; private connection: SerialPort; - private _portReadable: ReadableStream; - private _portWritable: WritableStream; + private pipePromise: Promise | null = null; + private abortController: AbortController; + private portReadable: ReadableStream; - private _lastStatus: Types.DeviceStatusEnum = + private lastStatus: Types.DeviceStatusEnum = Types.DeviceStatusEnum.DeviceDisconnected; - private _closingByUser = false; + private closingByUser = false; /** - * Prompts the user to select a serial port, opens it, and returns a transport. + * Create a new TransportWebSerial instance using a serial port. */ public static async create(baudRate?: number): Promise { const port = await navigator.serial.requestPort(); @@ -30,13 +30,16 @@ export class TransportWebSerial implements Types.Transport { } /** - * Creates a transport from an existing, user-provided {@link SerialPort}. + * Creates a new TransportWebSerial instance from an existing, provided {@link SerialPort}. + * Opens it if not already open. */ public static async createFromPort( port: SerialPort, baudRate?: number, ): Promise { - await port.open({ baudRate: baudRate || 115200 }); + if (!port.readable || !port.writable) { + await port.open({ baudRate: baudRate || 115200 }); + } return new TransportWebSerial(port); } @@ -45,26 +48,41 @@ export class TransportWebSerial implements Types.Transport { * @throws If the port lacks readable or writable streams. */ constructor(connection: SerialPort) { - const readable = connection.readable; - const writable = connection.writable; - if (!readable || !writable) { + if (!connection.readable || !connection.writable) { throw new Error("Stream not accessible"); } this.connection = connection; - this._portReadable = readable; - this._portWritable = writable; + this.portReadable = connection.readable; + this.abortController = new AbortController(); + const abortController = this.abortController; + + // Set up the pipe with abort signal for clean cancellation + this.pipePromise = Utils.toDeviceStream.readable + .pipeTo(connection.writable, { signal: this.abortController.signal }) + .catch((err) => { + // Ignore expected rejection when we cancel it via the AbortController. + if (abortController.signal.aborted) { + return; + } + console.error("Error piping data to serial port:", err); + this.connection.close().catch(() => {}); + this.emitStatus( + Types.DeviceStatusEnum.DeviceDisconnected, + "write-error", + ); + }); - Utils.toDeviceStream.readable.pipeTo(this._portWritable); this._toDevice = Utils.toDeviceStream.writable; + // Wrap + capture controller to inject status packets this._fromDevice = new ReadableStream({ start: async (ctrl) => { - this._fromDeviceController = ctrl; + this.fromDeviceController = ctrl; this.emitStatus(Types.DeviceStatusEnum.DeviceConnecting); - const transformed = this._portReadable.pipeThrough( + const transformed = this.portReadable.pipeThrough( Utils.fromDeviceStream(), ); const reader = transformed.getReader(); @@ -72,9 +90,6 @@ export class TransportWebSerial implements Types.Transport { const onOsDisconnect = (ev: Event) => { const { port } = ev as unknown as { port?: SerialPort }; if (port && port === this.connection) { - if (this._closingByUser) { - return; - } this.emitStatus( Types.DeviceStatusEnum.DeviceDisconnected, "serial-disconnected", @@ -91,15 +106,20 @@ export class TransportWebSerial implements Types.Transport { if (done) { break; } - if (value) { - ctrl.enqueue(value); - } + ctrl.enqueue(value); } - } catch { - this.emitStatus( - Types.DeviceStatusEnum.DeviceDisconnected, - "read-error", - ); + ctrl.close(); + } catch (error) { + if (!this.closingByUser) { + this.emitStatus( + Types.DeviceStatusEnum.DeviceDisconnected, + "read-error", + ); + } + ctrl.error(error instanceof Error ? error : new Error(String(error))); + try { + await transformed.cancel(); + } catch {} } finally { reader.releaseLock(); navigator.serial.removeEventListener("disconnect", onOsDisconnect); @@ -109,25 +129,21 @@ export class TransportWebSerial implements Types.Transport { } /** Writable stream of bytes to the device. */ - get toDevice(): WritableStream { + public get toDevice(): WritableStream { return this._toDevice; } /** Readable stream of {@link Types.DeviceOutput} from the device. */ - get fromDevice(): ReadableStream { + public get fromDevice(): ReadableStream { return this._fromDevice; } - private enqueue(output: Types.DeviceOutput): void { - this._fromDeviceController?.enqueue(output); - } - private emitStatus(next: Types.DeviceStatusEnum, reason?: string): void { - if (next === this._lastStatus) { + if (next === this.lastStatus) { return; } - this._lastStatus = next; - this._fromDeviceController?.enqueue({ + this.lastStatus = next; + this.fromDeviceController?.enqueue({ type: "status", data: { status: next, reason }, }); @@ -136,13 +152,76 @@ export class TransportWebSerial implements Types.Transport { /** * Closes the serial port and emits `DeviceDisconnected("user")`. */ - async disconnect(): Promise { + public async disconnect(): Promise { try { - this._closingByUser = true; + this.closingByUser = true; + + // Stop outbound piping + this.abortController.abort(); + if (this.pipePromise) { + await this.pipePromise; + } + + // Cancel any remaining streams + if (this._fromDevice && !this._fromDevice.locked) { + try { + await this._fromDevice.cancel(); + } catch { + // Stream cancellation might fail if already cancelled + } + } + await this.connection.close(); + } catch (error) { + // If we can't close cleanly, let the browser handle cleanup + console.warn("Could not cleanly disconnect serial port:", error); } finally { this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "user"); - this._closingByUser = false; + this.closingByUser = false; + } + } + + /** + * Reconnects the transport by creating a new AbortController and re-establishing + * the pipe connection. Only call this after disconnect() or if the connection failed. + */ + public async reconnect() { + this.emitStatus(Types.DeviceStatusEnum.DeviceConnecting, "reconnect"); + + try { + if (!this.connection.readable || !this.connection.writable) { + throw new Error("Stream not accessible"); + } + this.portReadable = this.connection.readable; + + // Create a new AbortController for the new connection + this.abortController = new AbortController(); + const abortController = this.abortController; + + // Re-establish the pipe connection + this.pipePromise = Utils.toDeviceStream.readable + .pipeTo(this.connection.writable, { + signal: this.abortController.signal, + }) + .catch((error) => { + if (abortController.signal.aborted) { + return; + } + console.error("Error piping data to serial port (reconnect):", error); + this.emitStatus( + Types.DeviceStatusEnum.DeviceDisconnected, + "write-error", + ); + }); + + this.emitStatus(Types.DeviceStatusEnum.DeviceConnected, "reconnected"); + } catch (error) { + // Couldn’t re-pipe + this.emitStatus( + Types.DeviceStatusEnum.DeviceDisconnected, + "reconnect-failed", + ); + throw error; } } }