Browse Source

Transport status events

Add symbol docs
Emit transport status events
Transport test suite
pull/790/head
philon- 11 months ago
committed by philon-
parent
commit
b94af56dfd
  1. 1
      .vscode/settings.json
  2. 3
      package.json
  3. 8
      packages/core/src/types.ts
  4. 13
      packages/core/src/utils/transform/decodePacket.ts
  5. 210
      packages/transport-http/src/transport.test.ts
  6. 227
      packages/transport-http/src/transport.ts
  7. 187
      packages/transport-node-serial/src/transport.test.ts
  8. 89
      packages/transport-node-serial/src/transport.ts
  9. 186
      packages/transport-node/src/transport.test.ts
  10. 105
      packages/transport-node/src/transport.ts
  11. 166
      packages/transport-web-bluetooth/src/transport.test.ts
  12. 148
      packages/transport-web-bluetooth/src/transport.ts
  13. 227
      packages/transport-web-serial/src/transport.test.ts
  14. 115
      packages/transport-web-serial/src/transport.ts
  15. 29
      packages/web/vitest.config.ts
  16. 329
      pnpm-lock.yaml
  17. 122
      tests/utils/transportContract.ts

1
.vscode/settings.json

@ -6,4 +6,5 @@
"search.exclude": {
"**/i18n/locales/*-*/**": true,
},
"vitest.workspaceConfig": "vitest.config.ts",
}

3
package.json

@ -39,6 +39,7 @@
"biome": "^0.3.3",
"tsdown": "^0.13.4",
"typescript": "^5.8.3",
"vitest": "^3.2.4"
"vitest": "^3.2.4",
"@meshtastic/core": "workspace:*"
}
}

8
packages/core/src/types.ts

@ -10,7 +10,12 @@ interface DebugLog {
data: string;
}
export type DeviceOutput = Packet | DebugLog;
interface StatusEvent {
type: "status";
data: { status: DeviceStatusEnum; reason?: string };
}
export type DeviceOutput = Packet | DebugLog | StatusEvent;
export interface Transport {
toDevice: WritableStream<Uint8Array>;
@ -101,6 +106,7 @@ export enum Emitter {
RemoveNodeByNum = 32,
SetCannedMessages = 33,
Disconnect = 34,
ConnectionStatus = 35,
}
export interface LogEvent {

13
packages/core/src/utils/transform/decodePacket.ts

@ -7,6 +7,19 @@ export const decodePacket = (device: MeshDevice) =>
new WritableStream<DeviceOutput>({
write(chunk) {
switch (chunk.type) {
case "status": {
const { status, reason } = chunk.data as {
status: Types.DeviceStatusEnum;
reason?: string;
};
device.updateDeviceStatus(status);
device.log.info(
Types.Emitter[Types.Emitter.ConnectionStatus],
`🔗 ${Types.DeviceStatusEnum[status]} ${reason ? `(${reason})` : ""}`,
);
break;
}
case "debug": {
break;
}

210
packages/transport-http/src/transport.test.ts

@ -0,0 +1,210 @@
import { describe, vi, expect, it, beforeEach, afterEach } from "vitest";
import { runTransportContract } from "../../../tests/utils/transportContract";
import { TransportHTTP } from "./transport";
function stubFetch() {
const inbox: Uint8Array[] = [];
let lastWritten: ArrayBuffer | undefined;
let forceNextReadToHang = false;
let forceNextReadToReturn500 = false;
function makeAbortAwareHang(signal?: AbortSignal): Promise<Response> {
return new Promise((_, reject) => {
const abort = () => reject(new DOMException("Aborted", "AbortError"));
if (signal?.aborted) {
abort();
return;
}
if (signal) {
signal.addEventListener("abort", abort, { once: true });
}
});
}
const mockFetch = vi.fn(async (url: string, init?: RequestInit) => {
const method = (init?.method ?? "GET").toUpperCase();
if (url.includes("/api/v1/toradio") && method === "OPTIONS") {
return { ok: true, status: 204 } as Response;
}
if (url.includes("/api/v1/toradio") && method === "PUT") {
lastWritten = init?.body as ArrayBuffer;
return { ok: true, status: 200 } as Response;
}
if (url.includes("/api/v1/fromradio") && method === "GET") {
if (forceNextReadToHang) {
forceNextReadToHang = false;
return makeAbortAwareHang(init?.signal ?? undefined);
}
if (forceNextReadToReturn500) {
forceNextReadToReturn500 = false;
return {
ok: false,
status: 500,
arrayBuffer: async () => new ArrayBuffer(0),
} as Response;
}
const next = inbox.shift() ?? new Uint8Array();
return {
ok: true,
status: 200,
arrayBuffer: async () => next.buffer,
} as Response;
}
return { ok: true, status: 200 } as Response;
});
vi.stubGlobal("fetch", mockFetch);
return {
pushIncoming: (u8: Uint8Array) => inbox.push(u8),
assertLastWritten: (u8: Uint8Array) => {
const got = new Uint8Array(lastWritten || new ArrayBuffer(0));
expect(got).toEqual(u8);
},
forceReadErrorOnce: () => {
forceNextReadToReturn500 = true;
},
forceReadTimeoutOnce: () => {
forceNextReadToHang = true;
},
getMock: () => mockFetch,
cleanup: () => vi.unstubAllGlobals(),
};
}
async function tickNextTimer() {
try {
await vi.advanceTimersToNextTimerAsync();
} catch {
await new Promise((r) => setTimeout(r, 5));
}
}
describe("TransportHTTP (contract)", () => {
runTransportContract({
name: "TransportHTTP",
setup: () => {
vi.useFakeTimers();
},
teardown: () => {
vi.useRealTimers();
vi.restoreAllMocks();
vi.unstubAllGlobals();
},
create: async () => {
(globalThis as unknown as { __http: ReturnType<typeof stubFetch> }).__http = stubFetch();
const transport = await TransportHTTP.create("127.0.0.1:80", false);
await tickNextTimer();
return transport;
},
pushIncoming: async (bytes) => {
(globalThis as unknown as { __http: ReturnType<typeof stubFetch> }).__http.pushIncoming(bytes);
await tickNextTimer();
},
assertLastWritten: (bytes) => {
(globalThis as unknown as { __http: ReturnType<typeof stubFetch> }).__http.assertLastWritten(bytes);
},
triggerDisconnect: async () => {
(globalThis as unknown as { __http: ReturnType<typeof stubFetch> }).__http.forceReadErrorOnce();
await tickNextTimer();
},
});
});
describe("TransportHTTP (extras)", () => {
let httpStub: ReturnType<typeof stubFetch> | undefined;
beforeEach(() => {
vi.useFakeTimers();
});
afterEach(() => {
vi.useRealTimers();
vi.restoreAllMocks();
httpStub?.cleanup();
httpStub = undefined;
});
async function createTransport(): Promise<TransportHTTP> {
httpStub = stubFetch();
const transport = await TransportHTTP.create("127.0.0.1:80", false);
await tickNextTimer();
return transport;
}
async function advanceOnePoll() {
await tickNextTimer();
}
it("emits DeviceDisconnected with reason 'read-timeout' when GET /fromradio hangs", async () => {
const transport = await createTransport();
const reader = transport.fromDevice.getReader();
httpStub!.forceReadTimeoutOnce();
await tickNextTimer();
await vi.advanceTimersByTimeAsync(8000);
let sawReadTimeout = false;
for (let i = 0; i < 6; i++) {
const { value } = await reader.read();
if (value?.type === "status" && value.data.reason === "read-timeout") {
sawReadTimeout = true;
break;
}
}
expect(sawReadTimeout).toBe(true);
reader.releaseLock();
await transport.disconnect();
});
it("stops polling after disconnect()", async () => {
const transport = await createTransport();
const fetchMock = httpStub!.getMock();
const callsBeforeDisconnect = fetchMock.mock.calls.length;
await transport.disconnect();
await advanceOnePoll();
await vi.runOnlyPendingTimersAsync();
const callsAfterDisconnect = fetchMock.mock.calls.length;
expect(callsAfterDisconnect).toBe(callsBeforeDisconnect);
});
it("emits DeviceDisconnected with reason 'read-timeout' when GET /fromradio hangs", async () => {
const transport = await createTransport();
const reader = transport.fromDevice.getReader();
httpStub!.forceReadTimeoutOnce();
await vi.advanceTimersToNextTimerAsync();
await vi.advanceTimersByTimeAsync(8000);
await Promise.resolve();
await Promise.resolve();
let sawReadTimeout = false;
for (let i = 0; i < 6; i++) {
const { value } = await reader.read();
if (value?.type === "status" && value.data.reason === "read-timeout") {
sawReadTimeout = true;
break;
}
}
expect(sawReadTimeout).toBe(true);
reader.releaseLock();
await transport.disconnect();
});
});

227
packages/transport-http/src/transport.ts

@ -1,14 +1,82 @@
import type { Types } from "@meshtastic/core";
import { Types } from "@meshtastic/core";
const FETCH_INTERVAL_MS = 3000;
const READ_TIMEOUT_MS = 7000;
const WRITE_TIMEOUT_MS = 4000;
function toArrayBuffer(uint8array: Uint8Array): ArrayBuffer {
if (
uint8array.buffer instanceof ArrayBuffer &&
uint8array.byteOffset === 0 &&
uint8array.byteLength === uint8array.buffer.byteLength
) {
return uint8array.buffer;
}
return uint8array.slice().buffer;
}
function createTimeoutController(ms: number): {
controller: AbortController;
clear: () => void;
} {
const controller = new AbortController();
const id = setTimeout(() => controller.abort(), ms);
const clear = () => clearTimeout(id);
return { controller, clear };
}
function composeAbortSignals(
a: AbortSignal,
b: AbortSignal,
): { signal: AbortSignal; cleanup: () => void } {
const controller = new AbortController();
const abortIfNeeded = () => controller.abort();
if (a.aborted || b.aborted) {
controller.abort();
}
a.addEventListener("abort", abortIfNeeded);
b.addEventListener("abort", abortIfNeeded);
const cleanup = () => {
a.removeEventListener("abort", abortIfNeeded);
b.removeEventListener("abort", abortIfNeeded);
};
return { signal: controller.signal, cleanup };
}
/**
* Provides HTTP(S) transport for Meshtastic devices.
*
* Implements {@link Types.Transport} using the device's HTTP API.
* Polls `/api/v1/fromradio` for incoming packets and writes to `/api/v1/toradio`.
*/
export class TransportHTTP implements Types.Transport {
private _toDevice: WritableStream<Uint8Array>;
private _fromDevice: ReadableStream<Types.DeviceOutput>;
private _fromDeviceController?: ReadableStreamDefaultController<Types.DeviceOutput>;
private url: string;
private receiveBatchRequests: boolean;
private fetchInterval: number;
private fetching: boolean;
private interval: ReturnType<typeof setInterval> | undefined;
private _inflightReadController?: AbortController;
private _inflightReadStartedAt = 0;
private _lastStatus: Types.DeviceStatusEnum =
Types.DeviceStatusEnum.DeviceDisconnected;
private _closingByUser = false;
/**
* Probe the device and return a connected HTTP transport.
*
* @param address Hostname or IP address (with optional port).
* @param tls Use HTTPS if true, HTTP otherwise.
*/
public static async create(
address: string,
tls?: boolean,
@ -17,97 +85,204 @@ export class TransportHTTP implements Types.Transport {
await fetch(`${connectionUrl}/api/v1/toradio`, {
method: "OPTIONS",
});
await Promise.resolve();
return new TransportHTTP(connectionUrl);
}
/**
* Construct a new HTTP transport for the given device URL.
*
* @param url Base URL of the device (`http://host:port` or `https://host:port`).
*/
constructor(url: string) {
this.url = url;
this.receiveBatchRequests = false;
this.fetchInterval = 3000;
this.fetchInterval = FETCH_INTERVAL_MS;
this.fetching = false;
this._toDevice = new WritableStream<Uint8Array>({
write: async (chunk) => {
try {
await this.writeToRadio(chunk);
} catch (error) {
if (!this._closingByUser) {
this.emitStatus(
Types.DeviceStatusEnum.DeviceDisconnected,
error instanceof DOMException && error.name === "AbortError"
? "write-timeout"
: "write-error",
);
}
throw error;
}
},
});
let controller: ReadableStreamDefaultController<Types.DeviceOutput>;
this._fromDevice = new ReadableStream<Types.DeviceOutput>({
start: (ctrl) => {
controller = ctrl;
this._fromDeviceController = ctrl;
this.emitStatus(Types.DeviceStatusEnum.DeviceConnecting);
},
cancel: () => {
if (this.interval) {
clearInterval(this.interval);
}
this.interval = undefined;
},
});
this.interval = setInterval(async () => {
if (this.fetching) {
// We still have the previous request open
if (
this._inflightReadController &&
Date.now() - this._inflightReadStartedAt > READ_TIMEOUT_MS + 250
) {
try {
this._inflightReadController.abort();
} catch {}
}
return;
}
this.fetching = true;
try {
await this.readFromRadio(controller);
} catch {
// TODO: Emit disconnection events for certain types of errors
await this.readFromRadio();
} catch (error) {
if (!this._closingByUser) {
this.emitStatus(
Types.DeviceStatusEnum.DeviceDisconnected,
error instanceof DOMException && error.name === "AbortError"
? "read-timeout"
: "read-error",
);
}
} finally {
this.fetching = false;
}
}, this.fetchInterval);
}
private async readFromRadio(
controller: ReadableStreamDefaultController<Types.DeviceOutput>,
): Promise<void> {
/** Poll `/api/v1/fromradio` and enqueue incoming packets. */
private async readFromRadio(): Promise<void> {
let readBuffer = new ArrayBuffer(1);
while (readBuffer.byteLength > 0) {
const inflight = new AbortController();
this._inflightReadController = inflight;
this._inflightReadStartedAt = Date.now();
const { controller: timeoutCtrl, clear } =
createTimeoutController(READ_TIMEOUT_MS);
const { signal, cleanup } = composeAbortSignals(
inflight.signal,
timeoutCtrl.signal,
);
try {
const response = await fetch(
`${this.url}/api/v1/fromradio?all=${
this.receiveBatchRequests ? "true" : "false"
}`,
`${this.url}/api/v1/fromradio?all=${this.receiveBatchRequests ? "true" : "false"}`,
{
method: "GET",
headers: {
Accept: "application/x-protobuf",
},
headers: { Accept: "application/x-protobuf" },
signal,
},
);
if (!response.ok) {
throw new Error(
`fromradio ${response.status} ${response.statusText}`,
);
}
if (this._lastStatus === Types.DeviceStatusEnum.DeviceDisconnected) {
this.emitStatus(Types.DeviceStatusEnum.DeviceConnected);
}
readBuffer = await response.arrayBuffer();
if (readBuffer.byteLength > 0) {
controller.enqueue({
this._fromDeviceController?.enqueue({
type: "packet",
data: new Uint8Array(readBuffer),
});
}
} finally {
cleanup();
clear();
this._inflightReadController = undefined;
}
}
}
/** Write a protobuf-encoded request to `/api/v1/toradio`. */
private async writeToRadio(data: Uint8Array): Promise<void> {
await fetch(`${this.url}/api/v1/toradio`, {
const { controller: timeoutCtrl, clear } =
createTimeoutController(WRITE_TIMEOUT_MS);
try {
const response = await fetch(`${this.url}/api/v1/toradio`, {
method: "PUT",
headers: {
"Content-Type": "application/x-protobuf",
},
body: data,
headers: { "Content-Type": "application/x-protobuf" },
body: toArrayBuffer(data),
signal: timeoutCtrl.signal,
});
if (!response.ok) {
throw new Error(`toradio ${response.status} ${response.statusText}`);
}
} catch (error) {
if (!this._closingByUser) {
this.emitStatus(
Types.DeviceStatusEnum.DeviceDisconnected,
error instanceof DOMException && error.name === "AbortError"
? "write-timeout"
: "write-error",
);
}
throw error;
} finally {
clear();
}
}
/** Writable stream of bytes to the device. */
get toDevice(): WritableStream<Uint8Array> {
return this._toDevice;
}
/** Readable stream of {@link Types.DeviceOutput} from the device. */
get fromDevice(): ReadableStream<Types.DeviceOutput> {
return this._fromDevice;
}
/**
* Stop polling and emit `DeviceDisconnected("user")`.
*/
disconnect(): Promise<void> {
this.fetching = false;
this._closingByUser = true;
if (this.interval) {
clearInterval(this.interval);
}
this.interval = undefined;
this.fetching = false;
try {
this._inflightReadController?.abort();
} catch {}
this._inflightReadController = undefined;
this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "user");
return Promise.resolve();
}
private emitStatus(next: Types.DeviceStatusEnum, reason?: string): void {
if (next === this._lastStatus) {
return;
}
this._lastStatus = next;
this._fromDeviceController?.enqueue({
type: "status",
data: { status: next, reason },
});
}
}

187
packages/transport-node-serial/src/transport.test.ts

@ -0,0 +1,187 @@
import { describe, vi, expect, beforeEach, afterEach, it } from "vitest";
import { Duplex } from "node:stream";
import type { SerialPort } from "serialport";
import { Types, Utils } from "@meshtastic/core";
import { runTransportContract } from "../../../tests/utils/transportContract";
import { TransportNodeSerial } from "./transport";
function isStatusEvent(
output: Types.DeviceOutput | undefined,
): output is Extract<Types.DeviceOutput, { type: "status" }> {
return output !== undefined && output.type === "status";
}
class FakeSerialPort extends Duplex {
public lastWritten: Uint8Array | undefined;
constructor() {
super({ objectMode: false });
}
_read() {}
_write(
chunk: Buffer,
_encoding: BufferEncoding,
callback: (error?: Error | null) => void,
) {
this.lastWritten = new Uint8Array(
chunk.buffer,
chunk.byteOffset,
chunk.byteLength,
);
callback();
}
pushIncoming(data: Uint8Array) {
const buf = Buffer.from(data.buffer, data.byteOffset, data.byteLength);
this.push(buf);
}
emitErrorOnce(message = "simulated serial error") {
this.emit("error", new Error(message));
}
emitClose() {
this.emit("close");
}
close() {
this.destroy();
this.emit("close");
}
}
function stubCoreTransforms() {
const toDevice = new TransformStream<Uint8Array, Uint8Array>({
transform(chunk, controller) {
controller.enqueue(chunk);
},
});
const fromDeviceFactory = () =>
new TransformStream<Uint8Array, Types.DeviceOutput>({
transform(chunk, controller) {
controller.enqueue({ type: "packet", data: chunk });
},
});
// Utils.toDeviceStream is a getter
vi.spyOn(Utils, "toDeviceStream", "get").mockReturnValue(
toDevice as unknown as typeof Utils.toDeviceStream,
);
vi.spyOn(Utils, "fromDeviceStream").mockImplementation(
() =>
fromDeviceFactory() as unknown as TransformStream<
Uint8Array,
Types.DeviceOutput
>,
);
return {
restore: () => vi.restoreAllMocks(),
};
}
describe("TransportNodeSerial (contract)", () => {
let transformsStub: { restore: () => void } | undefined;
beforeEach(() => {
transformsStub = stubCoreTransforms();
});
afterEach(() => {
transformsStub?.restore();
});
runTransportContract({
name: "TransportNodeSerial",
setup: () => {},
teardown: () => {
vi.restoreAllMocks();
},
create: async () => {
const fakePort = new FakeSerialPort();
const transport = new TransportNodeSerial(
fakePort as unknown as SerialPort,
);
await Promise.resolve();
(globalThis as unknown as { __fakePort: FakeSerialPort }).__fakePort =
fakePort;
return transport;
},
pushIncoming: async (bytes) => {
(globalThis as unknown as { __fakePort: FakeSerialPort }).__fakePort.pushIncoming(
bytes,
);
await Promise.resolve();
},
assertLastWritten: (bytes) => {
const port =
(globalThis as unknown as { __fakePort: FakeSerialPort }).__fakePort;
expect(port.lastWritten).toBeDefined();
expect(port.lastWritten).toEqual(bytes);
},
triggerDisconnect: async () => {
(globalThis as unknown as { __fakePort: FakeSerialPort }).__fakePort.emitErrorOnce(
"test-disconnect",
);
await Promise.resolve();
},
});
});
describe("TransportNodeSerial (extras)", () => {
let transformsStub: { restore: () => void } | undefined;
beforeEach(() => {
transformsStub = stubCoreTransforms();
});
afterEach(() => {
transformsStub?.restore();
});
it("emits DeviceDisconnected with reason 'port-closed' on close event", async () => {
const fakePort = new FakeSerialPort();
const transport = new TransportNodeSerial(
fakePort as unknown as SerialPort,
);
const reader = transport.fromDevice.getReader();
await Promise.resolve();
const first = await reader.read();
expect(isStatusEvent(first.value)).toBe(true);
if (isStatusEvent(first.value)) {
expect(first.value.data.status).toBe(
Types.DeviceStatusEnum.DeviceConnecting,
);
}
const second = await reader.read();
expect(isStatusEvent(second.value)).toBe(true);
if (isStatusEvent(second.value)) {
expect(second.value.data.status).toBe(
Types.DeviceStatusEnum.DeviceConnected,
);
}
fakePort.emitClose();
await Promise.resolve();
let sawClosed = false;
for (let i = 0; i < 6; i++) {
const { value } = await reader.read();
if (isStatusEvent(value) && value.data.reason === "port-closed") {
sawClosed = true;
break;
}
}
expect(sawClosed).toBe(true);
reader.releaseLock();
await transport.disconnect();
});
});

89
packages/transport-node-serial/src/transport.ts

@ -1,20 +1,34 @@
import { Readable, Writable } from "node:stream";
import type { Types } from "@meshtastic/core";
import { Utils } from "@meshtastic/core";
import { Types, Utils } from "@meshtastic/core";
import { SerialPort } from "serialport";
/**
* Node.js Serial transport for Meshtastic.
*
* Implements {@link Types.Transport} on top of a Node `SerialPort`.
* Use {@link TransportNodeSerial.create} for a convenient factory, or
* `new TransportNodeSerial(port)` if you already have an open port.
*/
export class TransportNodeSerial implements Types.Transport {
private readonly _toDevice: WritableStream<Uint8Array>;
private readonly _fromDevice: ReadableStream<Types.DeviceOutput>;
private _fromDeviceController?: ReadableStreamDefaultController<Types.DeviceOutput>;
private port: SerialPort | undefined;
private _lastStatus: Types.DeviceStatusEnum =
Types.DeviceStatusEnum.DeviceDisconnected;
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).
* @returns A promise that resolves with a connected TransportNode instance.
*/
public static create(path: string, baudRate = 115200): Promise<TransportNodeSerial> {
public static create(
path: string,
baudRate = 115200,
): Promise<TransportNodeSerial> {
return new Promise((resolve, reject) => {
const port = new SerialPort({
path,
@ -43,25 +57,55 @@ export class TransportNodeSerial implements Types.Transport {
this.port = port;
this.port.on("error", (err) => {
console.error("Serial port connection error:", err);
this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "port-error");
});
this.port.on("close", () => {
if (this._closingByUser) {
return;
}
this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "port-closed");
});
const fromDeviceSource = Readable.toWeb(
port,
) as ReadableStream<Uint8Array>;
this._fromDevice = fromDeviceSource.pipeThrough(Utils.fromDeviceStream());
const fromDeviceSource = Readable.toWeb(port) as ReadableStream<Uint8Array>;
const transformed = fromDeviceSource.pipeThrough(Utils.fromDeviceStream());
this._fromDevice = new ReadableStream<Types.DeviceOutput>({
start: async (ctrl) => {
this._fromDeviceController = ctrl;
this.emitStatus(Types.DeviceStatusEnum.DeviceConnecting);
this.emitStatus(Types.DeviceStatusEnum.DeviceConnected);
const reader = transformed.getReader();
try {
while (true) {
const { value, done } = await reader.read();
if (done) {
break;
}
if (value) {
ctrl.enqueue(value);
}
}
} catch {
this.emitStatus(
Types.DeviceStatusEnum.DeviceDisconnected,
"read-error",
);
} finally {
reader.releaseLock();
}
},
});
// Stream for data going FROM the application TO the Meshtastic device.
const toDeviceTransform = Utils.toDeviceStream;
this._toDevice = toDeviceTransform.writable;
// The readable end of the transform is then piped to the Node.js SerialPort connection.
// A similar assertion is needed here because `Writable.toWeb` also returns
// a generically typed stream (`WritableStream<any>`).
toDeviceTransform.readable
.pipeTo(Writable.toWeb(port) as WritableStream<Uint8Array>)
.catch((err) => {
console.error("Error piping data to serial port:", err);
this.port.close(err as Error);
});
}
@ -79,9 +123,30 @@ export class TransportNodeSerial implements Types.Transport {
return this._fromDevice;
}
/**
* Disconnect from the serial port and emit `DeviceDisconnected("user")`.
* Safe to call multiple times.
*/
disconnect() {
this.port.close();
try {
this._closingByUser = true;
this.port?.close();
} finally {
this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "user");
this.port = undefined;
this._closingByUser = false;
}
return Promise.resolve();
}
private emitStatus(next: Types.DeviceStatusEnum, reason?: string): void {
if (next === this._lastStatus) {
return;
}
this._lastStatus = next;
this._fromDeviceController?.enqueue({
type: "status",
data: { status: next, reason },
});
}
}

186
packages/transport-node/src/transport.test.ts

@ -0,0 +1,186 @@
import { describe, vi, expect, beforeEach, afterEach, it } from "vitest";
import { Duplex } from "node:stream";
import type { Socket } from "node:net";
import { runTransportContract } from "../../../tests/utils/transportContract";
import { TransportNode } from "./transport";
import { Utils, Types } from "@meshtastic/core";
function isStatusEvent(
out: Types.DeviceOutput | undefined,
): out is Extract<Types.DeviceOutput, { type: "status" }> {
return !!out && (out as any).type === "status";
}
class FakeSocket extends Duplex {
public lastWritten: Uint8Array | undefined;
constructor() {
super({ objectMode: false });
}
_read() {}
_write(
chunk: Buffer,
_encoding: BufferEncoding,
callback: (error?: Error | null) => void,
) {
this.lastWritten = new Uint8Array(
chunk.buffer,
chunk.byteOffset,
chunk.byteLength,
);
callback();
}
pushIncoming(data: Uint8Array) {
const buf = Buffer.from(data.buffer, data.byteOffset, data.byteLength);
this.push(buf);
}
emitErrorOnce(message = "simulated error") {
this.emit("error", new Error(message));
}
emitClose() {
this.emit("close");
}
override destroy(error?: Error) {
super.destroy(error);
this.emit("close");
return this;
}
}
function stubCoreTransforms() {
const toDevice = new TransformStream<Uint8Array, Uint8Array>({
transform(chunk, controller) {
controller.enqueue(chunk);
},
});
const fromDeviceFactory = () =>
new TransformStream<Uint8Array, Types.DeviceOutput>({
transform(chunk, controller) {
controller.enqueue({ type: "packet", data: chunk });
},
});
vi.spyOn(Utils, "toDeviceStream", "get").mockReturnValue(
toDevice as unknown as typeof Utils.toDeviceStream,
);
vi
.spyOn(Utils, "fromDeviceStream")
.mockImplementation(
() =>
fromDeviceFactory() as unknown as TransformStream<
Uint8Array,
Types.DeviceOutput
>,
);
return {
restore: () => vi.restoreAllMocks(),
};
}
describe("TransportNode (contract)", () => {
let transformsStub: { restore: () => void } | undefined;
beforeEach(() => {
transformsStub = stubCoreTransforms();
});
afterEach(() => {
transformsStub?.restore();
});
runTransportContract({
name: "TransportNode",
setup: () => {},
teardown: () => {
vi.restoreAllMocks();
},
create: async () => {
const fakeSocket = new FakeSocket();
const transport = new TransportNode(fakeSocket as unknown as Socket);
await Promise.resolve();
(globalThis as unknown as { __nodeSock: FakeSocket }).__nodeSock =
fakeSocket;
return transport;
},
pushIncoming: async (bytes) => {
(globalThis as unknown as { __nodeSock: FakeSocket }).__nodeSock.pushIncoming(
bytes,
);
await Promise.resolve();
},
assertLastWritten: (bytes) => {
const sock = (globalThis as unknown as { __nodeSock: FakeSocket })
.__nodeSock;
expect(sock.lastWritten).toBeDefined();
expect(sock.lastWritten).toEqual(bytes);
},
triggerDisconnect: async () => {
(globalThis as unknown as { __nodeSock: FakeSocket }).__nodeSock.emitErrorOnce(
"test-disconnect",
);
await Promise.resolve();
},
});
});
describe("TransportNode (extras)", () => {
let transformsStub: { restore: () => void } | undefined;
beforeEach(() => {
transformsStub = stubCoreTransforms();
});
afterEach(() => {
transformsStub?.restore();
});
it("emits DeviceDisconnected with reason 'socket-closed' on close event", async () => {
const fakeSocket = new FakeSocket();
const transport = new TransportNode(fakeSocket as unknown as Socket);
const reader = transport.fromDevice.getReader();
await Promise.resolve();
const first = await reader.read();
expect(isStatusEvent(first.value)).toBe(true);
if (isStatusEvent(first.value)) {
expect(first.value.data.status).toBe(
Types.DeviceStatusEnum.DeviceConnecting,
);
}
const second = await reader.read();
expect(isStatusEvent(second.value)).toBe(true);
if (isStatusEvent(second.value)) {
expect(second.value.data.status).toBe(
Types.DeviceStatusEnum.DeviceConnected,
);
}
fakeSocket.emitClose();
await Promise.resolve();
let sawClosed = false;
for (let i = 0; i < 6; i++) {
const { value } = await reader.read();
if (isStatusEvent(value) && value.data.reason === "socket-closed") {
sawClosed = true;
break;
}
}
expect(sawClosed).toBe(true);
reader.releaseLock();
await transport.disconnect();
});
});

105
packages/transport-node/src/transport.ts

@ -1,13 +1,25 @@
import { Socket } from "node:net";
import { Readable, Writable } from "node:stream";
import type { Types } from "@meshtastic/core";
import { Utils } from "@meshtastic/core";
import { Types, Utils } from "@meshtastic/core";
/**
* Node.js TCP transport for Meshtastic.
*
* Implements {@link Types.Transport} on top of a Node `net.Socket`.
* Use {@link TransportNode.create} to open a new connection, or
* construct directly with an existing socket.
*/
export class TransportNode implements Types.Transport {
private readonly _toDevice: WritableStream<Uint8Array>;
private readonly _fromDevice: ReadableStream<Types.DeviceOutput>;
private _fromDeviceController?: ReadableStreamDefaultController<Types.DeviceOutput>;
private socket: Socket | undefined;
private _lastStatus: Types.DeviceStatusEnum =
Types.DeviceStatusEnum.DeviceDisconnected;
private _closingByUser = false;
/**
* Creates and connects a new TransportNode instance.
* @param hostname - The IP address or hostname of the Meshtastic device.
@ -38,47 +50,110 @@ export class TransportNode implements Types.Transport {
*/
constructor(connection: Socket) {
this.socket = connection;
this.socket.on("error", (err) => {
console.error("Socket connection error:", err);
if (!this._closingByUser) {
this.emitStatus(
Types.DeviceStatusEnum.DeviceDisconnected,
"socket-error",
);
}
});
this.socket.on("close", () => {
if (this._closingByUser) {
return; // suppress close-derived disconnect in user flow
}
this.emitStatus(
Types.DeviceStatusEnum.DeviceDisconnected,
"socket-closed",
);
});
const fromDeviceSource = Readable.toWeb(
connection,
) as ReadableStream<Uint8Array>;
this._fromDevice = fromDeviceSource.pipeThrough(Utils.fromDeviceStream());
const transformed = fromDeviceSource.pipeThrough(Utils.fromDeviceStream());
this._fromDevice = new ReadableStream<Types.DeviceOutput>({
start: async (ctrl) => {
this._fromDeviceController = ctrl;
this.emitStatus(Types.DeviceStatusEnum.DeviceConnecting);
this.emitStatus(Types.DeviceStatusEnum.DeviceConnected);
const reader = transformed.getReader();
try {
while (true) {
const { value, done } = await reader.read();
if (done) {
break;
}
if (value) {
ctrl.enqueue(value);
}
}
} catch {
if (!this._closingByUser) {
this.emitStatus(
Types.DeviceStatusEnum.DeviceDisconnected,
"read-error",
);
}
} finally {
reader.releaseLock();
}
},
});
// Stream for data going FROM the application TO the Meshtastic device.
const toDeviceTransform = Utils.toDeviceStream;
this._toDevice = toDeviceTransform.writable;
// The readable end of the transform is then piped to the Node.js socket.
// A similar assertion is needed here because `Writable.toWeb` also returns
// a generically typed stream (`WritableStream<any>`).
toDeviceTransform.readable
.pipeTo(Writable.toWeb(connection) as WritableStream<Uint8Array>)
.catch((err) => {
console.error("Error piping data to socket:", err);
this.socket.destroy(err as Error);
const error = err instanceof Error ? err : new Error(String(err));
this.socket?.destroy(error);
});
}
/**
* The WritableStream to send data to the Meshtastic device.
*/
/** WritableStream to send data to the device. */
public get toDevice(): WritableStream<Uint8Array> {
return this._toDevice;
}
/**
* The ReadableStream to receive data from the Meshtastic device.
*/
/** ReadableStream to receive data from the device. */
public get fromDevice(): ReadableStream<Types.DeviceOutput> {
return this._fromDevice;
}
disconnect() {
this.socket.destroy();
/**
* Disconnect from the TCP socket and emit `DeviceDisconnected("user")`.
* Safe to call multiple times.
*/
disconnect(): Promise<void> {
try {
this._closingByUser = true;
this.socket?.destroy();
} finally {
this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "user");
this.socket = undefined;
this._closingByUser = false;
}
return Promise.resolve();
}
private emitStatus(next: Types.DeviceStatusEnum, reason?: string): void {
if (next === this._lastStatus) {
return;
}
this._lastStatus = next;
this._fromDeviceController?.enqueue({
type: "status",
data: { status: next, reason },
});
}
}

166
packages/transport-web-bluetooth/src/transport.test.ts

@ -0,0 +1,166 @@
import { describe, vi, expect, beforeEach, afterEach } from "vitest";
import { runTransportContract } from "../../../tests/utils/transportContract";
import { TransportWebBluetooth } from "./transport";
class MiniEmitter {
private listeners = new Map<string, Set<(e: Event) => void>>();
addEventListener(type: string, listener: (e: Event) => void) {
if (!this.listeners.has(type)) this.listeners.set(type, new Set());
this.listeners.get(type)!.add(listener);
}
removeEventListener(type: string, listener: (e: Event) => void) {
this.listeners.get(type)?.delete(listener);
}
dispatchEvent(event: Event) {
this.listeners.get(event.type)?.forEach((l) => l(event));
}
}
function stubWebBluetooth() {
const incomingQueue: Uint8Array[] = [];
let lastWritten: Uint8Array | undefined;
// fromRadioCharacteristic: read bytes from queue, one buffer per read
const fromRadioCharacteristic: BluetoothRemoteGATTCharacteristic = {
async readValue() {
const next = incomingQueue.shift() ?? new Uint8Array();
return new DataView(
next.buffer,
next.byteOffset,
next.byteLength,
) as unknown as DataView;
},
addEventListener() {},
removeEventListener() {},
} as unknown as BluetoothRemoteGATTCharacteristic;
// characteristicvaluechanged event plumbing (fromNumCharacteristic)
const charEmitter = new MiniEmitter();
const fromNumCharacteristic: BluetoothRemoteGATTCharacteristic = {
async startNotifications() {
return this;
},
addEventListener(type: string, listener: (e: Event) => void) {
charEmitter.addEventListener(type, listener);
},
removeEventListener(type: string, listener: (e: Event) => void) {
charEmitter.removeEventListener(type, listener);
},
} as unknown as BluetoothRemoteGATTCharacteristic;
const toRadioCharacteristic: BluetoothRemoteGATTCharacteristic = {
async writeValue(bufferSource: BufferSource) {
const u8 =
bufferSource instanceof ArrayBuffer
? new Uint8Array(bufferSource)
: new Uint8Array(
bufferSource.buffer,
bufferSource.byteOffset,
bufferSource.byteLength,
);
lastWritten = new Uint8Array(u8);
},
} as unknown as BluetoothRemoteGATTCharacteristic;
// Primary service returns our three characteristics by UUID
const primaryService: BluetoothRemoteGATTService = {
async getCharacteristic(uuid: string) {
if (uuid === TransportWebBluetooth.ToRadioUuid) return toRadioCharacteristic;
if (uuid === TransportWebBluetooth.FromRadioUuid) return fromRadioCharacteristic;
if (uuid === TransportWebBluetooth.FromNumUuid) return fromNumCharacteristic;
throw new Error("Unknown characteristic: " + uuid);
},
} as unknown as BluetoothRemoteGATTService;
// Device-level emitter to deliver gattserverdisconnected
const deviceEmitter = new MiniEmitter();
// GATT server with readonly connected
let isConnected = true;
const gattServer: BluetoothRemoteGATTServer = {
get connected() {
return isConnected;
},
async connect() {
isConnected = true;
return gattServer;
},
disconnect() {
isConnected = false;
deviceEmitter.dispatchEvent(new Event("gattserverdisconnected"));
},
async getPrimaryService() {
return primaryService;
},
device: {
addEventListener: (...args: Parameters<EventTarget["addEventListener"]>) =>
deviceEmitter.addEventListener(args[0] as string, args[1] as (e: Event) => void),
removeEventListener: (...args: Parameters<EventTarget["removeEventListener"]>) =>
deviceEmitter.removeEventListener(args[0] as string, args[1] as (e: Event) => void),
} as unknown as BluetoothDevice,
} as unknown as BluetoothRemoteGATTServer;
const fakeDevice: BluetoothDevice = {
async watchAdvertisements() {},
gatt: gattServer,
} as unknown as BluetoothDevice;
const fakeNavigator = {
bluetooth: {
async requestDevice() {
return fakeDevice;
},
},
};
vi.stubGlobal("navigator", Object.assign({}, globalThis.navigator, fakeNavigator));
// helper actions for tests/contract
return {
pushIncoming: (u8: Uint8Array) => {
incomingQueue.push(u8);
charEmitter.dispatchEvent(new Event("characteristicvaluechanged"));
},
assertLastWritten: (u8: Uint8Array) => {
expect(lastWritten).toBeDefined();
expect(lastWritten).toEqual(u8);
},
// simulate underlying link drop (OS-level disconnect)
triggerGattDisconnect: () => {
isConnected = false;
deviceEmitter.dispatchEvent(new Event("gattserverdisconnected"));
},
cleanup: () => {
vi.unstubAllGlobals();
},
};
}
describe("TransportWebBluetooth (contract)", () => {
runTransportContract({
name: "TransportWebBluetooth",
setup: () => {},
teardown: () => {
(globalThis as unknown as { __ble?: ReturnType<typeof stubWebBluetooth> }).__ble?.cleanup();
(globalThis as unknown as { __ble?: ReturnType<typeof stubWebBluetooth> }).__ble = undefined;
vi.restoreAllMocks();
vi.unstubAllGlobals();
},
create: async () => {
(globalThis as unknown as { __ble: ReturnType<typeof stubWebBluetooth> }).__ble = stubWebBluetooth();
return await TransportWebBluetooth.create();
},
pushIncoming: async (bytes) => {
(globalThis as unknown as { __ble: ReturnType<typeof stubWebBluetooth> }).__ble.pushIncoming(bytes);
await Promise.resolve();
},
assertLastWritten: (bytes) => {
(globalThis as unknown as { __ble: ReturnType<typeof stubWebBluetooth> }).__ble.assertLastWritten(bytes);
},
triggerDisconnect: async () => {
(globalThis as unknown as { __ble: ReturnType<typeof stubWebBluetooth> }).__ble.triggerGattDisconnect();
await Promise.resolve();
},
});
});

148
packages/transport-web-bluetooth/src/transport.ts

@ -1,5 +1,23 @@
import type { Types } from "@meshtastic/core";
import { Types } from "@meshtastic/core";
function toArrayBuffer(uint8array: Uint8Array): ArrayBuffer {
if (
uint8array.buffer instanceof ArrayBuffer &&
uint8array.byteOffset === 0 &&
uint8array.byteLength === uint8array.buffer.byteLength
) {
return uint8array.buffer;
}
return uint8array.slice().buffer;
}
/**
* Provides Web Bluetooth transport for Meshtastic devices.
*
* Implements the {@link Types.Transport} contract using the Web Bluetooth API.
* Use {@link TransportWebBluetooth.create} or {@link TransportWebBluetooth.createFromDevice}
* to construct an instance.
*/
export class TransportWebBluetooth implements Types.Transport {
private _toDevice: WritableStream<Uint8Array>;
private _fromDevice: ReadableStream<Types.DeviceOutput>;
@ -11,11 +29,23 @@ export class TransportWebBluetooth implements Types.Transport {
private fromNumCharacteristic: BluetoothRemoteGATTCharacteristic;
private gattServer: BluetoothRemoteGATTServer;
private _lastStatus: Types.DeviceStatusEnum =
Types.DeviceStatusEnum.DeviceDisconnected;
private _closingByUser = false;
/** UUID for the "toRadio" write characteristic. */
static ToRadioUuid = "f75c76d2-129e-4dad-a1dd-7866124401e7";
/** UUID for the "fromRadio" read characteristic. */
static FromRadioUuid = "2c55e69e-4993-11ed-b878-0242ac120002";
/** UUID for the "fromNum" notification characteristic. */
static FromNumUuid = "ed9da18c-a800-4f66-a670-aa7547e34453";
/** UUID for the Meshtastic GATT service. */
static ServiceUuid = "6ba1b218-15a8-461f-9fa8-5dcae273eafd";
/**
* Prompts the user to select a Bluetooth device, connects it, and returns a transport.
*/
public static async create(): Promise<TransportWebBluetooth> {
const device = await navigator.bluetooth.requestDevice({
filters: [{ services: [TransportWebBluetooth.ServiceUuid] }],
@ -23,17 +53,25 @@ export class TransportWebBluetooth implements Types.Transport {
return await TransportWebBluetooth.prepareConnection(device);
}
/**
* Creates a transport from an existing, user-provided {@link BluetoothDevice}.
*/
public static async createFromDevice(
device: BluetoothDevice,
): Promise<TransportWebBluetooth> {
return await TransportWebBluetooth.prepareConnection(device);
}
/**
* Prepares and connects to a {@link BluetoothDevice}, resolving its GATT server
* and characteristics, then returning a transport.
*
* @throws if required services or characteristics are missing.
*/
public static async prepareConnection(
device: BluetoothDevice,
): Promise<TransportWebBluetooth> {
const gattServer = await device.gatt?.connect();
if (!gattServer) {
throw new Error("Failed to connect to GATT server");
}
@ -41,7 +79,6 @@ export class TransportWebBluetooth implements Types.Transport {
const service = await gattServer.getPrimaryService(
TransportWebBluetooth.ServiceUuid,
);
const toRadioCharacteristic = await service.getCharacteristic(
TransportWebBluetooth.ToRadioUuid,
);
@ -60,8 +97,6 @@ export class TransportWebBluetooth implements Types.Transport {
throw new Error("Failed to find required characteristics");
}
console.log("Connected to device", device.name);
return new TransportWebBluetooth(
toRadioCharacteristic,
fromRadioCharacteristic,
@ -70,6 +105,10 @@ export class TransportWebBluetooth implements Types.Transport {
);
}
/**
* Create a transport from resolved GATT characteristics and server.
* Prefer using the static factory methods instead.
*/
constructor(
toRadioCharacteristic: BluetoothRemoteGATTCharacteristic,
fromRadioCharacteristic: BluetoothRemoteGATTCharacteristic,
@ -81,22 +120,42 @@ export class TransportWebBluetooth implements Types.Transport {
this.fromNumCharacteristic = fromNumCharacteristic;
this.gattServer = gattServer;
this._fromDevice = new ReadableStream({
this._fromDevice = new ReadableStream<Types.DeviceOutput>({
start: (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._toDevice = new WritableStream({
this._toDevice = new WritableStream<Uint8Array>({
write: async (chunk) => {
await this.toRadioCharacteristic.writeValue(chunk);
try {
const ab = toArrayBuffer(chunk);
await this.toRadioCharacteristic.writeValue(ab);
if (this._isFirstWrite && this._fromDeviceController) {
if (this._isFirstWrite) {
this._isFirstWrite = false;
setTimeout(() => {
this.readFromRadio(this._fromDeviceController!);
}, 50);
setTimeout(() => this.readFromRadio(), 50);
}
} catch (error) {
this.emitStatus(
Types.DeviceStatusEnum.DeviceDisconnected,
"write-error",
);
throw error;
}
},
});
@ -104,26 +163,54 @@ export class TransportWebBluetooth implements Types.Transport {
this.fromNumCharacteristic.addEventListener(
"characteristicvaluechanged",
() => {
if (this._fromDeviceController) {
this.readFromRadio(this._fromDeviceController);
}
this.readFromRadio();
},
);
this.fromNumCharacteristic.startNotifications();
this.fromNumCharacteristic
.startNotifications()
.then(() => {
this.emitStatus(Types.DeviceStatusEnum.DeviceConnected);
})
.catch(() => {
this.emitStatus(
Types.DeviceStatusEnum.DeviceDisconnected,
"notify-failed",
);
});
}
/** Writable stream of bytes to the device. */
get toDevice(): WritableStream<Uint8Array> {
return this._toDevice;
}
/** Readable stream of {@link Types.DeviceOutput} from the device. */
get fromDevice(): ReadableStream<Types.DeviceOutput> {
return this._fromDevice;
}
protected async readFromRadio(
controller: ReadableStreamDefaultController<Types.DeviceOutput>,
): Promise<void> {
/**
* Closes the GATT connection and emits `DeviceDisconnected("user")`.
*/
disconnect(): Promise<void> {
try {
this._closingByUser = true;
this.gattServer.disconnect();
} finally {
this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "user");
this._closingByUser = false;
}
return Promise.resolve();
}
private async readFromRadio(): Promise<void> {
const controller = this._fromDeviceController;
if (!controller) {
return;
}
try {
let hasMoreData = true;
while (hasMoreData && this.fromRadioCharacteristic) {
const value = await this.fromRadioCharacteristic.readValue();
@ -131,15 +218,28 @@ export class TransportWebBluetooth implements Types.Transport {
hasMoreData = false;
continue;
}
controller.enqueue({
this.enqueue({
type: "packet",
data: new Uint8Array(value.buffer),
});
}
} catch {
this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "read-error");
}
}
disconnect(): Promise<void> {
this.gattServer.disconnect();
return Promise.resolve();
private emitStatus(next: Types.DeviceStatusEnum, reason?: string): void {
if (next === this._lastStatus) {
return;
}
this._lastStatus = next;
this._fromDeviceController?.enqueue({
type: "status",
data: { status: next, reason },
});
}
private enqueue(output: Types.DeviceOutput): void {
this._fromDeviceController?.enqueue(output);
}
}

227
packages/transport-web-serial/src/transport.test.ts

@ -0,0 +1,227 @@
import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
import { TransportWebSerial } from "./transport";
import { Types, Utils } from "@meshtastic/core";
import { runTransportContract } from "../../../tests/utils/transportContract";
function stubCoreTransforms() {
const toDevice = new TransformStream<Uint8Array, Uint8Array>({
transform(chunk, controller) {
controller.enqueue(chunk);
},
});
// maps raw bytes -> DeviceOutput.packet
const fromDeviceFactory = () =>
new TransformStream<Uint8Array, Types.DeviceOutput>({
transform(chunk, controller) {
controller.enqueue({ type: "packet", data: chunk });
},
});
const restoreTo = vi
.spyOn(Utils, "toDeviceStream", "get")
.mockReturnValue(toDevice as unknown as typeof Utils.toDeviceStream);
const restoreFrom = vi
.spyOn(Utils, "fromDeviceStream")
.mockImplementation(
() =>
fromDeviceFactory() as unknown as TransformStream<
Uint8Array,
Types.DeviceOutput
>,
);
return {
restore: () => {
restoreTo.mockRestore();
restoreFrom.mockRestore();
},
};
}
type SerialDisconnectHandler = (ev: { port?: any }) => void;
function stubNavigatorSerial() {
type SerialDisconnectHandler = (ev: { port?: any }) => void;
const handlers = new Set<SerialDisconnectHandler>();
const serialStub = {
addEventListener: (type: string, handler: EventListenerOrEventListenerObject) => {
if (type === "disconnect") handlers.add(handler as any as SerialDisconnectHandler);
},
removeEventListener: (type: string, handler: EventListenerOrEventListenerObject) => {
if (type === "disconnect") handlers.delete(handler as any as SerialDisconnectHandler);
},
dispatchDisconnect(port: any) {
for (const h of handlers) h({ port });
},
requestPort: vi.fn(async () => new FakeSerialPort()),
};
const nav: any = (globalThis as any).navigator ?? {};
const hadNavigator = !!(globalThis as any).navigator;
const originalSerial = nav.serial;
if (!hadNavigator) {
Object.defineProperty(globalThis as any, "navigator", {
value: nav,
configurable: true,
writable: false,
});
}
Object.defineProperty(nav, "serial", {
value: serialStub,
configurable: true,
enumerable: true,
writable: true,
});
return {
serialStub,
restore: () => {
if (hadNavigator) {
if (originalSerial === undefined) {
delete (globalThis as any).navigator.serial;
} else {
Object.defineProperty((globalThis as any).navigator, "serial", {
value: originalSerial,
configurable: true,
enumerable: true,
writable: true,
});
}
} else {
delete (globalThis as any).navigator;
}
},
};
}
class FakeSerialPort {
readable: ReadableStream<Uint8Array>;
writable: WritableStream<Uint8Array>;
lastWritten?: Uint8Array;
private _readController!: ReadableStreamDefaultController<Uint8Array>;
constructor() {
this.readable = new ReadableStream<Uint8Array>({
start: (controller) => {
this._readController = controller;
},
});
this.writable = new WritableStream<Uint8Array>({
write: async (chunk) => {
this.lastWritten = chunk;
},
});
}
open(_options?: { baudRate?: number }): Promise<void> {
return Promise.resolve();
}
close(): Promise<void> {
try {
this._readController.close();
} catch {}
return Promise.resolve();
}
pushIncoming(bytes: Uint8Array) {
this._readController.enqueue(bytes);
}
}
describe("TransportWebSerial (contract)", () => {
let transforms: { restore(): void } | undefined;
let navSerial: { serialStub: any; restore(): void } | undefined;
beforeEach(() => {
transforms = stubCoreTransforms();
navSerial = stubNavigatorSerial();
});
afterEach(() => {
transforms?.restore();
navSerial?.restore();
vi.restoreAllMocks();
});
runTransportContract({
name: "TransportWebSerial",
setup: () => {},
teardown: () => {},
create: async () => {
const fake = new FakeSerialPort();
const transport = await TransportWebSerial.createFromPort(fake as any);
(globalThis as any).__ws = { fake, serial: navSerial!.serialStub };
await Promise.resolve();
return transport;
},
pushIncoming: async (bytes) => {
(globalThis as any).__ws.fake.pushIncoming(bytes);
await Promise.resolve();
},
assertLastWritten: (bytes) => {
expect((globalThis as any).__ws.fake.lastWritten).toEqual(bytes);
},
triggerDisconnect: async () => {
(globalThis as any).__ws.serial.dispatchDisconnect(
(globalThis as any).__ws.fake,
);
await Promise.resolve();
},
});
});
describe("TransportWebSerial (extras)", () => {
let transforms: { restore(): void } | undefined;
let navSerial: { serialStub: any; restore(): void } | undefined;
beforeEach(() => {
transforms = stubCoreTransforms();
navSerial = stubNavigatorSerial();
});
afterEach(() => {
transforms?.restore();
navSerial?.restore();
vi.restoreAllMocks();
});
it("emits DeviceDisconnected('serial-disconnected') on OS disconnect event", async () => {
const fake = new FakeSerialPort();
const transport = await TransportWebSerial.createFromPort(fake as any);
(globalThis as any).__ws = { fake, serial: navSerial!.serialStub };
const reader = transport.fromDevice.getReader();
// drain statuses until connected
for (let i = 0; i < 3; i++) {
const { value } = await reader.read();
if (!value || value.type !== "status") break;
if (value.data.status === Types.DeviceStatusEnum.DeviceConnected) break;
}
// fire OS-level disconnect
navSerial!.serialStub.dispatchDisconnect(fake as any);
await Promise.resolve();
let saw = false;
for (let i = 0; i < 6; i++) {
const { value } = await reader.read();
if (value?.type === "status" && value.data.reason === "serial-disconnected") {
saw = true;
break;
}
}
expect(saw).toBe(true);
reader.releaseLock();
await transport.disconnect();
});
});

115
packages/transport-web-serial/src/transport.ts

@ -1,17 +1,37 @@
import type { Types } from "@meshtastic/core";
import { Utils } from "@meshtastic/core";
import { Types, Utils } from "@meshtastic/core";
/**
* Provides Web Serial transport for Meshtastic devices.
*
* Implements the {@link Types.Transport} contract using the Web Serial API.
* Use {@link TransportWebSerial.create} or {@link TransportWebSerial.createFromPort}
* to construct an instance.
*/
export class TransportWebSerial implements Types.Transport {
private _toDevice: WritableStream<Uint8Array>;
private _fromDevice: ReadableStream<Types.DeviceOutput>;
private _fromDeviceController?: ReadableStreamDefaultController<Types.DeviceOutput>;
private connection: SerialPort;
private _portReadable: ReadableStream<Uint8Array>;
private _portWritable: WritableStream<Uint8Array>;
private _lastStatus: Types.DeviceStatusEnum =
Types.DeviceStatusEnum.DeviceDisconnected;
private _closingByUser = false;
/**
* Prompts the user to select a serial port, opens it, and returns a transport.
*/
public static async create(baudRate?: number): Promise<TransportWebSerial> {
const port = await navigator.serial.requestPort();
await port.open({ baudRate: baudRate || 115200 });
return new TransportWebSerial(port);
}
/**
* Creates a transport from an existing, user-provided {@link SerialPort}.
*/
public static async createFromPort(
port: SerialPort,
baudRate?: number,
@ -20,30 +40,109 @@ export class TransportWebSerial implements Types.Transport {
return new TransportWebSerial(port);
}
/**
* Constructs a transport around a given {@link SerialPort}.
* @throws If the port lacks readable or writable streams.
*/
constructor(connection: SerialPort) {
if (!connection.readable || !connection.writable) {
const readable = connection.readable;
const writable = connection.writable;
if (!readable || !writable) {
throw new Error("Stream not accessible");
}
this.connection = connection;
this._portReadable = readable;
this._portWritable = writable;
Utils.toDeviceStream.readable.pipeTo(connection.writable);
Utils.toDeviceStream.readable.pipeTo(this._portWritable);
this._toDevice = Utils.toDeviceStream.writable;
this._fromDevice = connection.readable.pipeThrough(
this._fromDevice = new ReadableStream<Types.DeviceOutput>({
start: async (ctrl) => {
this._fromDeviceController = ctrl;
this.emitStatus(Types.DeviceStatusEnum.DeviceConnecting);
const transformed = this._portReadable.pipeThrough(
Utils.fromDeviceStream(),
);
const reader = transformed.getReader();
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",
);
}
};
navigator.serial.addEventListener("disconnect", onOsDisconnect);
this.emitStatus(Types.DeviceStatusEnum.DeviceConnected);
try {
while (true) {
const { value, done } = await reader.read();
if (done) {
break;
}
if (value) {
ctrl.enqueue(value);
}
}
} catch {
this.emitStatus(
Types.DeviceStatusEnum.DeviceDisconnected,
"read-error",
);
} finally {
reader.releaseLock();
navigator.serial.removeEventListener("disconnect", onOsDisconnect);
}
},
});
}
/** Writable stream of bytes to the device. */
get toDevice(): WritableStream<Uint8Array> {
return this._toDevice;
}
/** Readable stream of {@link Types.DeviceOutput} from the device. */
get fromDevice(): ReadableStream<Types.DeviceOutput> {
return this._fromDevice;
}
disconnect() {
return this.connection.close();
private enqueue(output: Types.DeviceOutput): void {
this._fromDeviceController?.enqueue(output);
}
private emitStatus(next: Types.DeviceStatusEnum, reason?: string): void {
if (next === this._lastStatus) {
return;
}
this._lastStatus = next;
this._fromDeviceController?.enqueue({
type: "status",
data: { status: next, reason },
});
}
/**
* Closes the serial port and emits `DeviceDisconnected("user")`.
*/
async disconnect(): Promise<void> {
try {
this._closingByUser = true;
await this.connection.close();
} finally {
this.emitStatus(Types.DeviceStatusEnum.DeviceDisconnected, "user");
this._closingByUser = false;
}
}
}

29
packages/web/vitest.config.ts

@ -1,23 +1,27 @@
import path from "node:path";
import process from "node:process";
import { fileURLToPath } from "node:url";
import react from "@vitejs/plugin-react";
import { enableMapSet } from "immer";
import { defineProject } from "vitest/config";
enableMapSet();
const __filename = fileURLToPath(import.meta.url);
const __dirname = path.dirname(__filename);
const pkgRoot = __dirname;
const srcDir = path.resolve(pkgRoot, "src");
const publicDir = path.resolve(pkgRoot, "public");
export default defineProject({
plugins: [react()],
resolve: {
alias: {
"@app": path.resolve(process.cwd(), "./packages/web/src"),
"@public": path.resolve(process.cwd(), "./packages/web/public"),
"@core": path.resolve(process.cwd(), "./packages/web/src/core"),
"@pages": path.resolve(process.cwd(), "./packages/web/src/pages"),
"@components": path.resolve(
process.cwd(),
"./packages/web/src/components",
),
"@layouts": path.resolve(process.cwd(), "./packages/web/src/layouts"),
"@app": srcDir,
"@public": publicDir,
"@core": path.resolve(srcDir, "core"),
"@pages": path.resolve(srcDir, "pages"),
"@components": path.resolve(srcDir, "components"),
"@layouts": path.resolve(srcDir, "layouts"),
},
},
test: {
@ -26,8 +30,7 @@ export default defineProject({
mockReset: true,
clearMocks: true,
restoreMocks: true,
root: path.resolve(process.cwd(), "./packages/web/src"),
include: ["**/*.{test,spec}.{ts,tsx}"],
setupFiles: ["./src/tests/setup.ts"],
include: ["src/**/*.{test,spec}.{ts,tsx}"],
setupFiles: [path.resolve(srcDir, "tests/setup.ts")],
},
});

329
pnpm-lock.yaml

@ -21,6 +21,9 @@ importers:
specifier: ^4.9.3
version: 4.9.3
devDependencies:
'@meshtastic/core':
specifier: workspace:*
version: link:packages/core
'@types/node':
specifier: ^22.16.4
version: 22.17.1
@ -66,8 +69,8 @@ importers:
'@meshtastic/core':
specifier: workspace:*
version: link:../core
'serialport':
specifier: npm:serialport@^13.0.0
serialport:
specifier: ^13.0.0
version: 13.0.0
packages/transport-web-bluetooth:
@ -821,12 +824,12 @@ packages:
resolution: {integrity: sha512-oGB+UxlgWcgQkgwo8GcEGwemoTFt3FIO9ababBmaGwXIoBKZ+GTy0pP185beGg7Llih/NSHSV2XAs1lnznocSg==}
engines: {node: '>= 8'}
'@oxc-project/[email protected]0.0':
resolution: {integrity: sha512-3rzy1bJAZ4s7zV9TKT60x119RwJDCDqEtCwK/Zc2qlm7wGhiIUxLLYUhE/mN91yB0u1kxm5sh4NjU12sPqQTpg==}
'@oxc-project/[email protected]1.0':
resolution: {integrity: sha512-zm/LDVOq9FEmHiuM8zO4DWirv0VP2Tv2VsgaiHby9nvpq+FVrcqNYgv+TysLKOITQXWZj/roluTxFvpkHP0Iuw==}
engines: {node: '>=6.9.0'}
'@oxc-project/[email protected]0.0':
resolution: {integrity: sha512-xxHQm8wfCv2e8EmtaDwpMeAHOWqgQDAYg+BJouLXSQt5oTKu9TIXrgNMGSrM2fLvKmECsRd9uUFAAD+hPyootA==}
'@oxc-project/[email protected]1.0':
resolution: {integrity: sha512-CnOqkybZK8z6Gx7Wb1qF7AEnSzbol1WwcIzxYOr8e91LytGOjo0wCpgoYWZo8sdbpqX+X+TJayIzo4Pv0R/KjA==}
'@quansync/[email protected]':
resolution: {integrity: sha512-vy/41FCdnIalPTQCb2Wl0ic1caMdzGus4ktDp+gpZesQNydXcx8nhh8qB3qMPbGkictOTaXgXEUUfQEm8DQYoA==}
@ -1343,81 +1346,81 @@ packages:
'@radix-ui/[email protected]':
resolution: {integrity: sha512-HPwpGIzkl28mWyZqG52jiqDJ12waP11Pa1lGoiyUkIEuMLBP0oeK/C89esbXrxsky5we7dfd8U58nm0SgAWpVw==}
'@rolldown/[email protected]1':
resolution: {integrity: sha512-0mFtKwOG7smn0HkvQ6h8j0m/ohkR7Fp5eMTJ2Pns/HSbePHuDpxMaQ4TjZ6arlVXxpeWZlAHeT5BeNsOA3iWTg==}
'@rolldown/[email protected]2':
resolution: {integrity: sha512-Gs+313LfR4Ka3hvifdag9r44WrdKQaohya7ZXUXzARF7yx0atzFlVZjsvxtKAw1Vmtr4hB/RjUD1jf73SW7zDw==}
cpu: [arm64]
os: [android]
'@rolldown/[email protected]1':
resolution: {integrity: sha512-BHfHJ8Nb5G7ZKJl6pimJacupONT4F7w6gmQHw41rouAnJF51ORDwGefWeb6OMLzGmJwzxlIVPERfnJf1EsMM7A==}
'@rolldown/[email protected]2':
resolution: {integrity: sha512-W8oMqzGcI7wKPXUtS3WJNXzbghHfNiuM1UBAGpVb+XlUCgYRQJd2PRGP7D3WGql3rR3QEhUvSyAuCBAftPQw6Q==}
cpu: [arm64]
os: [darwin]
'@rolldown/[email protected]1':
resolution: {integrity: sha512-4MiuRtExC08jHbSU/diIL+IuQP+3Ck1FbWAplK+ysQJ7fxT3DMxy5FmnIGfmhaqow8oTjb2GEwZJKgTRjZL1Vw==}
'@rolldown/[email protected]2':
resolution: {integrity: sha512-pM4c4sKUk37noJrnnDkJknLhCsfZu7aWyfe67bD0GQHfzAPjV16wPeD9CmQg4/0vv+5IfHYaa4VE536xbA+W0Q==}
cpu: [x64]
os: [darwin]
'@rolldown/[email protected]1':
resolution: {integrity: sha512-nffC1u7ccm12qlAea8ExY3AvqlaHy/o/3L4p5Es8JFJ3zJSs6e3DyuxGZZVdl9EVwsLxPPTvioIl4tEm2afwyw==}
'@rolldown/[email protected]2':
resolution: {integrity: sha512-M8SUgFlYb5kJJWcFC8gUMRiX4WLFxPKMed3SJ2YrxontgIrEcpizPU8nLNVsRYEStoSfKHKExpQw3OP6fm+5bw==}
cpu: [x64]
os: [freebsd]
'@rolldown/[email protected]1':
resolution: {integrity: sha512-LHmAaB3rB1GOJuHscKcL2Ts/LKLcb3YWTh2uQ/876rg/J9WE9kQ0kZ+3lRSYbth/YL8ln54j4JZmHpqQY3xptQ==}
'@rolldown/[email protected]2':
resolution: {integrity: sha512-FuQpbNC/hE//bvv29PFnk0AtpJzdPdYl5CMhlWPovd9g3Kc3lw9TrEPIbL7gRPUdhKAiq6rVaaGvOnXxsa0eww==}
cpu: [arm]
os: [linux]
'@rolldown/[email protected]1':
resolution: {integrity: sha512-oTDZVfqIAjLB2I1yTiLyyhfPPO6dky33sTblxTCpe+ZT55WizN3KDoBKJ4yXG8shI6I4bRShVu29Xg0yAjyQYw==}
'@rolldown/[email protected]2':
resolution: {integrity: sha512-hRZygRlaGCjcNTNY9GV7dDI18sG1dK3cc7ujHq72LoDad23zFDUGMQjiSxHWK+/r92iMV+j2MiHbvzayxqynsg==}
cpu: [arm64]
os: [linux]
'@rolldown/[email protected]1':
resolution: {integrity: sha512-duJ3IkEBj9Xe9NYW1n8Y3483VXHGi8zQ0ZsLbK8464EJUXLF7CXM8Ry+jkkUw+ZvA+Zu1E/+C6p2Y6T9el0C9g==}
'@rolldown/[email protected]2':
resolution: {integrity: sha512-HzgT6h+CXLs+GKAU0Wvkt3rvcv0CmDBsDjlPhh4GHysOKbG9NjpKYX2zvjx671E9pGbTvcPpwy7gGsy7xpu+8g==}
cpu: [arm64]
os: [linux]
'@rolldown/[email protected]':
resolution: {integrity: sha512-qdbmU5QSZ0uoLZBYMxiHsMQmizqtzFGTVPU5oyU1n0jU0Mo+mkSzqZuL8VBnjHOHzhVxZsoAGH9JjiRzCnoGVA==}
cpu: [arm64]
os: [openharmony]
'@rolldown/[email protected]':
resolution: {integrity: sha512-H7+r34TSV8udB2gAsebFM/YuEeNCkPGEAGJ1JE7SgI9XML6FflqcdKfrRSneQFsPaom/gCEc1g0WW5MZ0O3blw==}
'@rolldown/[email protected]':
resolution: {integrity: sha512-Ab/wbf6gdzphDbsg51UaxsC93foQ7wxhtg0SVCXd25BrV4MAJ1HoDtKN/f4h0maFmJobkqYub2DlmoasUzkvBg==}
cpu: [x64]
os: [linux]
'@rolldown/[email protected]1':
resolution: {integrity: sha512-zRm2YmzFVqbsmUsyyZnHfJrOlQUcWS/FJ5ZWL8Q1kZh5PnLBrTVZNpakIWwAxpN5gNEi9MmFd5YHocVJp8ps1Q==}
'@rolldown/[email protected]':
resolution: {integrity: sha512-VoxqGEfh5A1Yx+zBp/FR5QwAbtzbuvky2SVc+ii4g1gLD4zww6mt/hPi5zG+b88zYPFBKHpxMtsz9cWqXU5V5Q==}
cpu: [x64]
os: [linux]
'@rolldown/[email protected]':
resolution: {integrity: sha512-fM1eUIuHLsNJXRlWOuIIex1oBJ89I0skFWo5r/D3KSJ5gD9MBd3g4Hp+v1JGohvyFE+7ylnwRxSUyMEeYpA69A==}
'@rolldown/[email protected]':
resolution: {integrity: sha512-qZ1ViyOUDGbiZrSAJ/FIAhYUElDfVxxFW6DLT/w4KeoZN3HsF4jmRP95mXtl51/oGrqzU9l9Q2f7/P4O/o2ZZA==}
cpu: [arm64]
os: [openharmony]
'@rolldown/[email protected]':
resolution: {integrity: sha512-hEkG3wD+f3wytV0lqwb/uCrXc4r4Ny/DWJFJPfQR3VeMWplhWGgSHNwZc2Q7k86Yi36f9NNzzWmrIuvHI9lCVw==}
engines: {node: '>=14.0.0'}
cpu: [wasm32]
'@rolldown/[email protected]':
resolution: {integrity: sha512-4nftR9V2KHH3zjBwf6leuZZJQZ7v0d70ogjHIqB3SDsbDLvVEZiGSsSn2X6blSZRZeJSFzK0pp4kZ67zdZXwSw==}
'@rolldown/[email protected]2':
resolution: {integrity: sha512-k3MvDf8SiA7uP2ikP0unNouJ2YCrnwi7xcVW+RDgMp5YXVr3Xu6svmT3HGn0tkCKUuPmf+uy8I5uiHt5qWQbew==}
cpu: [arm64]
os: [win32]
'@rolldown/[email protected]1':
resolution: {integrity: sha512-0TQcKu9xZVHYALit+WJsSuADGlTFfOXhnZoIHWWQhTk3OgbwwbYcSoZUXjRdFmR6Wswn4csHtJGN1oYKeQ6/2g==}
'@rolldown/[email protected]2':
resolution: {integrity: sha512-wAi/FxGh7arDOUG45UmnXE1sZUa0hY4cXAO2qWAjFa3f7bTgz/BqwJ7XN5SUezvAJPNkME4fEpInfnBvM25a0w==}
cpu: [ia32]
os: [win32]
'@rolldown/[email protected]1':
resolution: {integrity: sha512-3zMICWwpZh1jrkkKDYIUCx/2wY3PXLICAS0AnbeLlhzfWPhCcpNK9eKhiTlLAZyTp+3kyipoi/ZSVIh+WDnBpQ==}
'@rolldown/[email protected]2':
resolution: {integrity: sha512-Ej0i4PZk8ltblZtzVK8ouaGUacUtxRmTm5S9794mdyU/tYxXjAJNseOfxrnHpMWKjMDrOKbqkPqJ52T9NR4LQQ==}
cpu: [x64]
os: [win32]
'@rolldown/[email protected]':
resolution: {integrity: sha512-+d0F4MKMCbeVUJwG96uQ4SgAznZNSq93I3V+9NHA4OpvqG8mRCpGdKmK8l/dl02h2CCDHwW2FqilnTyDcAnqjA==}
'@rolldown/[email protected]1':
resolution: {integrity: sha512-IaDZ9NhjOIOkYtm+hH0GX33h3iVZ2OeSUnFF0+7Z4+1GuKs4Kj5wK3+I2zNV9IPLfqV4XlwWif8SXrZNutxciQ==}
'@rolldown/[email protected]2':
resolution: {integrity: sha512-QReCdvxiUZAPkvp1xpAg62IeNzykOFA6syH2CnClif4YmALN1XKpB39XneL80008UbtMShthSVDKmrx05N1q/g==}
'@rollup/[email protected]':
resolution: {integrity: sha512-iKnFXr7NkdZAIHiIWE+BX5ULi/ucVFYWD6TbAV+rZctiRTY2PL6tsIKhoIOaoskiWAkgu+VsbXgUVDNLHf+InQ==}
@ -1523,6 +1526,70 @@ packages:
cpu: [x64]
os: [win32]
'@serialport/[email protected]':
resolution: {integrity: sha512-HAFzGhk9OuFMpuor7aT5G1ChPgn5qSsklTFOTUX72Rl6p0xwcSVsRtG/xaGp6bxpN7fI9D/S8THLBWbBgS6ldw==}
engines: {node: '>=12.0.0'}
'@serialport/[email protected]':
resolution: {integrity: sha512-r25o4Bk/vaO1LyUfY/ulR6hCg/aWiN6Wo2ljVlb4Pj5bqWGcSRC4Vse4a9AcapuAu/FeBzHCbKMvRQeCuKjzIQ==}
engines: {node: '>=18.0.0'}
'@serialport/[email protected]':
resolution: {integrity: sha512-CJaUd5bLvtM9c5dmO9rPBHPXTa9R2UwpkJ0wdh9JCYcbrPWsKz+ErvR0hBLeo7NPeiFdjFO4sonRljiw4d2XiA==}
engines: {node: ^12.22 || ^14.13 || >=16}
'@serialport/[email protected]':
resolution: {integrity: sha512-32yvqeTAqJzAEtX5zCrN1Mej56GJ5h/cVFsCDPbF9S1ZSC9FWjOqNAgtByseHfFTSTs/4ZBQZZcZBpolt8sUng==}
engines: {node: '>=20.0.0'}
'@serialport/[email protected]':
resolution: {integrity: sha512-RErAe57g9gvnlieVYGIn1xymb1bzNXb2QtUQd14FpmbQQYlcrmuRnJwKa1BgTCujoCkhtaTtgHlbBWOxm8U2uA==}
engines: {node: '>=20.0.0'}
'@serialport/[email protected]':
resolution: {integrity: sha512-gu26tVt5lQoybhorLTPsH2j2LnX3AOP2x/34+DUSTNaUTzu2fBXw+isVjQJpUBFWu6aeQRZw5bJol5X9Gxjblw==}
engines: {node: '>=12.0.0'}
'@serialport/[email protected]':
resolution: {integrity: sha512-Qqyb0FX1avs3XabQqNaZSivyVbl/yl0jywImp7ePvfZKLwx7jBZjvL+Hawt9wIG6tfq6zbFM24vzCCK7REMUig==}
engines: {node: '>=20.0.0'}
'@serialport/[email protected]':
resolution: {integrity: sha512-a0w0WecTW7bD2YHWrpTz1uyiWA2fDNym0kjmPeNSwZ2XCP+JbirZt31l43m2ey6qXItTYVuQBthm75sPVeHnGA==}
engines: {node: '>=20.0.0'}
'@serialport/[email protected]':
resolution: {integrity: sha512-60ZDDIqYRi0Xs2SPZUo4Jr5LLIjtb+rvzPKMJCohrO6tAqSDponcNpcB1O4W21mKTxYjqInSz+eMrtk0LLfZIg==}
engines: {node: '>=8.6.0'}
'@serialport/[email protected]':
resolution: {integrity: sha512-O7cywCWC8PiOMvo/gglEBfAkLjp/SENEML46BXDykfKP5mTPM46XMaX1L0waWU6DXJpBgjaL7+yX6VriVPbN4w==}
engines: {node: '>=12.0.0'}
'@serialport/[email protected]':
resolution: {integrity: sha512-dov3zYoyf0dt1Sudd1q42VVYQ4WlliF0MYvAMA3MOyiU1IeG4hl0J6buBA2w4gl3DOCC05tGgLDN/3yIL81gsA==}
engines: {node: '>=20.0.0'}
'@serialport/[email protected]':
resolution: {integrity: sha512-JNUQA+y2Rfs4bU+cGYNqOPnNMAcayhhW+XJZihSLQXOHcZsFnOa2F9YtMg9VXRWIcnHldHYtisp62Etjlw24bw==}
engines: {node: '>=20.0.0'}
'@serialport/[email protected]':
resolution: {integrity: sha512-m7HpIf56G5XcuDdA3DB34Z0pJiwxNRakThEHjSa4mG05OnWYv0IG8l2oUyYfuGMowQWaVnQ+8r+brlPxGVH+eA==}
engines: {node: '>=20.0.0'}
'@serialport/[email protected]':
resolution: {integrity: sha512-fUHZEExm6izJ7rg0A1yjXwu4sOzeBkPAjDZPfb+XQoqgtKAk+s+HfICiYn7N2QU9gyaeCO8VKgWwi+b/DowYOg==}
engines: {node: '>=20.0.0'}
'@serialport/[email protected]':
resolution: {integrity: sha512-DoXJ3mFYmyD8X/8931agJvrBPxqTaYDsPoly9/cwQSeh/q4EjQND9ySXBxpWz5WcpyCU4jOuusqCSAPsbB30Eg==}
engines: {node: '>=20.0.0'}
'@serialport/[email protected]':
resolution: {integrity: sha512-F7xLJKsjGo2WuEWMSEO1SimRcOA+WtWICsY13r0ahx8s2SecPQH06338g28OT7cW7uRXI7oEQAk62qh5gHJW3g==}
engines: {node: '>=20.0.0'}
'@standard-schema/[email protected]':
resolution: {integrity: sha512-e7Mew686owMaPJVNNLs55PUvgz371nKgwsc4vxE49zsODpJEnxgxRo2y/OKrqueavXgZNMDVj3DdHFlaSAeU8g==}
@ -2555,6 +2622,15 @@ packages:
resolution: {integrity: sha512-jRFi8UDGo6j+odZiEpjazZaWqEal3w/basFjQHQEwVtZJGDpxbH1MeYluwCS8Xq5wmLJooDlMgvVarmWfGM44g==}
engines: {node: '>=0.10'}
[email protected]:
resolution: {integrity: sha512-6WTZ/IxCY/T6BALoZHaE4ctp9xm+Z5kY/pzYaCHRFeyVhojxlrm+46y68HA6hr0TcwEssoxNiDEUJQjfPZ/RYA==}
engines: {node: '>=6.0'}
peerDependencies:
supports-color: '*'
peerDependenciesMeta:
supports-color:
optional: true
[email protected]:
resolution: {integrity: sha512-KcKCqiftBJcZr++7ykoDIEwSa3XWowTfNPo92BYxjXiyYEVrUQh2aLyhxBCwww+heortUFxEJYcRzosstTEBYQ==}
engines: {node: '>=6.0'}
@ -3228,6 +3304,10 @@ packages:
no-[email protected]:
resolution: {integrity: sha512-fgAN3jGAh+RoxUGZHTSOLJIqUc2wmoBwGR4tbpNAKmmovFoWq0OdRkb0VkldReO2a2iBT/OEulG9XSUc10r3zg==}
[email protected]:
resolution: {integrity: sha512-8VOpLHFrOQlAH+qA0ZzuGRlALRA6/LVh8QJldbrC4DY0hXoMP0l4Acq8TzFC018HztWiRqyCEj2aTWY2UvnJUg==}
engines: {node: ^18 || ^20 || >= 21}
[email protected]:
resolution: {integrity: sha512-c4FRfUm/dbcWZ7U+1Wq0AwCyFL+3nt2bEw05wfxSz+DWpWsitgmSgYmy2dQdWyKC1694ELPqMs/YzUSNozLt8A==}
engines: {node: 4.x || >=6.0.0}
@ -3237,6 +3317,10 @@ packages:
encoding:
optional: true
[email protected]:
resolution: {integrity: sha512-LA4ZjwlnUblHVgq0oBF3Jl/6h/Nvs5fzBLwdEF4nuxnFdsfajde4WfxtJr3CaiH+F6ewcIB/q4jQ4UzPyid+CQ==}
hasBin: true
[email protected]:
resolution: {integrity: sha512-RaBPP3+51hPne/OolXxcz89iYvQvKOydaqoePpOgXcrOKZhjVIzmpKZz+Hd/RBO2/zN2q6CNJhQzucVz+u3Jyw==}
@ -3572,8 +3656,8 @@ packages:
vue-tsc:
optional: true
[email protected]1:
resolution: {integrity: sha512-M2Q+RfG0FMJeSW3RSFTbvtjGVTcQpTQvN247D0EMSsPkpZFoinopR9oAnQiwgogQyzDuvKNnbyCbQQlmNAzSoQ==}
[email protected]2:
resolution: {integrity: sha512-vxI2sPN07MMaoYKlFrVva5qZ1Y7DAZkgp7MQwTnyHt4FUMz9Sh+YeCzNFV9JYHI6ZNwoGWLCfCViE3XVsRC1cg==}
hasBin: true
[email protected]:
@ -3618,6 +3702,10 @@ packages:
engines: {node: '>=10'}
hasBin: true
[email protected]:
resolution: {integrity: sha512-PHpnTd8isMGPfFTZNCzOZp9m4mAJSNWle9Jxu6BPTcWq7YXl5qN7tp8Sgn0h+WIGcD6JFz5QDgixC2s4VW7vzg==}
engines: {node: '>=20.0.0'}
[email protected]:
resolution: {integrity: sha512-0QvCV2lM3aj/U3YozDiVwx9zpH0q8A60CTWIv4Jszj/givcudPb48B+rkU5D51NJ0pTpweGMttHjboPa9/zoIQ==}
engines: {node: '>=10'}
@ -4613,9 +4701,9 @@ snapshots:
'@nodelib/fs.scandir': 2.1.5
fastq: 1.19.1
'@oxc-project/[email protected]0.0': {}
'@oxc-project/[email protected]1.0': {}
'@oxc-project/[email protected]0.0': {}
'@oxc-project/[email protected]1.0': {}
'@quansync/[email protected]':
dependencies:
@ -5170,53 +5258,53 @@ snapshots:
'@radix-ui/[email protected]': {}
'@rolldown/[email protected]1':
'@rolldown/[email protected]2':
optional: true
'@rolldown/[email protected]1':
'@rolldown/[email protected]2':
optional: true
'@rolldown/[email protected]1':
'@rolldown/[email protected]2':
optional: true
'@rolldown/[email protected]1':
'@rolldown/[email protected]2':
optional: true
'@rolldown/[email protected]1':
'@rolldown/[email protected]2':
optional: true
'@rolldown/[email protected]1':
'@rolldown/[email protected]2':
optional: true
'@rolldown/[email protected]1':
'@rolldown/[email protected]2':
optional: true
'@rolldown/binding-linux-[email protected]':
'@rolldown/binding-linux-[email protected]':
optional: true
'@rolldown/binding-linux-x64-[email protected]':
'@rolldown/binding-linux-x64-[email protected]':
optional: true
'@rolldown/binding-[email protected]':
'@rolldown/binding-[email protected]':
optional: true
'@rolldown/[email protected]1':
'@rolldown/[email protected]2':
dependencies:
'@napi-rs/wasm-runtime': 1.0.3
optional: true
'@rolldown/[email protected]1':
'@rolldown/[email protected]2':
optional: true
'@rolldown/[email protected]1':
'@rolldown/[email protected]2':
optional: true
'@rolldown/[email protected]1':
'@rolldown/[email protected]2':
optional: true
'@rolldown/[email protected]': {}
'@rolldown/[email protected]1': {}
'@rolldown/[email protected]2': {}
'@rollup/[email protected]':
dependencies:
@ -5283,6 +5371,60 @@ snapshots:
'@rollup/[email protected]':
optional: true
'@serialport/[email protected]':
dependencies:
'@serialport/bindings-interface': 1.2.2
debug: 4.4.1
transitivePeerDependencies:
- supports-color
'@serialport/[email protected]':
dependencies:
'@serialport/bindings-interface': 1.2.2
'@serialport/parser-readline': 12.0.0
debug: 4.4.0
node-addon-api: 8.3.0
node-gyp-build: 4.8.4
transitivePeerDependencies:
- supports-color
'@serialport/[email protected]': {}
'@serialport/[email protected]': {}
'@serialport/[email protected]': {}
'@serialport/[email protected]': {}
'@serialport/[email protected]': {}
'@serialport/[email protected]': {}
'@serialport/[email protected]': {}
'@serialport/[email protected]':
dependencies:
'@serialport/parser-delimiter': 12.0.0
'@serialport/[email protected]':
dependencies:
'@serialport/parser-delimiter': 13.0.0
'@serialport/[email protected]': {}
'@serialport/[email protected]': {}
'@serialport/[email protected]': {}
'@serialport/[email protected]': {}
'@serialport/[email protected]':
dependencies:
'@serialport/bindings-interface': 1.2.2
debug: 4.4.0
transitivePeerDependencies:
- supports-color
'@standard-schema/[email protected]': {}
'@tailwindcss/[email protected]':
@ -7117,6 +7259,10 @@ snapshots:
dependencies:
assert-plus: 1.0.0
[email protected]:
dependencies:
ms: 2.1.3
[email protected]:
dependencies:
ms: 2.1.3
@ -7758,10 +7904,14 @@ snapshots:
lower-case: 2.0.2
tslib: 2.8.1
[email protected]: {}
[email protected]:
dependencies:
whatwg-url: 5.0.0
[email protected]: {}
[email protected]:
dependencies:
css-select: 4.3.0
@ -8065,7 +8215,7 @@ snapshots:
[email protected]: {}
[email protected]([email protected]1)([email protected]):
[email protected]([email protected]2)([email protected]):
dependencies:
'@babel/generator': 7.28.0
'@babel/parser': 7.28.0
@ -8075,34 +8225,34 @@ snapshots:
debug: 4.4.1
dts-resolver: 2.1.1
get-tsconfig: 4.10.1
rolldown: 1.0.0-beta.31
rolldown: 1.0.0-beta.32
optionalDependencies:
typescript: 5.9.2
transitivePeerDependencies:
- oxc-resolver
- supports-color
[email protected]1:
[email protected]2:
dependencies:
'@oxc-project/runtime': 0.80.0
'@oxc-project/types': 0.80.0
'@rolldown/pluginutils': 1.0.0-beta.31
'@oxc-project/runtime': 0.81.0
'@oxc-project/types': 0.81.0
'@rolldown/pluginutils': 1.0.0-beta.32
ansis: 4.1.0
optionalDependencies:
'@rolldown/binding-android-arm64': 1.0.0-beta.31
'@rolldown/binding-darwin-arm64': 1.0.0-beta.31
'@rolldown/binding-darwin-x64': 1.0.0-beta.31
'@rolldown/binding-freebsd-x64': 1.0.0-beta.31
'@rolldown/binding-linux-arm-gnueabihf': 1.0.0-beta.31
'@rolldown/binding-linux-arm64-gnu': 1.0.0-beta.31
'@rolldown/binding-linux-arm64-musl': 1.0.0-beta.31
'@rolldown/binding-linux-arm64-ohos': 1.0.0-beta.31
'@rolldown/binding-linux-x64-gnu': 1.0.0-beta.31
'@rolldown/binding-linux-x64-musl': 1.0.0-beta.31
'@rolldown/binding-wasm32-wasi': 1.0.0-beta.31
'@rolldown/binding-win32-arm64-msvc': 1.0.0-beta.31
'@rolldown/binding-win32-ia32-msvc': 1.0.0-beta.31
'@rolldown/binding-win32-x64-msvc': 1.0.0-beta.31
'@rolldown/binding-android-arm64': 1.0.0-beta.32
'@rolldown/binding-darwin-arm64': 1.0.0-beta.32
'@rolldown/binding-darwin-x64': 1.0.0-beta.32
'@rolldown/binding-freebsd-x64': 1.0.0-beta.32
'@rolldown/binding-linux-arm-gnueabihf': 1.0.0-beta.32
'@rolldown/binding-linux-arm64-gnu': 1.0.0-beta.32
'@rolldown/binding-linux-arm64-musl': 1.0.0-beta.32
'@rolldown/binding-linux-x64-gnu': 1.0.0-beta.32
'@rolldown/binding-linux-x64-musl': 1.0.0-beta.32
'@rolldown/binding-openharmony-arm64': 1.0.0-beta.32
'@rolldown/binding-wasm32-wasi': 1.0.0-beta.32
'@rolldown/binding-win32-arm64-msvc': 1.0.0-beta.32
'@rolldown/binding-win32-ia32-msvc': 1.0.0-beta.32
'@rolldown/binding-win32-x64-msvc': 1.0.0-beta.32
[email protected]:
dependencies:
@ -8158,6 +8308,25 @@ snapshots:
[email protected]: {}
[email protected]:
dependencies:
'@serialport/binding-mock': 10.2.2
'@serialport/bindings-cpp': 13.0.0
'@serialport/parser-byte-length': 13.0.0
'@serialport/parser-cctalk': 13.0.0
'@serialport/parser-delimiter': 13.0.0
'@serialport/parser-inter-byte-timeout': 13.0.0
'@serialport/parser-packet-length': 13.0.0
'@serialport/parser-readline': 13.0.0
'@serialport/parser-ready': 13.0.0
'@serialport/parser-regex': 13.0.0
'@serialport/parser-slip-encoder': 13.0.0
'@serialport/parser-spacepacket': 13.0.0
'@serialport/stream': 13.0.0
debug: 4.4.0
transitivePeerDependencies:
- supports-color
[email protected]([email protected]):
dependencies:
seroval: 1.3.2
@ -8396,8 +8565,8 @@ snapshots:
diff: 8.0.2
empathic: 2.0.0
hookable: 5.5.3
rolldown: 1.0.0-beta.31
rolldown-plugin-dts: 0.15.6([email protected]1)([email protected])
rolldown: 1.0.0-beta.32
rolldown-plugin-dts: 0.15.6([email protected]2)([email protected])
semver: 7.7.2
tinyexec: 1.0.1
tinyglobby: 0.2.14

122
tests/utils/transportContract.ts

@ -0,0 +1,122 @@
import { Types } from "@meshtastic/core";
import { describe, expect, it } from "vitest";
export interface TransportContract {
name: string;
create: () => Promise<Types.Transport>;
setup?: () => void | Promise<void>;
teardown?: () => void | Promise<void>;
pushIncoming?: (bytes: Uint8Array) => void | Promise<void>;
assertLastWritten?: (bytes: Uint8Array) => void;
triggerDisconnect?: () => void | Promise<void>;
}
async function readUntilType(
reader: ReadableStreamDefaultReader<Types.DeviceOutput>,
expectedType: Types.DeviceOutput["type"],
maxReads = 20,
): Promise<Types.DeviceOutput> {
for (let i = 0; i < maxReads; i++) {
const { value, done } = await reader.read();
if (done) {
break;
}
if (value && value.type === expectedType) {
return value;
}
}
throw new Error(
`Did not receive a '${expectedType}' event within ${maxReads} reads`,
);
}
export function runTransportContract(contract: TransportContract) {
describe(contract.name, () => {
it("reads packets from fromDevice", async () => {
await contract.setup?.();
const transport = await contract.create();
const reader = transport.fromDevice.getReader();
const sampleBytes = new Uint8Array([0x01, 0x02, 0x03]);
await contract.pushIncoming?.(sampleBytes);
const packetEvent = await readUntilType(reader, "packet");
expect("data" in packetEvent ? packetEvent.data : undefined).toEqual(
sampleBytes,
);
reader.releaseLock();
await contract.teardown?.();
});
it("writes bytes to toDevice", async () => {
await contract.setup?.();
const transport = await contract.create();
const writer = transport.toDevice.getWriter();
const outgoingBytes = new Uint8Array([0xaa, 0xbb]);
await writer.write(outgoingBytes);
await writer.close();
contract.assertLastWritten?.(outgoingBytes);
await contract.teardown?.();
});
it("disconnect() emits DeviceDisconnected('user')", async () => {
await contract.setup?.();
const transport = await contract.create();
const reader = transport.fromDevice.getReader();
// Trigger user disconnect
await transport.disconnect();
// Read a few events and assert we eventually see the user disconnect.
let sawUser = false;
for (let i = 0; i < 10; i++) {
const { value } = await reader.read();
if (
value &&
value.type === "status" &&
value.data.status === Types.DeviceStatusEnum.DeviceDisconnected &&
value.data.reason === "user"
) {
sawUser = true;
break;
}
}
expect(sawUser).toBe(true);
reader.releaseLock();
await contract.teardown?.();
});
it("emits DeviceDisconnected when the underlying link drops", async () => {
await contract.setup?.();
const transport = await contract.create();
const reader = transport.fromDevice.getReader();
await contract.triggerDisconnect?.();
// As above, read a few events and assert we eventually see "disconnected"
let sawDrop = false;
for (let i = 0; i < 10; i++) {
const { value } = await reader.read();
if (
value &&
value.type === "status" &&
value.data.status === Types.DeviceStatusEnum.DeviceDisconnected
) {
sawDrop = true;
break;
}
}
expect(sawDrop).toBe(true);
reader.releaseLock();
await contract.teardown?.();
});
});
}
Loading…
Cancel
Save