144 lines
4.4 KiB
TypeScript
144 lines
4.4 KiB
TypeScript
/**
|
|
* In-memory `DriveTransport` for tests.
|
|
*
|
|
* Modelled on the FakeWebSocketBroker pattern from drive-webrtc: a
|
|
* single `Broker` object knows about every transport instance, so
|
|
* an `advertise` call on transport A reaches every subscriber on
|
|
* transport B that is registered for the same broker.
|
|
*
|
|
* `requestChunk` resolves with the bytes the broker has stored under
|
|
* `(peerId, fileId, chunkId)`. Tests stage bytes via `broker.serve(...)`.
|
|
*/
|
|
|
|
import type { PeerId } from "@suaveplan/types";
|
|
import type {
|
|
AdvertisementHandler,
|
|
ChunkAdvertisement,
|
|
DriveTransport,
|
|
Unsubscribe,
|
|
} from "../../src/types/index.js";
|
|
|
|
export interface ChunkBytesSource {
|
|
/** Bytes to return, OR a thrown error / promise rejection. */
|
|
readonly bytes?: Uint8Array;
|
|
readonly throwReason?: string;
|
|
/** Optional delay in ms before resolving. */
|
|
readonly delayMs?: number;
|
|
}
|
|
|
|
export class Broker {
|
|
public readonly transports = new Set<FakeDriveTransport>();
|
|
public readonly chunkStore = new Map<string, ChunkBytesSource>();
|
|
|
|
public serve(
|
|
peerId: PeerId,
|
|
fileId: string,
|
|
chunkId: number,
|
|
source: ChunkBytesSource
|
|
): void {
|
|
this.chunkStore.set(`${peerId}:${fileId}:${chunkId}`, source);
|
|
}
|
|
|
|
public deliverAdvertisement(
|
|
senderId: PeerId,
|
|
advert: ChunkAdvertisement
|
|
): void {
|
|
for (const transport of this.transports) {
|
|
for (const handler of transport.handlers) {
|
|
if (transport.localPeerId === senderId) continue;
|
|
handler(senderId, advert);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
export class FakeDriveTransport implements DriveTransport {
|
|
public readonly id: string;
|
|
public readonly localPeerId: PeerId;
|
|
public readonly handlers = new Set<AdvertisementHandler>();
|
|
public readonly knownPeers = new Set<PeerId>();
|
|
readonly #broker: Broker;
|
|
|
|
public constructor(args: {
|
|
readonly broker: Broker;
|
|
readonly localPeerId: PeerId;
|
|
readonly id?: string;
|
|
}) {
|
|
this.id = args.id ?? "fake";
|
|
this.localPeerId = args.localPeerId;
|
|
this.#broker = args.broker;
|
|
this.#broker.transports.add(this);
|
|
}
|
|
|
|
public addKnownPeer(peerId: PeerId): void {
|
|
this.knownPeers.add(peerId);
|
|
}
|
|
|
|
public async requestChunk(
|
|
peerId: PeerId,
|
|
fileId: string,
|
|
chunkId: number,
|
|
signal?: AbortSignal
|
|
): Promise<Uint8Array> {
|
|
const key = `${peerId}:${fileId}:${chunkId}`;
|
|
const source = this.#broker.chunkStore.get(key);
|
|
if (source === undefined) {
|
|
throw new Error(`no chunk staged at ${key}`);
|
|
}
|
|
if (source.delayMs !== undefined && source.delayMs > 0) {
|
|
await new Promise<void>((resolve, reject) => {
|
|
const handle = setTimeout(() => {
|
|
signal?.removeEventListener("abort", onAbort);
|
|
resolve();
|
|
}, source.delayMs);
|
|
const onAbort = (): void => {
|
|
clearTimeout(handle);
|
|
reject(new Error("aborted"));
|
|
};
|
|
if (signal?.aborted) {
|
|
onAbort();
|
|
} else {
|
|
signal?.addEventListener("abort", onAbort, { once: true });
|
|
}
|
|
});
|
|
}
|
|
if (source.throwReason !== undefined) {
|
|
throw new Error(source.throwReason);
|
|
}
|
|
if (source.bytes === undefined) {
|
|
throw new Error(`no bytes staged at ${key}`);
|
|
}
|
|
return source.bytes;
|
|
}
|
|
|
|
public async advertise(
|
|
peerId: PeerId,
|
|
advert: ChunkAdvertisement
|
|
): Promise<void> {
|
|
// Deliver to every other transport's subscribers via the broker.
|
|
for (const transport of this.#broker.transports) {
|
|
if (transport === this) continue;
|
|
if (transport.localPeerId !== peerId) continue;
|
|
for (const handler of transport.handlers) {
|
|
handler(this.localPeerId, advert);
|
|
}
|
|
}
|
|
}
|
|
|
|
public onAdvertisement(handler: AdvertisementHandler): Unsubscribe {
|
|
this.handlers.add(handler);
|
|
return () => {
|
|
this.handlers.delete(handler);
|
|
};
|
|
}
|
|
|
|
public peers(): Iterable<PeerId> {
|
|
return this.knownPeers;
|
|
}
|
|
|
|
public dispose(): void {
|
|
this.#broker.transports.delete(this);
|
|
this.handlers.clear();
|
|
}
|
|
}
|