rabbitmq

Documentation

Index

• type RabbitMQ
  • func NewRabbitMQ(ctx context.Context, cfg RabbitMQConfig) (*RabbitMQ, error)
  • func (r *RabbitMQ) Connect() error
  • func (r *RabbitMQ) Disconnect() error
  • func (r *RabbitMQ) Install(args ...any) error
  • func (r *RabbitMQ) Publish(ctx context.Context, message any, attributes map[string]string) (string, error)
  • func (r *RabbitMQ) RegisterReceiver(receiver port.PubSubReceiver)
  • func (r *RabbitMQ) StartReceiving(ctx context.Context)
  • func (r *RabbitMQ) Uninstall() error
• type RabbitMQConfig
  • func (c *RabbitMQConfig) SetDefaults() map[string]any
  • func (c *RabbitMQConfig) SetEnvBindings() map[string]string
• type RabbitMQLoader
  • func (l *RabbitMQLoader) Init(args ...any) (port.Library, error)
  • func (a *RabbitMQLoader) Name() string
  • func (a *RabbitMQLoader) SetName(name string)
• type RabbitMQMessage
  • func (m *RabbitMQMessage) GetAttributes() map[string]string
  • func (m *RabbitMQMessage) GetData() []byte
  • func (m *RabbitMQMessage) GetID() string
  • func (m *RabbitMQMessage) GetPublishTime() time.Time

Types

type RabbitMQ

type RabbitMQ struct {
	Connection *amqp.Connection
	Channel    *amqp.Channel
	Config     RabbitMQConfig
	Receivers  []port.PubSubReceiver
}

func NewRabbitMQ

func NewRabbitMQ(ctx context.Context, cfg RabbitMQConfig) (*RabbitMQ, error)

func (*RabbitMQ) Connect

func (r *RabbitMQ) Connect() error

func (*RabbitMQ) Disconnect

func (r *RabbitMQ) Disconnect() error

func (*RabbitMQ) Install

func (r *RabbitMQ) Install(args ...any) error

func (*RabbitMQ) Publish

func (r *RabbitMQ) Publish(ctx context.Context, message any, attributes map[string]string) (string, error)

func (*RabbitMQ) RegisterReceiver

func (r *RabbitMQ) RegisterReceiver(receiver port.PubSubReceiver)

func (*RabbitMQ) StartReceiving

func (r *RabbitMQ) StartReceiving(ctx context.Context)

func (*RabbitMQ) Uninstall

func (r *RabbitMQ) Uninstall() error

type RabbitMQConfig

type RabbitMQConfig struct {
	config.PubSubConfig

	Uri           string `mapstructure:"uri"`           // amqp://user:pass@host:port/vhost
	ExchangeType  string `mapstructure:"exchange_type"` // direct, fanout, topic, headers, none (tanpa exchage = message queue biasa)
	Durable       bool   `mapstructure:"durable"`
	AutoDelete    bool   `mapstructure:"auto_delete"`
	Exclusive     bool   `mapstructure:"exclusive"`
	PrefetchCount int    `mapstructure:"prefetch_count"`
}

func (*RabbitMQConfig) SetDefaults

func (c *RabbitMQConfig) SetDefaults() map[string]any

func (*RabbitMQConfig) SetEnvBindings

func (c *RabbitMQConfig) SetEnvBindings() map[string]string

type RabbitMQLoader

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

func (*RabbitMQLoader) Init

func (l *RabbitMQLoader) Init(args ...any) (port.Library, error)

func (*RabbitMQLoader) Name

func (a *RabbitMQLoader) Name() string

func (*RabbitMQLoader) SetName

func (a *RabbitMQLoader) SetName(name string)

type RabbitMQMessage

type RabbitMQMessage struct {
	ID          string
	Data        []byte
	PublishTime time.Time
	Attributes  map[string]string
}

func (*RabbitMQMessage) GetAttributes

func (m *RabbitMQMessage) GetAttributes() map[string]string

func (*RabbitMQMessage) GetData

func (m *RabbitMQMessage) GetData() []byte

func (*RabbitMQMessage) GetID

func (m *RabbitMQMessage) GetID() string

func (*RabbitMQMessage) GetPublishTime

func (m *RabbitMQMessage) GetPublishTime() time.Time