pubsub

Documentation

Index

• func RunConsumer[T any](ctx context.Context, subscriber Subscriber[T], consumer Consumer[T], ...) error
• type Consumer
• type PubSub
  • func NewPubSub[T any](client *redis.Client, channel string) *PubSub[T]
  • func (p *PubSub[T]) Channel() string
  • func (p *PubSub[T]) Collect(ch chan<- prometheus.Metric)
  • func (p *PubSub[T]) Describe(ch chan<- *prometheus.Desc)
  • func (p *PubSub[T]) Ping(ctx context.Context) error
  • func (p *PubSub[T]) Publish(ctx context.Context, message T) error
  • func (p *PubSub[T]) Subscribe(ctx context.Context) Subscription[T]
  • func (p *PubSub[T]) SubscribeWithCallback(ctx context.Context, f func(context.Context, T))
• type Subscriber
• type Subscription

Functions

func RunConsumer

func RunConsumer[T any](
	ctx context.Context,
	subscriber Subscriber[T],
	consumer Consumer[T],
	logger *slog.Logger,
) error

Types

type Consumer

type Consumer[T any] interface {
	Consume(context.Context, T)
}

type PubSub

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

func NewPubSub

func NewPubSub[T any](client *redis.Client, channel string) *PubSub[T]

func (*PubSub[T]) Channel

func (p *PubSub[T]) Channel() string

func (*PubSub[T]) Collect

func (p *PubSub[T]) Collect(ch chan<- prometheus.Metric)

func (*PubSub[T]) Describe

func (p *PubSub[T]) Describe(ch chan<- *prometheus.Desc)

func (*PubSub[T]) Ping

func (p *PubSub[T]) Ping(ctx context.Context) error

func (*PubSub[T]) Publish

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

func (*PubSub[T]) Subscribe

func (p *PubSub[T]) Subscribe(ctx context.Context) Subscription[T]

func (*PubSub[T]) SubscribeWithCallback

func (p *PubSub[T]) SubscribeWithCallback(ctx context.Context, f func(context.Context, T))

type Subscriber

type Subscriber[T any] interface {
	SubscribeWithCallback(ctx context.Context, f func(context.Context, T))
	Channel() string
	Ping(context.Context) error
}

type Subscription

type Subscription[T any] interface {
	Channel() <-chan T
	Close() error
}