xrabbitmq

Documentation

Index

• type Client
  • func NewClient(logger xlog.Logger, config Config) (*Client, error)
  • func (c *Client) Publish(ctx context.Context, payload Payload) error
  • func (c *Client) Stream(ctx context.Context, queue string, routingKeys []string, ready chan<- struct{}, ...)
• type Config
  • func (c *Config) ResetToDefault()
• type Connection
  • func NewConnection(url string, logger xlog.Logger) *Connection
  • func (c *Connection) Connect() error
  • func (c *Connection) GetChannel() (*amqp091.Channel, error)
• type Payload

Types

type Client

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

Client is the RabbitMQ client that manages publishing and consuming

func NewClient

func NewClient(
	logger xlog.Logger,
	config Config,
) (*Client, error)

NewClient creates a new RabbitMQ client

func (*Client) Publish

func (c *Client) Publish(ctx context.Context, payload Payload) error

func (*Client) Stream

func (c *Client) Stream(
	ctx context.Context,
	queue string,
	routingKeys []string,
	ready chan<- struct{},
	numConsumers int,
	callback func(context.Context, amqp091.Delivery) error,
)

Stream sets up N consumers on the specified queue with the given routing key

type Config

type Config struct {
	URL          string        `yaml:"url"`
	ExchangeName string        `yaml:"exchange_name"`
	RetryDelay   time.Duration `yaml:"retry_delay"`
}

func (*Config) ResetToDefault

func (c *Config) ResetToDefault()

type Connection

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

func NewConnection

func NewConnection(
	url string,
	logger xlog.Logger,
) *Connection

func (*Connection) Connect

func (c *Connection) Connect() error

Connect establishes the connection to RabbitMQ

func (*Connection) GetChannel

func (c *Connection) GetChannel() (*amqp091.Channel, error)

GetChannel returns a new channel from the connection

type Payload

type Payload struct {
	Topic       string
	ContentType string
	MessageID   string
	Timestamp   time.Time
	Body        []byte
}