mirror
/
meshastic.web
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Improve stream handling

pull/790/head
philon- 11 months ago
committed by philon-
parent
9e288bac10
commit
7f2bd8d2f4
5 changed files with 342 additions and 195 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 104
      packages/transport-http/src/transport.ts
  2. 79
      packages/transport-node-serial/src/transport.ts
  3. 59
      packages/transport-node/src/transport.ts
  4. 114
      packages/transport-web-bluetooth/src/transport.ts
  5. 147
      packages/transport-web-serial/src/transport.ts

104
packages/transport-http/src/transport.ts
View File

@ -24,7 +24,7 @@ function toArrayBuffer(uint8array: Uint8Array): ArrayBuffer {
export class TransportHTTP implements Types.Transport { export class TransportHTTP implements Types.Transport {
private _toDevice: WritableStream<Uint8Array>; private _toDevice: WritableStream<Uint8Array>;
private _fromDevice: ReadableStream<Types.DeviceOutput>; private _fromDevice: ReadableStream<Types.DeviceOutput>;
private _fromDeviceController?: ReadableStreamDefaultController<Types.DeviceOutput>; private fromDeviceController?: ReadableStreamDefaultController<Types.DeviceOutput>;
private url: string; private url: string;
private receiveBatchRequests: boolean; private receiveBatchRequests: boolean;
@ -32,12 +32,11 @@ export class TransportHTTP implements Types.Transport {
private fetching: boolean; private fetching: boolean;
private interval: ReturnType<typeof setInterval> | undefined; private interval: ReturnType<typeof setInterval> | undefined;
private _inflightReadController?: AbortController; private inflightReadController?: AbortController;
private _inflightReadStartedAt = 0;
private _lastStatus: Types.DeviceStatusEnum = private lastStatus: Types.DeviceStatusEnum =
Types.DeviceStatusEnum.DeviceDisconnected; Types.DeviceStatusEnum.DeviceDisconnected;
private _closingByUser = false; private closingByUser = false;
/** /**
* Probe the device and return a connected HTTP transport. * Probe the device and return a connected HTTP transport.
@ -72,13 +71,10 @@ export class TransportHTTP implements Types.Transport {
try { try {
await this.writeToRadio(chunk); await this.writeToRadio(chunk);
} catch (error) { } catch (error) {
if (!this._closingByUser) { if (!this.closingByUser) {
this.emitStatus( this.emitStatus(
Types.DeviceStatusEnum.DeviceDisconnected, Types.DeviceStatusEnum.DeviceDisconnected,
error instanceof DOMException && this.isTimeoutOrAbort(error) ? "write-timeout" : "write-error",
(error.name === "AbortError" || error.name === "TimeoutError")
? "write-timeout"
: "write-error",
); );
} }
throw error; throw error;
@ -88,8 +84,15 @@ export class TransportHTTP implements Types.Transport {
this._fromDevice = new ReadableStream<Types.DeviceOutput>({ this._fromDevice = new ReadableStream<Types.DeviceOutput>({
start: (ctrl) => { start: (ctrl) => {
this._fromDeviceController = ctrl; this.fromDeviceController = ctrl;
this.emitStatus(Types.DeviceStatusEnum.DeviceConnecting); this.emitStatus(Types.DeviceStatusEnum.DeviceConnecting);
// Start polling immediately
void this.safePoll();
this.interval = setInterval(
() => void this.safePoll(),
this.fetchInterval,
);
}, },
cancel: () => { cancel: () => {
if (this.interval) { if (this.interval) {
@ -98,29 +101,6 @@ export class TransportHTTP implements Types.Transport {
this.interval = undefined; 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. */ /** Poll `/api/v1/fromradio` and enqueue incoming packets. */
@ -129,8 +109,7 @@ export class TransportHTTP implements Types.Transport {
while (readBuffer.byteLength > 0) { while (readBuffer.byteLength > 0) {
const inflight = new AbortController(); const inflight = new AbortController();
this._inflightReadController = inflight; this.inflightReadController = inflight;
this._inflightReadStartedAt = Date.now();
const signal = AbortSignal.any([ const signal = AbortSignal.any([
inflight.signal, 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(); readBuffer = await response.arrayBuffer();
if (readBuffer.byteLength > 0) { if (readBuffer.byteLength > 0) {
this._fromDeviceController?.enqueue({ this.fromDeviceController?.enqueue({
type: "packet", type: "packet",
data: new Uint8Array(readBuffer), data: new Uint8Array(readBuffer),
}); });
} }
} finally { } 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}`); throw new Error(`toradio ${response.status} ${response.statusText}`);
} }
} catch (error) { } catch (error) {
if (!this._closingByUser) { if (!this.closingByUser) {
this.emitStatus( this.emitStatus(
Types.DeviceStatusEnum.DeviceDisconnected, Types.DeviceStatusEnum.DeviceDisconnected,
error instanceof DOMException && this.isTimeoutOrAbort(error) ? "write-timeout" : "write-error",
(error.name === "AbortError" || error.name === "TimeoutError")
? "write-timeout"
: "write-error",
); );
} }
throw error; throw error;
@ -210,7 +184,7 @@ export class TransportHTTP implements Types.Transport {
* Stop polling and emit `DeviceDisconnected("user")`. * Stop polling and emit `DeviceDisconnected("user")`.
*/ */
disconnect(): Promise<void> { disconnect(): Promise<void> {
this._closingByUser = true; this.closingByUser = true;
if (this.interval) { if (this.interval) {
clearInterval(this.interval); clearInterval(this.interval);
@ -219,9 +193,9 @@ export class TransportHTTP implements Types.Transport {
this.fetching = false; this.fetching = false;
try { try {
this._inflightReadController?.abort(); this.inflightReadController?.abort();
} catch {} } catch {}
this._inflightReadController = undefined; this.inflightReadController = undefined;
this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "user"); this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "user");
@ -229,13 +203,41 @@ export class TransportHTTP implements Types.Transport {
} }
private emitStatus(next: Types.DeviceStatusEnum, reason?: string): void { private emitStatus(next: Types.DeviceStatusEnum, reason?: string): void {
if (next === this._lastStatus) { if (next === this.lastStatus) {
return; return;
} }
this._lastStatus = next; this.lastStatus = next;
this._fromDeviceController?.enqueue({ this.fromDeviceController?.enqueue({
type: "status", type: "status",
data: { status: next, reason }, 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<void> {
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;
}
}
} }

79
packages/transport-node-serial/src/transport.ts
View File

@ -12,17 +12,18 @@ import { SerialPort } from "serialport";
export class TransportNodeSerial implements Types.Transport { export class TransportNodeSerial implements Types.Transport {
private readonly _toDevice: WritableStream<Uint8Array>; private readonly _toDevice: WritableStream<Uint8Array>;
private readonly _fromDevice: ReadableStream<Types.DeviceOutput>; private readonly _fromDevice: ReadableStream<Types.DeviceOutput>;
private _fromDeviceController?: ReadableStreamDefaultController<Types.DeviceOutput>; private fromDeviceController?: ReadableStreamDefaultController<Types.DeviceOutput>;
private port: SerialPort | undefined; private port: SerialPort | undefined;
private pipePromise?: Promise<void>;
private _lastStatus: Types.DeviceStatusEnum = private abortController: AbortController;
private lastStatus: Types.DeviceStatusEnum =
Types.DeviceStatusEnum.DeviceDisconnected; Types.DeviceStatusEnum.DeviceDisconnected;
private _closingByUser = false; private closingByUser = false;
/** /**
* Creates and connects a new TransportNode instance. * Creates and connects a new TransportNode instance.
* @param path - Path to the serial device * @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. * @returns A promise that resolves with a connected TransportNode instance.
*/ */
public static create( public static create(
@ -60,7 +61,7 @@ export class TransportNodeSerial implements Types.Transport {
this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "port-error"); this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "port-error");
}); });
this.port.on("close", () => { this.port.on("close", () => {
if (this._closingByUser) { if (this.closingByUser) {
return; return;
} }
this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "port-closed"); this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "port-closed");
@ -69,9 +70,12 @@ export class TransportNodeSerial implements Types.Transport {
const fromDeviceSource = Readable.toWeb(port) as ReadableStream<Uint8Array>; const fromDeviceSource = Readable.toWeb(port) as ReadableStream<Uint8Array>;
const transformed = fromDeviceSource.pipeThrough(Utils.fromDeviceStream()); const transformed = fromDeviceSource.pipeThrough(Utils.fromDeviceStream());
this.abortController = new AbortController();
const controller = this.abortController;
this._fromDevice = new ReadableStream<Types.DeviceOutput>({ this._fromDevice = new ReadableStream<Types.DeviceOutput>({
start: async (ctrl) => { start: async (ctrl) => {
this._fromDeviceController = ctrl; this.fromDeviceController = ctrl;
this.emitStatus(Types.DeviceStatusEnum.DeviceConnecting); this.emitStatus(Types.DeviceStatusEnum.DeviceConnecting);
this.emitStatus(Types.DeviceStatusEnum.DeviceConnected); this.emitStatus(Types.DeviceStatusEnum.DeviceConnected);
@ -83,15 +87,24 @@ export class TransportNodeSerial implements Types.Transport {
if (done) { if (done) {
break; break;
} }
if (value) {
ctrl.enqueue(value); ctrl.enqueue(value);
} }
} ctrl.close();
} catch { } catch (error) {
if (this.closingByUser) {
ctrl.close(); // graceful EOF on user
} else {
this.emitStatus( this.emitStatus(
Types.DeviceStatusEnum.DeviceDisconnected, Types.DeviceStatusEnum.DeviceDisconnected,
"read-error", "read-error",
); );
ctrl.error(
error instanceof Error ? error : new Error(String(error)),
);
}
try {
await transformed.cancel();
} catch {}
} finally { } finally {
reader.releaseLock(); reader.releaseLock();
} }
@ -99,13 +112,24 @@ export class TransportNodeSerial implements Types.Transport {
}); });
// Stream for data going FROM the application TO the Meshtastic device. // Stream for data going FROM the application TO the Meshtastic device.
const toDeviceTransform = Utils.toDeviceStream; this._toDevice = Utils.toDeviceStream.writable;
this._toDevice = toDeviceTransform.writable;
this.pipePromise = Utils.toDeviceStream.readable
toDeviceTransform.readable .pipeTo(Writable.toWeb(port) as WritableStream<Uint8Array>, {
.pipeTo(Writable.toWeb(port) as WritableStream<Uint8Array>) signal: controller.signal,
.catch((err) => { })
console.error("Error piping data to serial port:", err); .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")`. * Disconnect from the serial port and emit `DeviceDisconnected("user")`.
* Safe to call multiple times. * Safe to call multiple times.
*/ */
disconnect() { async disconnect() {
try {
this.closingByUser = true;
this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "user");
this.abortController?.abort();
await this.pipePromise?.catch(() => {});
try { try {
this._closingByUser = true;
this.port?.close(); this.port?.close();
} catch {}
} finally { } finally {
this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "user");
this.port = undefined; this.port = undefined;
this._closingByUser = false; this.closingByUser = false;
} }
return Promise.resolve();
} }
private emitStatus(next: Types.DeviceStatusEnum, reason?: string): void { private emitStatus(next: Types.DeviceStatusEnum, reason?: string): void {
if (next === this._lastStatus) { if (next === this.lastStatus) {
return; return;
} }
this._lastStatus = next; this.lastStatus = next;
this._fromDeviceController?.enqueue({ this.fromDeviceController?.enqueue({
type: "status", type: "status",
data: { status: next, reason }, data: { status: next, reason },
}); });

59
packages/transport-node/src/transport.ts
View File

@ -12,13 +12,14 @@ import { Types, Utils } from "@meshtastic/core";
export class TransportNode implements Types.Transport { export class TransportNode implements Types.Transport {
private readonly _toDevice: WritableStream<Uint8Array>; private readonly _toDevice: WritableStream<Uint8Array>;
private readonly _fromDevice: ReadableStream<Types.DeviceOutput>; private readonly _fromDevice: ReadableStream<Types.DeviceOutput>;
private _fromDeviceController?: ReadableStreamDefaultController<Types.DeviceOutput>; private fromDeviceController?: ReadableStreamDefaultController<Types.DeviceOutput>;
private socket: Socket | undefined; private socket: Socket | undefined;
private pipePromise?: Promise<void>;
private _lastStatus: Types.DeviceStatusEnum = private abortController: AbortController;
private lastStatus: Types.DeviceStatusEnum =
Types.DeviceStatusEnum.DeviceDisconnected; Types.DeviceStatusEnum.DeviceDisconnected;
private _closingByUser = false; private closingByUser = false;
/** /**
* Creates and connects a new TransportNode instance. * Creates and connects a new TransportNode instance.
@ -53,7 +54,7 @@ export class TransportNode implements Types.Transport {
this.socket.on("error", (err) => { this.socket.on("error", (err) => {
console.error("Socket connection error:", err); console.error("Socket connection error:", err);
if (!this._closingByUser) { if (!this.closingByUser) {
this.emitStatus( this.emitStatus(
Types.DeviceStatusEnum.DeviceDisconnected, Types.DeviceStatusEnum.DeviceDisconnected,
"socket-error", "socket-error",
@ -62,7 +63,7 @@ export class TransportNode implements Types.Transport {
}); });
this.socket.on("close", () => { this.socket.on("close", () => {
if (this._closingByUser) { if (this.closingByUser) {
return; // suppress close-derived disconnect in user flow return; // suppress close-derived disconnect in user flow
} }
this.emitStatus( this.emitStatus(
@ -71,6 +72,9 @@ export class TransportNode implements Types.Transport {
); );
}); });
this.abortController = new AbortController();
const abortController = this.abortController;
const fromDeviceSource = Readable.toWeb( const fromDeviceSource = Readable.toWeb(
connection, connection,
) as ReadableStream<Uint8Array>; ) as ReadableStream<Uint8Array>;
@ -78,7 +82,7 @@ export class TransportNode implements Types.Transport {
this._fromDevice = new ReadableStream<Types.DeviceOutput>({ this._fromDevice = new ReadableStream<Types.DeviceOutput>({
start: async (ctrl) => { start: async (ctrl) => {
this._fromDeviceController = ctrl; this.fromDeviceController = ctrl;
this.emitStatus(Types.DeviceStatusEnum.DeviceConnecting); this.emitStatus(Types.DeviceStatusEnum.DeviceConnecting);
this.emitStatus(Types.DeviceStatusEnum.DeviceConnected); this.emitStatus(Types.DeviceStatusEnum.DeviceConnected);
@ -90,19 +94,22 @@ export class TransportNode implements Types.Transport {
if (done) { if (done) {
break; break;
} }
if (value) {
ctrl.enqueue(value); ctrl.enqueue(value);
} }
}
ctrl.close(); ctrl.close();
} catch (error) { } catch (error) {
if (!this._closingByUser) { if (this.closingByUser) {
ctrl.close();
} else {
this.emitStatus( this.emitStatus(
Types.DeviceStatusEnum.DeviceDisconnected, Types.DeviceStatusEnum.DeviceDisconnected,
"read-error", "read-error",
); );
ctrl.error(
error instanceof Error ? error : new Error(String(error)),
);
} }
ctrl.error(error instanceof Error ? error : new Error(String(error)));
try { try {
await transformed.cancel(); await transformed.cancel();
} catch {} } catch {}
@ -116,9 +123,14 @@ export class TransportNode implements Types.Transport {
const toDeviceTransform = Utils.toDeviceStream; const toDeviceTransform = Utils.toDeviceStream;
this._toDevice = toDeviceTransform.writable; this._toDevice = toDeviceTransform.writable;
toDeviceTransform.readable this.pipePromise = toDeviceTransform.readable
.pipeTo(Writable.toWeb(connection) as WritableStream<Uint8Array>) .pipeTo(Writable.toWeb(connection) as WritableStream<Uint8Array>, {
signal: abortController.signal,
})
.catch((err) => { .catch((err) => {
if (abortController.signal.aborted) {
return;
}
console.error("Error piping data to socket:", err); console.error("Error piping data to socket:", err);
const error = err instanceof Error ? err : new Error(String(err)); const error = err instanceof Error ? err : new Error(String(err));
this.socket?.destroy(error); this.socket?.destroy(error);
@ -139,24 +151,29 @@ export class TransportNode implements Types.Transport {
* Disconnect from the TCP socket and emit `DeviceDisconnected("user")`. * Disconnect from the TCP socket and emit `DeviceDisconnected("user")`.
* Safe to call multiple times. * Safe to call multiple times.
*/ */
disconnect(): Promise<void> { async disconnect(): Promise<void> {
try { 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(); this.socket?.destroy();
} finally { } finally {
this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "user");
this.socket = undefined; this.socket = undefined;
this._closingByUser = false; this.closingByUser = false;
} }
return Promise.resolve();
} }
private emitStatus(next: Types.DeviceStatusEnum, reason?: string): void { private emitStatus(next: Types.DeviceStatusEnum, reason?: string): void {
if (next === this._lastStatus) { if (next === this.lastStatus) {
return; return;
} }
this._lastStatus = next; this.lastStatus = next;
this._fromDeviceController?.enqueue({ this.fromDeviceController?.enqueue({
type: "status", type: "status",
data: { status: next, reason }, data: { status: next, reason },
}); });

114
packages/transport-web-bluetooth/src/transport.ts
View File

@ -21,19 +21,18 @@ function toArrayBuffer(uint8array: Uint8Array): ArrayBuffer {
export class TransportWebBluetooth implements Types.Transport { export class TransportWebBluetooth implements Types.Transport {
private _toDevice: WritableStream<Uint8Array>; private _toDevice: WritableStream<Uint8Array>;
private _fromDevice: ReadableStream<Types.DeviceOutput>; private _fromDevice: ReadableStream<Types.DeviceOutput>;
private _fromDeviceController?: ReadableStreamDefaultController<Types.DeviceOutput>; private fromDeviceController?: ReadableStreamDefaultController<Types.DeviceOutput>;
private _isFirstWrite = true;
private toRadioCharacteristic: BluetoothRemoteGATTCharacteristic; private toRadioCharacteristic: BluetoothRemoteGATTCharacteristic;
private fromRadioCharacteristic: BluetoothRemoteGATTCharacteristic; private fromRadioCharacteristic: BluetoothRemoteGATTCharacteristic;
private fromNumCharacteristic: BluetoothRemoteGATTCharacteristic; private fromNumCharacteristic: BluetoothRemoteGATTCharacteristic;
private gattServer: BluetoothRemoteGATTServer; private gattServer: BluetoothRemoteGATTServer;
private _lastStatus: Types.DeviceStatusEnum = private lastStatus: Types.DeviceStatusEnum =
Types.DeviceStatusEnum.DeviceDisconnected; Types.DeviceStatusEnum.DeviceDisconnected;
private _closingByUser = false; private closingByUser = false;
private reading = false;
/** UUID for the "toRadio" write characteristic. */ /** UUID for the "toRadio" write characteristic. */
static ToRadioUuid = "f75c76d2-129e-4dad-a1dd-7866124401e7"; static ToRadioUuid = "f75c76d2-129e-4dad-a1dd-7866124401e7";
/** UUID for the "fromRadio" read characteristic. */ /** UUID for the "fromRadio" read characteristic. */
@ -43,6 +42,19 @@ export class TransportWebBluetooth implements Types.Transport {
/** UUID for the Meshtastic GATT service. */ /** UUID for the Meshtastic GATT service. */
static ServiceUuid = "6ba1b218-15a8-461f-9fa8-5dcae273eafd"; 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. * 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.gattServer = gattServer;
this._fromDevice = new ReadableStream<Types.DeviceOutput>({ this._fromDevice = new ReadableStream<Types.DeviceOutput>({
start: (ctrl) => { start: async (ctrl) => {
this._fromDeviceController = ctrl; this.fromDeviceController = ctrl;
this.emitStatus(Types.DeviceStatusEnum.DeviceConnecting); this.emitStatus(Types.DeviceStatusEnum.DeviceConnecting);
this.gattServer.device.addEventListener( this.gattServer.device.addEventListener(
"gattserverdisconnected", "gattserverdisconnected",
() => { this.onGattDisconnected,
if (this._closingByUser) { );
return;
} 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( this.emitStatus(
Types.DeviceStatusEnum.DeviceDisconnected, Types.DeviceStatusEnum.DeviceDisconnected,
"gatt-disconnected", "notify-failed",
); );
}, this.gattServer.device.removeEventListener(
"gattserverdisconnected",
this.onGattDisconnected,
); );
}
}, },
}); });
@ -145,11 +169,7 @@ export class TransportWebBluetooth implements Types.Transport {
try { try {
const ab = toArrayBuffer(chunk); const ab = toArrayBuffer(chunk);
await this.toRadioCharacteristic.writeValue(ab); await this.toRadioCharacteristic.writeValue(ab);
void this.readFromRadio(); // ensure we read any response
if (this._isFirstWrite) {
this._isFirstWrite = false;
setTimeout(() => this.readFromRadio(), 50);
}
} catch (error) { } catch (error) {
this.emitStatus( this.emitStatus(
Types.DeviceStatusEnum.DeviceDisconnected, 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. */ /** Writable stream of bytes to the device. */
@ -195,20 +196,32 @@ export class TransportWebBluetooth implements Types.Transport {
*/ */
disconnect(): Promise<void> { disconnect(): Promise<void> {
try { 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(); this.gattServer.disconnect();
} finally { } finally {
this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "user"); this.closingByUser = false;
this._closingByUser = false;
} }
return Promise.resolve(); return Promise.resolve();
} }
private async readFromRadio(): Promise<void> { private async readFromRadio(): Promise<void> {
const controller = this._fromDeviceController; if (this.reading) {
if (!controller) {
return; return;
} }
this.reading = true;
try { try {
let hasMoreData = true; let hasMoreData = true;
@ -224,23 +237,30 @@ export class TransportWebBluetooth implements Types.Transport {
}); });
} }
} catch (error) { } catch (error) {
this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "read-error"); if (!this.closingByUser) {
this.emitStatus(
Types.DeviceStatusEnum.DeviceDisconnected,
"read-error",
);
}
throw error; throw error;
} finally {
this.reading = false;
} }
} }
private emitStatus(next: Types.DeviceStatusEnum, reason?: string): void { private emitStatus(next: Types.DeviceStatusEnum, reason?: string): void {
if (next === this._lastStatus) { if (next === this.lastStatus) {
return; return;
} }
this._lastStatus = next; this.lastStatus = next;
this._fromDeviceController?.enqueue({ this.fromDeviceController?.enqueue({
type: "status", type: "status",
data: { status: next, reason }, data: { status: next, reason },
}); });
} }
private enqueue(output: Types.DeviceOutput): void { private enqueue(output: Types.DeviceOutput): void {
this._fromDeviceController?.enqueue(output); this.fromDeviceController?.enqueue(output);
} }
} }

147
packages/transport-web-serial/src/transport.ts
View File

@ -10,18 +10,18 @@ import { Types, Utils } from "@meshtastic/core";
export class TransportWebSerial implements Types.Transport { export class TransportWebSerial implements Types.Transport {
private _toDevice: WritableStream<Uint8Array>; private _toDevice: WritableStream<Uint8Array>;
private _fromDevice: ReadableStream<Types.DeviceOutput>; private _fromDevice: ReadableStream<Types.DeviceOutput>;
private _fromDeviceController?: ReadableStreamDefaultController<Types.DeviceOutput>; private fromDeviceController?: ReadableStreamDefaultController<Types.DeviceOutput>;
private connection: SerialPort; private connection: SerialPort;
private _portReadable: ReadableStream<Uint8Array>; private pipePromise: Promise<void> | null = null;
private _portWritable: WritableStream<Uint8Array>; private abortController: AbortController;
private portReadable: ReadableStream<Uint8Array>;
private _lastStatus: Types.DeviceStatusEnum = private lastStatus: Types.DeviceStatusEnum =
Types.DeviceStatusEnum.DeviceDisconnected; 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<TransportWebSerial> { public static async create(baudRate?: number): Promise<TransportWebSerial> {
const port = await navigator.serial.requestPort(); 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( public static async createFromPort(
port: SerialPort, port: SerialPort,
baudRate?: number, baudRate?: number,
): Promise<TransportWebSerial> { ): Promise<TransportWebSerial> {
if (!port.readable || !port.writable) {
await port.open({ baudRate: baudRate || 115200 }); await port.open({ baudRate: baudRate || 115200 });
}
return new TransportWebSerial(port); return new TransportWebSerial(port);
} }
@ -45,26 +48,41 @@ export class TransportWebSerial implements Types.Transport {
* @throws If the port lacks readable or writable streams. * @throws If the port lacks readable or writable streams.
*/ */
constructor(connection: SerialPort) { constructor(connection: SerialPort) {
const readable = connection.readable; if (!connection.readable || !connection.writable) {
const writable = connection.writable;
if (!readable || !writable) {
throw new Error("Stream not accessible"); throw new Error("Stream not accessible");
} }
this.connection = connection; this.connection = connection;
this._portReadable = readable; this.portReadable = connection.readable;
this._portWritable = writable; 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; this._toDevice = Utils.toDeviceStream.writable;
// Wrap + capture controller to inject status packets
this._fromDevice = new ReadableStream<Types.DeviceOutput>({ this._fromDevice = new ReadableStream<Types.DeviceOutput>({
start: async (ctrl) => { start: async (ctrl) => {
this._fromDeviceController = ctrl; this.fromDeviceController = ctrl;
this.emitStatus(Types.DeviceStatusEnum.DeviceConnecting); this.emitStatus(Types.DeviceStatusEnum.DeviceConnecting);
const transformed = this._portReadable.pipeThrough( const transformed = this.portReadable.pipeThrough(
Utils.fromDeviceStream(), Utils.fromDeviceStream(),
); );
const reader = transformed.getReader(); const reader = transformed.getReader();
@ -72,9 +90,6 @@ export class TransportWebSerial implements Types.Transport {
const onOsDisconnect = (ev: Event) => { const onOsDisconnect = (ev: Event) => {
const { port } = ev as unknown as { port?: SerialPort }; const { port } = ev as unknown as { port?: SerialPort };
if (port && port === this.connection) { if (port && port === this.connection) {
if (this._closingByUser) {
return;
}
this.emitStatus( this.emitStatus(
Types.DeviceStatusEnum.DeviceDisconnected, Types.DeviceStatusEnum.DeviceDisconnected,
"serial-disconnected", "serial-disconnected",
@ -91,15 +106,20 @@ export class TransportWebSerial implements Types.Transport {
if (done) { if (done) {
break; break;
} }
if (value) {
ctrl.enqueue(value); ctrl.enqueue(value);
} }
} ctrl.close();
} catch { } catch (error) {
if (!this.closingByUser) {
this.emitStatus( this.emitStatus(
Types.DeviceStatusEnum.DeviceDisconnected, Types.DeviceStatusEnum.DeviceDisconnected,
"read-error", "read-error",
); );
}
ctrl.error(error instanceof Error ? error : new Error(String(error)));
try {
await transformed.cancel();
} catch {}
} finally { } finally {
reader.releaseLock(); reader.releaseLock();
navigator.serial.removeEventListener("disconnect", onOsDisconnect); navigator.serial.removeEventListener("disconnect", onOsDisconnect);
@ -109,25 +129,21 @@ export class TransportWebSerial implements Types.Transport {
} }
/** Writable stream of bytes to the device. */ /** Writable stream of bytes to the device. */
get toDevice(): WritableStream<Uint8Array> { public get toDevice(): WritableStream<Uint8Array> {
return this._toDevice; return this._toDevice;
} }
/** Readable stream of {@link Types.DeviceOutput} from the device. */ /** Readable stream of {@link Types.DeviceOutput} from the device. */
get fromDevice(): ReadableStream<Types.DeviceOutput> { public get fromDevice(): ReadableStream<Types.DeviceOutput> {
return this._fromDevice; return this._fromDevice;
} }
private enqueue(output: Types.DeviceOutput): void {
this._fromDeviceController?.enqueue(output);
}
private emitStatus(next: Types.DeviceStatusEnum, reason?: string): void { private emitStatus(next: Types.DeviceStatusEnum, reason?: string): void {
if (next === this._lastStatus) { if (next === this.lastStatus) {
return; return;
} }
this._lastStatus = next; this.lastStatus = next;
this._fromDeviceController?.enqueue({ this.fromDeviceController?.enqueue({
type: "status", type: "status",
data: { status: next, reason }, data: { status: next, reason },
}); });
@ -136,13 +152,76 @@ export class TransportWebSerial implements Types.Transport {
/** /**
* Closes the serial port and emits `DeviceDisconnected("user")`. * Closes the serial port and emits `DeviceDisconnected("user")`.
*/ */
async disconnect(): Promise<void> { public async disconnect(): Promise<void> {
try { 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(); 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 { } finally {
this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "user"); 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;
} }
} }
} }

Write Preview
Loading…
Cancel
Save