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

*Add prettier

This commit is contained in:
Derrick Hammer 2022-11-15 14:28:28 -05:00
parent 8d57c62994
commit aab5254ef2
Signed by: pcfreak30
GPG Key ID: C997C339BE476FF2
3 changed files with 230 additions and 184 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
package.json
View File

@ -16,6 +16,7 @@
"b4a": "^1.6.1",
"debug": "^4.3.4",
"hyperswarm": "^4.3.5",
"prettier": "^2.7.1",
"protoc": "^1.1.3",
"sodium-universal": "^3.1.0",
"tape": "^5.6.1"

77
src/index.ts
View File

@ -7,16 +7,16 @@ import debug0 from "debug";
import Protomux from "protomux";
import { Packet, PacketType } from "./messages.js";
// @ts-ignore
import c from "compact-encoding"
import b4a from "b4a"
import c from "compact-encoding";
import b4a from "b4a";
const debug = debug0('dht-flood')
const debug = debug0("dht-flood");
const LRU_SIZE = 255
const TTL = 255
const PROTOCOL = "lumeweb.flood"
const LRU_SIZE = 255;
const TTL = 255;
const PROTOCOL = "lumeweb.flood";
const FLOOD_SYMBOL = Symbol.for(PROTOCOL)
const FLOOD_SYMBOL = Symbol.for(PROTOCOL);
export default class DHTFlood extends EventEmitter {
private id: Buffer;
@ -30,16 +30,16 @@ export default class DHTFlood extends EventEmitter {
ttl = TTL,
messageNumber = 0,
id = crypto.randomBytes(32),
swarm = null
swarm = null,
} = {}) {
super()
super();
this.id = id
this.ttl = ttl
this.messageNumber = messageNumber
this.lru = new LRU(lruSize)
this.id = id;
this.ttl = ttl;
this.messageNumber = messageNumber;
this.lru = new LRU(lruSize);
if (!swarm) {
throw new Error('swarm is required');
throw new Error("swarm is required");
}
this.swarm = swarm;
@ -49,28 +49,37 @@ export default class DHTFlood extends EventEmitter {
});
}
private handleMessage({originId, messageNumber, ttl, data}: PacketType, messenger: any) {
private handleMessage(
{ originId, messageNumber, ttl, data }: PacketType,
messenger: any
) {
const originIdBuf = b4a.from(originId) as Buffer;
// Ignore messages from ourselves
if (originIdBuf.equals(this.id)) return debug('Got message from self', originId, messageNumber)
if (originIdBuf.equals(this.id))
return debug("Got message from self", originId, messageNumber);
// Ignore messages we've already seen
const key = originIdBuf.toString('hex') + messageNumber
if (this.lru.get(key)) return debug('Got message that was already seen', originId, messageNumber)
this.lru.set(key, true)
const key = originIdBuf.toString("hex") + messageNumber;
if (this.lru.get(key))
return debug(
"Got message that was already seen",
originId,
messageNumber
);
this.lru.set(key, true);
this.emit('message', data, originId, messageNumber)
this.emit("message", data, originId, messageNumber);
if (ttl <= 0) return debug('Got message at end of TTL', originId, messageNumber, ttl)
if (ttl <= 0)
return debug("Got message at end of TTL", originId, messageNumber, ttl);
messenger.send({
originId,
messageNumber,
data,
ttl: ttl - 1
ttl: ttl - 1,
});
}
private setupPeer(peer: any) {
@ -83,10 +92,10 @@ export default class DHTFlood extends EventEmitter {
chan = mux.createChannel({
protocol: PROTOCOL,
async onopen() {
self.emit('peer-open', peer)
self.emit("peer-open", peer);
},
async ondestroy() {
self.emit('peer-remove', peer)
self.emit("peer-remove", peer);
},
});
peer[FLOOD_SYMBOL] = chan;
@ -95,18 +104,20 @@ export default class DHTFlood extends EventEmitter {
chan = peer[FLOOD_SYMBOL];
if (!chan) {
throw new Error('could not find channel');
throw new Error("could not find channel");
}
if (!chan.messages.length) {
chan.addMessage({
encoding: {
preencode: (state: any, m: any) => c.raw.preencode(state, Packet.toBinary(Packet.create(m))),
encode: (state: any, m: any) => c.raw.encode(state, Packet.toBinary(Packet.create(m))),
preencode: (state: any, m: any) =>
c.raw.preencode(state, Packet.toBinary(Packet.create(m))),
encode: (state: any, m: any) =>
c.raw.encode(state, Packet.toBinary(Packet.create(m))),
decode: (state: any) => Packet.fromBinary(c.raw.decode(state)),
},
onmessage: (msg: any) => this.handleMessage(msg, chan.messages[0]),
})
});
}
if (!chan.opened) {
@ -117,8 +128,8 @@ export default class DHTFlood extends EventEmitter {
}
broadcast(data: any, ttl = this.ttl) {
this.messageNumber++
const {id, messageNumber} = this
this.messageNumber++;
const { id, messageNumber } = this;
for (const peer of this.swarm.connections.values()) {
const message = this.setupPeer(peer);
@ -126,8 +137,8 @@ export default class DHTFlood extends EventEmitter {
originId: id,
messageNumber,
ttl,
data
})
data,
});
}
}
}

58
src/messages.ts
View File

@ -37,20 +37,39 @@ 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: 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?: 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 });
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;
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) {
@ -69,15 +88,27 @@ class Packet$Type extends MessageType<PacketType> {
default:
let u = options.readUnknownField;
if (u === "throw")
throw new globalThis.Error(`Unknown field ${fieldNo} (wire type ${wireType}) for ${this.typeName}`);
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);
(u === true ? UnknownFieldHandler.onRead : u)(
this.typeName,
message,
fieldNo,
wireType,
d
);
}
}
return message;
}
internalBinaryWrite(message: PacketType, writer: IBinaryWriter, options: BinaryWriteOptions): IBinaryWriter {
internalBinaryWrite(
message: PacketType,
writer: IBinaryWriter,
options: BinaryWriteOptions
): IBinaryWriter {
/* bytes originId = 1; */
if (message.originId.length)
writer.tag(1, WireType.LengthDelimited).bytes(message.originId);
@ -85,14 +116,17 @@ class Packet$Type extends MessageType<PacketType> {
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);
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);
(u == true ? UnknownFieldHandler.onWrite : u)(
this.typeName,
message,
writer
);
return writer;
}
}