rabbitmq

Documentation

Index

• func NewRabbitMQ(tag string, tracing tracing.Tracing, lo logger.Logger, ...) database.RabbitMQ
• type Consumer
  • func NewConsumer(tracing tracing.Tracing, lg logger.Logger, dialer *Dialer, ...) *Consumer
  • func (c *Consumer) Init()
  • func (c *Consumer) OnSession(queue string, sub Session, messages chan<- Message)
  • func (c *Consumer) Subscribe(sessions chan chan Session, messages chan<- Message)
  • func (c *Consumer) Write() chan<- Message
• type Dialer
  • func NewDialer(id string, lg logger.Logger) *Dialer
  • func (c *Dialer) Dial(cx context.Context, config interfaces.RabbitMQProviderConfig, onError OnError, ...)
  • func (c *Dialer) IsConnected() bool
• type Message
• type MsgSend
• type OnConnected
• type OnError
• type PendingQue
• type Producer
  • func NewProducer(lg logger.Logger, dialer *Dialer, config database.RabbitMQProviderConfig, ...) *Producer
  • func (c *Producer) Init()
  • func (c *Producer) Publish(sessions chan chan Session)
  • func (c *Producer) SendingData(id string, key string, body interface{}, headers map[string]interface{}, ...)
• type RabbitMQ
  • func (c *RabbitMQ) Consumer(options database.RabbitMQOptions, callback database.ConsumerCallback)
  • func (c *RabbitMQ) Producer(options database.RabbitMQOptions)
  • func (c *RabbitMQ) Push(ctx context.Context, id, exchange, key string, body interface{}, ...) error
• type Session
  • func (s *Session) Close() error
• type Stores
  • func NewStore() *Stores
  • func (c *Stores) Delete(id string)
  • func (c *Stores) Get(id string) (cb database.ConsumerCallback, ok bool)
  • func (c *Stores) LoadClient(id string) (client *RabbitMQ, ok bool)
  • func (c *Stores) LoadClientByTag(ta string) (client *RabbitMQ, ok bool)
  • func (c *Stores) Put(id string, cb database.ConsumerCallback)
  • func (c *Stores) StoreClient(client *RabbitMQ)

Functions

func NewRabbitMQ

func NewRabbitMQ(tag string, tracing tracing.Tracing,
	lo logger.Logger, config database.RabbitMQProviderConfig) database.RabbitMQ

Types

type Consumer

type Consumer struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func NewConsumer

func NewConsumer(tracing tracing.Tracing, lg logger.Logger, dialer *Dialer, config database.RabbitMQProviderConfig,
	options database.RabbitMQOptions,
	callback database.ConsumerCallback,
	store *Stores) *Consumer

func (*Consumer) Init

func (c *Consumer) Init()

func (*Consumer) OnSession

func (c *Consumer) OnSession(queue string, sub Session, messages chan<- Message)

func (*Consumer) Subscribe

func (c *Consumer) Subscribe(sessions chan chan Session, messages chan<- Message)

func (*Consumer) Write

func (c *Consumer) Write() chan<- Message

type Dialer

type Dialer struct {
	Session chan chan Session

	sync.RWMutex
	// contains filtered or unexported fields
}

func NewDialer

func NewDialer(id string, lg logger.Logger) *Dialer

func (*Dialer) Dial

func (c *Dialer) Dial(cx context.Context, config interfaces.RabbitMQProviderConfig, onError OnError, onConnected OnConnected)

Dial - redial connects to the URL, exiting the program when no longer possible

func (*Dialer) IsConnected

func (c *Dialer) IsConnected() bool

type Message

type Message struct {
	Headers map[string]interface{}
	Body    []byte
}

type MsgSend

type MsgSend struct {
	ID      string
	Data    []byte
	Name    string
	Headers map[string]interface{}
}

type OnConnected

type OnConnected func()

type OnError

type OnError func(error)

type PendingQue

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

type Producer

type Producer struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func NewProducer

func NewProducer(lg logger.Logger, dialer *Dialer,
	config database.RabbitMQProviderConfig,
	options database.RabbitMQOptions, store *Stores) *Producer

func (*Producer) Init

func (c *Producer) Init()

func (*Producer) Publish

func (c *Producer) Publish(sessions chan chan Session)

func (*Producer) SendingData

func (c *Producer) SendingData(id string, key string, body interface{}, headers map[string]interface{}, cb database.ConsumerCallback)

type RabbitMQ

type RabbitMQ struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func (*RabbitMQ) Consumer

func (c *RabbitMQ) Consumer(options database.RabbitMQOptions, callback database.ConsumerCallback)

func (*RabbitMQ) Producer

func (c *RabbitMQ) Producer(options database.RabbitMQOptions)

func (*RabbitMQ) Push

func (c *RabbitMQ) Push(ctx context.Context, id, exchange, key string, body interface{}, cb database.ConsumerCallback) error

type Session

type Session struct {
	*amqp.Connection
	*amqp.Channel
}

func (*Session) Close

func (s *Session) Close() error

Close tears the connection down, taking the channel with it.

type Stores

type Stores struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func NewStore

func NewStore() *Stores

func (*Stores) Delete

func (c *Stores) Delete(id string)

func (*Stores) Get

func (c *Stores) Get(id string) (cb database.ConsumerCallback, ok bool)

func (*Stores) LoadClient

func (c *Stores) LoadClient(id string) (client *RabbitMQ, ok bool)

func (*Stores) LoadClientByTag

func (c *Stores) LoadClientByTag(ta string) (client *RabbitMQ, ok bool)

func (*Stores) Put

func (c *Stores) Put(id string, cb database.ConsumerCallback)

func (*Stores) StoreClient

func (c *Stores) StoreClient(client *RabbitMQ)