queuefx

Feb 25, 2025

Changes in this version

type AmqpBroker

+ func (broker *AmqpBroker) Consume(ctx context.Context, queueName string, config ConsumerConfig) (<-chan Message, <-chan error)

+ type ConsumerConfig struct

+ Args map[string]any

+ AutoAck bool

+ Exclusive bool

+ NoLocal bool

+ NoWait bool

+ func DefaultConsumerConfig() ConsumerConfig

+ type Message struct

+ Body []byte

+ Headers map[string]any

+ func (m *Message) Ack() error

+ func (m *Message) Nack(requeue bool) error

Feb 12, 2025

Changes in this version

type Registry

+ func (registry *Registry) GetDefault() Broker

+ func (registry *Registry) GetNamed(name string) Broker

Feb 6, 2025

Changes in this version

+ const DefaultBroker

+ var ErrUnableToDetermineDialect = errors.New("unable to determine dialect")

+ var ErrUnknownProvider = errors.New("unknown provider")

+ type AmqpBroker struct

+ func NewAmqpBroker(ctx context.Context, dialect Dialect, dsn string) (*AmqpBroker, error)

+ func (broker *AmqpBroker) GetDialect() Dialect

+ func (broker *AmqpBroker) Publish(ctx context.Context, name string, body []byte) error

+ func (broker *AmqpBroker) QueueDeclare(ctx context.Context, name string) (string, error)

+ type Broker interface

+ GetDialect func() Dialect

+ Publish func(ctx context.Context, name string, body []byte) error

+ QueueDeclare func(ctx context.Context, name string) (string, error)

+ type Config struct

+ Brokers map[string]ConfigBroker

+ type ConfigBroker struct

+ DSN string

+ Provider string

+ type Dialect string

+ const DialectAmqp

+ func DetermineDialect(provider string, dsn string) (Dialect, error)

+ type Registry struct

+ func NewRegistry(logger *slog.Logger) *Registry

+ func (registry *Registry) AddConnection(ctx context.Context, name string, provider string, dsn string) error

+ func (registry *Registry) GetDefaultSql() Broker

+ func (registry *Registry) GetNamedSql(name string) Broker

+ func (registry *Registry) LoadFromConfig(ctx context.Context, config *Config) error