rpc-client/dist/query/streaming.js

77 lines
2.4 KiB
JavaScript
Raw Normal View History

2022-08-28 06:33:49 +00:00
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;
2022-08-31 23:59:06 +00:00
_canceled = false;
2022-08-28 06:33:49 +00:00
constructor(network, relay, query, options) {
super(network, relay, query, options);
this._options = options;
}
2022-08-31 23:59:06 +00:00
cancel() {
this._canceled = true;
}
2022-08-28 06:33:49 +00:00
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;
2022-08-31 23:59:06 +00:00
const finish = () => {
relay = relay;
this._responses[relay] = {};
resolve(null);
socket.end();
};
2022-08-28 06:33:49 +00:00
socket.on("data", (res) => {
relay = relay;
if (timer && timer.close) {
clearTimeout(timer);
}
2022-08-31 23:59:06 +00:00
if (this._canceled) {
finish();
return;
}
2022-08-28 06:33:49 +00:00
const response = unpack(res);
if (response && response.error) {
this._errors[relay] = response.error;
return reject(null);
}
if (response?.data.done) {
2022-08-31 23:59:06 +00:00
finish();
2022-08-28 06:33:49 +00:00
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);
});
}
}