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.

Basic Use

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 our dispatcher 
    dispatcher := events.DispatcherFor[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 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.
	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

	// 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]

	// 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 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.