rabbitmq_cli

Documentation

Index

• Constants
• Variables
• func StartRabbitWithFile(rbConfigs map[string]*RabbitConfig) error
• type ExchangeKind
  • func (kind ExchangeKind) String() string
• type Handler
• type PublishOption
• type PublishOptions
• type RabbitChannel
  • func PickupRabbitClient(ctx context.Context, rabbitName string) (*RabbitChannel, error)
  • func (r *RabbitChannel) Bind(queue, bindKey, exchange string) error
  • func (r *RabbitChannel) Close() error
  • func (r *RabbitChannel) Consume(queue string, autoAck bool) (<-chan amqp.Delivery, error)
  • func (r *RabbitChannel) DeclareExchange(name string, kind string, durable, autoDelete bool) error
  • func (r *RabbitChannel) DeclareQueue(name string, durable, autoDelete bool) (amqp.Queue, error)
  • func (r *RabbitChannel) DeleteExchange(name string) error
  • func (r *RabbitChannel) DeleteQueue(name string) (int, error)
  • func (r *RabbitChannel) DioPublish(topic string, msg []byte, opts ...PublishOption) error
  • func (r *RabbitChannel) DioSubscribe(topic string, autoAck bool, handler Handler, opts ...SubscribeOption) error
  • func (r *RabbitChannel) GetContext() context.Context
  • func (r *RabbitChannel) GetPrefix() string
  • func (r *RabbitChannel) Publish(exchange, routingKey string, msg []byte, deliveryMode uint8) error
• type RabbitConfig
• type Resource
  • func (r *Resource) Close()
• type ResourcePool
  • func NewResourcePool(config *RabbitConfig) (*ResourcePool, error)
  • func (rp *ResourcePool) Get() (*Resource, error)
• type SubscribeOption
• type SubscribeOptions

Constants

const (
	DefaultMQTimeout = time.Second * 2
	DefaultIdleTime  = 0
	DefaultHeartbeat = time.Second * 2
	DefaultCapacity  = 10
	MaxCapacity      = 30
)

Variables

var (
	ErrorOutOfCapacity = errors.New("rabbitMQ resource pool has got the Max Capacity. ")
)

Functions

func StartRabbitWithFile

func StartRabbitWithFile(rbConfigs map[string]*RabbitConfig) error

Types

type ExchangeKind

type ExchangeKind string

const (
	ExchangeKindFanout  ExchangeKind = "fanout"
	ExchangeKindDirect  ExchangeKind = "direct"
	ExchangeKindTopic   ExchangeKind = "topic"
	ExchangeKindHeaders ExchangeKind = "headers"
)

func (ExchangeKind) String

func (kind ExchangeKind) String() string

type Handler

type Handler func(amqp.Delivery) error

定义回调函数

type PublishOption

type PublishOption func(*PublishOptions)

type PublishOptions

type PublishOptions struct{}

定义publish opts

type RabbitChannel

type RabbitChannel struct {
	*amqp.Channel
	// contains filtered or unexported fields
}

func PickupRabbitClient

func PickupRabbitClient(ctx context.Context, rabbitName string) (*RabbitChannel, error)

获取rabbit客户端
ctx
rabbitName 配置中的信息

Caution!!! This function return a channel, you should close it after use.

func (*RabbitChannel) Bind

func (r *RabbitChannel) Bind(queue, bindKey, exchange string) error

 exchange和queue绑定
 参数：
	queue 队列名称
	bindkey 绑定的key
	exchange 交换器名称
 返回值：
	error 操作期间产生的错误

func (*RabbitChannel) Close

func (r *RabbitChannel) Close() error

关闭channel

func (*RabbitChannel) Consume

func (r *RabbitChannel) Consume(queue string, autoAck bool) (<-chan amqp.Delivery, error)

 消费消息队列
 参数：
	queue 队列名称
	autoAck	是否自动回复ack
 返回值：
	Delivery 传递消息的单向通道，可以通过读取该通道获取接收到的消息
	error 操作期间产生的错误

func (*RabbitChannel) DeclareExchange

func (r *RabbitChannel) DeclareExchange(name string, kind string, durable, autoDelete bool) error

 定义exchange
 参数：
	name exchange名称
	kind exchange种类
	durable 是否持久化
	autoDelete 当所有绑定的队列都与交换器解绑后，交换器会自动删除
 返回值：
	error 操作期间产生的错误

func (*RabbitChannel) DeclareQueue

func (r *RabbitChannel) DeclareQueue(name string, durable, autoDelete bool) (amqp.Queue, error)

 定义队列
 参数：
	name 队列的名称
	durable 是否持久化
	autoDelete 当所有消费者都断开时，队列会自动删除
 返回值：
	error 操作期间产生的错误

func (*RabbitChannel) DeleteExchange

func (r *RabbitChannel) DeleteExchange(name string) error

  删除 exchange
  参数：
	name exchange名称
  返回值：
	error 操作期间产生的错误

func (*RabbitChannel) DeleteQueue

func (r *RabbitChannel) DeleteQueue(name string) (int, error)

  删除队列
  参数：
	name 队列名称
  返回值：
	int 清除的消息数
	error 操作期间产生的错误

func (*RabbitChannel) DioPublish

func (r *RabbitChannel) DioPublish(topic string, msg []byte, opts ...PublishOption) error

  简单发布消息
  参数：
	topic 发布消息的topic
 	msg 消息内容
    opts 预留opts操作

func (*RabbitChannel) DioSubscribe

func (r *RabbitChannel) DioSubscribe(topic string, autoAck bool, handler Handler, opts ...SubscribeOption) error

  简单消费消息
  参数：
	topic 消费消息的topic
    opts 预留opts操作

func (*RabbitChannel) GetContext

func (r *RabbitChannel) GetContext() context.Context

func (*RabbitChannel) GetPrefix

func (r *RabbitChannel) GetPrefix() string

func (*RabbitChannel) Publish

func (r *RabbitChannel) Publish(exchange, routingKey string, msg []byte, deliveryMode uint8) error

 发送消息到指定的exchange
 参数：
	exchange 制定交换机
	routingKey 路由key
 	msg 消息内容
    deliveryMode 持久化 Transient (0 or 1) or Persistent (2)
 返回值：
	error 操作期间产生的错误

type RabbitConfig

type RabbitConfig struct {
	Key string
	URL string
}

type Resource

type Resource struct {
	*amqp.Connection
	// contains filtered or unexported fields
}

Resource adapts a client connection to a Vitess Resource.

func (*Resource) Close

func (r *Resource) Close()

Close is put the conn to the poll

type ResourcePool

type ResourcePool struct {
	*pools.ResourcePool
}

func NewResourcePool

func NewResourcePool(config *RabbitConfig) (*ResourcePool, error)

NewResourcePool create a resource pool

func (*ResourcePool) Get

func (rp *ResourcePool) Get() (*Resource, error)

type SubscribeOption

type SubscribeOption func(*SubscribeOptions)

type SubscribeOptions

type SubscribeOptions struct{}

定义subscribe