dht-data/dist/dhtOnlineBase.js

227 lines
7.6 KiB
JavaScript

"use strict";
var __importDefault = (this && this.__importDefault) || function (mod) {
return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
const events_1 = __importDefault(require("events"));
// @ts-ignore
const jsnetworkx_1 = require("jsnetworkx");
const codecs_1 = __importDefault(require("codecs"));
const messages_js_1 = require("./messages.js");
const DEFAULT_ENCODING = "json";
class DHTOnlineBase extends events_1.default {
id;
bootstrapped;
graph;
connectedTo;
encoding;
constructor(id, { encoding = DEFAULT_ENCODING } = {}) {
super();
if (!id)
throw new TypeError("Must provide id for self");
this.id = id;
this.bootstrapped = false;
this.graph = new jsnetworkx_1.DiGraph();
this.connectedTo = new Set();
this._data = {};
this.encoding = (0, codecs_1.default)(encoding || DEFAULT_ENCODING);
this._online = [this._maybeHexify(this.id)];
}
_data;
get data() {
return { ...this._data };
}
set data(value) {
this._data = value;
this._setPeer(this.id, value);
this._broadcastData();
}
_online;
get online() {
return this._online;
}
broadcast(data, ttl) {
throw new TypeError("Broadcast has not been implemented");
}
getPeerData(id) {
return this.graph.node.get(this._maybeHexify(id));
}
onAddPeer(id) {
const stringId = id.toString("hex");
if (this.connectedTo.has(stringId)) {
return;
} // Already know we're connected here
this.connectedTo.add(stringId);
this._addPeerConnection(this.id, id);
this.emit("peer-add", id);
this._recalculate();
this.broadcast(messages_js_1.Message.toBinary(messages_js_1.Message.create({
type: messages_js_1.Type.CONNECTED,
id,
})));
this._broadcastData();
if (this.bootstrapped) {
return;
}
// If this is the first person we've met, get their graph
this.broadcast(messages_js_1.Message.toBinary(messages_js_1.Message.create({
type: messages_js_1.Type.BOOTSTRAP_REQUEST,
})), 0);
}
onRemovePeer(id) {
this.connectedTo.delete(id.toString("hex"));
this._removePeerConnection(this.id, id);
this.emit("peer-remove");
this._recalculate();
this.broadcast(messages_js_1.Message.toBinary(messages_js_1.Message.create({
type: messages_js_1.Type.DISCONNECTED,
id,
})));
}
onGetBroadcast(message, id) {
let decoded;
try {
decoded = messages_js_1.Message.fromBinary(message);
}
catch {
return;
}
const { type } = decoded;
if (!type) {
throw new Error("Missing Type In Message");
}
if (type === messages_js_1.Type.STATE) {
const { data: rawData } = decoded;
const data = this.encoding.decode(rawData);
this._setPeer(id, data);
this.emit("peer-data", data, id);
this._recalculate();
}
else if (type === messages_js_1.Type.CONNECTED) {
const { id: toId } = decoded;
this._addPeerConnection(id, Buffer.from(toId));
this.emit("peer-add-seen", id, toId);
this._recalculate();
}
else if (type === messages_js_1.Type.DISCONNECTED) {
const { id: toId } = decoded;
this._removePeerConnection(id, Buffer.from(toId));
this.emit("peer-remove-seen", id, toId);
this._recalculate();
}
else if (type === messages_js_1.Type.BOOTSTRAP_REQUEST) {
const bootstrap = this._getBootstrapInfo();
this.broadcast(messages_js_1.Message.toBinary(messages_js_1.Message.create({
type: messages_js_1.Type.BOOTSTRAP_RESPONSE,
bootstrap,
})), 0);
}
else if (type === messages_js_1.Type.BOOTSTRAP_RESPONSE) {
const { bootstrap } = decoded;
this._bootstrapFrom(bootstrap);
}
}
_broadcastData() {
const rawData = this._data;
if (!Object.keys(rawData).length) {
return;
}
const data = this.encoding.encode(rawData);
this.broadcast(messages_js_1.Message.toBinary(messages_js_1.Message.create({
type: messages_js_1.Type.STATE,
data,
})));
}
_hasSeenPeer(id) {
return this.graph.hasNode(this._maybeHexify(id));
}
_setPeer(id, data) {
this.graph.addNode(this._maybeHexify(id), data);
}
_ensurePeer(id) {
id = this._maybeHexify(id);
if (!this._hasSeenPeer(id)) {
this._setPeer(id, {});
}
}
_addPeerConnection(origin, destination) {
this._ensurePeer(origin);
this._ensurePeer(destination);
this.graph.addEdge(this._maybeHexify(origin), this._maybeHexify(destination));
}
_removePeerConnection(origin, destination) {
try {
this._ensurePeer(origin);
this._ensurePeer(destination);
this.graph.removeEdge(origin.toString("hex"), destination.toString("hex"));
}
catch (e) {
if (e.name !== "JSNetworkXError")
throw e;
}
}
_bootstrapFrom(bootstrap) {
if (this.bootstrapped) {
return;
}
for (const id in bootstrap) {
const { data, connectedTo } = bootstrap[id];
const parsedData = data ? this.encoding.decode(data) : null;
let peerData = parsedData || {};
if (id === this.id.toString("hex"))
continue;
// If we're already tracking them
if (this._hasSeenPeer(id)) {
// See what data we already have for them
// Add their existing data to what we got from the bootstrap
const existingPeerData = this.getPeerData(id);
peerData = { ...existingPeerData, ...peerData };
}
this._setPeer(id, peerData);
for (const connection of connectedTo) {
this._addPeerConnection(id, Buffer.from(connection));
}
}
this.emit("bootstrapped");
this._recalculate();
}
_getBootstrapInfo() {
const state = {};
for (const [id, rawData] of this.graph.nodes(true)) {
const connectedTo = this.graph
.neighbors(id)
.map((id) => Buffer.from(id, "hex"));
const data = rawData ? this.encoding.encode(rawData) : null;
state[id] = { data, connectedTo };
}
return state;
}
// Calculate who's online and emit an event
_recalculate() {
const online = this.graph.nodes().filter((id) => {
return (0, jsnetworkx_1.hasPath)(this.graph, {
source: this._maybeHexify(this.id),
target: id,
});
});
const offline = this.graph.nodes().filter((id) => {
return !(0, jsnetworkx_1.hasPath)(this.graph, {
source: this._maybeHexify(this.id),
target: id,
});
});
for (const id of offline) {
this.graph.removeNode(id);
}
this._online = online;
this.emit("online", online);
}
_maybeHexify(data) {
if (Buffer.isBuffer(data)) {
return data.toString("hex");
}
return data;
}
}
exports.default = DHTOnlineBase;