events

Documentation

Overview

Package event provides event handling functionality for the hop framework. It includes an event bus for publishing and subscribing to events, as well as helper functions for working with typed event payloads.

Index

• func IsPayloadType[T any](e Event) bool
• func MustPayloadAs[T any](e Event) T
• func PayloadAs[T any](e Event) (T, error)
• func PayloadAsMap(e Event) (map[string]any, error)
• func PayloadAsSlice(e Event) ([]any, error)
• func PayloadMapAs[T any](e Event) (map[string]T, error)
• func PayloadSliceAs[T any](e Event) ([]T, error)
• type Bus
  • func NewEventBus(logger *slog.Logger) *Bus
  • func (b *Bus) Emit(ctx context.Context, signature string, payload any)
  • func (b *Bus) EmitSync(ctx context.Context, signature string, payload any)
  • func (b *Bus) On(signature string, handler Handler)
• type Event
  • func NewEvent(signature string, payload any) Event
• type Handler
  • func HandlePayload[T any](handler func(context.Context, T)) Handler

Examples

• Bus (Basic)
• Bus (ContextCancellation)
• Bus (MultipleHandlers)
• Bus (SyncEmit)
• Bus (Wildcards)
• HandlePayload
• NewEvent
• PayloadAs
• PayloadAsMap
• PayloadAsSlice
• PayloadMapAs
• PayloadSliceAs

Functions

func IsPayloadType

func IsPayloadType[T any](e Event) bool

IsPayloadType checks if an event's payload is of the specified type T.

func MustPayloadAs

func MustPayloadAs[T any](e Event) T

MustPayloadAs converts an event's payload to the specified type T. Panics if the conversion fails.

func PayloadAs

func PayloadAs[T any](e Event) (T, error)

PayloadAs safely converts an event's payload to the specified type T. Returns the typed payload and any conversion error.

Example

package main

import (
	"fmt"

	"github.com/patrickward/hop/events"
)

func main() {
	// Example event with a structured payload
	type UserCreated struct {
		ID   string
		Name string
	}

	evt := events.NewEvent("user.created", UserCreated{
		ID:   "123",
		Name: "John Doe",
	})

	// Safe conversion with error handling
	user, err := events.PayloadAs[UserCreated](evt)
	if err != nil {
		fmt.Printf("Error: %v\n", err)
		return
	}
	fmt.Printf("User created: %s\n", user.Name)
}

Output:

User created: John Doe

func PayloadAsMap

func PayloadAsMap(e Event) (map[string]any, error)

PayloadAsMap is a convenience function for working with map[string]any payloads, which are common when dealing with JSON data.

Example

package main

import (
	"fmt"

	"github.com/patrickward/hop/events"
)

func main() {
	// Create an event with a map payload
	evt := events.NewEvent("config.updated", map[string]any{
		"database": "postgres",
		"port":     5432,
	})

	// Use the convenience function for map payloads
	config, err := events.PayloadAsMap(evt)
	if err != nil {
		fmt.Printf("Error: %v\n", err)
		return
	}

	if dbName, ok := config["database"].(string); ok {
		fmt.Printf("Database: %s\n", dbName)
	}
}

Output:

Database: postgres

func PayloadAsSlice

func PayloadAsSlice(e Event) ([]any, error)

PayloadAsSlice is a convenience function for working with []any payloads.

Example

package main

import (
	"fmt"

	"github.com/patrickward/hop/events"
)

func main() {
	// Create an event with a slice payload
	evt := events.NewEvent("users.updated", []any{
		"john",
		"jane",
	})

	// Use the convenience function for slice payloads
	users, err := events.PayloadAsSlice(evt)
	if err != nil {
		fmt.Printf("Error: %v\n", err)
		return
	}

	fmt.Printf("Users: %v\n", users)

}

Output:

Users: [john jane]

func PayloadMapAs

func PayloadMapAs[T any](e Event) (map[string]T, error)

PayloadMapAs converts a map payload into a map with typed values. Returns an error if the payload is not a map or if any value cannot be converted to type T.

Example:

type User struct { ID string }
userMap, err := PayloadMapAs[User](event)

Example

package main

import (
	"fmt"
	"sort"
	"sync"

	"github.com/patrickward/hop/events"
)

func main() {
	// Define a type we want to convert to
	type Region struct {
		Code string
		Name string
	}

	var mu sync.Mutex
	var results []string

	// Create an event with a map payload
	evt := events.NewEvent("regions.updated", map[string]any{
		"us-east": Region{Code: "USE", Name: "US East"},
		"us-west": Region{Code: "USW", Name: "US West"},
	})

	// Convert the payload to a map of Regions
	regions, err := events.PayloadMapAs[Region](evt)
	if err != nil {
		fmt.Printf("Error: %v\n", err)
		return
	}

	// Process the typed regions
	for key, region := range regions {
		mu.Lock()
		results = append(results, fmt.Sprintf("Region %s: %s (%s)", key, region.Name, region.Code))
		mu.Unlock()
	}

	// Sort and print results for consistent output
	sort.Strings(results)
	for _, result := range results {
		fmt.Println(result)
	}

}

Output:

Region us-east: US East (USE)
Region us-west: US West (USW)

func PayloadSliceAs

func PayloadSliceAs[T any](e Event) ([]T, error)

PayloadSliceAs converts a slice payload into a slice of typed elements. Returns an error if the payload is not a slice or if any element cannot be converted to type T.

Example:

type User struct { ID string }
users, err := PayloadSliceAs[User](event)

Example

package main

import (
	"fmt"

	"github.com/patrickward/hop/events"
)

func main() {
	// Define a type we want to convert to
	type User struct {
		ID   string
		Name string
	}

	// Create an event with a slice payload
	evt := events.NewEvent("users.imported", []any{
		User{ID: "1", Name: "Alice"},
		User{ID: "2", Name: "Bob"},
	})

	// Convert the payload to a slice of Users
	users, err := events.PayloadSliceAs[User](evt)
	if err != nil {
		fmt.Printf("Error: %v\n", err)
		return
	}

	// Process the typed users
	for _, user := range users {
		fmt.Printf("User: %s (ID: %s)\n", user.Name, user.ID)
	}
}

Output:

User: Alice (ID: 1)
User: Bob (ID: 2)

Types

type Bus

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

Bus manages event publishing and subscription

Example (Basic)

package main

import (
	"context"
	"fmt"
	"log/slog"
	"os"
	"time"

	"github.com/patrickward/hop/events"
)

func main() {
	// Create a new event bus with a basic logger
	logger := slog.New(slog.NewTextHandler(os.Stdout, nil))
	bus := events.NewEventBus(logger)

	// Register an event handler
	bus.On("user.login", func(ctx context.Context, event events.Event) {
		payload := event.Payload.(map[string]string)
		fmt.Printf("User logged in: %s\n", payload["username"])
	})

	// Emit an event
	bus.Emit(context.Background(), "user.login", map[string]string{
		"username": "alice",
	})

	// Wait for async handler to complete
	time.Sleep(10 * time.Millisecond)
}

Output:

User logged in: alice

Example (ContextCancellation)

package main

import (
	"context"
	"fmt"
	"log/slog"
	"os"
	"time"

	"github.com/patrickward/hop/events"
)

func main() {
	logger := slog.New(slog.NewTextHandler(os.Stdout, nil))
	bus := events.NewEventBus(logger)

	// Create a context with cancellation
	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
	defer cancel()

	// Register a handler that respects context cancellation
	bus.On("long.task", func(ctx context.Context, event events.Event) {
		select {
		case <-ctx.Done():
			fmt.Println("Task cancelled")
			return
		case <-time.After(100 * time.Millisecond):
			fmt.Println("Task completed")
		}
	})

	// Emit event with cancellable context
	bus.EmitSync(ctx, "long.task", nil)
}

Output:

Task cancelled

Example (MultipleHandlers)

package main

import (
	"context"
	"fmt"
	"log/slog"
	"os"
	"sort"
	"sync"

	"github.com/patrickward/hop/events"
)

func main() {
	logger := slog.New(slog.NewTextHandler(os.Stdout, nil))
	bus := events.NewEventBus(logger)

	var mu sync.Mutex
	var results []string

	// Register multiple handlers for the same event
	bus.On("notification.sent", func(ctx context.Context, event events.Event) {
		mu.Lock()
		results = append(results, "Logging notification")
		mu.Unlock()
	})

	bus.On("notification.sent", func(ctx context.Context, event events.Event) {
		mu.Lock()
		results = append(results, "Sending analytics")
		mu.Unlock()
	})

	bus.On("notification.sent", func(ctx context.Context, event events.Event) {
		mu.Lock()
		results = append(results, "Updating cache")
		mu.Unlock()
	})

	// Emit event synchronously - all handlers will be called
	bus.EmitSync(context.Background(), "notification.sent", nil)

	// Sort and print results
	sort.Strings(results)
	for _, result := range results {
		fmt.Println(result)
	}

}

Output:

Logging notification
Sending analytics
Updating cache

Example (SyncEmit)

package main

import (
	"context"
	"fmt"
	"log/slog"
	"os"
	"time"

	"github.com/patrickward/hop/events"
)

func main() {
	logger := slog.New(slog.NewTextHandler(os.Stdout, nil))
	bus := events.NewEventBus(logger)

	// Register event handlers
	bus.On("task.process", func(ctx context.Context, event events.Event) {
		fmt.Println("Processing task...")
		time.Sleep(10 * time.Millisecond)
		fmt.Println("Task completed")
	})

	// EmitSync will wait for all handlers to complete
	fmt.Println("Starting task")
	bus.EmitSync(context.Background(), "task.process", nil)
	fmt.Println("All processing complete")

}

Output:

Starting task
Processing task...
Task completed
All processing complete

Example (Wildcards)

package main

import (
	"context"
	"fmt"
	"log/slog"
	"os"
	"sort"
	"sync"

	"github.com/patrickward/hop/events"
)

func main() {
	logger := slog.New(slog.NewTextHandler(os.Stdout, nil))
	bus := events.NewEventBus(logger)

	var mu sync.Mutex
	var results []string

	// Register handlers with wildcards
	bus.On("user.*", func(ctx context.Context, event events.Event) {
		mu.Lock()
		results = append(results, fmt.Sprintf("User event: %s", event.Signature))
		mu.Unlock()
	})

	bus.On("*.created", func(ctx context.Context, event events.Event) {
		mu.Lock()
		results = append(results, fmt.Sprintf("Created event: %s", event.Signature))
		mu.Unlock()
	})

	// Emit events synchronously
	bus.EmitSync(context.Background(), "user.created", nil)
	bus.EmitSync(context.Background(), "user.deleted", nil)
	bus.EmitSync(context.Background(), "post.created", nil)

	// Sort and print results
	sort.Strings(results)
	for _, result := range results {
		fmt.Println(result)
	}

}

Output:

Created event: post.created
Created event: user.created
User event: user.created
User event: user.deleted

func NewEventBus

func NewEventBus(logger *slog.Logger) *Bus

NewEventBus creates a new event bus

func (*Bus) Emit

func (b *Bus) Emit(ctx context.Context, signature string, payload any)

Emit sends an event to all registered handlers asynchronously

func (*Bus) EmitSync

func (b *Bus) EmitSync(ctx context.Context, signature string, payload any)

EmitSync sends an event and waits for all handlers to complete

func (*Bus) On

func (b *Bus) On(signature string, handler Handler)

On registers a handler for an event signature Supports wildcards: "hop.*" or "*.system.start"

type Event

type Event struct {
	ID        string    `json:"id"`
	Signature string    `json:"signature"` // e.g. "hop.system.start"
	Payload   any       `json:"payload,omitempty"`
	Timestamp time.Time `json:"timestamp"`
}

Event represents a system event with a simplified structure

func NewEvent

func NewEvent(signature string, payload any) Event

NewEvent creates an event with the given signature and optional payload

Example

package main

import (
	"fmt"

	"github.com/patrickward/hop/events"
)

func main() {
	// Create a new event with a payload
	evt := events.NewEvent("user.created", map[string]string{
		"id":    "123",
		"email": "user@example.com",
	})

	fmt.Printf("Signature: %s\n", evt.Signature)
}

Output:

Signature: user.created

type Handler

type Handler func(ctx context.Context, event Event)

Handler processes an event

func HandlePayload

func HandlePayload[T any](handler func(context.Context, T)) Handler

HandlePayload creates an event handler that automatically converts the payload to the specified type T and calls the provided typed handler function. If type conversion fails, logs the error and returns without calling the handler.

Example

type UserCreated struct {
	ID   string
	Name string
}

logger := newTestLogger(os.Stderr)
// Create a test event bus
bus := events.NewEventBus(logger) // You'd normally pass a logger here

// Register handler with automatic payload conversion
bus.On("user.created", events.HandlePayload[UserCreated](func(ctx context.Context, user UserCreated) {
	fmt.Printf("Processing user: %s\n", user.Name)
}))

// Emit an event
ctx := context.Background()
bus.EmitSync(ctx, "user.created", UserCreated{
	ID:   "123",
	Name: "John Doe",
})

Output:

Processing user: John Doe