event_bus.go

package api

import (
	"log"
	"sync"
	"time"
)

type EventBus struct {
	mutex     sync.RWMutex
	listeners []chan HashedEntity
	closed    bool
}

func (receiver *EventBus) Send(entity HashedEntity) {
	receiver.mutex.RLock()
	defer receiver.mutex.RUnlock()
	if receiver.closed {
		return
	}
	for _, ch := range receiver.listeners {
		ch <- entity
	}
}

func (receiver *EventBus) Receive() <-chan HashedEntity {
	receiver.mutex.Lock()
	defer receiver.mutex.Unlock()
	ch := make(chan HashedEntity, 1)
	receiver.listeners = append(receiver.listeners, ch)
	return ch
}

func (receiver *EventBus) Close() {
	receiver.mutex.Lock()
	defer receiver.mutex.Unlock()
	if !receiver.closed {
		for _, ch := range receiver.listeners {
			close(ch)
		}
		receiver.closed = true
	}
}

func NewEventBus() *EventBus {
	eb := &EventBus{}
	eb.listeners = make([]chan HashedEntity, 0)
	return eb
}

func test() {
	eb := NewEventBus()
	ch := eb.Receive()
	go func() {
		time.Sleep(2 * time.Second)
		log.Println("received ", <-ch)
	}()
	time.Sleep(3 * time.Second)
	eb.Send(HashedEntity{})
	time.Sleep(4 * time.Second)
	log.Printf("done")
}