event

Documentation

Overview

Event bus

Example (FanOutBus)

Create a FanOutBus and two receivers.

bus := NewFanOutBus(1024)
bus.GoDispatch()

recvFoo := bus.NewRecv("foo", 1024)

go func() {
	// send event
	bus.C <- "hello"
}()

go func() {
	// receive event
	for v := range recvFoo.C {
		fmt.Println(v)
	}
}()

// make a new Receiver on the fly
recvBar := bus.NewRecv("bar", 1024)
go func() {
	// receive event
	for v := range recvBar.C {
		fmt.Println(v)
	}
}()

go func() {
	// close Receiver if you want
	recvBar.Close()
}()

go func() {
	// close the bus, also close all the receivers.
	bus.Close()
}()

Index

• type FanOutBus
  • func NewFanOutBus(bufSize int) *FanOutBus
  • func (b *FanOutBus) Close()
  • func (b *FanOutBus) GoDispatch()
  • func (b *FanOutBus) NewRecv(name string, bufSize int) *Receiver
  • func (b *FanOutBus) NewRecvStrategy(name string, bufSize int, fullStrategy FullStrategy) *Receiver
• type FullStrategy
• type Receiver
  • func (r *Receiver) Close()
  • func (r *Receiver) CollectTimeout(timeout time.Duration) (data []interface{})
  • func (r *Receiver) Drain() (data []interface{})

Examples

• Package (FanOutBus)

Types

type FanOutBus

type FanOutBus struct {
	C chan<- interface{} // a channel to send events
	// contains filtered or unexported fields
}

A fan-out event bus. Sending a event to it, it will dispatch events to all Receivers.

func NewFanOutBus

func NewFanOutBus(bufSize int) *FanOutBus

NewFanOutBus make a new FanOutBus

bufSize: capacity of FanOutBus.C, if 0 then it's unbuffered

func (*FanOutBus) Close

func (b *FanOutBus) Close()

Close FanOutBus.C, close all Receiver.C, deregister Receivers

func (*FanOutBus) GoDispatch

func (b *FanOutBus) GoDispatch()

Start dispatching events

func (*FanOutBus) NewRecv

func (b *FanOutBus) NewRecv(name string, bufSize int) *Receiver

Make a new *Receiver and registered to event bus with fullStrategy of Drop. See NewRecvStrategy for more information.

func (*FanOutBus) NewRecvStrategy

func (b *FanOutBus) NewRecvStrategy(name string, bufSize int, fullStrategy FullStrategy) *Receiver

NewRecvStrategy make a new *Receiver and registered to event bus. You can make a Receiver at any time except event bus is closed. Any events sent to event bus before Receiver making maybe or maybe not be sent to them.

name: receiver's name
bufSize: capacity of Receiver.C, if 0 then it's unbuffered
fullStrategy: if Receiver.C is full, Drop the event or Block the goroutine.

Be careful if choose Block, the whole event bus will be blocked if any Receiver can't catch-up. i.e, if Receiver A blocks, other receivers cannot receive event until Receiver A proceeds.

type FullStrategy

type FullStrategy int

Strategy for when Receiver.C is full

const (
	// drop event if if Receiver.C is full
	Drop FullStrategy = iota
	// blocks until there is room for Receiver.C(buffered) or there is a receiver waiting on Receiver.C(unbuffered)
	Block FullStrategy = iota
)

type Receiver

type Receiver struct {
	C <-chan interface{} // a channel for receiving events
	// contains filtered or unexported fields
}

Event receiver

func (*Receiver) Close

func (r *Receiver) Close()

Close Receiver.C, deregistered itself from event bus

func (*Receiver) CollectTimeout

func (r *Receiver) CollectTimeout(timeout time.Duration) (data []interface{})

CollectTimeout collects elements of Receiver.C until timeout

func (*Receiver) Drain

func (r *Receiver) Drain() (data []interface{})

Drain drains Receiver.C until it's empty