worker

Documentation

Overview

Package worker implements an event management system for publishing and subscribing to events.

Index

• Variables
• type EventManager
  • func NewEventManager(eventTypes ...entities.EventType) *EventManager
  • func (e *EventManager) Publish(ctx context.Context, event entities.Event) error
  • func (e *EventManager) Run(ctx context.Context)
  • func (e *EventManager) RunSubscription(ctx context.Context, sub Subscriber) error
  • func (e *EventManager) Stop()
  • func (e *EventManager) Subscribe(eventType entities.EventType) (id int, ch <-chan entities.Event, err error)
  • func (e *EventManager) Unsubscribe(eventType entities.EventType, id int) error
• type Subscriber

Variables

var (
	ErrUnknownEventTypeErr = errors.New("unsupported event type")
	ErrNotSubscribedErr    = errors.New("not subscribed")
)

Types

type EventManager

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

EventManager manages event publication and subscription.

func NewEventManager

func NewEventManager(eventTypes ...entities.EventType) *EventManager

NewEventManager creates a new EventManager for the given event types.

The EventManager facilitates the publication and subscription of events. It ensures that only supported event types are processed and provides thread-safe methods for managing subscriptions and event delivery.

Parameters: - eventTypes: A variadic list of entities.EventType values that the manager should support.

Returns: - A pointer to the newly created EventManager instance.

Example usage:

em := NewEventManager(
	entities.EventType("UserCreated"),
	entities.EventType("UserDeleted"),
)

func (*EventManager) Publish

func (e *EventManager) Publish(ctx context.Context, event entities.Event) error

Publish sends an event to all subscribers of its type.

This method ensures that the event is only published if its type is supported by the EventManager. Events are delivered asynchronously to prevent blocking the publisher.

Parameters: - event: The event to be published. Must implement the entities.Event interface.

Returns: - An error if the event type is unsupported.

Example usage:

err := em.Publish(myEvent)
if err != nil {
	log.Fatalf("Failed to publish event: %v", err)
}

func (*EventManager) Run

func (e *EventManager) Run(ctx context.Context)

Run processes events and sends them to appropriate subscribers.

This is a blocking method and should be run in a separate goroutine. It listens for published events and distributes them to subscribers of matching types.

Parameters: - ctx: A context to control the lifecycle of the Run method. Canceling the context will stop event processing.

Example usage:

ctx, cancel := context.WithCancel(context.Background())
go em.Run(ctx)
// Perform operations...
cancel()

func (*EventManager) RunSubscription

func (e *EventManager) RunSubscription(ctx context.Context, sub Subscriber) error

RunSubscription manages a single subscription, forwarding events to the Subscriber.

This is a blocking method and should be run in a separate goroutine. It allows a Subscriber implementation to process events in its own context. It ensures that the subscription is cleaned up when the context is canceled or an error occurs.

Parameters: - ctx: A context to control the lifecycle of the subscription. - sub: The Subscriber implementation to handle events of a specific type.

Returns: - An error if event handling fails or the subscription encounters an issue.

Example usage:

subscriber := MySubscriber{}
go func() {
	err := em.RunSubscription(ctx, subscriber)
	if err != nil {
		log.Printf("Subscription error: %v", err)
	}
}()

func (*EventManager) Stop

func (e *EventManager) Stop()

Stop stops the EventManager by closing its event channel.

This method should be called to gracefully terminate the EventManager's operations.

func (*EventManager) Subscribe

func (e *EventManager) Subscribe(eventType entities.EventType) (id int, ch <-chan entities.Event, err error)

Subscribe registers a new subscription for the specified event type.

A subscription allows a caller to receive events of a specific type via a dedicated channel. The caller must manage the lifecycle of the channel.

Parameters: - eventType: The type of events to subscribe to.

Returns: - A unique subscription ID. - A channel to receive events of the specified type. - An error if the event type is unsupported.

Example usage:

id, ch, err := em.Subscribe(entities.EventType("UserCreated"))
if err != nil {
	log.Fatalf("Failed to subscribe: %v", err)
}

func (*EventManager) Unsubscribe

func (e *EventManager) Unsubscribe(eventType entities.EventType, id int) error

Unsubscribe removes a subscription identified by its event type and ID.

This method ensures that the channel associated with the subscription is closed and the subscription is removed from the internal registry.

Parameters: - eventType: The type of the event subscription. - id: The unique ID of the subscription to be removed.

Returns: - An error if the event type is unsupported or the subscription ID is not found.

Example usage:

err := em.Unsubscribe(entities.EventType("UserCreated"), subscriptionID)
if err != nil {
	log.Printf("Failed to unsubscribe: %v", err)
}

type Subscriber

type Subscriber interface {
	// HandleEvent processes the received event.
	HandleEvent(ctx context.Context, event entities.Event) error

	// EventType returns the type of events this subscriber is interested in.
	EventType() entities.EventType
}

Subscriber defines the interface for handling events of a specific type.