riverkfq

Documentation

Overview

Package riverkfq provides River based queues that offer Kafka publishing guarantees.

Index

• type MaxWorkers
• type Option
  • func Pool(p *pgxpool.Pool) Option
  • func WithProducer(p Producer) Option
• type Producer
• type PublishArgs
  • func (args PublishArgs) Kind() string
• type PublishQueue
  • func NewPublishQueue(opts ...Option) (*PublishQueue, error)
  • func (q *PublishQueue) Add(ctx context.Context, msgs ...*xkafka.Message) error
  • func (q *PublishQueue) AddTx(ctx context.Context, tx pgx.Tx, msgs ...*xkafka.Message) error
  • func (q *PublishQueue) Run(ctx context.Context) error
• type PublishWorker
  • func NewPublishWorker(producer Producer) *PublishWorker
  • func (w *PublishWorker) Work(ctx context.Context, job *river.Job[PublishArgs]) error

Types

type MaxWorkers

type MaxWorkers int

MaxWorkers sets the maximum number of workers. Default is 1.

type Option

type Option interface {
	// contains filtered or unexported methods
}

Option defines interface for configuring riverkfq

func Pool

func Pool(p *pgxpool.Pool) Option

Pool sets the database connection pool.

func WithProducer

func WithProducer(p Producer) Option

WithProducer sets the Kafka producer.

type Producer

type Producer interface {
	Publish(ctx context.Context, msg *xkafka.Message) error
}

Producer is a Kafka producer.

type PublishArgs

type PublishArgs struct {
	Msg *xkafka.Message `json:"msg"`
}

PublishArgs is the job payload for publishing messages to Kafka.

func (PublishArgs) Kind

func (args PublishArgs) Kind() string

Kind returns the kind of the job.

type PublishQueue

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

PublishQueue is a thin wrapper around river.Client that enqueues messages are asynchronously published to Kafka.

func NewPublishQueue

func NewPublishQueue(opts ...Option) (*PublishQueue, error)

NewPublishQueue creates a new PublishQueue.

func (*PublishQueue) Add

func (q *PublishQueue) Add(ctx context.Context, msgs ...*xkafka.Message) error

Add enqueues one or more messages.

func (*PublishQueue) AddTx

func (q *PublishQueue) AddTx(ctx context.Context, tx pgx.Tx, msgs ...*xkafka.Message) error

AddTx enqueues one or more messages in a transaction.

func (*PublishQueue) Run

func (q *PublishQueue) Run(ctx context.Context) error

Run starts the client and begins publishing messages to Kafka.

type PublishWorker

type PublishWorker struct {
	river.WorkerDefaults[PublishArgs]
	// contains filtered or unexported fields
}

PublishWorker is a worker that publishes messages to Kafka.

func NewPublishWorker

func NewPublishWorker(producer Producer) *PublishWorker

NewPublishWorker creates a new PublishWorker.

func (*PublishWorker) Work

func (w *PublishWorker) Work(ctx context.Context, job *river.Job[PublishArgs]) error

Work executes the job.