activemq

Documentation

Overview

Package activemq provides an ActiveMQ backend for the ling-base/mq broker abstraction. It communicates with the broker using the STOMP protocol (Streaming Text Oriented Messaging Protocol) via the go-stomp/stomp client library.

ActiveMQ auto-creates destinations (queues/topics) on first use, so the topology-management methods on Broker (DeclareExchange, DeclareQueue, Bind, etc.) are no-ops and always return nil.

Basic usage

broker, _ := activemq.New(activemq.DefaultConfig())
_ = broker.Connect()
defer broker.Close()

producer, _ := broker.Producer("/queue/events", mq.PublishOptions{})
_ = producer.Publish(ctx, &mq.Message{Body: []byte("hello")})

consumer, _ := broker.Consumer("/queue/events", mq.ConsumeOptions{
    Handler: func(ctx context.Context, d mq.Delivery) error {
        fmt.Println(string(d.Body()))
        return d.Ack()
    },
})
_ = consumer.Start(ctx)

Index

• type Broker
  • func New(cfg Config) (*Broker, error)
  • func (b *Broker) Bind(queue, exchange, routingKey string) error
  • func (b *Broker) Close() error
  • func (b *Broker) Connect() error
  • func (b *Broker) Consumer(destination string, opts mq.ConsumeOptions) (mq.Consumer, error)
  • func (b *Broker) DeclareExchange(name string, opts mq.ExchangeOptions) error
  • func (b *Broker) DeclareQueue(name string, opts mq.QueueOptions) error
  • func (b *Broker) DeleteExchange(name string) error
  • func (b *Broker) DeleteQueue(name string) error
  • func (b *Broker) IsConnected() bool
  • func (b *Broker) Metrics() mq.Metrics
  • func (b *Broker) Producer(destination string, opts mq.PublishOptions) (mq.Producer, error)
  • func (b *Broker) Unbind(queue, exchange, routingKey string) error
• type Config
  • func DefaultConfig() Config
• type Consumer
  • func (c *Consumer) IsRunning() bool
  • func (c *Consumer) Start(ctx context.Context) error
  • func (c *Consumer) Stop(timeout time.Duration) error
• type Producer
  • func (p *Producer) Close() error
  • func (p *Producer) Publish(ctx context.Context, msg *mq.Message) error

Types

type Broker

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

Broker implements mq.Broker for ActiveMQ over STOMP.

func New

func New(cfg Config) (*Broker, error)

New creates a new ActiveMQ broker. The connection is not established until Connect is called.

func (*Broker) Bind

func (b *Broker) Bind(queue, exchange, routingKey string) error

Bind is a no-op; ActiveMQ routing is destination-based.

func (*Broker) Close

func (b *Broker) Close() error

Close shuts down the broker, disconnecting the STOMP connection and closing all producers and consumers. It is safe to call multiple times.

func (*Broker) Connect

func (b *Broker) Connect() error

Connect establishes the STOMP connection to the ActiveMQ broker.

func (*Broker) Consumer

func (b *Broker) Consumer(destination string, opts mq.ConsumeOptions) (mq.Consumer, error)

Consumer creates a consumer for the given destination. The consumer is not started; call Start to begin receiving messages.

func (*Broker) DeclareExchange

func (b *Broker) DeclareExchange(name string, opts mq.ExchangeOptions) error

DeclareExchange is a no-op; ActiveMQ auto-creates destinations.

func (*Broker) DeclareQueue

func (b *Broker) DeclareQueue(name string, opts mq.QueueOptions) error

DeclareQueue is a no-op; ActiveMQ auto-creates destinations.

func (*Broker) DeleteExchange

func (b *Broker) DeleteExchange(name string) error

DeleteExchange is a no-op.

func (*Broker) DeleteQueue

func (b *Broker) DeleteQueue(name string) error

DeleteQueue is a no-op. STOMP does not provide a standard way to delete a destination; use the ActiveMQ JMX/admin API if required.

func (*Broker) IsConnected

func (b *Broker) IsConnected() bool

IsConnected reports whether the broker is currently connected.

func (*Broker) Metrics

func (b *Broker) Metrics() mq.Metrics

Metrics returns a snapshot of broker metrics.

func (*Broker) Producer

func (b *Broker) Producer(destination string, opts mq.PublishOptions) (mq.Producer, error)

Producer creates or returns a cached producer for the given destination (e.g. "/queue/events" or "/topic/news").

func (*Broker) Unbind

func (b *Broker) Unbind(queue, exchange, routingKey string) error

Unbind is a no-op.

type Config

type Config struct {
	// Addr is the network address of the STOMP transport, e.g.
	// "localhost:61613". Required.
	Addr string

	// Network is the transport network ("tcp", "tcp4", "tcp6").
	// Default: "tcp".
	Network string

	// Login is the STOMP login (username). Optional.
	Login string

	// Passcode is the STOMP passcode (password). Optional.
	Passcode string

	// Vhost is the STOMP "host" header value (virtual host). If empty,
	// the host is derived from the remote address.
	Vhost string

	// Heartbeat is the desired STOMP heart-beat interval for both send
	// and receive directions. Default: 10s. Set to 0 to use the library
	// default (1 minute).
	Heartbeat time.Duration

	// ConnectTimeout is the timeout for establishing the underlying TCP
	// connection. Default: 10s.
	ConnectTimeout time.Duration
}

Config configures an ActiveMQ (STOMP) broker.

func DefaultConfig

func DefaultConfig() Config

DefaultConfig returns a Config with sensible defaults for a local ActiveMQ broker exposing the STOMP transport on port 61613.

type Consumer

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

Consumer implements mq.Consumer for ActiveMQ over STOMP. It subscribes to a destination and dispatches received messages to a handler.

func (*Consumer) IsRunning

func (c *Consumer) IsRunning() bool

IsRunning reports whether the consumer is actively consuming.

func (*Consumer) Start

func (c *Consumer) Start(ctx context.Context) error

Start subscribes to the destination and begins dispatching messages to the handler. The consumer runs until ctx is cancelled or Stop is called.

func (*Consumer) Stop

func (c *Consumer) Stop(timeout time.Duration) error

Stop gracefully stops consuming, waiting for in-flight handlers to complete up to the given timeout. It is safe to call multiple times.

type Producer

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

Producer implements mq.Producer for ActiveMQ over STOMP. It sends messages to a fixed destination using the broker's shared STOMP connection.

func (*Producer) Close

func (p *Producer) Close() error

Close releases producer resources. After Close, Publish returns ErrClosed. It is safe to call multiple times.

func (*Producer) Publish

func (p *Producer) Publish(ctx context.Context, msg *mq.Message) error

Publish sends a message to the producer's destination. The mq.Message fields are mapped onto STOMP SEND frame headers.