Server/bus.go

package lucifer3

import (
	"log"
	"sync"
)

type ServiceKey struct{}

type Event interface {
	EventName() string
}

type EventBus struct {
	mu        sync.Mutex
	listeners []*serviceListener
	signal    chan struct{}
}

// JoinCallback joins the event bus for a moment.
func (b *EventBus) JoinCallback(cb func(event Event, sender ServiceID) bool) {
	b.Join(newCallbackService(cb))
}

func (b *EventBus) Join(service Service) {
	// Take the signal here so that it will receive the events sent by the calling function
	// so that they're processed without having to wait for a new event.
	signal := b.signalCh()

	b.mu.Lock()
	listener := &serviceListener{
		bus:     b,
		queue:   make([]queuedEvent, 0, 16),
		service: service,
	}

	go listener.run(signal)

	b.listeners = append(b.listeners, listener)
	b.mu.Unlock()
}

func (b *EventBus) Send(event Event, sender ServiceID, recipient *ServiceID) {
	b.mu.Lock()
	defer b.mu.Unlock()

	deleteList := make([]int, 0, 0)
	for i, listener := range b.listeners {
		if !listener.service.Active() {
			deleteList = append(deleteList, i-len(deleteList))
			continue
		}
		if recipient != nil && *recipient != listener.service.ServiceID() {
			continue
		}

		listener.mu.Lock()
		listener.queue = append(listener.queue, queuedEvent{
			event:  event,
			sender: sender,
		})
		listener.mu.Unlock()
	}

	for i := range deleteList {
		b.listeners = append(b.listeners[:i], b.listeners[i+1:]...)
	}

	if recipient != nil {
		log.Printf("%s (-> %s): %s", &sender, recipient, event.EventName())
	} else {
		log.Printf("%s: %s", &sender, event.EventName())
	}

	if b.signal != nil {
		close(b.signal)
		b.signal = nil
	}
}

func (b *EventBus) signalCh() <-chan struct{} {
	b.mu.Lock()
	defer b.mu.Unlock()

	if b.signal == nil {
		b.signal = make(chan struct{})
	}

	return b.signal
}

type serviceListener struct {
	mu      sync.Mutex
	bus     *EventBus
	queue   []queuedEvent
	service Service
}

func (l *serviceListener) run(signal <-chan struct{}) {
	queue := make([]queuedEvent, 0, 16)

	for {
		// Listen for the signal, but stop here if the service has marked
		// itself as inactive.
		<-signal
		if !l.service.Active() {
			return
		}

		// Take a new signal before copying the queue to avoid delay
		// in case a message is about to be enqueued right now!
		signal = l.bus.signalCh()

		// Take a copy of the queue.
		l.mu.Lock()
		for _, message := range l.queue {
			queue = append(queue, message)
		}
		l.queue = l.queue[:0]
		l.mu.Unlock()

		// Handle the messages in the queue, but stop mid-queue if the previous message
		// or external circumstances marked the service inactive.
		for _, message := range queue {
			if !l.service.Active() {
				return
			}

			l.service.Handle(l.bus, message.event, message.sender)
		}
		queue = queue[:0]
	}
}

type queuedEvent struct {
	event  Event
	sender ServiceID
}