dispatch

README

godispatch

This is a general purpose object dispatcher for go.

It can be used to either asynchronously dispatch queues of objects or synchronously dispatch single objects to multiple handlers.

Go Get

go get -u github.com/markus-wa/godispatch

Example

import (
	"fmt"
	dp "github.com/markus-wa/godispatch"
)

func main() {
	d := dp.Dispatcher{}
	// Register a handler for string (not *string!)
	// We get the string Type by calling Elem() on reflect.Type *string)
	// This is faster than doing reflect.TypeOf("")
	d.RegisterHandler(func(s string) {
		fmt.Println("Handled string", s)
	})
	d.RegisterHandler(func(obj interface{}) {
		fmt.Println("Handled object", obj)
	})

	d.Dispatch("Hello")
	// Prints (in this order - as the object handler was registered after the string handler)
	// "Handled string Hello"
	// "Handled object Hello"
	d.Dispatch(123)
	// Prints "Handled object 123"
}

Queue Example

type Event struct {
	reference int
	message   string
}

type TriggerEvent struct{}

func main() {
	d := dp.Dispatcher{}
	// If you wanted to handle pointers of the Event just remove .Elem(),
	// use *Event for the type assertion and send pointers
	d.RegisterHandler(func(e Event) {
		fmt.Println("Handled Event", e)
		// Handle event
	})
	d.RegisterHandler(func(te TriggerEvent) {
		// Do stuff when we receive a 'TriggerEvent'
	})

	// Buffered to improve performance by avoiding locking
	q := make(chan interface{}, 5)
	q2 := make(chan interface{}, 5)
	q3 := make(chan interface{}, 5)

	// Add queues to dispatcher
	d.AddQueues(q, q2, q3)

	// Send some events
	for i := 0; i < 10; i++ {
		q <- Event{i, "abc"}
		q <- TriggerEvent{}
		q2 <- Event{i, "def"}
		q3 <- Event{i, "geh"}
		// Events that are not in the same queue will be handled concurrently
		d.SyncQueues(q)
		// Do stuff that requires events in q (but not q2 & q3) to be handled
	}
	d.SyncAllQueues()
	// Do stuff that requires events of q, q2 & q3 to be handled

	// Maybe send some more events . . .
	q <- TriggerEvent{}

	// Remove queues q & q2
	d.RemoveQueues(q, q2)

	q3 <- Event{}

	// Also remove q3
	d.RemoveAllQueues()
}

Documentation

Overview

Package dispatch provides a general purpose object dispatcher. It can be used to asynchronously dispatch queues (channels) or synchronously dispatch single objects. For example one could use it to dispatch events or messages.

Index

• Variables
• type Config
• type ConsumerCodePanic
• type Dispatcher
  • func NewDispatcherWithConfig(cfg Config) *Dispatcher
  • func (d *Dispatcher) AddQueues(queues ...chan interface{})
  • func (d *Dispatcher) Dispatch(object interface{})
  • func (d *Dispatcher) RegisterHandler(handler interface{}) HandlerIdentifier
  • func (d *Dispatcher) RemoveAllQueues()
  • func (d *Dispatcher) RemoveQueues(queues ...chan interface{}) error
  • func (d *Dispatcher) SyncAllQueues()
  • func (d *Dispatcher) SyncQueues(queues ...chan interface{}) error
  • func (d *Dispatcher) UnregisterAllHandlers()
  • func (d *Dispatcher) UnregisterHandler(identifier HandlerIdentifier)
• type HandlerIdentifier
• type PanicHandler

Variables

var ErrQueueNotFound = errors.New("one or more queues not found")

Types

type Config

type Config struct {
	PanicHandler PanicHandler // A function that handles panics in goroutines created by Dispatcher.AddQueues()
}

Config contains configuration parameters for a Dispatcher.

type ConsumerCodePanic

type ConsumerCodePanic interface {
	String() string
	Value() interface{}
}

ConsumerCodePanic is the type which identifies panics in consumer code (e.g. RegisterHandler).

Example:

d := dp.Dispatcher{}
d.RegisterHandler(func(a *A) {
	fmt.Println(a.val)
})

var err interface{}
func() {
	defer func() {
		err = recover()
	}()

	var a *A
	d.Dispatch(a)
}()

switch err.(type) {
case dp.ConsumerCodePanic:
	fmt.Println("something went wrong inside consumer code")
default:
	fmt.Println("something went wrong in the library")
}

type Dispatcher

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

Dispatcher is used to register handlers and dispatch objects.

func NewDispatcherWithConfig

func NewDispatcherWithConfig(cfg Config) *Dispatcher

NewDispatcherWithConfig creates a new dispatcher with a given config and returns it.

func (*Dispatcher) AddQueues

func (d *Dispatcher) AddQueues(queues ...chan interface{})

AddQueues adds channels as 'object-queues'. All objects sent to the passed queues will be dispatched to their handlers in a separate go routine per channel.

func (*Dispatcher) Dispatch

func (d *Dispatcher) Dispatch(object interface{})

Dispatch dispatches an object to all it's handlers in the order in which the handlers were registered.

func (*Dispatcher) RegisterHandler

func (d *Dispatcher) RegisterHandler(handler interface{}) HandlerIdentifier

RegisterHandler registers an object handler (func) for the type of objects assignable to it's input parameter type. If the handler registers a new handler in the function body, the new handler will only be active for new Dispatch() calls. Calling this method clears the internal type/handler mapping cache for interface handlers. Returns a unique identifier which can be used to remove the handler via UnregisterHandler().

func (*Dispatcher) RemoveAllQueues

func (d *Dispatcher) RemoveAllQueues()

RemoveAllQueues removes all queues from the Dispatcher without closing them.

func (*Dispatcher) RemoveQueues

func (d *Dispatcher) RemoveQueues(queues ...chan interface{}) error

RemoveQueues removes the given queues from the Dispatcher without closing them.

func (*Dispatcher) SyncAllQueues

func (d *Dispatcher) SyncAllQueues()

SyncAllQueues calls SyncQueues() for all queues in this Dispatcher.

func (*Dispatcher) SyncQueues

func (d *Dispatcher) SyncQueues(queues ...chan interface{}) error

SyncQueues syncs the channels dispatch routines to the current go routine. This ensures all objects received in the passed channels up to this point will be handled before continuing.

func (*Dispatcher) UnregisterAllHandlers

func (d *Dispatcher) UnregisterAllHandlers()

UnregisterAllHandlers removes all handlers previously registered via RegisterHandler.

func (*Dispatcher) UnregisterHandler

func (d *Dispatcher) UnregisterHandler(identifier HandlerIdentifier)

UnregisterHandler unregisters a handler by it's identifier (as returned by RegisterHandler()). Unregistering is done via identifiers because functions can't be compared in Go.

type HandlerIdentifier

type HandlerIdentifier *int

HandlerIdentifier uniquely identifies a handler

type PanicHandler

type PanicHandler func(v interface{})

PanicHandler is a function that gets called when a panic occurs during dispatching.