diff --git a/index.js b/index.js index aa0e3ab..763dd1a 100644 --- a/index.js +++ b/index.js @@ -1,7 +1,7 @@ import * as dotenv from 'dotenv' dotenv.config() -import { Server } from './src/Server.js' +import { Server } from './src/Server.js' const { PPP2_PORT, diff --git a/src/Peer.js b/src/Peer.js index eb707d4..f1f5500 100644 --- a/src/Peer.js +++ b/src/Peer.js @@ -2,7 +2,12 @@ import { Container, AuthenticationRequest, AuthenticationResponse, - Message + Message, + PublishRequest, + PublishResponse, + DiscoverRequest, + DiscoverResponse, + DiscoverPeer } from './Protocol.js' import { EventEmitter } from 'node:events' @@ -21,8 +26,10 @@ export class Peer extends EventEmitter { this.#salt = salt this.#msgMap = new Map() - this.#msgMap.set('AuthenticationRequest', this.#onauth) + this.#msgMap.set('AuthenticationRequest', this.#onauthrequest) this.#msgMap.set('Message', this.#onmessage) + this.#msgMap.set('PublishRequest', this.#onpublishrequest) + this.#msgMap.set('DiscoverRequest', this.#ondiscoverrequest) this.#setupListeners() } @@ -51,14 +58,15 @@ export class Peer extends EventEmitter { } }) - this.on('incoming', ({ source, message }) => { - this.#send({ - id: this.#id++, - message: Message.create({ - address: source.address.toString(10), - port: message.Message.port, - data: message.Message.data - }) + } + + onincoming(source, message) { + this.#send({ + id: this.#id++, + message: Message.create({ + address: source.address.toString(10), + port: message.Message.port, + data: message.Message.data }) }) } @@ -74,7 +82,7 @@ export class Peer extends EventEmitter { this.#socket.send(buf) } - #onauth(message) { + #onauthrequest(message) { const { seed } = AuthenticationRequest.toObject(message.AuthenticationRequest, { longs: String }) if(this.#address !== null) { this.#send({ @@ -105,9 +113,73 @@ export class Peer extends EventEmitter { #onmessage(message) { const { address } = Message.toObject(message.Message, { longs: String }) - this.emit('outgoing', { - destination: BigInt(address), - message + const bigint = BigInt(address) + + const destPeer = this.server.peers.find(p => p.address === bigint) + if(destPeer) { + destPeer.onincoming(this, message) + } + } + + #publish(port, flags) { + const service = this.server.discover.find(sv => sv.peer === this && sv.port === port) + if(service) { + return false + } + + this.server.discover.push({ + peer: this, + port, + flags + }) + return true + } + + #unpublish(port) { + this.server.discover = this.server.discover.filter(sv => !(sv.peer === this && sv.port === port)) + return true + } + + #onpublishrequest(message) { + const { port, publish, flags } = PublishRequest.toObject(message.PublishRequest) + + let ret + if(publish) { + ret = this.#publish(port, flags) + } else { + ret = this.#unpublish(port) + } + + this.#send({ + id: message.id, + message: PublishResponse.create({ + status: ret ? 0: 1 + }) + }) + } + + #ondiscoverrequest(message) { + const { limit, flags } = DiscoverRequest.toObject(message.DiscoverRequest) + let services = this.server.discover.filter(sv => sv.flags.find(flag => flags.includes(flag))) + + if(limit !== 0) { + services = services.slice(0, limit) + } + + const discoverPeers = services.map(sv => { + return DiscoverPeer.create({ + port: sv.port, + address: sv.peer.address.toString(10), + flags: sv.flags + }) + }) + + this.#send({ + id: message.id, + message: DiscoverResponse.create({ + status: 0, + peers: discoverPeers + }) }) } diff --git a/src/Protocol.js b/src/Protocol.js index 76d53d6..fa7ad99 100644 --- a/src/Protocol.js +++ b/src/Protocol.js @@ -23,8 +23,9 @@ export const [ PublishRequest, PublishResponse, DiscoverRequest, - DiscoverResponse -] = await import_proto('2.0/PublishDiscover.proto', ['PublishRequest', 'PublishResponse', 'DiscoverRequest', 'DiscoverResponse']) + DiscoverResponse, + DiscoverPeer +] = await import_proto('2.0/PublishDiscover.proto', ['PublishRequest', 'PublishResponse', 'DiscoverRequest', 'DiscoverResponse', 'DiscoverPeer']) export const messages = [ Container, diff --git a/src/Server.js b/src/Server.js index 4769d9b..29e4ac0 100644 --- a/src/Server.js +++ b/src/Server.js @@ -19,28 +19,21 @@ const DEFAULT_OPTIONS = { } export class Server { - #salt = null - #wss = null - #options = Object.create(null) - #peers = [] + #salt = null + #wss = null + #options = Object.create(null) constructor({ deflate = false, salt }) { + this.discover = [] + this.peers = [] + if(deflate) { this.#options.perMessageDeflate = DEFAULT_OPTIONS } + this.#salt = salt } - #randomizeAddress() { - let address - - while(!address || this.#peers.find(peer => peer.address.compare(address) === 0)) { - address = randomBytes(8) - } - - return address - } - listen(port) { const wss = new WebSocketServer({ ...this.#options, @@ -53,17 +46,12 @@ export class Server { socket }) - peer.on('outgoing', ({ destination, message }) => { - const destPeer = this.#peers.find(p => p.address === destination) - if(destPeer) { - destPeer.emit('incoming', { source: peer, message }) - } - }) - - this.#peers.push(peer) + peer.server = this + this.peers.push(peer) socket.on('close', () => { - this.#peers = this.#peers.filter(p => p !== peer) + this.peers = this.peers.filter(p => p !== peer) + this.discover = this.discover.filter(sv => sv.peer !== peer) }) })