confrabbit

package
v0.4.8 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 22, 2026 License: MIT Imports: 20 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	With  = mq.With[ProducerMessage, ConsumerMessage]
	From  = mq.From[ProducerMessage, ConsumerMessage]
	Must  = mq.Must[ProducerMessage, ConsumerMessage]
	Carry = mq.Carry[ProducerMessage, ConsumerMessage]
)

Functions

func WithPubExchange 

func WithPubExchange(name, kind string) mq.OptionApplier

WithPubExchange explicit declares the destination Exchange for the publisher. 'name' is the exchange name, 'kind' could be "direct", "fanout", "topic" or "headers".

func WithPubRoutingKey 

func WithPubRoutingKey(routingKey string) mq.OptionApplier

WithPubRoutingKey is an alias of WithPubTopic for clearer intention in RabbitMQ.

func WithPubTimeout 

func WithPubTimeout(d time.Duration) mq.OptionApplier

func WithPubTopic 

func WithPubTopic(topic string) mq.OptionApplier

WithPubTopic sets the topic for publisher. NOTE: In RabbitMQ, Topic concept is mapped to 'Routing Key'.

func WithPublishCallback 

func WithPublishCallback(f mq.AsyncPubCallback[ProducerMessage]) mq.OptionApplier

func WithRabbitConsumerOptions 

func WithRabbitConsumerOptions(opts ...func(*rabbitmq.ConsumerOptions)) mq.OptionApplier

func WithRabbitPublisherOptions 

func WithRabbitPublisherOptions(opts ...func(*rabbitmq.PublisherOptions)) mq.OptionApplier

func WithSubBufferSize 

func WithSubBufferSize(n uint16) mq.OptionApplier

func WithSubCallback 

func WithSubCallback(f mq.SubCallback[ConsumerMessage]) mq.OptionApplier

func WithSubConsumeMode 

func WithSubConsumeMode(mode mq.ConsumeHandleMode) mq.OptionApplier

func WithSubDisableAutoAck 

func WithSubDisableAutoAck() mq.OptionApplier

func WithSubExchange 

func WithSubExchange(name, kind string) mq.OptionApplier

WithSubExchange explicit declares the source Exchange for the consumer, and implicitly binds the Queue to this Exchange using the Queue name as RoutingKey.

func WithSubHasher 

func WithSubHasher(h mq.Hasher) mq.OptionApplier

func WithSubQueue 

func WithSubQueue(queue string) mq.OptionApplier

func WithSubRoutingKey 

func WithSubRoutingKey(routingKey string) mq.OptionApplier

WithSubRoutingKey explicit sets the routing key for consumer binding. Note: If you want to use wildcards like `*.info` or `#`, use this option along with `WithSubExchange(name, "topic")`.

func WithSubWorker 

func WithSubWorker(n uint16) mq.OptionApplier

func WithSyncPublish 

func WithSyncPublish() mq.OptionApplier

Types

type ConsumerMessage 

type ConsumerMessage interface {
	mq.HasTopic
	mq.HasPayload
	mq.HasExtra
	mq.HasPartitionKey
	mq.HasPublishedAt
	mq.HasConsumedAt
	mq.CanRefreshConsumedAt
	mq.HasLatency
	mq.HasUnderlying[amqp.Delivery]
}

func NewConsumerMessage 

func NewConsumerMessage(u amqp.Delivery) ConsumerMessage

type Endpoint 

type Endpoint struct {
	types.Endpoint[Option]

	mq.ResourceManager `env:"-"`
	// contains filtered or unexported fields
}

func (*Endpoint) Close 

func (e *Endpoint) Close() error

func (*Endpoint) Init 

func (e *Endpoint) Init(ctx context.Context) error

func (*Endpoint) LivenessCheck 

func (e *Endpoint) LivenessCheck(ctx context.Context) (v liveness.Result)

func (*Endpoint) NewConsumer 

func (e *Endpoint) NewConsumer(ctx context.Context, options ...mq.OptionApplier) (_ mq.Consumer[ConsumerMessage], err error)

func (*Endpoint) NewProducer 

func (e *Endpoint) NewProducer(ctx context.Context, options ...mq.OptionApplier) (_ mq.Producer[ProducerMessage], err error)

func (*Endpoint) SetDefault 

func (e *Endpoint) SetDefault()

type Error 

type Error int8

Error presents +genx:code 

const (
	ECODE_UNDEFINED             Error = iota
	ECODE__CLI_CLOSED                 // client closed
	ECODE__SUB_CLOSED                 // subscriber closed
	ECODE__SUB_BOOTED                 // subscriber is already booted
	ECODE__SUB_HANDLER_PANICKED       // subscriber handler panicked
	ECODE__PUB_CLOSED                 // publisher closed
	ECODE__PUB_INVALID_MESSAGE        // publisher got invalid message
	ECODE__SUB_UNSUBSCRIBED           // subscriber unsubscribed
)

type Option 

type Option struct {
	Addresses []string
	Shuffle   bool

	ReconnectInterval types.Duration `url:",default=5s"`
	Vhost             string         `url:",default=/"`
	TLS               conftls.X509KeyPair

	PubTimeout types.Duration `url:",default=2s"`
	// contains filtered or unexported fields
}

func (*Option) ClientOptions 

func (o *Option) ClientOptions() []func(*rabbitmq.ConnectionOptions)

func (*Option) PubOption 

func (o *Option) PubOption(appliers ...mq.OptionApplier) *PubOption

func (*Option) SetDefault 

func (o *Option) SetDefault()

func (*Option) SubOption 

func (o *Option) SubOption(appliers ...mq.OptionApplier) *SubOption

func (*Option) URLs 

func (o *Option) URLs(main string) []string

type ProducerMessage 

type ProducerMessage interface {
	mq.HasTopic
	mq.CanSetTopic
	mq.HasPayload
	mq.CanSetPayload
	mq.HasExtra
	mq.CanAppendExtra
	mq.HasPublishedAt
	mq.CanRefreshPublishedAt
	mq.HasDelay
	mq.CanSetDelay
	mq.HasPartitionKey
	mq.CanSetPartitionKey
	mq.HasUnderlying[*amqp.Publishing]
}

func NewProducerMessage 

func NewProducerMessage(topic string, payload []byte) ProducerMessage

type PubOption 

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

func (*PubOption) OptionScheme 

func (*PubOption) OptionScheme() string

type SubOption 

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

func (*SubOption) OptionScheme 

func (*SubOption) OptionScheme() string

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.