messaging

Documentation

Index

• Variables
• type AckManager
  • func NewAckManager(logger log.Logger) AckManager
  • func NewContinuousAckManager(logger log.Logger) AckManager
• type Client
• type CloseableProducer
• type Consumer
  • func NewNoopConsumer() Consumer
• type Message
• type Producer
  • func NewMetricProducer(producer Producer, metricsClient metrics.Client) Producer
  • func NewNoopProducer() Producer

Variables

var (
	// ErrMessageSizeLimit indicate that message is rejected by server due to size limitation
	ErrMessageSizeLimit = errors.New("message was too large, server rejected it to avoid allocation error")
)

Types

type AckManager

type AckManager interface {
	// Read an item into backlog for processing for ack
	ReadItem(id int64) error
	// Get current max ID from read items
	GetReadLevel() int64
	// Set current max ID from read items
	SetReadLevel(readLevel int64)
	// Mark an item as done processing, and remove from backlog
	AckItem(id int64) (ackLevel int64)
	// Get current max level that can safely ack
	GetAckLevel() int64
	// Set current max level that can safely ack
	SetAckLevel(ackLevel int64)
	// GetBacklogCount return the of items that are waiting for ack
	GetBacklogCount() int64
}

AckManager convert out of order acks into ackLevel movement.

func NewAckManager

func NewAckManager(logger log.Logger) AckManager

NewAckManager returns a AckManager without monitoring the itemIDs continousness. For example, our internal matching task queue doesn't guarantee it.

func NewContinuousAckManager

func NewContinuousAckManager(logger log.Logger) AckManager

NewContinuousAckManager returns a ContinuousAckManager it will emit error logs for itemIDs being incontinuous This is useful for some message queue system that guarantees continuousness that we want to monitor it's behaving correctly

type Client

type Client interface {
	NewConsumer(appName, consumerName string) (Consumer, error)
	NewProducer(appName string) (Producer, error)
}

Client is the interface used to abstract out interaction with messaging system for replication

type CloseableProducer

type CloseableProducer interface {
	Producer
	Close() error
}

CloseableProducer is a Producer that can be closed

type Consumer

type Consumer interface {
	// Start starts the consumer
	Start() error
	// Stop stops the consumer
	Stop()
	// Messages return the message channel for this consumer
	Messages() <-chan Message
}

Consumer is the unified interface for both internal and external kafka clients

func NewNoopConsumer

func NewNoopConsumer() Consumer

NewNoopProducer returns a no-op message consumer

type Message

type Message interface {
	// Value is a mutable reference to the message's value
	Value() []byte
	// Partition is the ID of the partition from which the message was read.
	Partition() int32
	// Offset is the message's offset.
	Offset() int64
	// Ack marks the message as successfully processed.
	Ack() error
	// Nack marks the message processing as failed and the message will be retried or sent to DLQ.
	Nack() error
}

Message is the unified interface for a Kafka message

type Producer

type Producer interface {
	Publish(ctx context.Context, message interface{}) error
}

Producer is the interface used to send replication tasks to other clusters through replicator

func NewMetricProducer

func NewMetricProducer(
	producer Producer,
	metricsClient metrics.Client,
) Producer

NewMetricProducer creates a new instance of producer that emits metrics

func NewNoopProducer

func NewNoopProducer() Producer

NewNoopProducer returns a no-op message producer