Queues and Asynchronous Processing

Motivation

The go-events library provides a framework for building asynchronous, event-driven systems using queues.

The library was built based on the following design principles:

• Simplicity: It should be easy to use and integrate into existing applications.
• Flexibility: It should support different queue backends and message formats.
• Scalability: It should be able to handle high volumes of events.
• Reliability: It should ensure that events are processed reliably, even in the event of failures.

Core Concepts

Queues

Queues are used to store events that need to be processed asynchronously. The library supports the following queue backends:

• Redis: pkg/queue/redis/redis_queue.go
• NATS: pkg/queue/nats/nats_queue.go
• Kafka: pkg/queue/kafka/kafka_queue.go
• InMemory: pkg/queue/inmemory/inmemory_queue.go

Event Processors

Event processors are responsible for consuming events from queues and processing them. The library provides a simple interface for defining event processors:

type EventProcessor interface {
              Process(event *Event) error
          }
          

Event Handlers

Event handlers are responsible for handling specific types of events. The library provides a mechanism for registering event handlers:

func (e *Events) RegisterHandler(eventTypeName string, handler EventProcessor) error
          

Event Dispatcher

The event dispatcher is responsible for dispatching events to the appropriate event handlers. The library provides a default event dispatcher:

type EventDispatcher interface {
              Dispatch(event *Event) error
          }
          

Usage

Creating a Queue

The following example shows how to create a Redis queue:

import (
              "context"
          
              "github.com/docker/go-events/pkg/queue"
              "github.com/docker/go-events/pkg/queue/redis"
          )
          
          func main() {
              ctx := context.Background()
              queue, err := redis.NewRedisQueue(ctx, "redis://localhost:6379", "my-queue")
              if err != nil {
                  // Handle error
              }
          
              // Use the queue
          }
          

Creating an Event Processor

The following example shows how to create an event processor that logs events to the console:

import (
              "fmt"
          
              "github.com/docker/go-events/pkg/event"
          )
          
          type LogEventProcessor struct{}
          
          func (p *LogEventProcessor) Process(event *event.Event) error {
              fmt.Printf("Received event: %+v\n", event)
              return nil
          }
          

Registering an Event Handler

The following example shows how to register an event handler for events of type “user.created”:

import (
              "github.com/docker/go-events/pkg/event"
          )
          
          func main() {
              events := NewEvents()
              err := events.RegisterHandler("user.created", &LogEventProcessor{})
              if err != nil {
                  // Handle error
              }
          
              // Use the events instance
          }
          

Dispatching an Event

The following example shows how to dispatch an event to the queue:

import (
              "context"
          
              "github.com/docker/go-events/pkg/event"
              "github.com/docker/go-events/pkg/queue"
          )
          
          func main() {
              ctx := context.Background()
              queue, err := redis.NewRedisQueue(ctx, "redis://localhost:6379", "my-queue")
              if err != nil {
                  // Handle error
              }
          
              // Create an event
              event := event.NewEvent("user.created", map[string]interface{}{
                  "userId": "123",
                  "name":   "John Doe",
              })
          
              // Dispatch the event
              err = queue.Enqueue(ctx, event)
              if err != nil {
                  // Handle error
              }
          }
          

Consuming Events

The following example shows how to consume events from the queue:

import (
              "context"
          
              "github.com/docker/go-events/pkg/event"
              "github.com/docker/go-events/pkg/queue"
          )
          
          func main() {
              ctx := context.Background()
              queue, err := redis.NewRedisQueue(ctx, "redis://localhost:6379", "my-queue")
              if err != nil {
                  // Handle error
              }
          
              // Create an event processor
              processor := &LogEventProcessor{}
          
              // Consume events from the queue
              for {
                  event, err := queue.Dequeue(ctx)
                  if err != nil {
                      // Handle error
                  }
          
                  // Process the event
                  err = processor.Process(event)
                  if err != nil {
                      // Handle error
                  }
              }
          }
          

Additional Information

• The library provides several configuration options for customizing queue behavior, including:
  • Queue name: This is the name of the queue to use.
  • Connection string: This is the connection string to use for the queue backend.
  • Message format: This specifies the format of messages to be stored in the queue.
• The library provides a mechanism for handling errors during event processing.
• The library supports asynchronous event processing, which allows you to process events in the background without blocking the main thread.

Contributing

Contributions to the go-events library are welcome! Please see the CONTRIBUTING.md for more information.