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(ctx context.Context, uri string, connectFn AMQPConnectFunc) (*AMQPProducer, error)
  • func (p *AMQPProducer) Publish(ctx context.Context, msg AMQPMessage) 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 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(ctx context.Context, uri string, connectFn AMQPConnectFunc) (*AMQPProducer, error)

func (*AMQPProducer) Publish

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

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
}