eventstream

Documentation

Index

• type AckFn
• type EventBatcher
• type EventStream
• type JetstreamEventStream
  • func NewJetstreamEventStream(opts *configuration.JetstreamConfig, consumerOpts ...jsm.ConsumerOption) (*JetstreamEventStream, error)
  • func (c *JetstreamEventStream) Check() error
  • func (c *JetstreamEventStream) Close() error
  • func (c *JetstreamEventStream) Publish(events []*api.EventMessage) []error
  • func (c *JetstreamEventStream) Subscribe(queue string, callback func(event *Message) error) error
• type Message
• type StanClient
• type StanClientConnection
  • func NewStanClientConnection(stanClusterID, clientID string, servers []string, options ...stan.Option) (*StanClientConnection, error)
  • func (c *StanClientConnection) Check() error
  • func (c *StanClientConnection) Close() error
  • func (c *StanClientConnection) PublishAsync(subject string, data []byte, ah stan.AckHandler) (string, error)
  • func (c *StanClientConnection) QueueSubscribe(subject, qgroup string, cb stan.MsgHandler, opts ...stan.SubscriptionOption) error
• type StanEventStream
  • func NewStanEventStream(subject string, stanClient StanClient, ...) *StanEventStream
  • func (stream *StanEventStream) Close() error
  • func (stream *StanEventStream) Publish(events []*api.EventMessage) []error
  • func (stream *StanEventStream) Subscribe(queue string, callback func(event *Message) error) error
• type TimedEventBatcher
  • func NewTimedEventBatcher(batchSize int, maxTimeBetweenBatches time.Duration, timeout time.Duration) *TimedEventBatcher
  • func (b *TimedEventBatcher) Register(callback eventBatchCallback)
  • func (b *TimedEventBatcher) Report(event *Message) error
  • func (b *TimedEventBatcher) Stop() error

Types

type AckFn

type AckFn func() error

type EventBatcher

type EventBatcher interface {
	Register(callback eventBatchCallback)
	Report(event *Message) error
	Stop() error
}

type EventStream

type EventStream interface {
	Publish(events []*api.EventMessage) []error
	Subscribe(queue string, callback func(event *Message) error) error
	Close() error
}

type JetstreamEventStream

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

func NewJetstreamEventStream

func NewJetstreamEventStream(
	opts *configuration.JetstreamConfig,
	consumerOpts ...jsm.ConsumerOption) (*JetstreamEventStream, error)

func (*JetstreamEventStream) Check

func (c *JetstreamEventStream) Check() error

func (*JetstreamEventStream) Close

func (c *JetstreamEventStream) Close() error

func (*JetstreamEventStream) Publish

func (c *JetstreamEventStream) Publish(events []*api.EventMessage) []error

func (*JetstreamEventStream) Subscribe

func (c *JetstreamEventStream) Subscribe(queue string, callback func(event *Message) error) error

type Message

type Message struct {
	EventMessage *api.EventMessage
	Ack          AckFn
}

type StanClient

type StanClient interface {
	PublishAsync(subject string, data []byte, ah stan.AckHandler) (string, error)
	QueueSubscribe(subject, queue string, cb stan.MsgHandler, opts ...stan.SubscriptionOption) error
	Close() error
}

type StanClientConnection

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

func NewStanClientConnection

func NewStanClientConnection(
	stanClusterID, clientID string, servers []string,
	options ...stan.Option) (*StanClientConnection, error)

func (*StanClientConnection) Check

func (c *StanClientConnection) Check() error

func (*StanClientConnection) Close

func (c *StanClientConnection) Close() error

func (*StanClientConnection) PublishAsync

func (c *StanClientConnection) PublishAsync(subject string, data []byte, ah stan.AckHandler) (string, error)

func (*StanClientConnection) QueueSubscribe

func (c *StanClientConnection) QueueSubscribe(subject, qgroup string, cb stan.MsgHandler, opts ...stan.SubscriptionOption) error

type StanEventStream

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

func NewStanEventStream

func NewStanEventStream(subject string, stanClient StanClient, subscriptionOptions ...stan.SubscriptionOption) *StanEventStream

func (*StanEventStream) Close

func (stream *StanEventStream) Close() error

func (*StanEventStream) Publish

func (stream *StanEventStream) Publish(events []*api.EventMessage) []error

func (*StanEventStream) Subscribe

func (stream *StanEventStream) Subscribe(queue string, callback func(event *Message) error) error

type TimedEventBatcher

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

func NewTimedEventBatcher

func NewTimedEventBatcher(batchSize int, maxTimeBetweenBatches time.Duration, timeout time.Duration) *TimedEventBatcher

func (*TimedEventBatcher) Register

func (b *TimedEventBatcher) Register(callback eventBatchCallback)

Process is started when the callback is registered

func (*TimedEventBatcher) Report

func (b *TimedEventBatcher) Report(event *Message) error

func (*TimedEventBatcher) Stop

func (b *TimedEventBatcher) Stop() error