pgqueue

Documentation

Overview

Package pgqueue provides a generic PostgreSQL queue implementation using the pgq library. It allows publishing and subscribing to messages of any type, with automatic queue initialization.

Example Usage:

// Create a new Subscriber for a specific payload type.
sub := pgqueue.NewSubscriber[MyPayload](db)

// Define a handler function to process messages.
handler := func(payload MyPayload) error {
    fmt.Printf("Received message: %+v\n", payload)
    return nil
}

// Subscribe to a queue.
err := sub.Subscribe(context.Background(), "my_queue", handler)
if err != nil {
    log.Fatal("Failed to start subscriber: ", err)
}

Index

• func InitQueue(db *sql.DB, queueName string) error
• type Config
  • func LoadConfig() (*Config, error)
• type Publisher
  • func NewPublisher[T any](db *sql.DB) *Publisher[T]
  • func (p *Publisher[T]) Publish(ctx context.Context, queueName string, customMetaData *pgq.Metadata, payload T) ([]uuid.UUID, error)
• type Subscriber
  • func NewSubscriber[T any](db *sql.DB) *Subscriber[T]
  • func (s *Subscriber[T]) Subscribe(ctx context.Context, queueName string, ...) error

Functions

func InitQueue

func InitQueue(db *sql.DB, queueName string) error

Types

type Config

type Config struct {
	PostgresDSN string `cf_env:"POSTGRES_DSN"`
}

func LoadConfig

func LoadConfig() (*Config, error)

type Publisher

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

func NewPublisher

func NewPublisher[T any](db *sql.DB) *Publisher[T]

NewPublisher creates a new Publisher with support for generics. Accepts a database connection (*sql.DB) and returns a pointer to Publisher[T].

func (*Publisher[T]) Publish

func (p *Publisher[T]) Publish(ctx context.Context, queueName string, customMetaData *pgq.Metadata, payload T) ([]uuid.UUID, error)

Publish publishes a message to the queue. Accepts a context (context.Context), a queue name (queueName), and a payload of any type (T). Returns an error if something goes wrong.

type Subscriber

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

func NewSubscriber

func NewSubscriber[T any](db *sql.DB) *Subscriber[T]

NewSubscriber creates a new Subscriber for a specific payload type. It accepts a database connection (*sql.DB) and returns a pointer to Subscriber[T].

Example:

sub := pgqueue.NewSubscriber[MyPayload](db)

func (*Subscriber[T]) Subscribe

func (s *Subscriber[T]) Subscribe(ctx context.Context, queueName string, handler func(payload T, meta pgq.Metadata) error) error

Subscribe subscribes to a queue and processes messages using the provided handler. It accepts a context (context.Context), a queue name (queueName), and a message handler (handler). The handler function processes messages of type T and returns an error if processing fails.