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 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 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 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
}