broker

Documentation

Index

• Constants
• func NewContext(ctx context.Context, b Broker) context.Context
• func ToContext(c Setter, b Broker)
• type AMQPBroker
  • func NewAMQPBroker(cfg configuration.Broker) (*AMQPBroker, error)
  • func (broker *AMQPBroker) Publish(ctx context.Context, ev Event) error
  • func (broker *AMQPBroker) Run(ctx context.Context, handlers map[string]Handler) error
  • func (broker *AMQPBroker) Stop()
• type AMQPConsumer
  • func NewAMQPConsumer(ctx context.Context, pool amqpx.Pooler, queueName string, handler Handler) (*AMQPConsumer, error)
  • func (consumer *AMQPConsumer) Start(ctx context.Context) error
  • func (consumer *AMQPConsumer) Stop() error
• type Broker
  • func FromContext(c context.Context) Broker
  • func Load(cfg configuration.Broker) (Broker, error)
• type Event
• type Handler
• type Setter
• type UserCreatedEvent
  • func (UserCreatedEvent) Name() string
  • func (e *UserCreatedEvent) ToBytes() ([]byte, error)

Constants

const (
	// WarpExchange transfer expired delayed messages to direct queue
	// We use default amq.direct exchange for convenience
	WarpExchange = "amq.direct"
)

Functions

func NewContext

func NewContext(ctx context.Context, b Broker) context.Context

func ToContext

func ToContext(c Setter, b Broker)

ToContext adds the Config to this context if it supports the Setter interface.

Types

type AMQPBroker

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

AMQPBroker service

func NewAMQPBroker

func NewAMQPBroker(cfg configuration.Broker) (*AMQPBroker, error)

NewAMQPBroker creates an AMQP broker

func (*AMQPBroker) Publish

func (broker *AMQPBroker) Publish(ctx context.Context, ev Event) error

Publish emits a message to a given topic

func (*AMQPBroker) Run

func (broker *AMQPBroker) Run(ctx context.Context, handlers map[string]Handler) error

func (*AMQPBroker) Stop

func (broker *AMQPBroker) Stop()

Publish emits a message to a given topic

type AMQPConsumer

type AMQPConsumer struct {
	Channel   *amqp.Channel
	QueueName string
	Consumer  string
	Handler   Handler
	// contains filtered or unexported fields
}

AMQPConsumer listen on a given queue for AMQP events

func NewAMQPConsumer

func NewAMQPConsumer(ctx context.Context, pool amqpx.Pooler, queueName string, handler Handler) (*AMQPConsumer, error)

NewAMQPConsumer returns an AMQP Consumer

func (*AMQPConsumer) Start

func (consumer *AMQPConsumer) Start(ctx context.Context) error

Start launch events listening

func (*AMQPConsumer) Stop

func (consumer *AMQPConsumer) Stop() error

Stop interrupts events listening

type Broker

type Broker interface {
	Publish(ctx context.Context, event Event) error
	Run(ctx context.Context, handlers map[string]Handler) error
	Stop()
}

func FromContext

func FromContext(c context.Context) Broker

FromContext returns the Config associated with this context.

func Load

func Load(cfg configuration.Broker) (Broker, error)

type Event

type Event interface {
	Name() string
	ToBytes() ([]byte, error)
}

type Handler

type Handler func(ctx context.Context, message []byte)

type Setter

type Setter interface {
	Set(string, interface{})
}

Setter defines a context that enables setting values.

type UserCreatedEvent

type UserCreatedEvent struct {
	Username string `json:"username"`
}

func (UserCreatedEvent) Name

func (UserCreatedEvent) Name() string

func (*UserCreatedEvent) ToBytes

func (e *UserCreatedEvent) ToBytes() ([]byte, error)