rabbitmq

Documentation

Index

• func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, ...) adapter.MessageAdapter
• func NewEnvConfig() adapter.EnvConfigAccessor
• type Adapter
  • func (a *Adapter) ConsumeMessages(channel *wabbit.Channel, queue *wabbit.Queue, logger *zap.Logger) (<-chan wabbit.Delivery, error)
  • func (a *Adapter) CreateChannel(conn wabbit.Conn, connTest *amqptest.Conn, logger *zap.Logger) (wabbit.Channel, error)
  • func (a *Adapter) CreateConn(user string, password string, logger *zap.Logger) (wabbit.Conn, error)
  • func (a *Adapter) PollForMessages(channel *wabbit.Channel, queue *wabbit.Queue, stopCh <-chan struct{}) error
  • func (a *Adapter) Start(ctx context.Context) error
  • func (a *Adapter) StartAmqpClient(ch *wabbit.Channel) (*wabbit.Queue, error)
• type ChannelConfig
• type ExchangeConfig
• type QueueConfig

Functions

func NewAdapter

func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, httpMessageSender *kncloudevents.HTTPMessageSender, reporter source.StatsReporter) adapter.MessageAdapter

func NewEnvConfig

func NewEnvConfig() adapter.EnvConfigAccessor

Types

type Adapter

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

func (*Adapter) ConsumeMessages

func (a *Adapter) ConsumeMessages(channel *wabbit.Channel,
	queue *wabbit.Queue, logger *zap.Logger) (<-chan wabbit.Delivery, error)

func (*Adapter) CreateChannel

func (a *Adapter) CreateChannel(conn wabbit.Conn, connTest *amqptest.Conn,
	logger *zap.Logger) (wabbit.Channel, error)

func (*Adapter) CreateConn

func (a *Adapter) CreateConn(user string, password string, logger *zap.Logger) (wabbit.Conn, error)

func (*Adapter) PollForMessages

func (a *Adapter) PollForMessages(channel *wabbit.Channel,
	queue *wabbit.Queue, stopCh <-chan struct{}) error

func (*Adapter) Start

func (a *Adapter) Start(ctx context.Context) error

func (*Adapter) StartAmqpClient

func (a *Adapter) StartAmqpClient(ch *wabbit.Channel) (*wabbit.Queue, error)

type ChannelConfig

type ChannelConfig struct {
	Parallelism int  `envconfig:"RABBITMQ_CHANNEL_CONFIG_PARALLELISM" default:"1" required:"false"`
	GlobalQos   bool `envconfig:"RABBITMQ_CHANNEL_CONFIG_QOS_GLOBAL" required:"false"`
}

type ExchangeConfig

type ExchangeConfig struct {
	Name       string `envconfig:"RABBITMQ_EXCHANGE_CONFIG_NAME" required:"false"`
	Type       string `envconfig:"RABBITMQ_EXCHANGE_CONFIG_TYPE" required:"true"`
	Durable    bool   `envconfig:"RABBITMQ_EXCHANGE_CONFIG_DURABLE" required:"false"`
	AutoDelete bool   `envconfig:"RABBITMQ_EXCHANGE_CONFIG_AUTO_DELETE" required:"false"`
}

type QueueConfig

type QueueConfig struct {
	Name       string `envconfig:"RABBITMQ_QUEUE_CONFIG_NAME" required:"true"`
	RoutingKey string `envconfig:"RABBITMQ_ROUTING_KEY" required:"true"`
	Durable    bool   `envconfig:"RABBITMQ_QUEUE_CONFIG_DURABLE" required:"false"`
	AutoDelete bool   `envconfig:"RABBITMQ_QUEUE_CONFIG_AUTO_DELETE" required:"false"`
}