rpc-client/dist/query/streaming.js

76 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 } from "timers";
2022-08-28 06:33:49 +00:00
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) => {
2022-08-31 23:59:06 +00:00
const finish = () => {
relay = relay;
this._responses[relay] = {};
resolve(null);
socket.end();
};
2022-09-01 00:56:56 +00:00
const listener = (res) => {
2022-08-28 06:33:49 +00:00
relay = relay;
if (this._timeoutTimer) {
clearTimeout(this._timeoutTimer);
this._timeoutTimer = null;
}
2022-08-31 23:59:06 +00:00
if (this._canceled) {
2022-09-01 00:45:09 +00:00
socket.write(pack({ cancel: true }));
2022-09-01 00:56:56 +00:00
socket.off("data", listener);
2022-08-31 23:59:06 +00:00
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);
2022-09-01 00:56:56 +00:00
};
socket.on("data", listener);
2022-08-28 06:33:49 +00:00
socket.on("error", (error) => {
relay = relay;
this._errors[relay] = error;
reject({ error });
});
socket.write("rpc");
socket.write(pack(this._query));
});
}
}