subscribe

Documentation

Index

• type Broadcast
  • func NewBroadcast() *Broadcast
  • func (b *Broadcast) Publish(ctx context.Context, message *Message)
  • func (b *Broadcast) RegisterSession(sessionId string, pub Publisher)
  • func (b *Broadcast) UnregisterSession(id string)
• type BufferedBroadcast
  • func NewBufferedBroadcast(n int) *BufferedBroadcast
  • func (bb *BufferedBroadcast) Publish(ctx context.Context, msg *Message)
  • func (bb *BufferedBroadcast) RegisterSession(ctx context.Context, pub Publisher)
• type FilterFunc
  • func FilterAnd(filterFuncs ...FilterFunc) FilterFunc
  • func FilterKeys(keys []string) (FilterFunc, error)
  • func FilterMaxAge(age string) (FilterFunc, error)
  • func FilterMinOffset(minOffset int64) FilterFunc
• type FilterPublisher
  • func NewFilterPublisher(pub Publisher) *FilterPublisher
  • func (p *FilterPublisher) Publish(ctx context.Context, message *Message)
  • func (p *FilterPublisher) SetFilter(f FilterFunc)
• type Header
• type KafkaClient
• type KafkaSubscriber
  • func NewKafkaSubscriber(cli KafkaClient, pub Publisher) *KafkaSubscriber
  • func (k *KafkaSubscriber) Subscribe(ctx context.Context)
• type Message
• type Publisher
• type Publishers
  • func (pubs Publishers) Publish(ctx context.Context, message *Message)
• type SenderStream
• type StreamPublisher
  • func NewStreamPublisher(stream grpc.ServerStreamingServer[protobuf.Message]) *StreamPublisher
  • func (s *StreamPublisher) Publish(ctx context.Context, message *Message)

Types

type Broadcast

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

func NewBroadcast

func NewBroadcast() *Broadcast

func (*Broadcast) Publish

func (b *Broadcast) Publish(ctx context.Context, message *Message)

func (*Broadcast) RegisterSession

func (b *Broadcast) RegisterSession(sessionId string, pub Publisher)

func (*Broadcast) UnregisterSession

func (b *Broadcast) UnregisterSession(id string)

type BufferedBroadcast

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

func NewBufferedBroadcast

func NewBufferedBroadcast(n int) *BufferedBroadcast

func (*BufferedBroadcast) Publish

func (bb *BufferedBroadcast) Publish(ctx context.Context, msg *Message)

func (*BufferedBroadcast) RegisterSession

func (bb *BufferedBroadcast) RegisterSession(ctx context.Context, pub Publisher)

type FilterFunc

type FilterFunc func(*Message) bool

func FilterAnd

func FilterAnd(filterFuncs ...FilterFunc) FilterFunc

func FilterKeys

func FilterKeys(keys []string) (FilterFunc, error)

func FilterMaxAge

func FilterMaxAge(age string) (FilterFunc, error)

func FilterMinOffset

func FilterMinOffset(minOffset int64) FilterFunc

type FilterPublisher

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

func NewFilterPublisher

func NewFilterPublisher(pub Publisher) *FilterPublisher

func (*FilterPublisher) Publish

func (p *FilterPublisher) Publish(ctx context.Context, message *Message)

func (*FilterPublisher) SetFilter

func (p *FilterPublisher) SetFilter(f FilterFunc)

type Header

type Header struct {
	Key   string
	Value string
}

type KafkaClient

type KafkaClient interface {
	PollFetches(ctx context.Context) kgo.Fetches
}

type KafkaSubscriber

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

func NewKafkaSubscriber

func NewKafkaSubscriber(cli KafkaClient, pub Publisher) *KafkaSubscriber

func (*KafkaSubscriber) Subscribe

func (k *KafkaSubscriber) Subscribe(ctx context.Context)

type Message

type Message struct {
	Key       string
	Value     []byte
	Headers   []Header
	Timestamp time.Time
	Topic     string
	Partition int32
	Offset    int64
}

type Publisher

type Publisher interface {
	Publish(ctx context.Context, message *Message)
}

type Publishers

type Publishers []Publisher

func (Publishers) Publish

func (pubs Publishers) Publish(ctx context.Context, message *Message)

type SenderStream

type SenderStream interface {
	grpc.ServerStreamingServer[protobuf.Message]
}

type StreamPublisher

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

func NewStreamPublisher

func NewStreamPublisher(stream grpc.ServerStreamingServer[protobuf.Message]) *StreamPublisher

func (*StreamPublisher) Publish

func (s *StreamPublisher) Publish(ctx context.Context, message *Message)