evtbus

README

event-bus

An event bus for consuming and producing events. The library allows for slow consumers not to block the bus and stop other consumers from receiving events, it also handles having fast producers that are overwhelming the system.

##Documentation See godoc

##Installation

go get github.com/go-home-iot/event-bus

##Package

import "github.com/go-home-iot/event-bus"

##Usage See bus_test.go for examples on how to use this library.

##Version History ###0.1.0 Initial release

Documentation

Index

• Variables
• type Bus
  • func NewBus(capacity, consumerCapacity int) *Bus
  • func (b *Bus) AddConsumer(c Consumer)
  • func (b *Bus) AddEnqueueFilter(ID string, f func(Event) bool, delay time.Duration)
  • func (b *Bus) AddProducer(p Producer)
  • func (b *Bus) Enqueue(e Event) error
  • func (b *Bus) RemoveConsumer(c Consumer)
  • func (b *Bus) RemoveEnqueueFilter(ID string)
  • func (b *Bus) RemoveProducer(p Producer)
  • func (b *Bus) Stop()
• type Consumer
• type Event
• type Producer

Variables

var ErrBusFull = errors.New("Event bus is full, no more events can be added at this time")

ErrBusFull indicates the bus is full and can't handle any more events at this time.

Types

type Bus

type Bus struct {

	// Capacity is the number of events that can be in the bus at one time before incoming
	// events will be ignored
	Capacity int

	// ConsumerCapacity is the number of events a consumer can queue. If the consumer is slow
	// at processing events, you should set this to some high number. Once this number is reached
	// the bus will throw away events to the slow consumer so that other consumers are not blocked
	ConsumerCapacity int
	// contains filtered or unexported fields
}

Bus is a type that facilitates sending events from multiple producers to multiple consumers. It is non blocking, so fast producers won't break the bus and slow consumers won't block other consumers from receiving events

func NewBus

func NewBus(capacity, consumerCapacity int) *Bus

NewBus returns an initialized bus

func (*Bus) AddConsumer

func (b *Bus) AddConsumer(c Consumer)

AddConsumer adds a consumer to the bus, once added the consumer will start to receive events from the bus

func (*Bus) AddEnqueueFilter

func (b *Bus) AddEnqueueFilter(ID string, f func(Event) bool, delay time.Duration)

AddEnqueueFilter adds an enqueue filter. If a filter with the ID already exists it is replaced by the new filter

func (*Bus) AddProducer

func (b *Bus) AddProducer(p Producer)

AddProducer adds a producer to the bus and registers it

func (*Bus) Enqueue

func (b *Bus) Enqueue(e Event) error

Enqueue adds an event to the event bus. It is non blocking, if there is not enough capacity in the bus to add a new event, the method returns an error

func (*Bus) RemoveConsumer

func (b *Bus) RemoveConsumer(c Consumer)

RemoveConsumer removes a consumer from the bus, once removed consumers will no longer receive events

func (*Bus) RemoveEnqueueFilter

func (b *Bus) RemoveEnqueueFilter(ID string)

RemoveEnqueueFilter removes an enqueue filter

func (*Bus) RemoveProducer

func (b *Bus) RemoveProducer(p Producer)

RemoveProducer removes a producer from the bus

func (*Bus) Stop

func (b *Bus) Stop()

Stop removes all of the consumers and producers and stops processing events. After calling this method the Bus is no longer usable and you should create a new one of you need another bus

type Consumer

type Consumer interface {
	// Name should return a friendly name for the consumer.  This field
	// may be useful for debugging
	ConsumerName() string

	// StartConsuming is called when the consumer is added to the event bus.
	// The channel which is passed in will recieve all of the events in the bus
	StartConsuming(chan Event)

	// StopConsuming is called when the consumer is removed from the event bus
	StopConsuming()
}

Consumer is an interface that describes a type that consumes events

type Event

type Event interface {
	fmt.Stringer
}

Event is an interface describing an event in the system

type Producer

type Producer interface {
	// Name should return a friendly name for the consumer.  This field
	// may be useful for debugging
	ProducerName() string

	// StartProducing is called when the procuder is added to the bus
	StartProducing(*Bus)

	// StopProducing is called when the producer is removed from the bus
	StopProducing()
}

Producer is an interface that describes a type that produces events on the event bus