messagequeue

Documentation

Overview

Package messagequeue provides message queue publisher and consumer interfaces with implementations for Google Pub/Sub, Redis, and Amazon SQS.

Index

• Variables
• type Consumer
• type ConsumerFunc
• type ConsumerProvider
• type Publisher
• type PublisherProvider

Variables

var (
	// ErrEmptyTopicName is returned when a topic name is empty.
	ErrEmptyTopicName = platformerrors.New("empty topic name")

	// ErrConsumerAlreadyRegistered is returned when a second consumer is
	// requested for a topic a provider already has one for.
	//
	// Providers cache consumers by topic, and the cache used to win silently:
	// the second caller got the first caller's consumer, wired to the first
	// caller's handler, and their own handler was never invoked for any message.
	// Nothing failed and nothing logged — the messages simply went somewhere else.
	//
	// One consumer per topic per provider is the rule; a caller that wants two
	// behaviors for one topic multiplexes inside its own handler.
	ErrConsumerAlreadyRegistered = platformerrors.New("a consumer is already registered for this topic")
)

Types

type Consumer

type Consumer interface {
	Consume(ctx context.Context, errs chan<- error)
}

Consumer reads messages off a queue and hands each to its handler.

Stopping

Consume runs until ctx is done. There is no separate stop channel: every implementation turned one into a context cancellation immediately, so it was a second way to say the same thing — and a `chan bool` at that, which is bidirectional, so nothing stopped a caller from receiving on it and stealing the stop signal from the consumer.

Delivery semantics

These differ by backend, and the difference is load-bearing rather than an implementation detail a caller can ignore:

• redis is at-most-once. It is pub/sub: a message delivered while no consumer is running is gone, and a handler that fails does not get the message again. Do not use it for work that must not be lost.
• sqs, pubsub and kafka are at-least-once. A handler must therefore be idempotent — see the idempotency package — because redelivery is normal operation, not an error case.

Handler errors are reported on errs, and what happens next also differs: kafka stops the consumer, because its commits are cumulative by offset and committing past a failed message would lose it; the others log the failure and continue with the next message.

errs is send-only and must be drained. A consumer whose error channel is not being read does not block forever — it also selects on ctx — but it does discard errors while nobody is listening.

type ConsumerFunc

type ConsumerFunc func(context.Context, []byte) error

ConsumerFunc is a function type that handles consumed messages.

type ConsumerProvider

type ConsumerProvider interface {
	Close()
	NewConsumer(ctx context.Context, topic string, handlerFunc ConsumerFunc) (Consumer, error)
}

ConsumerProvider is a function that provides a Consumer for a given topic.

One consumer per topic: a second NewConsumer for a topic that already has one returns ErrConsumerAlreadyRegistered rather than silently handing back the first caller's consumer, wired to the first caller's handler.

type Publisher

type Publisher interface {
	// Stop halts all publishing.
	Stop()
	// Publish writes a message onto a message queue.
	Publish(ctx context.Context, data any) error
	// PublishAsync writes a message onto a message queue, logging any error
	// instead of returning it.
	//
	// "Async" names the error handling, not the delivery: it publishes on the
	// calling goroutine and returns when the publish has finished, exactly as
	// Publish does. A caller that wants the publish off its own goroutine has
	// to arrange that itself.
	PublishAsync(ctx context.Context, data any)
}

Publisher writes messages onto a queue.

type PublisherProvider

type PublisherProvider interface {
	Close()
	Ping(ctx context.Context) error
	NewPublisher(ctx context.Context, topic string) (Publisher, error)
}

PublisherProvider is a function that provides a Publisher for a given topic.