internal

Documentation

Index

• Constants
• func MustGet[T any](typ enum.PluginType) T
• func MustGetBroker(gapOpts *gap.Options) broker.Broker
• func MustGetStorage(gapOpts *gap.Options) storage.Storage
• func Pair(key string, value string) pair
• func Register[T any](typ enum.PluginType, value T)
• type Event
• type EventPub
  • func (e *EventPub) Bind(txer tx.Txer) (EventPublisher, error)
  • func (p *EventPub) Publish(ctx context.Context, event Event, args ...any) error
• type EventPublisher
• type Headers
  • func (h *Headers) Add(args ...any)
  • func (h *Headers) Value() map[string]string
• type Pub
  • func NewPub[T any](opts *gap.Options, broker broker.Broker, storage storage.Storage) *Pub[T]
  • func (p *Pub[T]) Bind(txer tx.Txer) (Publisher[T], error)
  • func (p *Pub[T]) Publish(ctx context.Context, topic string, msg T, args ...any) error
• type Publisher
• type Pump
  • func NewPump(gapOpts *gap.Options, storage storage.Storage, broker broker.Broker) *Pump
  • func (p *Pump) PollingHandle(enCh <-chan *entity.Envelope) <-chan *entity.Envelope
  • func (p *Pump) PollingSend()
• type Sub
  • func NewSub(opts *gap.Options, broker broker.Broker, storage storage.Storage) *Sub
  • 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 MustGet

func MustGet[T any](typ enum.PluginType) T

func MustGetBroker

func MustGetBroker(gapOpts *gap.Options) broker.Broker

func MustGetStorage

func MustGetStorage(gapOpts *gap.Options) storage.Storage

func Pair

func Pair(key string, value string) pair

func Register

func Register[T any](typ enum.PluginType, value T)

Types

type Event

type Event interface {
	Topic() string
}

type EventPub

type EventPub struct {
	*Pub[Event]
}

func (*EventPub) Bind

func (e *EventPub) Bind(txer tx.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 tx.Txer) (EventPublisher, error)

	Publish(ctx context.Context, event Event, args ...any) 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 Pub

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

func NewPub

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

func (*Pub[T]) Bind

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

Bind implements Pub.

func (*Pub[T]) Publish

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

Publish implements Pub.

type Publisher

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

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

type Pump

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

func NewPump

func NewPump(gapOpts *gap.Options, storage storage.Storage, broker broker.Broker) *Pump

func (*Pump) PollingHandle

func (p *Pump) PollingHandle(enCh <-chan *entity.Envelope) <-chan *entity.Envelope

func (*Pump) PollingSend

func (p *Pump) PollingSend()

type Sub

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

func NewSub

func NewSub(opts *gap.Options, broker broker.Broker, storage storage.Storage) *Sub

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].