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 AMQPChanPublisher
• type AMQPConnectFunc
  • func NewAMQPConnectFunc(setup func(c *amqp.Channel) error) AMQPConnectFunc
• type AMQPMessage
• type AMQPProducer
  • func NewAMQPProducer(uri string, connectFn AMQPConnectFunc) (*AMQPProducer, error)
  • func (p *AMQPProducer) Publish(ctx context.Context, msg AMQPMessage) error
  • func (p *AMQPProducer) Shutdown(ctx context.Context) error
• type BindingArgs
• type ConsumeOptions
• type Handler
• type Producer
  • func NewAMQPExchangeProducer(ctx context.Context, uri, exchange, keyNs string) (Producer, error)
  • func NewAMQPQueueProducer(ctx context.Context, uri, queue string) (Producer, error)
• type RawStreamOptions
• 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")
)

var ByteCapacity = stream.ByteCapacity{}

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 AMQPChanPublisher

type AMQPChanPublisher interface {
	Publish(exchange, key string, mandatory, immediate bool, msg amqp.Publishing) error
}

type AMQPConnectFunc

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

func NewAMQPConnectFunc

func NewAMQPConnectFunc(setup func(c *amqp.Channel) error) AMQPConnectFunc

type AMQPMessage

type AMQPMessage struct {
	Exchange, Key string
	Body          interface{}
	Persistent    bool
}

type AMQPProducer

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

func NewAMQPProducer

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

func (*AMQPProducer) Publish

func (p *AMQPProducer) Publish(ctx context.Context, msg AMQPMessage) error

func (*AMQPProducer) Shutdown

func (p *AMQPProducer) Shutdown(ctx context.Context) error

Shutdown will try to gracefully stop the background event publishing process, by waiting until all the publish buffer is flushed to the remote broker and all message confirmations have been received. This function must be called only once in a producer or it will panic.

The Publish function must not be called concurrently with Shutdown or the events sent concurrently may be lost (concurrent Publish and Shutdown functions may succeed but the event never really gets sent).

type BindingArgs

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

type ConsumeOptions

type ConsumeOptions struct {
	Stream string
	*StreamOptions
	*stream.ConsumerOptions
	MemorizeOffset bool
}

type Handler

type Handler interface {
	HandleMessage(msg StreamMessage)
}

type Producer

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

func NewAMQPExchangeProducer

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

func NewAMQPQueueProducer

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

type RawStreamOptions

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

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
}