Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Abstract signaling server messages #62

Open
wants to merge 8 commits into
base: master
Choose a base branch
from
Open

Abstract signaling server messages #62

Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
4be6f1c
abstract signaling server client
byrond Nov 20, 2023
e4ab973
export publishSignalingMessage for use by derived classes
byrond Nov 21, 2023
4725aeb
make handleMessage a method of SignalingConn
byrond Nov 21, 2023
f2c1847
remove publishSignalingMessage export
byrond Nov 28, 2023
7736019
only setup client if undefined
byrond Nov 28, 2023
dec3d27
add joinAllRooms and addSignalingConn methods
byrond Dec 5, 2023
a31abed
combine joinAllRooms and handleConnect
byrond Dec 6, 2023
19bf460
remove commented conditional
byrond Dec 6, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
198 changes: 117 additions & 81 deletions src/y-webrtc.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -269,8 +269,8 @@ const broadcastRoomMessage = (room, m) => {
const announceSignalingInfo = room => {
signalingConns.forEach(conn => {
// only subscribe if connection is established, otherwise the conn automatically subscribes to all rooms
if (conn.connected) {
conn.send({ type: 'subscribe', topics: [room.name] })
if (conn.connected()) {
conn.subscribe(room.name)
if (room.webrtcConns.size < room.provider.maxConns) {
publishSignalingMessage(conn, room, { type: 'announce', from: room.peerId })
}
Expand Down Expand Up @@ -412,8 +412,8 @@ export class Room {
disconnect () {
// signal through all available signaling connections
signalingConns.forEach(conn => {
if (conn.connected) {
conn.send({ type: 'unsubscribe', topics: [this.name] })
if (conn.connected()) {
conn.unsubscribe(this.name)
}
})
awarenessProtocol.removeAwarenessStates(this.awareness, [this.doc.clientID], 'disconnect')
Expand Down Expand Up @@ -466,96 +466,130 @@ const openRoom = (doc, provider, name, key) => {
const publishSignalingMessage = (conn, room, data) => {
if (room.key) {
cryptoutils.encryptJson(data, room.key).then(data => {
conn.send({ type: 'publish', topic: room.name, data: buffer.toBase64(data) })
conn.publish(room.name, buffer.toBase64(data))
})
} else {
conn.send({ type: 'publish', topic: room.name, data })
conn.publish(room.name, data)
}
}

export class SignalingConn extends ws.WebsocketClient {
export class SignalingConn {
constructor (url) {
super(url)
this.url = url
/**
* @type {Set<WebrtcProvider>}
*/
this.providers = new Set()
this.on('connect', () => {
log(`connected (${url})`)
const topics = Array.from(rooms.keys())
this.send({ type: 'subscribe', topics })
rooms.forEach(room =>
publishSignalingMessage(this, room, { type: 'announce', from: room.peerId })
)
this.setupClient()
}

setupClient() {
this.client = new ws.WebsocketClient(this.url)
this.client.on('connect', () => {
log(`connected (${this.url})`)
this.handleConnect()
})
this.on('message', m => {
this.client.on('message', m => {
switch (m.type) {
case 'publish': {
const roomName = m.topic
const room = rooms.get(roomName)
if (room == null || typeof roomName !== 'string') {
return
this.handleMessage(m.topic, m.data)
}
}
})
this.client.on('disconnect', () => log(`disconnect (${this.url})`))
}

handleConnect() {
const topics = Array.from(rooms.keys())
topics.forEach(topic =>
this.subscribe(topic)
)
Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Azure Web PubSub cannot join multiple groups, but that shouldn't stop the built-in client from doing so. Maybe instead of the forEach this needs a subscribeMulti() method or let subscribe() detect whether topic is an array in derived classes and do the iteration there?

rooms.forEach(room =>
publishSignalingMessage(this, room, { type: 'announce', from: room.peerId })
)
}

handleMessage (roomName, data) {
const room = rooms.get(roomName)
if (room == null || typeof roomName !== 'string') {
return
}
const execMessage = data => {
const webrtcConns = room.webrtcConns
const peerId = room.peerId
if (data == null || data.from === peerId || (data.to !== undefined && data.to !== peerId) || room.bcConns.has(data.from)) {
// ignore messages that are not addressed to this conn, or from clients that are connected via broadcastchannel
return
}
const emitPeerChange = webrtcConns.has(data.from)
? () => {}
: () =>
room.provider.emit('peers', [{
removed: [],
added: [data.from],
webrtcPeers: Array.from(room.webrtcConns.keys()),
bcPeers: Array.from(room.bcConns)
}])
switch (data.type) {
case 'announce':
if (webrtcConns.size < room.provider.maxConns) {
map.setIfUndefined(webrtcConns, data.from, () => new WebrtcConn(this, true, data.from, room))
emitPeerChange()
}
const execMessage = data => {
const webrtcConns = room.webrtcConns
const peerId = room.peerId
if (data == null || data.from === peerId || (data.to !== undefined && data.to !== peerId) || room.bcConns.has(data.from)) {
// ignore messages that are not addressed to this conn, or from clients that are connected via broadcastchannel
return
}
const emitPeerChange = webrtcConns.has(data.from)
? () => {}
: () =>
room.provider.emit('peers', [{
removed: [],
added: [data.from],
webrtcPeers: Array.from(room.webrtcConns.keys()),
bcPeers: Array.from(room.bcConns)
}])
switch (data.type) {
case 'announce':
if (webrtcConns.size < room.provider.maxConns) {
map.setIfUndefined(webrtcConns, data.from, () => new WebrtcConn(this, true, data.from, room))
emitPeerChange()
}
break
case 'signal':
if (data.signal.type === 'offer') {
const existingConn = webrtcConns.get(data.from)
if (existingConn) {
const remoteToken = data.token
const localToken = existingConn.glareToken
if (localToken && localToken > remoteToken) {
log('offer rejected: ', data.from)
return
}
// if we don't reject the offer, we will be accepting it and answering it
existingConn.glareToken = undefined
}
}
if (data.signal.type === 'answer') {
log('offer answered by: ', data.from)
const existingConn = webrtcConns.get(data.from)
existingConn.glareToken = undefined
}
if (data.to === peerId) {
map.setIfUndefined(webrtcConns, data.from, () => new WebrtcConn(this, false, data.from, room)).peer.signal(data.signal)
emitPeerChange()
}
break
break
case 'signal':
if (data.signal.type === 'offer') {
const existingConn = webrtcConns.get(data.from)
if (existingConn) {
const remoteToken = data.token
const localToken = existingConn.glareToken
if (localToken && localToken > remoteToken) {
log('offer rejected: ', data.from)
return
}
// if we don't reject the offer, we will be accepting it and answering it
existingConn.glareToken = undefined
}
}
if (room.key) {
if (typeof m.data === 'string') {
cryptoutils.decryptJson(buffer.fromBase64(m.data), room.key).then(execMessage)
}
} else {
execMessage(m.data)
if (data.signal.type === 'answer') {
log('offer answered by: ', data.from)
const existingConn = webrtcConns.get(data.from)
existingConn.glareToken = undefined
}
}
if (data.to === peerId) {
map.setIfUndefined(webrtcConns, data.from, () => new WebrtcConn(this, false, data.from, room)).peer.signal(data.signal)
emitPeerChange()
}
break
}
})
this.on('disconnect', () => log(`disconnect (${url})`))
}
if (room.key) {
if (typeof data === 'string') {
cryptoutils.decryptJson(buffer.fromBase64(data), room.key).then(execMessage)
}
} else {
execMessage(data)
}
}

connected () {
return this.client.connected
}

subscribe (roomName) {
this.client.send({ type: 'subscribe', topics: [roomName] })
}

unsubscribe (roomName) {
this.client.send({ type: 'unsubscribe', topics: [roomName] })
}

publish (roomName, message) {
this.client.send({ type: 'publish', topic: roomName, data: message })
}

destroy () {
this.client.destroy()
}
}

Expand Down Expand Up @@ -633,11 +667,7 @@ export class WebrtcProvider extends Observable {

connect () {
this.shouldConnect = true
this.signalingUrls.forEach(url => {
const signalingConn = map.setIfUndefined(signalingConns, url, () => new SignalingConn(url))
this.signalingConns.push(signalingConn)
signalingConn.providers.add(this)
})
this.signalingUrls.forEach(url => this.addSignalingConn(url, () => new SignalingConn(url)))
if (this.room) {
this.room.connect()
}
Expand Down Expand Up @@ -666,4 +696,10 @@ export class WebrtcProvider extends Observable {
})
super.destroy()
}

addSignalingConn(url, conn) {
Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

conn is technically a function. Maybe we need to call it createT or something similar?

const signalingConn = map.setIfUndefined(signalingConns, url, conn)
this.signalingConns.push(signalingConn)
signalingConn.providers.add(this)
}
}