rpc-client/dist/query/streaming.js

65 lines
2.1 KiB
JavaScript

import SimpleRpcQuery from "./simple.js";
import { Buffer } from "buffer";
import { isPromise } from "../util.js";
import { clearTimeout, setTimeout } from "timers";
import { pack, unpack } from "msgpackr";
export default class StreamingRpcQuery extends SimpleRpcQuery {
_options;
constructor(network, relay, query, options) {
super(network, relay, query, options);
this._options = options;
}
async queryRelay(relay) {
let socket;
let relayKey = relay;
if (relay === "string") {
relayKey = Buffer.from(relay, "hex");
}
if (relay instanceof Buffer) {
relayKey = relay;
relay = relay.toString("hex");
}
try {
socket = this._network.dht.connect(relayKey);
if (isPromise(socket)) {
socket = await socket;
}
}
catch (e) {
return;
}
return new Promise((resolve, reject) => {
let timer;
socket.on("data", (res) => {
relay = relay;
if (timer && timer.close) {
clearTimeout(timer);
}
socket.end();
const response = unpack(res);
if (response && response.error) {
this._errors[relay] = response.error;
return reject(null);
}
if (response?.data.done) {
this._responses[relay] = {};
resolve(null);
return;
}
this._options.streamHandler(response?.data.data);
});
socket.on("error", (error) => {
relay = relay;
this._errors[relay] = error;
reject({ error });
});
socket.write("rpc");
socket.write(pack(this._query));
timer = setTimeout(() => {
this._errors[relay] = "timeout";
reject(null);
}, (this._options.relayTimeout || this._network.relayTimeout) * 1000);
});
}
}