fix: fixed events buffering, now no event is lost with multiple listeners (fixes #266)

modules/api_rest.go

@@ -24,7 +24,6 @@ type RestAPI struct {
 	keyFile      string
 	useWebsocket bool
 	upgrader     websocket.Upgrader
-	eventListener <-chan session.Event
 	quit         chan bool
 }
 
@@ -34,7 +33,6 @@ func NewRestAPI(s *session.Session) *RestAPI {
 		server:        &http.Server{},
 		quit:          make(chan bool),
 		useWebsocket:  false,
-		eventListener: s.Events.Listen(),
 		upgrader: websocket.Upgrader{
 			ReadBufferSize:  1024,
 			WriteBufferSize: 1024,

modules/api_rest_ws.go

@@ -68,6 +68,8 @@ func (api *RestAPI) streamWriter(ws *websocket.Conn, w http.ResponseWriter, r *h
 	log.Debug("Listening for events and streaming to ws endpoint ...")
 
 	pingTicker := time.NewTicker(pingPeriod)
+	listener := session.I.Events.Listen()
+	defer session.I.Events.Unlisten(listener)
 
 	for {
 		select {
@@ -75,7 +77,7 @@ func (api *RestAPI) streamWriter(ws *websocket.Conn, w http.ResponseWriter, r *h
 			if err := api.sendPing(ws); err != nil {
 				return
 			}
-		case event := <-api.eventListener:
+		case event := <-listener:
 			if err := api.streamEvent(ws, event); err != nil {
 				return
 			}

session/events.go

@@ -62,6 +62,19 @@ func (p *EventPool) Listen() <-chan Event {
 	return l
 }
 
+func (p *EventPool) Unlisten(listener <-chan Event) {
+	p.Lock()
+	defer p.Unlock()
+
+	for i, l := range p.listeners {
+		if l == listener {
+			close(l)
+			p.listeners = append(p.listeners[:i], p.listeners[i+1:]...)
+			return
+		}
+	}
+}
+
 func (p *EventPool) SetSilent(s bool) {
 	p.Lock()
 	defer p.Unlock()
@@ -83,16 +96,7 @@ func (p *EventPool) Add(tag string, data interface{}) {
 
 	// broadcast the event to every listener
 	for _, l := range p.listeners {
-		select {
+		l <- e
-		case l <- e:
-			// NOTE: Without this 'default', errors in sending the event
-			// to the listener would not empty the channel, therefore
-			// all operations would be stuck at some point (after the first
-			// event if not buffered or after the first N events if buffered)
-			//
-			// See https://github.com/bettercap/bettercap/issues/198
-		default:
-		}
 	}
 }