internal

Documentation

Index

• Constants
• func Pair(key string, value string) pair
• type Event
• type EventPub
  • func (e *EventPub) Bind(txer txer.Txer) (EventPublisher, error)
  • func (p *EventPub) Publish(ctx context.Context, event Event, args ...any) error
• type EventPublisher
• type GroupedSubs
  • func SingleGroupedSubs() *GroupedSubs
  • func (g *GroupedSubs) AddDependencyOtps(lst []gap.DependencyOptions)
  • func (g *GroupedSubs) AddHandlerOtps(lst []gap.HandlerOptions)
  • func (g *GroupedSubs) ListeningAll() error
  • func (g *GroupedSubs) SubscribeAll(gapOpts *gap.Options) error
• type Headers
  • func (h *Headers) Add(args ...any)
  • func (h *Headers) Value() map[string]string
• type OptionsGetter
• type Pub
  • func NewPub[T any](opts *gap.Options, producer broker.Producer, storage storage.Storage) *Pub[T]
  • func (p *Pub[T]) Bind(txer txer.Txer) (Publisher[T], error)
  • func (p *Pub[T]) Options() *gap.Options
  • func (p *Pub[T]) Publish(ctx context.Context, topic string, msg T, args ...any) error
  • func (p *Pub[T]) SendAndUpdate(ctx context.Context, envelope *entity.Envelope) error
• type Publisher
• type Sub
  • func NewSub(opts *gap.Options, groupOpts *gap.GroupOptions, consumer broker.Consumer, ...) *Sub
  • func (s *Sub) HandleAndUpdate(ctx context.Context, envelope *entity.Envelope) error
  • func (s *Sub) Listening() error
  • func (s *Sub) Subscribe(topic string, handler gap.Handler[[]byte]) error

Constants

const (
	KeysMessageID     = "gap-msg-id"
	KeysMessageType   = "gap-msg-type"
	KeysGroup         = "gap-group"
	KeysCorrelationID = "gap-corr-id"
)

Functions

func Pair

func Pair(key string, value string) pair

Types

type Event

type Event interface {
	Topic() string
}

type EventPub

type EventPub struct {
	*Pub[Event]
}

func (*EventPub) Bind

func (e *EventPub) Bind(txer txer.Txer) (EventPublisher, error)

func (*EventPub) Publish

func (p *EventPub) Publish(ctx context.Context, event Event, args ...any) error

type EventPublisher

type EventPublisher interface {
	Bind(txer txer.Txer) (EventPublisher, error)

	Publish(ctx context.Context, event Event, args ...any) error

	OptionsGetter
}

type GroupedSubs

type GroupedSubs struct {
	HandlerOtpsLst    []gap.HandlerOptions
	DependencyOtpsLst []gap.DependencyOptions
	// contains filtered or unexported fields
}

func SingleGroupedSubs

func SingleGroupedSubs() *GroupedSubs

func (*GroupedSubs) AddDependencyOtps

func (g *GroupedSubs) AddDependencyOtps(lst []gap.DependencyOptions)

func (*GroupedSubs) AddHandlerOtps

func (g *GroupedSubs) AddHandlerOtps(lst []gap.HandlerOptions)

func (*GroupedSubs) ListeningAll

func (g *GroupedSubs) ListeningAll() error

func (*GroupedSubs) SubscribeAll

func (g *GroupedSubs) SubscribeAll(gapOpts *gap.Options) error

type Headers

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

func (*Headers) Add

func (h *Headers) Add(args ...any)

func (*Headers) Value

func (h *Headers) Value() map[string]string

type OptionsGetter

type OptionsGetter interface {
	Options() *gap.Options
}

type Pub

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

func NewPub

func NewPub[T any](opts *gap.Options, producer broker.Producer, storage storage.Storage) *Pub[T]

func (*Pub[T]) Bind

func (p *Pub[T]) Bind(txer txer.Txer) (Publisher[T], error)

Bind implements Pub.

func (*Pub[T]) Options

func (p *Pub[T]) Options() *gap.Options

func (*Pub[T]) Publish

func (p *Pub[T]) Publish(ctx context.Context, topic string, msg T, args ...any) error

Publish implements Pub.

func (*Pub[T]) SendAndUpdate

func (p *Pub[T]) SendAndUpdate(ctx context.Context, envelope *entity.Envelope) error

type Publisher

type Publisher[T any] interface {
	Bind(txer txer.Txer) (Publisher[T], error)

	Publish(ctx context.Context, topic string, msg T, args ...any) error

	OptionsGetter
}

type Sub

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

func NewSub

func NewSub(opts *gap.Options, groupOpts *gap.GroupOptions, consumer broker.Consumer, storage storage.Storage) *Sub

func (*Sub) HandleAndUpdate

func (s *Sub) HandleAndUpdate(ctx context.Context, envelope *entity.Envelope) error

func (*Sub) Listening

func (s *Sub) Listening() error

func (*Sub) Subscribe

func (s *Sub) Subscribe(topic string, handler gap.Handler[[]byte]) error

Subscribe implements [Subscriber].