ghostream/stream/messaging.go

99 lines
2.1 KiB
Go
Raw Normal View History

2020-10-17 08:02:38 +00:00
// Package stream defines a structure to communication between inputs and outputs
package stream
import "sync"
// Stream makes packages able to subscribe to an incoming stream
type Stream struct {
// Incoming data come from this channel
2020-10-17 08:21:40 +00:00
Broadcast chan<- []byte
2020-10-17 08:02:38 +00:00
// Use a map to be able to delete an item
2020-10-17 10:26:24 +00:00
outputs map[chan []byte]struct{}
2020-10-17 08:02:38 +00:00
2020-10-17 11:43:16 +00:00
// Count clients for statistics
nbClients int
2020-10-17 08:02:38 +00:00
// Mutex to lock this ressource
lock sync.Mutex
}
// New creates a new stream.
func New() *Stream {
s := &Stream{}
2020-10-17 08:21:40 +00:00
broadcast := make(chan []byte, 64)
2020-10-17 08:02:38 +00:00
s.Broadcast = broadcast
2020-10-17 10:26:24 +00:00
s.outputs = make(map[chan []byte]struct{})
2020-10-17 11:43:16 +00:00
s.nbClients = 0
2020-10-17 08:02:38 +00:00
go s.run(broadcast)
return s
}
2020-10-17 08:21:40 +00:00
func (s *Stream) run(broadcast <-chan []byte) {
2020-10-17 08:02:38 +00:00
for msg := range broadcast {
func() {
s.lock.Lock()
defer s.lock.Unlock()
for output := range s.outputs {
select {
case output <- msg:
default:
2020-10-17 10:26:24 +00:00
// If full, do a ring buffer
<-output
output <- msg
2020-10-17 08:02:38 +00:00
}
}
}()
}
// Incoming chan has been closed, close all outputs
s.lock.Lock()
defer s.lock.Unlock()
for ch := range s.outputs {
delete(s.outputs, ch)
close(ch)
}
}
// Close the incoming chan, this will also delete all outputs
func (s *Stream) Close() {
close(s.Broadcast)
}
2020-10-17 11:43:16 +00:00
// Register a new output on a stream.
// If hidden in true, then do not count this client.
func (s *Stream) Register(output chan []byte) {
2020-10-17 08:02:38 +00:00
s.lock.Lock()
defer s.lock.Unlock()
s.outputs[output] = struct{}{}
}
2020-10-17 11:43:16 +00:00
// Unregister removes an output.
// If hidden in true, then do not count this client.
func (s *Stream) Unregister(output chan []byte) {
2020-10-17 08:02:38 +00:00
s.lock.Lock()
defer s.lock.Unlock()
// Make sure we did not already close this output
_, ok := s.outputs[output]
if ok {
delete(s.outputs, output)
close(output)
}
}
2020-10-17 11:03:49 +00:00
// ClientCount returns the number of clients
func (s *Stream) ClientCount() int {
2020-10-17 11:43:16 +00:00
return s.nbClients
2020-10-17 11:03:49 +00:00
}
// IncrementClientCount increments the number of clients
func (s *Stream) IncrementClientCount() {
s.nbClients++
}
// DecrementClientCount decrements the number of clients
func (s *Stream) DecrementClientCount() {
s.nbClients--
}