84 lines
1.3 KiB
Go
84 lines
1.3 KiB
Go
package nexus
|
|
|
|
import (
|
|
"sync"
|
|
|
|
"dynatron.me/x/stillbox/pkg/gordio/calls"
|
|
"dynatron.me/x/stillbox/pkg/pb"
|
|
)
|
|
|
|
type Nexus struct {
|
|
sync.RWMutex
|
|
|
|
clients map[*client]struct{}
|
|
|
|
*wsManager
|
|
|
|
callCh chan *calls.Call
|
|
}
|
|
|
|
type Registry interface {
|
|
NewClient(Connection) Client
|
|
Register(Client)
|
|
Unregister(Client)
|
|
}
|
|
|
|
func New() *Nexus {
|
|
n := &Nexus{
|
|
clients: make(map[*client]struct{}),
|
|
callCh: make(chan *calls.Call),
|
|
}
|
|
|
|
n.wsManager = newWsManager(n)
|
|
|
|
return n
|
|
}
|
|
|
|
func (n *Nexus) Go(done <-chan struct{}) {
|
|
for {
|
|
select {
|
|
case call, ok := <-n.callCh:
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
n.broadcastCallToClients(call)
|
|
case <-done:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func (n *Nexus) BroadcastCall(call *calls.Call) {
|
|
n.callCh <- call
|
|
}
|
|
|
|
func (n *Nexus) broadcastCallToClients(call *calls.Call) {
|
|
message := &pb.Message{
|
|
ToClientMessage: &pb.Message_Call{Call: call.ToPB()},
|
|
}
|
|
n.Lock()
|
|
defer n.Unlock()
|
|
|
|
for cl, _ := range n.clients {
|
|
if cl.Send(message) {
|
|
// we already hold the lock, and the channel is closed anyway
|
|
delete(n.clients, cl)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (n *Nexus) Register(c Client) {
|
|
n.Lock()
|
|
defer n.Unlock()
|
|
|
|
n.clients[c.(*client)] = struct{}{}
|
|
}
|
|
|
|
func (n *Nexus) Unregister(c Client) {
|
|
n.Lock()
|
|
defer n.Unlock()
|
|
|
|
cl := c.(*client)
|
|
delete(n.clients, cl)
|
|
}
|