LumeWeb
/
dht-flood
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: LumeWeb:993eff5398ce3018dc52f281d71750676d4cb700
Branches Tags
LumeWeb:master
..
compare: LumeWeb:ebaba22213d1b9f051af9f9a0ec04ee28e7cfa61
Branches Tags
LumeWeb:master

No commits in common. "993eff5398ce3018dc52f281d71750676d4cb700" and "ebaba22213d1b9f051af9f9a0ec04ee28e7cfa61" have entirely different histories.
993eff5398 ... ebaba22213

10 changed files with 159 additions and 117 deletions
Show Stats Expand all files Collapse all files

1
.yarnrc.yml
View File

@ -1 +0,0 @@
nodeLinker: node-modules

3
dist/index.d.ts vendored
View File

@ -1,6 +1,7 @@
/// <reference types="node" />
/// <reference types="node" />
import EventEmitter from "events";
export declare const FLOOD_SYMBOL: unique symbol;
export default class DHTFlood extends EventEmitter {
private id;
private ttl;
@ -8,8 +9,6 @@ export default class DHTFlood extends EventEmitter {
private lru;
private swarm;
private protocol;
private symbol;
private socketMap;
constructor({ lruSize, ttl, messageNumber, id, swarm, protocol, }?: {
lruSize?: number | undefined;
ttl?: number | undefined;

2
dist/index.d.ts.map vendored
View File

@ -1 +1 @@
{"version":3,"file":"index.d.ts","sourceRoot":"","sources":["../src/index.ts"],"names":[],"mappings":";;AAAA,OAAO,YAAY,MAAM,QAAQ,CAAC;AAkBlC,MAAM,CAAC,OAAO,OAAO,QAAS,SAAQ,YAAY;IAChD,OAAO,CAAC,EAAE,CAAS;IACnB,OAAO,CAAC,GAAG,CAAS;IACpB,OAAO,CAAC,aAAa,CAAS;IAC9B,OAAO,CAAC,GAAG,CAAM;IACjB,OAAO,CAAC,KAAK,CAAM;IACnB,OAAO,CAAC,QAAQ,CAAS;IACzB,OAAO,CAAC,MAAM,CAAS;IACvB,OAAO,CAAC,SAAS,CAAsC;gBAE3C,EACV,OAAkB,EAClB,GAAS,EACT,aAAiB,EACjB,EAA2B,EAC3B,KAAY,EACZ,QAAmB,GACpB;;;;;;;KAAK;IAqBN,OAAO,CAAC,aAAa;IAkCrB,OAAO,CAAC,SAAS;IAwDjB,SAAS,CAAC,IAAI,EAAE,GAAG,EAAE,GAAG,SAAW;IAenC,IAAI,CAAC,IAAI,EAAE,GAAG,EAAE,IAAI,EAAE,GAAG,EAAE,GAAG,SAAW;CAY1C"}
{"version":3,"file":"index.d.ts","sourceRoot":"","sources":["../src/index.ts"],"names":[],"mappings":";;AAAA,OAAO,YAAY,MAAM,QAAQ,CAAC;AAkBlC,eAAO,MAAM,YAAY,eAAuB,CAAC;AAIjD,MAAM,CAAC,OAAO,OAAO,QAAS,SAAQ,YAAY;IAChD,OAAO,CAAC,EAAE,CAAS;IACnB,OAAO,CAAC,GAAG,CAAS;IACpB,OAAO,CAAC,aAAa,CAAS;IAC9B,OAAO,CAAC,GAAG,CAAM;IACjB,OAAO,CAAC,KAAK,CAAM;IACnB,OAAO,CAAC,QAAQ,CAAS;gBAEb,EACV,OAAkB,EAClB,GAAS,EACT,aAAiB,EACjB,EAA2B,EAC3B,KAAY,EACZ,QAAmB,GACpB;;;;;;;KAAK;IAmBN,OAAO,CAAC,aAAa;IAkCrB,OAAO,CAAC,SAAS;IAsDjB,SAAS,CAAC,IAAI,EAAE,GAAG,EAAE,GAAG,SAAW;IAenC,IAAI,CAAC,IAAI,EAAE,GAAG,EAAE,IAAI,EAAE,GAAG,EAAE,GAAG,SAAW;CAY1C"}

18
dist/index.js vendored
View File

@ -3,6 +3,7 @@ var __importDefault = (this && this.__importDefault) || function (mod) {
return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.FLOOD_SYMBOL = void 0;
const events_1 = __importDefault(require("events"));
const crypto_1 = __importDefault(require("crypto"));
// @ts-ignore
@ -18,6 +19,8 @@ const debug = (0, debug_1.default)("dht-flood");
const LRU_SIZE = 255;
const TTL = 255;
const PROTOCOL = "lumeweb.flood";
exports.FLOOD_SYMBOL = Symbol.for(PROTOCOL);
const closedMap = new Set();
class DHTFlood extends events_1.default {
id;
ttl;
@ -25,8 +28,6 @@ class DHTFlood extends events_1.default {
lru;
swarm;
protocol;
symbol;
socketMap = new Set();
constructor({ lruSize = LRU_SIZE, ttl = TTL, messageNumber = 0, id = crypto_1.default.randomBytes(32), swarm = null, protocol = PROTOCOL, } = {}) {
super();
this.id = id;
@ -42,7 +43,6 @@ class DHTFlood extends events_1.default {
const mux = protomux_1.default.from(peer);
mux.pair({ protocol: this.protocol }, () => this.setupPeer(peer));
});
this.symbol = Symbol.for(this.protocol);
}
handleMessage({ originId, messageNumber, ttl, data }, messenger) {
const originIdBuf = b4a_1.default.from(originId);
@ -77,21 +77,19 @@ class DHTFlood extends events_1.default {
},
});
if (chan) {
// @ts-ignore
peer[this.symbol] = chan;
peer[exports.FLOOD_SYMBOL] = chan;
}
}
if (!this.socketMap.has(peer)) {
if (!closedMap.has(peer)) {
const close = () => {
self.emit("peer-remove", peer);
peer.removeListener("close", close);
this.socketMap.delete(peer);
closedMap.delete(peer);
};
peer.on("close", close);
this.socketMap.add(peer);
closedMap.add(peer);
}
// @ts-ignore
chan = peer[this.symbol];
chan = peer[exports.FLOOD_SYMBOL];
if (!chan) {
throw new Error("could not find channel");
}

10
dist/messages.d.ts vendored
View File

@ -7,7 +7,7 @@ import { MessageType } from "@protobuf-ts/runtime";
/**
* @generated from protobuf message Packet
*/
export interface Packet {
export interface PacketType {
/**
* @generated from protobuf field: bytes originId = 1;
*/
@ -25,11 +25,11 @@ export interface Packet {
*/
data: Uint8Array;
}
declare class Packet$Type extends MessageType<Packet> {
declare class Packet$Type extends MessageType<PacketType> {
constructor();
create(value?: PartialMessage<Packet>): Packet;
internalBinaryRead(reader: IBinaryReader, length: number, options: BinaryReadOptions, target?: Packet): Packet;
internalBinaryWrite(message: Packet, writer: IBinaryWriter, options: BinaryWriteOptions): IBinaryWriter;
create(value?: PartialMessage<PacketType>): PacketType;
internalBinaryRead(reader: IBinaryReader, length: number, options: BinaryReadOptions, target?: PacketType): PacketType;
internalBinaryWrite(message: PacketType, writer: IBinaryWriter, options: BinaryWriteOptions): IBinaryWriter;
}
/**
* @generated MessageType for protobuf message Packet

2
dist/messages.d.ts.map vendored
View File

@ -1 +1 @@
{"version":3,"file":"messages.d.ts","sourceRoot":"","sources":["../src/messages.ts"],"names":[],"mappings":"AAGA,OAAO,KAAK,EAAE,kBAAkB,EAAE,MAAM,sBAAsB,CAAC;AAC/D,OAAO,KAAK,EAAE,aAAa,EAAE,MAAM,sBAAsB,CAAC;AAE1D,OAAO,KAAK,EAAE,iBAAiB,EAAE,MAAM,sBAAsB,CAAC;AAC9D,OAAO,KAAK,EAAE,aAAa,EAAE,MAAM,sBAAsB,CAAC;AAE1D,OAAO,KAAK,EAAE,cAAc,EAAE,MAAM,sBAAsB,CAAC;AAG3D,OAAO,EAAE,WAAW,EAAE,MAAM,sBAAsB,CAAC;AACnD;;GAEG;AACH,MAAM,WAAW,MAAM;IACnB;;OAEG;IACH,QAAQ,EAAE,UAAU,CAAC;IACrB;;OAEG;IACH,aAAa,EAAE,MAAM,CAAC;IACtB;;OAEG;IACH,GAAG,EAAE,MAAM,CAAC;IACZ;;OAEG;IACH,IAAI,EAAE,UAAU,CAAC;CACpB;AAED,cAAM,WAAY,SAAQ,WAAW,CAAC,MAAM,CAAC;;IASzC,MAAM,CAAC,KAAK,CAAC,EAAE,cAAc,CAAC,MAAM,CAAC,GAAG,MAAM;IAO9C,kBAAkB,CAAC,MAAM,EAAE,aAAa,EAAE,MAAM,EAAE,MAAM,EAAE,OAAO,EAAE,iBAAiB,EAAE,MAAM,CAAC,EAAE,MAAM,GAAG,MAAM;IA4B9G,mBAAmB,CAAC,OAAO,EAAE,MAAM,EAAE,MAAM,EAAE,aAAa,EAAE,OAAO,EAAE,kBAAkB,GAAG,aAAa;CAkB1G;AACD;;GAEG;AACH,eAAO,MAAM,MAAM,aAAoB,CAAC"}
{"version":3,"file":"messages.d.ts","sourceRoot":"","sources":["../src/messages.ts"],"names":[],"mappings":"AAGA,OAAO,KAAK,EAAE,kBAAkB,EAAE,MAAM,sBAAsB,CAAC;AAC/D,OAAO,KAAK,EAAE,aAAa,EAAE,MAAM,sBAAsB,CAAC;AAE1D,OAAO,KAAK,EAAE,iBAAiB,EAAE,MAAM,sBAAsB,CAAC;AAC9D,OAAO,KAAK,EAAE,aAAa,EAAE,MAAM,sBAAsB,CAAC;AAE1D,OAAO,KAAK,EAAE,cAAc,EAAE,MAAM,sBAAsB,CAAC;AAG3D,OAAO,EAAE,WAAW,EAAE,MAAM,sBAAsB,CAAC;AACnD;;GAEG;AACH,MAAM,WAAW,UAAU;IACzB;;OAEG;IACH,QAAQ,EAAE,UAAU,CAAC;IACrB;;OAEG;IACH,aAAa,EAAE,MAAM,CAAC;IACtB;;OAEG;IACH,GAAG,EAAE,MAAM,CAAC;IACZ;;OAEG;IACH,IAAI,EAAE,UAAU,CAAC;CAClB;AAED,cAAM,WAAY,SAAQ,WAAW,CAAC,UAAU,CAAC;;IAc/C,MAAM,CAAC,KAAK,CAAC,EAAE,cAAc,CAAC,UAAU,CAAC,GAAG,UAAU;IAetD,kBAAkB,CAChB,MAAM,EAAE,aAAa,EACrB,MAAM,EAAE,MAAM,EACd,OAAO,EAAE,iBAAiB,EAC1B,MAAM,CAAC,EAAE,UAAU,GAClB,UAAU;IAqCb,mBAAmB,CACjB,OAAO,EAAE,UAAU,EACnB,MAAM,EAAE,aAAa,EACrB,OAAO,EAAE,kBAAkB,GAC1B,aAAa;CAqBjB;AACD;;GAEG;AACH,eAAO,MAAM,MAAM,aAAoB,CAAC"}

21
dist/messages.js vendored
View File

@ -11,14 +11,27 @@ class Packet$Type extends runtime_5.MessageType {
constructor() {
super("Packet", [
{ no: 1, name: "originId", kind: "scalar", T: 12 /*ScalarType.BYTES*/ },
{ no: 2, name: "messageNumber", kind: "scalar", T: 13 /*ScalarType.UINT32*/ },
{
no: 2,
name: "messageNumber",
kind: "scalar",
T: 13 /*ScalarType.UINT32*/,
},
{ no: 3, name: "ttl", kind: "scalar", T: 13 /*ScalarType.UINT32*/ },
{ no: 4, name: "data", kind: "scalar", T: 12 /*ScalarType.BYTES*/ }
{ no: 4, name: "data", kind: "scalar", T: 12 /*ScalarType.BYTES*/ },
]);
}
create(value) {
const message = { originId: new Uint8Array(0), messageNumber: 0, ttl: 0, data: new Uint8Array(0) };
globalThis.Object.defineProperty(message, runtime_4.MESSAGE_TYPE, { enumerable: false, value: this });
const message = {
originId: new Uint8Array(0),
messageNumber: 0,
ttl: 0,
data: new Uint8Array(0),
};
globalThis.Object.defineProperty(message, runtime_4.MESSAGE_TYPE, {
enumerable: false,
value: this,
});
if (value !== undefined)
(0, runtime_3.reflectionMergePartial)(this, message, value);
return message;

3
package.json
View File

@ -10,13 +10,14 @@
"protomux-rpc": "^1.3.0"
},
"devDependencies": {
"@protobuf-ts/plugin": "^2.8.2",
"@protobuf-ts/plugin": "^2.8.1",
"@types/b4a": "^1.6.0",
"@types/debug": "^4.1.7",
"b4a": "^1.6.1",
"debug": "^4.3.4",
"hyperswarm": "^4.3.5",
"prettier": "^2.7.1",
"protoc": "^1.1.3",
"sleep-promise": "^9.1.0",
"sodium-universal": "^3.1.0",
"tape": "^5.6.1"

24
src/index.ts
View File

@ -5,7 +5,7 @@ import LRU from "lru";
import debug0 from "debug";
// @ts-ignore
import Protomux from "protomux";
import { Packet } from "./messages.js";
import { Packet, PacketType } from "./messages.js";
// @ts-ignore
import c from "compact-encoding";
import b4a from "b4a";
@ -16,6 +16,10 @@ const LRU_SIZE = 255;
const TTL = 255;
const PROTOCOL = "lumeweb.flood";
export const FLOOD_SYMBOL = Symbol.for(PROTOCOL);
const closedMap = new Set();
export default class DHTFlood extends EventEmitter {
private id: Buffer;
private ttl: number;
@ -23,8 +27,6 @@ export default class DHTFlood extends EventEmitter {
private lru: LRU;
private swarm: any;
private protocol: string;
private symbol: Symbol;
private socketMap: Set<Function> = new Set<Function>();
constructor({
lruSize = LRU_SIZE,
@ -50,12 +52,10 @@ export default class DHTFlood extends EventEmitter {
const mux = Protomux.from(peer);
mux.pair({ protocol: this.protocol }, () => this.setupPeer(peer));
});
this.symbol = Symbol.for(this.protocol);
}
private handleMessage(
{ originId, messageNumber, ttl, data }: Packet,
{ originId, messageNumber, ttl, data }: PacketType,
messenger: any
) {
const originIdBuf = b4a.from(originId) as Buffer;
@ -101,24 +101,22 @@ export default class DHTFlood extends EventEmitter {
},
});
if (chan) {
// @ts-ignore
peer[this.symbol] = chan;
peer[FLOOD_SYMBOL] = chan;
}
}
if (!this.socketMap.has(peer)) {
if (!closedMap.has(peer)) {
const close = () => {
self.emit("peer-remove", peer);
peer.removeListener("close", close);
this.socketMap.delete(peer);
closedMap.delete(peer);
};
peer.on("close", close);
this.socketMap.add(peer);
closedMap.add(peer);
}
// @ts-ignore
chan = peer[this.symbol];
chan = peer[FLOOD_SYMBOL];
if (!chan) {
throw new Error("could not find channel");

192
src/messages.ts
View File

@ -1,4 +1,4 @@
// @generated by protobuf-ts 2.8.2
// @generated by protobuf-ts 2.8.1
// @generated from protobuf file "messages.proto" (syntax proto2)
// tslint:disable
import type { BinaryWriteOptions } from "@protobuf-ts/runtime";
@ -14,87 +14,121 @@ import { MessageType } from "@protobuf-ts/runtime";
/**
* @generated from protobuf message Packet
*/
export interface Packet {
/**
* @generated from protobuf field: bytes originId = 1;
*/
originId: Uint8Array;
/**
* @generated from protobuf field: uint32 messageNumber = 2;
*/
messageNumber: number;
/**
* @generated from protobuf field: uint32 ttl = 3;
*/
ttl: number;
/**
* @generated from protobuf field: bytes data = 4;
*/
data: Uint8Array;
export interface PacketType {
/**
* @generated from protobuf field: bytes originId = 1;
*/
originId: Uint8Array;
/**
* @generated from protobuf field: uint32 messageNumber = 2;
*/
messageNumber: number;
/**
* @generated from protobuf field: uint32 ttl = 3;
*/
ttl: number;
/**
* @generated from protobuf field: bytes data = 4;
*/
data: Uint8Array;
}
// @generated message type with reflection information, may provide speed optimized methods
class Packet$Type extends MessageType<Packet> {
constructor() {
super("Packet", [
{ no: 1, name: "originId", kind: "scalar", T: 12 /*ScalarType.BYTES*/ },
{ no: 2, name: "messageNumber", kind: "scalar", T: 13 /*ScalarType.UINT32*/ },
{ no: 3, name: "ttl", kind: "scalar", T: 13 /*ScalarType.UINT32*/ },
{ no: 4, name: "data", kind: "scalar", T: 12 /*ScalarType.BYTES*/ }
]);
}
create(value?: PartialMessage<Packet>): Packet {
const message = { originId: new Uint8Array(0), messageNumber: 0, ttl: 0, data: new Uint8Array(0) };
globalThis.Object.defineProperty(message, MESSAGE_TYPE, { enumerable: false, value: this });
if (value !== undefined)
reflectionMergePartial<Packet>(this, message, value);
return message;
}
internalBinaryRead(reader: IBinaryReader, length: number, options: BinaryReadOptions, target?: Packet): Packet {
let message = target ?? this.create(), end = reader.pos + length;
while (reader.pos < end) {
let [fieldNo, wireType] = reader.tag();
switch (fieldNo) {
case /* bytes originId */ 1:
message.originId = reader.bytes();
break;
case /* uint32 messageNumber */ 2:
message.messageNumber = reader.uint32();
break;
case /* uint32 ttl */ 3:
message.ttl = reader.uint32();
break;
case /* bytes data */ 4:
message.data = reader.bytes();
break;
default:
let u = options.readUnknownField;
if (u === "throw")
throw new globalThis.Error(`Unknown field ${fieldNo} (wire type ${wireType}) for ${this.typeName}`);
let d = reader.skip(wireType);
if (u !== false)
(u === true ? UnknownFieldHandler.onRead : u)(this.typeName, message, fieldNo, wireType, d);
}
}
return message;
}
internalBinaryWrite(message: Packet, writer: IBinaryWriter, options: BinaryWriteOptions): IBinaryWriter {
/* bytes originId = 1; */
if (message.originId.length)
writer.tag(1, WireType.LengthDelimited).bytes(message.originId);
/* uint32 messageNumber = 2; */
if (message.messageNumber !== 0)
writer.tag(2, WireType.Varint).uint32(message.messageNumber);
/* uint32 ttl = 3; */
if (message.ttl !== 0)
writer.tag(3, WireType.Varint).uint32(message.ttl);
/* bytes data = 4; */
if (message.data.length)
writer.tag(4, WireType.LengthDelimited).bytes(message.data);
let u = options.writeUnknownFields;
if (u !== false)
(u == true ? UnknownFieldHandler.onWrite : u)(this.typeName, message, writer);
return writer;
class Packet$Type extends MessageType<PacketType> {
constructor() {
super("Packet", [
{ no: 1, name: "originId", kind: "scalar", T: 12 /*ScalarType.BYTES*/ },
{
no: 2,
name: "messageNumber",
kind: "scalar",
T: 13 /*ScalarType.UINT32*/,
},
{ no: 3, name: "ttl", kind: "scalar", T: 13 /*ScalarType.UINT32*/ },
{ no: 4, name: "data", kind: "scalar", T: 12 /*ScalarType.BYTES*/ },
]);
}
create(value?: PartialMessage<PacketType>): PacketType {
const message = {
originId: new Uint8Array(0),
messageNumber: 0,
ttl: 0,
data: new Uint8Array(0),
};
globalThis.Object.defineProperty(message, MESSAGE_TYPE, {
enumerable: false,
value: this,
});
if (value !== undefined)
reflectionMergePartial<PacketType>(this, message, value);
return message;
}
internalBinaryRead(
reader: IBinaryReader,
length: number,
options: BinaryReadOptions,
target?: PacketType
): PacketType {
let message = target ?? this.create(),
end = reader.pos + length;
while (reader.pos < end) {
let [fieldNo, wireType] = reader.tag();
switch (fieldNo) {
case /* bytes originId */ 1:
message.originId = reader.bytes();
break;
case /* uint32 messageNumber */ 2:
message.messageNumber = reader.uint32();
break;
case /* uint32 ttl */ 3:
message.ttl = reader.uint32();
break;
case /* bytes data */ 4:
message.data = reader.bytes();
break;
default:
let u = options.readUnknownField;
if (u === "throw")
throw new globalThis.Error(
`Unknown field ${fieldNo} (wire type ${wireType}) for ${this.typeName}`
);
let d = reader.skip(wireType);
if (u !== false)
(u === true ? UnknownFieldHandler.onRead : u)(
this.typeName,
message,
fieldNo,
wireType,
d
);
}
}
return message;
}
internalBinaryWrite(
message: PacketType,
writer: IBinaryWriter,
options: BinaryWriteOptions
): IBinaryWriter {
/* bytes originId = 1; */
if (message.originId.length)
writer.tag(1, WireType.LengthDelimited).bytes(message.originId);
/* uint32 messageNumber = 2; */
if (message.messageNumber !== 0)
writer.tag(2, WireType.Varint).uint32(message.messageNumber);
/* uint32 ttl = 3; */
if (message.ttl !== 0) writer.tag(3, WireType.Varint).uint32(message.ttl);
/* bytes data = 4; */
if (message.data.length)
writer.tag(4, WireType.LengthDelimited).bytes(message.data);
let u = options.writeUnknownFields;
if (u !== false)
(u == true ? UnknownFieldHandler.onWrite : u)(
this.typeName,
message,
writer
);
return writer;
}
}
/**
* @generated MessageType for protobuf message Packet