kernel-swarm-client/src/index.ts

335 lines
8.1 KiB
TypeScript
Raw Normal View History

2022-08-03 16:05:10 +00:00
import { Buffer } from "buffer";
2023-02-01 12:54:20 +00:00
import { Client, factory } from "@lumeweb/libkernel-universal";
import { DataFn, ErrTuple, hexToBuf } from "@siaweb/libweb";
import { blake2b } from "@noble/hashes/blake2b";
import b4a from "b4a";
2022-07-20 01:38:29 +00:00
import type { EventEmitter } from "eventemitter3";
2023-02-17 14:25:58 +00:00
// @ts-ignore
import Backoff from "backoff.js";
import { Mutex } from "async-mutex";
// @ts-ignore
import Protomux from "protomux";
import defer from "p-defer";
2023-02-01 12:54:20 +00:00
export class SwarmClient extends Client {
private useDefaultSwarm: boolean;
private id: number = 0;
private _autoReconnect: boolean;
2023-02-17 14:25:58 +00:00
private _connectBackoff: any;
private _ready?: Promise<void>;
private _connectionListener?: [
sendUpdate: DataFn,
response: Promise<ErrTuple>
];
2022-07-20 01:38:29 +00:00
2023-02-18 00:35:56 +00:00
private _topics: Set<Uint8Array> = new Set<Uint8Array>();
private _sockets: Map<number, Socket> = new Map<number, Socket>();
2023-02-18 00:35:56 +00:00
get dht() {
const self = this;
return {
async ready() {
return self.ready();
},
};
}
constructor(useDefaultDht = true, autoReconnect = false) {
2023-02-01 12:54:20 +00:00
super();
this.useDefaultSwarm = useDefaultDht;
this._autoReconnect = autoReconnect;
2023-02-17 14:25:58 +00:00
this._connectBackoff = new Backoff({
strategy: "fibo",
maxAttempts: Number.MAX_SAFE_INTEGER,
});
this._connectBackoff.on("retry", (error: any) => {
this.logErr(error);
});
2022-08-03 16:05:10 +00:00
}
get swarm(): number | undefined {
return this.useDefaultSwarm ? undefined : this.id;
}
2023-02-01 12:54:20 +00:00
public async connect(pubkey: string | Uint8Array): Promise<Socket> {
if (typeof pubkey === "string") {
const buf = hexToBuf(pubkey);
pubkey = this.handleErrorOrReturn(buf);
}
2023-02-01 12:54:20 +00:00
let existing = Array.from(this._sockets.values()).filter((socket) => {
return b4a.equals(socket.remotePublicKey, pubkey as Uint8Array);
});
if (existing.length) {
return existing[0] as Socket;
}
2023-02-01 12:54:20 +00:00
throw new Error("not implemented");
2022-08-03 16:05:10 +00:00
}
2023-02-01 13:47:19 +00:00
async init(): Promise<ErrTuple> {
return await this.callModuleReturn("init", { swarm: this.swarm });
2023-02-01 13:47:19 +00:00
}
async ready(): Promise<void> {
if (this._ready) {
return this._ready;
}
this._listen();
this._ready = this.callModuleReturn("ready", { swarm: this.swarm });
await this._ready;
this._ready = undefined;
2023-02-18 00:35:56 +00:00
for (const topic of this._topics) {
await this.join(topic);
}
}
async start(): Promise<void> {
2023-02-17 21:50:05 +00:00
await this._connectBackoff.run(() => this.init());
2023-02-17 14:25:58 +00:00
2023-02-17 13:08:43 +00:00
await this.ready();
2022-08-03 16:05:10 +00:00
}
private async _listen() {
if (!this._connectionListener) {
this._connectionListener = this.connectModule(
"listenConnections",
{ swarm: this.swarm },
async (socketId: any) => {
const socket =
2023-04-06 14:08:23 +00:00
this._sockets.get(socketId) ?? (await createSocket(socketId, this));
socket.on("close", () => {
this._sockets.delete(socketId);
});
if (!this._sockets.has(socketId)) {
this._sockets.set(socketId, socket);
}
this.emit("connection", socket);
2023-03-19 15:14:31 +00:00
}
);
}
2023-03-19 15:14:31 +00:00
await this._connectionListener[1];
this._connectionListener = undefined;
2023-02-17 14:25:58 +00:00
this.start();
}
2022-08-03 16:05:10 +00:00
public async addRelay(pubkey: string): Promise<void> {
2023-02-01 12:54:20 +00:00
return this.callModuleReturn("addRelay", { pubkey, swarm: this.swarm });
2022-08-03 16:05:10 +00:00
}
2022-08-03 16:05:10 +00:00
public async removeRelay(pubkey: string): Promise<void> {
2023-02-01 12:54:20 +00:00
return this.callModuleReturn("removeRelay", { pubkey, swarm: this.swarm });
2022-08-03 16:05:10 +00:00
}
2022-08-03 16:05:10 +00:00
public async clearRelays(): Promise<void> {
2023-02-01 12:54:20 +00:00
return this.callModuleReturn("clearRelays", { swarm: this.swarm });
2022-08-03 16:05:10 +00:00
}
public async getRelays(): Promise<string[]> {
2023-02-01 12:54:20 +00:00
return this.callModuleReturn("getRelays", { swarm: this.swarm });
}
2023-02-01 19:06:13 +00:00
public async join(topic: Buffer | Uint8Array | string): Promise<void> {
if (typeof topic === "string") {
2023-02-18 02:03:42 +00:00
topic = blake2b(topic, { dkLen: 32 });
}
2023-02-18 00:35:56 +00:00
this._topics.add(topic);
2023-02-01 19:06:13 +00:00
this.callModule("join", { id: this.id, topic });
}
2022-08-03 16:05:10 +00:00
}
interface SocketRawStream {
remoteHost: string;
remotePort: number;
remoteFamily: string;
}
2023-02-01 12:54:20 +00:00
export class Socket extends Client {
2022-08-03 16:05:10 +00:00
private id: number;
private eventUpdates: { [event: string]: DataFn[] } = {};
private syncMutex = new Mutex();
private swarm: SwarmClient;
private userData?: any = null;
constructor(id: number, swarm: SwarmClient) {
2022-08-03 16:05:10 +00:00
super();
this.id = id;
this.swarm = swarm;
2022-08-03 16:05:10 +00:00
}
private _remotePublicKey?: Uint8Array;
get remotePublicKey(): Uint8Array {
return this._remotePublicKey as Uint8Array;
}
private _rawStream?: Uint8Array;
get rawStream(): Uint8Array {
return this._rawStream as Uint8Array;
}
async setup() {
let info = await this.callModuleReturn("socketGetInfo", { id: this.id });
this._remotePublicKey = info.remotePublicKey;
this._rawStream = info.rawStream;
this._initSync();
}
private async _initSync() {
this.userData = null;
2023-04-06 18:38:07 +00:00
const mux = Protomux.from(this, { slave: true });
let updateSent = defer();
let updateReceived = defer();
const setup = defer();
const [update] = this.connectModule(
"syncProtomux",
{ id: this.id },
async (data: any) => {
if (data === true) {
updateSent.resolve();
updateSent = defer();
return;
}
await this.syncMutex.acquire();
["remote", "local"].forEach((field) => {
const rField = `_${field}`;
data[field].forEach((item: any) => {
if (!mux[rField][item]) {
while (item > mux[rField].length) {
mux[rField].push(null);
}
}
if (!mux[rField][item]) {
mux[rField][item] = null;
}
});
});
data.free.forEach((index: number) => {
if (mux._free[index] === null) {
mux._free[index] = undefined;
}
});
mux._free = mux._free.filter((item: any) => item !== undefined);
this.syncMutex.release();
setup.resolve();
updateReceived.resolve();
}
);
const send = async (mux: any) => {
update({
remote: Object.keys(mux._remote),
local: Object.keys(mux._local),
free: mux._free,
});
updateReceived = defer();
await updateSent.promise;
await updateReceived.promise;
};
mux.syncState = send.bind(undefined, mux);
await setup.promise;
this.swarm.emit("setup", this);
}
on<T extends EventEmitter.EventNames<string | symbol>>(
event: T,
fn: EventEmitter.EventListener<string | symbol, T>,
context?: any
): this {
2023-02-01 12:54:20 +00:00
const [update, promise] = this.connectModule(
"socketListenEvent",
{ id: this.id, event: event },
2022-08-03 16:05:10 +00:00
(data: any) => {
this.emit(event, data);
2022-08-03 16:05:10 +00:00
}
);
this.trackEvent(event as string, update);
2022-08-03 16:05:10 +00:00
promise.then(() => {
this.off(event as string, fn);
2022-08-03 16:05:10 +00:00
});
return super.on(event, fn, context) as this;
2022-08-03 16:05:10 +00:00
}
off<T extends EventEmitter.EventNames<string | symbol>>(
event: T,
fn?: EventEmitter.EventListener<string | symbol, T>,
context?: any,
once?: boolean
): this {
const updates = [...this.eventUpdates[event as string]];
this.eventUpdates[event as string] = [];
2022-08-03 16:05:10 +00:00
for (const func of updates) {
func();
2022-08-03 16:05:10 +00:00
}
return super.off(event, fn, context, once);
2022-08-03 16:05:10 +00:00
}
write(message: string | Buffer): void {
2023-02-06 08:33:44 +00:00
this.callModule("socketWrite", { id: this.id, message });
2022-08-03 16:05:10 +00:00
}
end(): void {
2023-02-01 12:54:20 +00:00
this.callModule("socketExists", { id: this.id }).then(
([exists]: ErrTuple) => {
if (exists) {
this.callModule("socketClose", { id: this.id });
}
}
);
2022-08-03 16:05:10 +00:00
}
private ensureEvent(event: string): void {
if (!(event in this.eventUpdates)) {
this.eventUpdates[event] = [];
}
}
private trackEvent(event: string, update: DataFn): void {
this.ensureEvent(event as string);
this.eventUpdates[event].push(update);
}
2022-07-20 01:38:29 +00:00
}
2023-02-01 12:54:20 +00:00
2023-04-05 08:07:46 +00:00
const MODULE = "_AVKgzVYC8Sb_qiTA6kw5BDzQ4Ch-8D4sldQJl8dXF9oTw";
2023-02-01 12:54:20 +00:00
export const createClient = factory<SwarmClient>(SwarmClient, MODULE);
const socketFactory = factory<Socket>(Socket, MODULE);
const createSocket = async (...args: any): Promise<Socket> => {
const socket = socketFactory(...args);
await socket.setup();
return socket;
};