event

Documentation

Index

• Constants
• Variables
• func NewConsumerOptions(name string, offset stream.OffsetSpecification) *stream.ConsumerOptions
• func ParseStreamOptions(raw RawStreamOptions) (*stream.StreamOptions, error)
• func TimestampOffset(t time.Time) stream.OffsetSpecification
• type AMQPChanOps
• type AMQPChanSetup
• type AMQPClient
  • func NewAMQPClient(uri string, connectFn AMQPConnectFunc) (AMQPClient, error)
• type AMQPConnectFunc
  • func NewAMQPConnectFunc(setup func(c AMQPChanSetup) error) AMQPConnectFunc
• type AMQPConsumer
  • func NewAMQPConsumer(uri string, connectFn AMQPConnectFunc) (AMQPConsumer, error)
• type AMQPMessage
• type AMQPMessageHandler
• type AMQPProducer
  • func NewAMQPProducer(uri string, connectFn AMQPConnectFunc) (AMQPProducer, error)
• type BindingArgs
• type ConsumeOptions
• type Handler
• type PublishResult
• type RawStreamOptions
• type SimpleProducer
  • func NewAMQPExchangeProducer(ctx context.Context, uri, exchange, keyNs string) (SimpleProducer, error)
  • func NewAMQPQueueProducer(ctx context.Context, uri, queue string) (SimpleProducer, error)
• type StreamConsumer
  • func NewStreamConsumer(streamUriStr, amqpUriStr string) (StreamConsumer, error)
• type StreamMessage
• type StreamOptions

Constants

const (
	PublishChannelSize   = 1024
	RetryMinDelay        = 5 * time.Second
	PublishLogSampleRate = 0.1
	MaxRetries           = 3
)

Variables

var (
	ErrProducerShuttingDown = errors.New("amqp: producer shutting down")
	ErrProducerClosed       = errors.New("amqp: producer closed")
	ErrMaxRetriesReached    = errors.New("amqp: publish max retries reached")
	ErrRetryQueueFull       = errors.New("amqp: retry queue full")
)

var ByteCapacity = stream.ByteCapacity{}

var ErrConsumerClosed = errors.New("amqp: consumer closed")

var OffsetSpec = stream.OffsetSpecification{}

Functions

func NewConsumerOptions

func NewConsumerOptions(name string, offset stream.OffsetSpecification) *stream.ConsumerOptions

func ParseStreamOptions

func ParseStreamOptions(raw RawStreamOptions) (*stream.StreamOptions, error)

func TimestampOffset

func TimestampOffset(t time.Time) stream.OffsetSpecification

Types

type AMQPChanOps

type AMQPChanOps interface {
	Publish(exchange, key string, mandatory, immediate bool, msg amqp.Publishing) error
	Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error)
}

type AMQPChanSetup

type AMQPChanSetup interface {
	ExchangeBind(destination, key, source string, noWait bool, args amqp.Table) error
	ExchangeDeclare(name, kind string, durable, autoDelete, internal, noWait bool, args amqp.Table) error
	ExchangeDeclarePassive(name, kind string, durable, autoDelete, internal, noWait bool, args amqp.Table) error
	ExchangeDelete(name string, ifUnused, noWait bool) error
	ExchangeUnbind(destination, key, source string, noWait bool, args amqp.Table) error
	QueueBind(name, key, exchange string, noWait bool, args amqp.Table) error
	QueueDeclare(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error)
	QueueDeclarePassive(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error)
	QueueDelete(name string, ifUnused, ifEmpty, noWait bool) (int, error)
	QueueInspect(name string) (amqp.Queue, error)
	QueueUnbind(name, key, exchange string, args amqp.Table) error
}

type AMQPClient

type AMQPClient interface {
	AMQPProducer
	AMQPConsumer
}

func NewAMQPClient

func NewAMQPClient(uri string, connectFn AMQPConnectFunc) (AMQPClient, error)

type AMQPConnectFunc

type AMQPConnectFunc func(ctx context.Context, uri string, confirms chan amqp.Confirmation, closed chan *amqp.Error) (AMQPChanOps, error)

func NewAMQPConnectFunc

func NewAMQPConnectFunc(setup func(c AMQPChanSetup) error) AMQPConnectFunc

type AMQPConsumer

type AMQPConsumer interface {
	Consume(queue string, concurrency int, handler AMQPMessageHandler) error
	Shutdown(context.Context) error
}

func NewAMQPConsumer

func NewAMQPConsumer(uri string, connectFn AMQPConnectFunc) (AMQPConsumer, error)

type AMQPMessage

type AMQPMessage struct {
	Exchange, Key string
	Body          interface{}
	Persistent    bool
	ResultChan    chan<- PublishResult
}

type AMQPMessageHandler

type AMQPMessageHandler func(amqp.Delivery) error

type AMQPProducer

type AMQPProducer interface {
	Publish(ctx context.Context, msg AMQPMessage) error
	Shutdown(context.Context) error
}

func NewAMQPProducer

func NewAMQPProducer(uri string, connectFn AMQPConnectFunc) (AMQPProducer, error)

type BindingArgs

type BindingArgs struct {
	Key      string
	Exchange string
	Args     amqp.Table
}

type ConsumeOptions

type ConsumeOptions struct {
	Stream string
	*StreamOptions
	*stream.ConsumerOptions
	// Whether to memorize the message offset in the stream and use it on
	// re-connections to continue from the last read message.
	MemorizeOffset bool
}

type Handler

type Handler interface {
	HandleMessage(msg StreamMessage)
}

type PublishResult

type PublishResult struct {
	Message AMQPMessage
	Error   error
}

type RawStreamOptions

type RawStreamOptions struct {
	MaxLengthBytes      string
	MaxSegmentSizeBytes string
	MaxAge              time.Duration
}

type SimpleProducer

type SimpleProducer interface {
	Publish(ctx context.Context, key string, body interface{}, persistent bool) error
}

func NewAMQPExchangeProducer

func NewAMQPExchangeProducer(ctx context.Context, uri, exchange, keyNs string) (SimpleProducer, error)

func NewAMQPQueueProducer

func NewAMQPQueueProducer(ctx context.Context, uri, queue string) (SimpleProducer, error)

type StreamConsumer

type StreamConsumer interface {
	ConsumeChan(ctx context.Context, opts ConsumeOptions) (<-chan StreamMessage, error)
	Consume(ctx context.Context, opts ConsumeOptions, handler Handler) error
	CheckConnection() error
	Close() error
}

func NewStreamConsumer

func NewStreamConsumer(streamUriStr, amqpUriStr string) (StreamConsumer, error)

type StreamMessage

type StreamMessage struct {
	stream.ConsumerContext
	*streamAmqp.Message
}

type StreamOptions

type StreamOptions struct {
	stream.StreamOptions
	Bindings []BindingArgs
}