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

Compare commits

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

4 Commits
ebaba22213 ... 993eff5398

Author SHA1 Message Date
Derrick Hammer 993eff5398
*Update dist 2022-12-30 00:03:56 -05:00
Derrick Hammer ef79936384
*Update to track symbol and sockets per instance 2022-12-30 00:02:47 -05:00
Derrick Hammer ecee330493
*Use yarn 2 2022-12-30 00:00:59 -05:00
Derrick Hammer 7b920bf2f5
*Update protobuf 2022-12-30 00:00:50 -05:00
10 changed files with 117 additions and 159 deletions
Show Stats Expand all files Collapse all files

1
.yarnrc.yml Normal file
View File

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

3
dist/index.d.ts vendored
View File

@ -1,7 +1,6 @@
/// <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;
@ -9,6 +8,8 @@ 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,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"}
{"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"}

18
dist/index.js vendored
View File

@ -3,7 +3,6 @@ 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
@ -19,8 +18,6 @@ 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;
@ -28,6 +25,8 @@ 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;
@ -43,6 +42,7 @@ 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,19 +77,21 @@ class DHTFlood extends events_1.default {
},
});
if (chan) {
peer[exports.FLOOD_SYMBOL] = chan;
// @ts-ignore
peer[this.symbol] = chan;
}
}
if (!closedMap.has(peer)) {
if (!this.socketMap.has(peer)) {
const close = () => {
self.emit("peer-remove", peer);
peer.removeListener("close", close);
closedMap.delete(peer);
this.socketMap.delete(peer);
};
peer.on("close", close);
closedMap.add(peer);
this.socketMap.add(peer);
}
chan = peer[exports.FLOOD_SYMBOL];
// @ts-ignore
chan = peer[this.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 PacketType {
export interface Packet {
/**
* @generated from protobuf field: bytes originId = 1;
*/
@ -25,11 +25,11 @@ export interface PacketType {
*/
data: Uint8Array;
}
declare class Packet$Type extends MessageType<PacketType> {
declare class Packet$Type extends MessageType<Packet> {
constructor();
create(value?: PartialMessage<PacketType>): PacketType;
internalBinaryRead(reader: IBinaryReader, length: number, options: BinaryReadOptions, target?: PacketType): PacketType;
internalBinaryWrite(message: PacketType, writer: IBinaryWriter, options: BinaryWriteOptions): IBinaryWriter;
create(value?: PartialMessage<Packet>): Packet;
internalBinaryRead(reader: IBinaryReader, length: number, options: BinaryReadOptions, target?: Packet): Packet;
internalBinaryWrite(message: Packet, 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,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"}
{"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"}

21
dist/messages.js vendored
View File

@ -11,27 +11,14 @@ 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,14 +10,13 @@
"protomux-rpc": "^1.3.0"
},
"devDependencies": {
"@protobuf-ts/plugin": "^2.8.1",
"@protobuf-ts/plugin": "^2.8.2",
"@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, PacketType } from "./messages.js";
import { Packet } from "./messages.js";
// @ts-ignore
import c from "compact-encoding";
import b4a from "b4a";
@ -16,10 +16,6 @@ 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;
@ -27,6 +23,8 @@ 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,
@ -52,10 +50,12 @@ 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 }: PacketType,
{ originId, messageNumber, ttl, data }: Packet,
messenger: any
) {
const originIdBuf = b4a.from(originId) as Buffer;
@ -101,22 +101,24 @@ export default class DHTFlood extends EventEmitter {
},
});
if (chan) {
peer[FLOOD_SYMBOL] = chan;
// @ts-ignore
peer[this.symbol] = chan;
}
}
if (!closedMap.has(peer)) {
if (!this.socketMap.has(peer)) {
const close = () => {
self.emit("peer-remove", peer);
peer.removeListener("close", close);
closedMap.delete(peer);
this.socketMap.delete(peer);
};
peer.on("close", close);
closedMap.add(peer);
this.socketMap.add(peer);
}
chan = peer[FLOOD_SYMBOL];
// @ts-ignore
chan = peer[this.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.1
// @generated by protobuf-ts 2.8.2
// @generated from protobuf file "messages.proto" (syntax proto2)
// tslint:disable
import type { BinaryWriteOptions } from "@protobuf-ts/runtime";
@ -14,121 +14,87 @@ import { MessageType } from "@protobuf-ts/runtime";
/**
* @generated from protobuf message Packet
*/
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;
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;
}
// @generated message type with reflection information, may provide speed optimized methods
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
);
}
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;
}
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