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 *amqp.Conn, connTest *amqptest.Conn, logger *zap.Logger) (wabbit.Channel, error)
  • func (a *Adapter) CreateConn(User string, Password string, logger *zap.Logger) (*amqp.Conn, error)
  • func (a *Adapter) JsonEncode(body []byte) (interface{}, 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 *amqp.Conn, connTest *amqptest.Conn,
	logger *zap.Logger) (wabbit.Channel, error)

func (*Adapter) CreateConn

func (a *Adapter) CreateConn(User string, Password string, logger *zap.Logger) (*amqp.Conn, error)

func (*Adapter) JsonEncode

func (a *Adapter) JsonEncode(body []byte) (interface{}, 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 {
	PrefetchCount int
	GlobalQos     bool
}

type ExchangeConfig

type ExchangeConfig struct {
	Name        string `envconfig:"RABBITMQ_EXCHANGE_CONFIG_NAME" required:"false"`
	TypeOf      string `envconfig:"RABBITMQ_EXCHANGE_CONFIG_TYPE" required:"true"`
	Durable     bool   `envconfig:"RABBITMQ_EXCHANGE_CONFIG_DURABLE" required:"false"`
	AutoDeleted bool   `envconfig:"RABBITMQ_EXCHANGE_CONFIG_AUTO_DELETED" required:"false"`
	Internal    bool   `envconfig:"RABBITMQ_EXCHANGE_CONFIG_INTERNAL" required:"false"`
	NoWait      bool   `envconfig:"RABBITMQ_EXCHANGE_CONFIG_NOWAIT" required:"false"`
}

type QueueConfig

type QueueConfig struct {
	Name             string `envconfig:"RABBITMQ_QUEUE_CONFIG_NAME" required:"false"`
	RoutingKey       string `envconfig:"RABBITMQ_ROUTING_KEY" required:"true"`
	Durable          bool   `envconfig:"RABBITMQ_QUEUE_CONFIG_DURABLE" required:"false"`
	DeleteWhenUnused bool   `envconfig:"RABBITMQ_QUEUE_CONFIG_AUTO_DELETED" required:"false"`
	Exclusive        bool   `envconfig:"RABBITMQ_QUEUE_CONFIG_EXCLUSIVE" required:"false"`
	NoWait           bool   `envconfig:"RABBITMQ_QUEUE_CONFIG_NOWAIT" required:"false"`
}