amqp

README

eventsourced AMQP event publisher

Package amqp provides an AMQP implementation of eventsourced.EventPublisher using the goamqp library.

Install

go get gitlab.com/unboundsoftware/eventsourced/amqp

Prerequisites

• An AMQP broker (e.g., RabbitMQ)
• The goamqp library

Getting Started

import (
    "github.com/sparetimecoders/goamqp"

    "gitlab.com/unboundsoftware/eventsourced/amqp"
    "gitlab.com/unboundsoftware/eventsourced/eventsourced"
)

// Create a goamqp publisher
goamqpPublisher := goamqp.NewPublisher()

// Create the AMQP event publisher
publisher, err := amqp.New(goamqpPublisher,
    amqp.WithTraceHandler(tracer),
)
if err != nil {
    log.Fatal(err)
}

// Use with a command handler
handler, err := eventsourced.NewHandler(ctx, aggregate, store,
    eventsourced.WithEventPublisher(publisher),
)

BackingPublisher

The BackingPublisher interface abstracts the underlying AMQP client:

type BackingPublisher interface {
    PublishWithContext(ctx context.Context, msg interface{}, headers ...goamqp.Header) error
}

The goamqp.Publisher type satisfies this interface.

External Event Unwrapping

When publishing an eventsourced.ExternalEvent, the publisher automatically unwraps it and publishes the inner event. This ensures that downstream consumers receive the original event type rather than the external wrapper.

Configuration Options

Option	Default	Description
WithTraceHandler	no-op	Add distributed tracing to publish operations

Documentation

Overview

Package amqp provides an AMQP-backed implementation of eventsourced.EventPublisher using the goamqp library.

Events are published to an AMQP exchange via a BackingPublisher. For events wrapped in eventsourced.ExternalEvent, the inner event is published instead of the wrapper.

Usage

goamqpPublisher := goamqp.NewPublisher()

publisher, err := amqp.New(goamqpPublisher,
    amqp.WithTraceHandler(tracer),
)
if err != nil {
    log.Fatal(err)
}

handler, err := eventsourced.NewHandler(ctx, aggregate, store,
    eventsourced.WithEventPublisher(publisher),
)

Configuration

Use functional options to configure the publisher:

• WithTraceHandler: Add distributed tracing to publish operations

Index

• Variables
• type Amqp
  • func New(p BackingPublisher, opts ...Option) (*Amqp, error)
  • func (a *Amqp) Publish(ctx context.Context, event eventsourced.Event) error
• type BackingPublisher
• type Option
  • func WithTraceHandler(traceHandler eventsourced.TraceHandler) Option
  • func WithTypeMapping(routingKey string, msgType any) Option

Variables

var ErrNoRoutingKey = fmt.Errorf("no routing key configured for message type")

ErrNoRoutingKey is returned when publishing an event whose type has no routing key registered via WithTypeMapping.

Types

type Amqp

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

Amqp is an implementation of eventsourced.EventPublisher that publishes events using AMQP via the go-messaging-amqp library. It supports distributed tracing.

func New

func New(p BackingPublisher, opts ...Option) (*Amqp, error)

New creates an AMQP event publisher using the provided backing publisher. Options register type-to-routing-key mappings (WithTypeMapping) and configure tracing (WithTraceHandler).

func (*Amqp) Publish

func (a *Amqp) Publish(ctx context.Context, event eventsourced.Event) error

Publish sends an event to the AMQP exchange using the routing key registered for its type via WithTypeMapping. If the event is an eventsourced.ExternalEvent, the inner wrapped event is published instead of the wrapper, since the wrapper is an internal representation.

type BackingPublisher

type BackingPublisher interface {
	Publish(ctx context.Context, routingKey string, msg any, headers ...goamqp.Header) error
}

BackingPublisher is the interface for the underlying AMQP publisher that sends messages to the exchange with an explicit routing key. This is satisfied by goamqp.Publisher.

type Option

type Option func(amqp *Amqp) error

Option is used to modify the publisher

func WithTraceHandler

func WithTraceHandler(traceHandler eventsourced.TraceHandler) Option

WithTraceHandler sets the trace handler to use on the publisher

func WithTypeMapping

func WithTypeMapping(routingKey string, msgType any) Option

WithTypeMapping registers the routing key used when publishing events of the given message type. Pointer types are dereferenced, so a mapping registered on either the value or the pointer type matches events published as either. Registering the same type twice is an error.