feat(discover): add discover

This commit is contained in:
2023-07-12 16:59:59 +02:00
parent 329b222e67
commit dedce2a891
4 changed files with 101 additions and 40 deletions
+1 -1
View File
@@ -1,7 +1,7 @@
import * as dotenv from 'dotenv' import * as dotenv from 'dotenv'
dotenv.config() dotenv.config()
import { Server } from './src/Server.js' import { Server } from './src/Server.js'
const { const {
PPP2_PORT, PPP2_PORT,
+86 -14
View File
@@ -2,7 +2,12 @@ import {
Container, Container,
AuthenticationRequest, AuthenticationRequest,
AuthenticationResponse, AuthenticationResponse,
Message Message,
PublishRequest,
PublishResponse,
DiscoverRequest,
DiscoverResponse,
DiscoverPeer
} from './Protocol.js' } from './Protocol.js'
import { EventEmitter } from 'node:events' import { EventEmitter } from 'node:events'
@@ -21,8 +26,10 @@ export class Peer extends EventEmitter {
this.#salt = salt this.#salt = salt
this.#msgMap = new Map() 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('Message', this.#onmessage)
this.#msgMap.set('PublishRequest', this.#onpublishrequest)
this.#msgMap.set('DiscoverRequest', this.#ondiscoverrequest)
this.#setupListeners() this.#setupListeners()
} }
@@ -51,14 +58,15 @@ export class Peer extends EventEmitter {
} }
}) })
this.on('incoming', ({ source, message }) => { }
this.#send({
id: this.#id++, onincoming(source, message) {
message: Message.create({ this.#send({
address: source.address.toString(10), id: this.#id++,
port: message.Message.port, message: Message.create({
data: message.Message.data 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) this.#socket.send(buf)
} }
#onauth(message) { #onauthrequest(message) {
const { seed } = AuthenticationRequest.toObject(message.AuthenticationRequest, { longs: String }) const { seed } = AuthenticationRequest.toObject(message.AuthenticationRequest, { longs: String })
if(this.#address !== null) { if(this.#address !== null) {
this.#send({ this.#send({
@@ -105,9 +113,73 @@ export class Peer extends EventEmitter {
#onmessage(message) { #onmessage(message) {
const { address } = Message.toObject(message.Message, { longs: String }) const { address } = Message.toObject(message.Message, { longs: String })
this.emit('outgoing', { const bigint = BigInt(address)
destination: BigInt(address),
message 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
})
}) })
} }
+3 -2
View File
@@ -23,8 +23,9 @@ export const [
PublishRequest, PublishRequest,
PublishResponse, PublishResponse,
DiscoverRequest, DiscoverRequest,
DiscoverResponse DiscoverResponse,
] = await import_proto('2.0/PublishDiscover.proto', ['PublishRequest', 'PublishResponse', 'DiscoverRequest', 'DiscoverResponse']) DiscoverPeer
] = await import_proto('2.0/PublishDiscover.proto', ['PublishRequest', 'PublishResponse', 'DiscoverRequest', 'DiscoverResponse', 'DiscoverPeer'])
export const messages = [ export const messages = [
Container, Container,
+11 -23
View File
@@ -19,28 +19,21 @@ const DEFAULT_OPTIONS = {
} }
export class Server { export class Server {
#salt = null #salt = null
#wss = null #wss = null
#options = Object.create(null) #options = Object.create(null)
#peers = []
constructor({ deflate = false, salt }) { constructor({ deflate = false, salt }) {
this.discover = []
this.peers = []
if(deflate) { if(deflate) {
this.#options.perMessageDeflate = DEFAULT_OPTIONS this.#options.perMessageDeflate = DEFAULT_OPTIONS
} }
this.#salt = salt this.#salt = salt
} }
#randomizeAddress() {
let address
while(!address || this.#peers.find(peer => peer.address.compare(address) === 0)) {
address = randomBytes(8)
}
return address
}
listen(port) { listen(port) {
const wss = new WebSocketServer({ const wss = new WebSocketServer({
...this.#options, ...this.#options,
@@ -53,17 +46,12 @@ export class Server {
socket socket
}) })
peer.on('outgoing', ({ destination, message }) => { peer.server = this
const destPeer = this.#peers.find(p => p.address === destination) this.peers.push(peer)
if(destPeer) {
destPeer.emit('incoming', { source: peer, message })
}
})
this.#peers.push(peer)
socket.on('close', () => { 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)
}) })
}) })