events

README

Eventing API

This event API is a very simple implementation of a global dispatch which leverages generic event type multicast dispatchers. This library requires go 1.18 for generics support.

Dispatchers

The main component of the events API is the Dispatcher[T] contract, which is implemented in this library as a multicast channel based dispatch. This library creates a global table that can be used to instantiate global event dispatchers as well as a method for instancing dispatchers individually.

Global Dispatcher Access

To access the global dispatcher for a specific type, use the following:

type MyEvent struct {
    Message string 
}

myEventDispatcher := events.GlobalDispatcherFor[MyEvent]()

This will allow you to retrieve a global dispatcher for the type MyEvent. Any events that dispatch over the global instance will be received by any event handlers on that dispatcher. Global dispatchers are instanced based on the type. For example, events.GlobalDispatcherFor[MyEvent]() will always return the exact dispatcher instance.

Instanced Dispatcher

To create a new instance of a dispatcher, use the following:

type MyEvent struct {
    Message string 
}

myEventDispatcher := events.NewDispatcher[MyEvent]()

This will create a new dispatcher instance for the type MyEvent. Any events that dispatch over the instance will be received by any event handlers on that dispatcher only. Instanced dispatchers do not share event streams with the global dispatcher for that type.

Instanced dispatchers should also be closed when they are no longer needed. This will close any open event streams and remove any existing event handlers automatically. NOTE: Do not call CloseEventStreams() on a global dispatcher.

type MyEvent struct {
    Message string 
}

// Create a new dispatcher instance
myEventDispatcher := events.NewDispatcher[MyEvent]()

// use the dispatcher... 

// shutdown the dispatcher
myEventDispatcher.CloseEventStreams()

Receiving Dispatched Events

Once you create a dispatcher instance, or retrieve a global dispatcher, you can add event handlers capable of receiving events that are dispatched via two methods:

Event Streams

Event streams are used internally in the default multicast dispatcher implementation. For channel based APIs, event streams provide direct access to the event receiving channel, so a simple range loop can be used to process all incoming events.

type MyEvent struct {
    Message string 
}

dispatcher := events.NewDispatcher[MyEvent]()
stream := dispatcher.NewEventStream()

// typically, a stream will be handled in a separate goroutine
go func() {
    for event := range stream.Stream() {
        fmt.Println(event.Message)
    }
}()

// dispatch an event 
dispatcher.Dispatch(MyEvent{Message: "Hello World!"})

// Output:
// Hello World!

Event Handlers

Event handlers are an abstraction over the event streams, and provide a simpler way to listen to events via function receivers.

type MyEvent struct {
    Message string 
}

dispatcher := events.NewDispatcher[MyEvent]()
id := dispatcher.AddEventHandler(func(event MyEvent) {
    fmt.Println(event.Message)
})

// dispatch an event 
dispatcher.Dispatch(MyEvent{Message: "Hello World!"})

// simulate some time passing 
time.Sleep(time.Second)

// remove the event handler when it's no longer needed
dispatcher.RemoveEventHandler(id)

You can also add new or remove existing handlers inside an event handler. A better way to write the previous example might be:

type MyEvent struct {
    Message string 
}

// handlerID should be captured by the handler 
var handlerID events.HandlerID

dispatcher := events.NewDispatcher[MyEvent]()
handlerID = dispatcher.AddEventHandler(func(event MyEvent) {
    fmt.Println(event.Message)
    dispatcher.RemoveEventHandler(handlerID)
})

// dispatch an event 
dispatcher.Dispatch(MyEvent{Message: "Hello World!"})

Filtering Events

The dispatcher supports filtering events based on a custom criteria for an event type. For example, you can have an event stream which only receives events that have a specific identifier, or events which have a Message string field that contains a specific substring. Let's use a similar example to the previous section, but with a filter:

Filtered Event Stream

type MyEvent struct {
    Message string 
}

dispatcher := events.NewDispatcher[MyEvent]()

// create a new event stream that only receives events with a Message that contains the substring "Wow"
stream := dispatcher.NewFilteredEventStream(func(event MyEvent) bool {
    return strings.Contains(event.Message, "Wow")
})

// typically, a stream will be handled in a separate goroutine
go func() {
    for event := range stream.Stream() {
        fmt.Println(event.Message)
    }
}()

// dispatch an event 
dispatcher.Dispatch(MyEvent{Message: "Hello World!"})
dispatcher.Dispatch(MyEvent{Message: "Hello! Wow!"})

// Output:
// Hello! Wow!

Note that filtered events actually prevent the event from ever dispatching over the stream, so it is more performant than adding a conditional check in a default handler for the event type.

Filtered Event Handler

Just like with event streams, you can add a filtered event handler as well by using:

type MyEvent struct {
    Message string 
}

eventHandler := func(event MyEvent) {
    fmt.Println(event.Message)
}

eventFilter := func(event MyEvent) bool { 
    return strings.Contains(event.Message, "Wow") 
}

dispatcher := events.NewDispatcher[MyEvent]()
id := dispatcher.AddFilteredEventHandler(eventHandler, eventFilter)

// dispatch an event 
dispatcher.Dispatch(MyEvent{Message: "Hello World!"})
dispatcher.Dispatch(MyEvent{Message: "Hello! Wow!"})

// simulate some time passing 
time.Sleep(time.Second)

// Remove is the same, just use the handler id provided
dispatcher.RemoveEventHandler(id)

Example

The following example is meant to demonstrate a very simplified use of the dispatcher. It receives two "types" of events, and could be re-written with filters to only receive specific events for specific Count types. It also leverages the global dispatcher for CountEvent (Note that SecondCounterTo doesn't accept a dispatcher parameter). This example could also be re-written to leverage a single instanced Dispatcher instead of the global dispatcher. Making the two previously mentioned changes may be a good programming exercise to better understand the events library.

import (
    "time"
    "fmt" 

    "github.com/kubecost/events"
)

const (
    CountEventTypeCount = "count"
    CountEventTypeFinished = "finished"
)

// CountEvent is our custom event payload
type CountEvent struct {
    Type string
    Value uint 
    Timestamp time.Time
}

// Counts up every second, dispatches an event dispatchEvery seconds 
func SecondCounterTo(target uint, dispatchEvery uint) {
    var count uint = 0

    // get a reference to the dispatcher for our count event
    dispatcher := events.GlobalDispatcherFor[CountEvent]()

    // Loop until our count reaches the target 
    for {    
        time.Sleep(time.Second)
        count++ 

        // dispatch when our count is divisble by dispatchEvery 
        if count % dispatchEvery == 0 {
            dispatcher.Dispatch(CountEvent{
                Type: CountEventTypeCount,
                Value: count,
                Timestamp: time.Now(),
            })
        }

        if count == target {
            dispatcher.Dispatch(CountEvent{
                Type: CountEventTypeFinished,
                Value: count,
                Timestamp: time.Now(),
            })
            return 
        }
    }
}

// Entry point 
func main() {
    // channel used to wait for counting complete 
    complete := make(chan bool) 

    // get a reference to the global dispatcher 
    dispatcher := events.GlobalDispatcherFor[CountEvent]()

    // create a handler that receives count events and logs them if they're Count types 
    // note that each handler has it's own goroutine context, so if you need to communicate 
    // outside of the handler, you must use channels
    dispatcher.AddEventHandler(func(event CountEvent) {
        if event.Type == CountEventTypeCount {
            fmt.Printf("Count: %d\n", event.Value)
        }
    })

    // create a handler that receives count events and waits for a Finished type
    // note that each handler has it's own goroutine context, so if you need to communicate 
    // outside of the handler, you must use channels
    dispatcher.AddEventHandler(func(event CountEvent) {
        if event.Type == CountEventTypeFinished {
            complete <- true 
        }
    })

    // run our slow counter to 10, count events every 3
    go SecondCounterTo(10, 3)

    // waits until complete is signaled 
    <- complete
}

Documentation

Index

• type Dispatcher
  • func GlobalDispatcherFor[T any]() Dispatcher[T]
  • func NewDispatcher[T any]() Dispatcher[T]
• type EventCondition
• type EventHandler
• type EventStream
• type HandlerID

Types

type Dispatcher

type Dispatcher[T any] interface {
	// Dispatch broadcasts the T event to any subscribed listeners asynchronously.
	Dispatch(event T)

	// DispatchSync is a special dispatch scenario which will block if any listeners are not yet receiving
	// events. This should be used if you need to guarantee that all receivers are processing events before
	// continuing. This method also has the added benefit of blocking until an event has been dispatched
	// over all event streams before returning.
	DispatchSync(event T)

	// AddEventHandler adds a new event handler method that is called whenever an event T is dispatched. A
	// unique HandlerID is returned that can be used to remove the handler.
	AddEventHandler(handler EventHandler[T]) HandlerID

	// AddFilteredEventHandler adds a new event handler method that is called whenever a dispatched event T
	// passes the provided condition. A unique HandlerID is returned that can be used to remove the handler.
	// Note that the condition will be checked prior to dispatch, which is more performant than filtering
	// in the event handler itself.
	AddFilteredEventHandler(handler EventHandler[T], condition EventCondition[T]) HandlerID

	// RemoveEventHandler removes an event handler that was added via AddEventHandler using it's HandlerID.
	// Returns true if the handler was successfully removed.
	RemoveEventHandler(id HandlerID) bool

	// NewEventStream returns an asynchronous event stream that can be used to receive dispatched events.
	NewEventStream() EventStream[T]

	// NewFilteredEventStream creates a new event stream that will only receive events that match the provided
	// condition. Note that the condition will be checked prior to dispatch, which is more performant than
	// filtering in the event handler itself.
	NewFilteredEventStream(condition EventCondition[T]) EventStream[T]

	// CloseEventStreams closes all listening event streams and shuts down the dispatcher
	CloseEventStreams()
}

Dispatcher[T] is an implementation prototype for an object capable of dispatching T instances to any subscribed listeners.

func GlobalDispatcherFor

func GlobalDispatcherFor[T any]() Dispatcher[T]

GlobalDispatcherFor[T] locates an existing global dispatcher for an event type, or creates a new one if one does not exist

func NewDispatcher

func NewDispatcher[T any]() Dispatcher[T]

NewDispatcher[T] creates a new multicast event dispatcher

type EventCondition

type EventCondition[T any] func(T) bool

EventCondition[T] is a type used to filter events that are dispatched to a specific handler.

type EventHandler

type EventHandler[T any] func(T)

EventHandler[T] is a type used to receive events from a dispatcher.

type EventStream

type EventStream[T any] interface {
	// Stream returns access to the event T channel where events will arrive.
	Stream() <-chan T

	// Close shuts down the event stream, closing the channel
	Close()

	// IsClosed is set to true if the event stream has been closed
	IsClosed() bool
}

EventStream[T] is an implementation prototype for an object capable of asynchronously listening for events dispatched via Dispatcher[T] implementation.

type HandlerID

type HandlerID string

HandlerID is a unique identifier assigned to a provided event handler. This is used to remove a handler from the dispatcher when it is no longer needed.