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) RegisterHandlerWithOptions(handlerFunc interface{}, opts HandlerOptions) 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 HandlerOptions
• type PanicHandler

Variables

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

Types

type Config

type Config struct {
	// PanicHandler handles panics in goroutines created by Dispatcher.AddQueues().
	// It receives the recovered value: a ConsumerCodePanic for handlers registered via RegisterHandler,
	// or the raw panic value for handlers registered with HandlerOptions.UnwrappedPanics.
	// If it is nil, panics are not recovered and crash the program.
	PanicHandler PanicHandler
}

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). Panics from handlers registered with HandlerOptions.UnwrappedPanics are not wrapped and propagate as-is.

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. If a handler panics, the remaining handlers are not called and the panic propagates to the caller, wrapped in a ConsumerCodePanic unless the handler was registered with HandlerOptions.UnwrappedPanics.

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(). Panics from the handler are wrapped in a ConsumerCodePanic, see RegisterHandlerWithOptions to change this.

func (*Dispatcher) RegisterHandlerWithOptions

func (d *Dispatcher) RegisterHandlerWithOptions(handlerFunc interface{}, opts HandlerOptions) HandlerIdentifier

RegisterHandlerWithOptions is like RegisterHandler but allows configuring the handler via HandlerOptions. The returned identifier can be used with UnregisterHandler() like any other.

With HandlerOptions.UnwrappedPanics set, a panic inside the handler propagates out of Dispatch() with its original value instead of being wrapped in a ConsumerCodePanic; handlers after it are not called. For objects dispatched via queues (see AddQueues()) the raw value is passed to Config.PanicHandler, or crashes the program if no PanicHandler is set (same as for wrapped panics).

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 or RegisterHandlerWithOptions.

func (*Dispatcher) UnregisterHandler

func (d *Dispatcher) UnregisterHandler(identifier HandlerIdentifier)

UnregisterHandler unregisters a handler by it's identifier (as returned by RegisterHandler() or RegisterHandlerWithOptions()). Unregistering is done via identifiers because functions can't be compared in Go. It may be called from inside a handler; handlers of an in-progress Dispatch() call are not affected, the change applies to subsequent Dispatch() calls.

type HandlerIdentifier

type HandlerIdentifier *int

HandlerIdentifier uniquely identifies a handler

type HandlerOptions

type HandlerOptions struct {
	// UnwrappedPanics makes panics from this handler propagate as-is instead of as ConsumerCodePanic.
	UnwrappedPanics bool
}

HandlerOptions configures a handler registered via RegisterHandlerWithOptions. The zero value gives the same behaviour as RegisterHandler.

type PanicHandler

type PanicHandler func(v interface{})

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