kernel-rpc-client/src/index.ts

281 lines
6.3 KiB
TypeScript
Raw Normal View History

2022-07-20 07:35:58 +00:00
import { ErrTuple } from "libskynet";
import type { RPCRequest, RPCResponse } from "@lumeweb/relay-types";
import {
RpcQueryOptions,
StreamHandlerFunction,
StreamingRpcQueryOptions,
} from "@lumeweb/dht-rpc-client";
import { Buffer } from "buffer";
2022-07-20 07:35:58 +00:00
const RPC_MODULE = "AQDaEPIo_lpdvz7AKbeafERBHR331RiyvweJ6OrFTplzyg";
let callModule: any, connectModule: any;
2022-07-20 07:35:58 +00:00
async function loadLibs() {
if (callModule && connectModule) {
return;
}
if (typeof window !== "undefined" && window?.document) {
const pkg = await import("libkernel");
callModule = pkg.callModule;
connectModule = pkg.connectModule;
} else {
const pkg = await import("libkmodule");
callModule = pkg.callModule;
connectModule = pkg.connectModule;
}
2022-07-20 07:35:58 +00:00
}
export class RpcNetwork {
private _actionQueue: [string, any][] = [];
private _addQueue: string[] = [];
private _removeQueue: string[] = [];
2022-08-31 18:42:26 +00:00
private _def: boolean;
constructor(def: boolean = true) {
this._def = def;
}
private _networkId: number = 0;
get networkId(): number {
return this._networkId;
}
2022-07-20 07:35:58 +00:00
get ready(): Promise<ErrTuple> {
2022-08-31 18:42:26 +00:00
let promise = loadLibs();
if (this._def) {
2022-08-31 19:25:18 +00:00
this._networkId = 1;
2022-08-31 18:42:26 +00:00
} else {
promise = promise
.then(() => callModule(RPC_MODULE, "createNetwork"))
.then((ret: ErrTuple) => (this._networkId = ret[0]));
}
return promise.then(() =>
callModule(RPC_MODULE, "ready", { network: this._networkId })
);
}
private static deleteItem(array: Array<any>, item: string): void {
if (array.includes(item)) {
let queue = new Set(array);
queue.delete(item);
[].splice.apply(array, [0, array.length].concat([...queue]) as any);
}
2022-07-20 07:35:58 +00:00
}
public addRelay(pubkey: string): void {
this._addQueue.push(pubkey);
this._addQueue = [...new Set(this._addQueue)];
RpcNetwork.deleteItem(this._removeQueue, pubkey);
2022-07-20 07:35:58 +00:00
}
public removeRelay(pubkey: string): void {
this._removeQueue.push(pubkey);
this._removeQueue = [...new Set(this._removeQueue)];
RpcNetwork.deleteItem(this._addQueue, pubkey);
2022-07-20 07:35:58 +00:00
}
public clearRelays(): void {
this._actionQueue.push(["clearRelays", {}]);
}
public wisdomQuery(
method: string,
module: string,
2022-07-20 07:35:58 +00:00
data: object | any[] = {},
bypassCache: boolean = false,
options: RpcQueryOptions = {}
): WisdomRpcQuery {
return new WisdomRpcQuery(
this,
{
method,
module,
data,
bypassCache,
},
options
).run();
}
public streamingQuery(
relay: Buffer | string,
method: string,
module: string,
streamHandler: StreamHandlerFunction,
data: object | any[] = {},
options: RpcQueryOptions = {}
): StreamingRpcQuery {
return new StreamingRpcQuery(
this,
relay,
{ method, module, data },
{ streamHandler, ...options }
).run();
}
public simpleQuery(
relay: Buffer | string,
method: string,
module: string,
data: object | any[] = {},
options: RpcQueryOptions = {}
): SimpleRpcQuery {
return new SimpleRpcQuery(
this,
relay,
{
method,
module,
data,
},
options
).run();
2022-07-20 07:35:58 +00:00
}
public async processQueue(): Promise<void> {
await loadLibs();
for (const action of this._actionQueue) {
try {
2022-08-31 18:42:26 +00:00
await callModule(RPC_MODULE, action[0], {
...action[1],
network: this._networkId,
});
} catch (e: any) {}
}
await Promise.allSettled(
this._removeQueue.map((item: string) =>
2022-08-31 18:42:26 +00:00
callModule(RPC_MODULE, "removeRelay", {
pubkey: item,
network: this._networkId,
})
)
);
await Promise.allSettled(
this._addQueue.map((item: string) =>
2022-08-31 18:42:26 +00:00
callModule(RPC_MODULE, "addRelay", {
pubkey: item,
network: this._networkId,
})
)
);
2022-07-20 07:35:58 +00:00
this._actionQueue = [];
this._removeQueue = [];
this._addQueue = [];
2022-07-20 07:35:58 +00:00
}
}
export abstract class RpcQueryBase {
protected _promise?: Promise<any>;
protected _network: RpcNetwork;
protected _query: RPCRequest;
protected _options: RpcQueryOptions;
protected _queryType: string;
constructor(
network: RpcNetwork,
query: RPCRequest,
options: RpcQueryOptions = {},
queryType: string
) {
this._network = network;
this._query = query;
this._options = options;
this._queryType = queryType;
}
2022-07-20 07:35:58 +00:00
2022-08-31 18:42:26 +00:00
get result(): Promise<RPCResponse> {
return (this._promise as Promise<any>).then((result): RPCResponse => {
if (result[1]) {
return { error: result[1] };
}
return result[0];
});
}
public run(): this {
2022-08-31 01:49:40 +00:00
this._promise = this._network.processQueue().then(() =>
callModule(RPC_MODULE, this._queryType, {
query: this._query,
options: this._options,
2022-08-31 18:42:26 +00:00
network: this._network.networkId,
2022-08-31 01:49:40 +00:00
})
);
return this;
2022-07-20 07:35:58 +00:00
}
}
export class SimpleRpcQuery extends RpcQueryBase {
2022-08-31 21:32:52 +00:00
private _relay: string | Buffer;
constructor(
network: RpcNetwork,
relay: string | Buffer,
query: RPCRequest,
options: RpcQueryOptions
) {
super(network, query, options, "simpleQuery");
2022-08-31 21:32:52 +00:00
this._relay = relay;
}
public run(): this {
this._promise = this._network.processQueue().then(() =>
callModule(RPC_MODULE, this._queryType, {
relay: this._relay,
query: this._query,
options: this._options,
network: this._network.networkId,
})
);
return this;
}
}
2022-08-31 21:32:52 +00:00
export class StreamingRpcQuery extends SimpleRpcQuery {
protected _options: StreamingRpcQueryOptions;
constructor(
network: RpcNetwork,
relay: string | Buffer,
query: RPCRequest,
options: StreamingRpcQueryOptions
) {
2022-08-31 21:32:52 +00:00
super(network, relay, query, options);
this._options = options;
2022-08-31 21:32:52 +00:00
this._queryType = "streamingQuery";
}
public run(): this {
2022-08-31 01:49:40 +00:00
this._promise = this._network.processQueue().then(() =>
connectModule(
RPC_MODULE,
this._queryType,
{
query: this._query,
options: this._options,
2022-08-31 18:42:26 +00:00
network: this._network.networkId,
2022-08-31 01:49:40 +00:00
},
this._options.streamHandler
)
);
return this;
}
}
export class WisdomRpcQuery extends RpcQueryBase {
constructor(
network: RpcNetwork,
query: RPCRequest,
options: RpcQueryOptions = {}
) {
super(network, query, options, "wisdomQuery");
}
}