pubsub

Documentation

Index

• type BufferingEnvelope
  • func (e *BufferingEnvelope) Size() int64
• type BufferingPubSub
  • func NewBufferingPubSub[T any](params BufferingPubSubParams) *BufferingPubSub[T]
  • func (p *BufferingPubSub[T]) Close(ctx context.Context) (err error)
  • func (p *BufferingPubSub[T]) Handle(ctx context.Context, envelope BufferingEnvelope) error
  • func (p *BufferingPubSub[T]) Publish(ctx context.Context, message T) error
  • func (p *BufferingPubSub[T]) Subscribe(ctx context.Context, subscriber Subscriber[T]) (err error)
• type BufferingPubSubParams
• type ChainedSubscriber
  • func NewChainedSubscriber[T any](ignoreErrors bool) *ChainedSubscriber[T]
  • func (c *ChainedSubscriber[T]) Add(subscriber Subscriber[T])
  • func (c *ChainedSubscriber[T]) Handle(ctx context.Context, message T) error
• type InMemoryPubSub
  • func NewInMemoryPubSub[T any]() *InMemoryPubSub[T]
  • func (p *InMemoryPubSub[T]) Close(ctx context.Context) error
  • func (p *InMemoryPubSub[T]) Publish(ctx context.Context, message T) error
  • func (p *InMemoryPubSub[T]) Subscribe(ctx context.Context, subscriber Subscriber[T]) error
• type InMemorySubscriber
  • func NewInMemorySubscriber[T any]() *InMemorySubscriber[T]
  • func (s *InMemorySubscriber[T]) Events() []T
  • func (s *InMemorySubscriber[T]) Handle(ctx context.Context, message T) error
• type PubSub
• type Subscriber
• type SubscriberFunc
  • func (f SubscriberFunc[T]) Handle(ctx context.Context, message T) error

Types

type BufferingEnvelope

type BufferingEnvelope struct {
	Offsets  []int64 `json:"offsets"`
	Payloads []byte  `json:"payloads"`
}

func (*BufferingEnvelope) Size

func (e *BufferingEnvelope) Size() int64

Size return the size of the buffered payloads in the envelope

type BufferingPubSub

type BufferingPubSub[T any] struct {
	// contains filtered or unexported fields
}

BufferingPubSub is a PubSub implementation that buffers messages in memory and flushes them to the delegate PubSub when the buffer is full or the buffer age is reached

func NewBufferingPubSub

func NewBufferingPubSub[T any](params BufferingPubSubParams) *BufferingPubSub[T]

func (*BufferingPubSub[T]) Close

func (p *BufferingPubSub[T]) Close(ctx context.Context) (err error)

func (*BufferingPubSub[T]) Handle

func (p *BufferingPubSub[T]) Handle(ctx context.Context, envelope BufferingEnvelope) error

func (*BufferingPubSub[T]) Publish

func (p *BufferingPubSub[T]) Publish(ctx context.Context, message T) error

func (*BufferingPubSub[T]) Subscribe

func (p *BufferingPubSub[T]) Subscribe(ctx context.Context, subscriber Subscriber[T]) (err error)

type BufferingPubSubParams

type BufferingPubSubParams struct {
	DelegatePubSub PubSub[BufferingEnvelope]
	MaxBufferSize  int64
	MaxBufferAge   time.Duration
}

type ChainedSubscriber

type ChainedSubscriber[T any] struct {
	// contains filtered or unexported fields
}

func NewChainedSubscriber

func NewChainedSubscriber[T any](ignoreErrors bool) *ChainedSubscriber[T]

func (*ChainedSubscriber[T]) Add

func (c *ChainedSubscriber[T]) Add(subscriber Subscriber[T])

Add subscriber to the chain

func (*ChainedSubscriber[T]) Handle

func (c *ChainedSubscriber[T]) Handle(ctx context.Context, message T) error

Handle message by calling all subscribers in the chain

type InMemoryPubSub

type InMemoryPubSub[T any] struct {
	// contains filtered or unexported fields
}

InMemoryPubSub is a simple in-memory pubsub implementation used for testing

func NewInMemoryPubSub

func NewInMemoryPubSub[T any]() *InMemoryPubSub[T]

func (*InMemoryPubSub[T]) Close

func (p *InMemoryPubSub[T]) Close(ctx context.Context) error

func (*InMemoryPubSub[T]) Publish

func (p *InMemoryPubSub[T]) Publish(ctx context.Context, message T) error

func (*InMemoryPubSub[T]) Subscribe

func (p *InMemoryPubSub[T]) Subscribe(ctx context.Context, subscriber Subscriber[T]) error

type InMemorySubscriber

type InMemorySubscriber[T any] struct {
	// contains filtered or unexported fields
}

InMemorySubscriber is a simple in-memory subscriber implementation used for testing

func NewInMemorySubscriber

func NewInMemorySubscriber[T any]() *InMemorySubscriber[T]

func (*InMemorySubscriber[T]) Events

func (s *InMemorySubscriber[T]) Events() []T

func (*InMemorySubscriber[T]) Handle

func (s *InMemorySubscriber[T]) Handle(ctx context.Context, message T) error

type PubSub

type PubSub[T any] interface {
	// Publish a message
	Publish(ctx context.Context, message T) error
	// Subscribe to messages
	Subscribe(ctx context.Context, subscriber Subscriber[T]) error
	// Close the PubSub and release resources, if any
	Close(ctx context.Context) error
}

PubSub enables publishing messages to subscribers

type Subscriber

type Subscriber[T any] interface {
	Handle(ctx context.Context, message T) error
}

Subscriber handles messages publishes to PubSub

type SubscriberFunc

type SubscriberFunc[T any] func(ctx context.Context, message T) error

SubscriberFunc is a helper function that implements Subscriber interface

func (SubscriberFunc[T]) Handle

func (f SubscriberFunc[T]) Handle(ctx context.Context, message T) error