messaging

Documentation

Index

• Variables
• func NewWatermillLogger() watermill.LoggerAdapter
• func WithBaggage() message.HandlerMiddleware
• func WithTracePropagation() message.HandlerMiddleware
• type BrokerCQRS
  • func NewBrokerCQRS(opts ...Option) (BrokerCQRS, error)
• type Option
  • func WithLogger(logger watermill.LoggerAdapter) Option
• type PubSub
• type WatermillLogger
  • func (l *WatermillLogger) Debug(msg string, fields watermill.LogFields)
  • func (l *WatermillLogger) Error(_ string, err error, fields watermill.LogFields)
  • func (l *WatermillLogger) Info(msg string, fields watermill.LogFields)
  • func (l *WatermillLogger) Trace(msg string, fields watermill.LogFields)
  • func (l *WatermillLogger) With(fields watermill.LogFields) watermill.LoggerAdapter

Variables

var (
	// ErrBuildRouter is returned when failed to build router.
	ErrBuildRouter errors.CustomError = errors.New("failed to build router").
					WithCode("BUILD_ROUTER_ERROR").
					WithKind(errors.KindInternal)

	// ErrBuildCommandBus is returned when failed to build command bus.
	ErrBuildCommandBus errors.CustomError = errors.New("failed to build command bus").
						WithCode("BUILD_COMMAND_BUS_ERROR").
						WithKind(errors.KindInternal)

	// ErrBuildCommandProcessor is returned when failed to build command processor.
	ErrBuildCommandProcessor errors.CustomError = errors.New("failed to build command processor").
								WithCode("BUILD_COMMAND_PROCESSOR_ERROR").
								WithKind(errors.KindInternal)

	// ErrBuildEventBus is returned when failed to build event bus.
	ErrBuildEventBus errors.CustomError = errors.New("failed to build event bus").
						WithCode("BUILD_EVENT_BUS_ERROR").
						WithKind(errors.KindInternal)

	// ErrBuildEventProcessor is returned when failed to build event processor.
	ErrBuildEventProcessor errors.CustomError = errors.New("failed to build event processor").
							WithCode("BUILD_EVENT_PROCESSOR_ERROR").
							WithKind(errors.KindInternal)

	// ErrAddCommandHandlers is returned when failed to add command handlers into command processor.
	ErrAddCommandHandlers errors.CustomError = errors.New("failed to add command handlers into command processor").
							WithCode("ADD_COMMAND_HANDLERS_ERROR").
							WithKind(errors.KindInternal)

	// ErrAddEventHandlers is returned when failed to add event handlers into event processor.
	ErrAddEventHandlers errors.CustomError = errors.New("failed to add event handlers into event processor").
						WithCode("ADD_EVENT_HANDLERS_ERROR").
						WithKind(errors.KindInternal)

	// ErrRunRouter is returned when failed to run router.
	ErrRunRouter errors.CustomError = errors.New("something went wrong with Message Router").
					WithCode("RUN_ROUTER_ERROR").
					WithKind(errors.KindInternal)

	// ErrSendCommand is returned when failed to send command to the command bus.
	ErrSendCommand errors.CustomError = errors.New("failed to send command to the command bus").
					WithCode("SEND_COMMAND_ERROR").
					WithKind(errors.KindInternal)

	// ErrSendEvent is returned when failed to send event to the event bus.
	ErrSendEvent errors.CustomError = errors.New("failed to send event to the event bus").
					WithCode("SEND_EVENT_ERROR").
					WithKind(errors.KindInternal)

	// ErrCreateBaggage is returned when failed to create baggage.
	ErrCreateBaggage errors.CustomError = errors.New("failed to create baggage").
						WithCode("CREATE_BAGGAGE_ERROR").
						WithKind(errors.KindInternal)

	// ErrCreateBaggageMember is returned when failed to create baggage member.
	ErrCreateBaggageMember errors.CustomError = errors.New("failed to create baggage member").
							WithCode("CREATE_BAGGAGE_MEMBER_ERROR").
							WithKind(errors.KindInternal)

	// ErrSetBaggageMember is returned when failed to set baggage member.
	ErrSetBaggageMember errors.CustomError = errors.New("failed to set baggage member").
						WithCode("SET_BAGGAGE_MEMBER_ERROR").
						WithKind(errors.KindInternal)
)

Functions

func NewWatermillLogger

func NewWatermillLogger() watermill.LoggerAdapter

NewWatermillLogger creates a new WatermillLogger instance.

func WithBaggage

func WithBaggage() message.HandlerMiddleware

WithBaggage is a middleware that adds baggage to the message metadata.

func WithTracePropagation

func WithTracePropagation() message.HandlerMiddleware

WithTracePropagation is a middleware that propagates the trace context to the message context.

Types

type BrokerCQRS

type BrokerCQRS interface {
	// Start starts the broker.
	// All Command and Event handlers must be added before calling this method.
	Start(ctx context.Context) error

	// Stop stops the broker.
	Stop(ctx context.Context) error

	// Running is closed when broker is running.
	// In other words: you can wait till broker is running using
	//
	//	fmt.Println("Starting broker")
	//	go r.Run(ctx)
	//	<- r.Running()
	//	fmt.Println("Broker is running")
	//
	// Warning: for historical reasons, this channel is not aware of broker closing.
	// The channel will be closed if the broker has been running and closed.
	Running(ctx context.Context) chan struct{}

	// AddCommandHandlers adds command handlers to the command processor.
	AddCommandHandlers(ctx context.Context, handlers ...cqrs.CommandHandler) error

	// AddEventHandlers adds event handlers to the event processor.
	AddEventHandlers(ctx context.Context, handlers ...cqrs.EventHandler) error

	// SendCommand sends a command to the command bus.
	SendCommand(ctx context.Context, command any) error

	// SendEvent sends an event to the event bus.
	SendEvent(ctx context.Context, event any) error
}

BrokerCQRS is a contract for a message broker that implements the CQRS pattern.

func NewBrokerCQRS

func NewBrokerCQRS(opts ...Option) (BrokerCQRS, error)

NewBrokerCQRS creates a new BrokerCQRS instance.

type Option

type Option func(broker *brokerCQRS)

Option is a function that configures the BrokerCQRS.

func WithLogger

func WithLogger(logger watermill.LoggerAdapter) Option

WithLogger configures the logger.

type PubSub

type PubSub interface {
	message.Publisher
	message.Subscriber
}

PubSub is a contract for a message broker that implements both Publisher and Subscriber.

type WatermillLogger

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

WatermillLogger is a logger adapter for the Watermill library.

func (*WatermillLogger) Debug

func (l *WatermillLogger) Debug(msg string, fields watermill.LogFields)

Debug logs a debug message with the given fields.

func (*WatermillLogger) Error

func (l *WatermillLogger) Error(_ string, err error, fields watermill.LogFields)

Error logs an error message with the given fields.

func (*WatermillLogger) Info

func (l *WatermillLogger) Info(msg string, fields watermill.LogFields)

Info logs an info message with the given fields.

func (*WatermillLogger) Trace

func (l *WatermillLogger) Trace(msg string, fields watermill.LogFields)

Trace logs a trace message with the given fields.

func (*WatermillLogger) With

func (l *WatermillLogger) With(fields watermill.LogFields) watermill.LoggerAdapter

With adds fields to be logged in all log messages.