eventbus

README

eventbus

A Multiconsumer/multiproducer bus.

go get gopkg.in/Eun/eventbus/v1

Example

package main

import (
	"fmt"

	"github.com/Eun/eventbus"
)

type IntegerEvent struct {
	I int
}

type ExitEvent struct{}

func consumer(consumerID int, bus *eventbus.EventBus, listener <-chan eventbus.Event) {
	defer bus.RemoveListener(listener)
	for ev := range listener {
		switch event := ev.(type) {
		case IntegerEvent:
			fmt.Printf("consumer %d: %d\n", consumerID, event.I)
		default:
			fmt.Printf("consumer %d: exit\n", consumerID)
			return
		}
	}
}

func main() {
	bus := eventbus.New()

	// add two consumers
	go consumer(1, bus, bus.AddListener())
	go consumer(2, bus, bus.AddListener())

	// raise 100 events
	for i := 0; i < 100; i++ {
		bus.Raise(IntegerEvent{I: i})
	}

	// add a third consumer, this one will only receive the last 10 items
	// because the HistorySize of the bus is set to 10
	go consumer(3, bus, bus.AddListener())

	// raise the exit event
	bus.Raise(ExitEvent{})

	// wait until all listeners have been removed
	bus.WaitAndClose()
}

Documentation

Index

• type Event
• type EventBus
  • func New(historySize ...int) *EventBus
  • func (bus *EventBus) AddListener(channelSize ...int) <-chan Event
  • func (bus *EventBus) Close()
  • func (bus *EventBus) Closed() bool
  • func (bus *EventBus) HistorySize() int
  • func (bus *EventBus) Raise(events ...Event)
  • func (bus *EventBus) RemoveListener(listeners ...<-chan Event)
  • func (bus *EventBus) Wait()
  • func (bus *EventBus) WaitAndClose()

Types

type Event

type Event interface{}

Event is the container that holds the event data

type EventBus

type EventBus struct {
	// contains filtered or unexported fields
}

EventBus is a multi consumer / multi producer bus

func New

func New(historySize ...int) *EventBus

New creates a new EventBus

func (*EventBus) AddListener

func (bus *EventBus) AddListener(channelSize ...int) <-chan Event

AddListener adds an listener to the bus all cached messages will be immediately passed to the listener

func (*EventBus) Close

func (bus *EventBus) Close()

Close stops new events from being trigerred, however the listeners need to close them self

func (*EventBus) Closed

func (bus *EventBus) Closed() bool

Closed returns if the

func (*EventBus) HistorySize

func (bus *EventBus) HistorySize() int

HistorySize returns the events that will be stored for future channels

func (*EventBus) Raise

func (bus *EventBus) Raise(events ...Event)

Raise raises an event

func (*EventBus) RemoveListener

func (bus *EventBus) RemoveListener(listeners ...<-chan Event)

RemoveListener removes an listener from the bus

func (*EventBus) Wait

func (bus *EventBus) Wait()

Wait waits until all listeners have been removed

func (*EventBus) WaitAndClose

func (bus *EventBus) WaitAndClose()

WaitAndClose waits until all listeners have been removed and closes the bus