streamredis

Documentation

Index

• Constants
• type Consumer
  • func NewConsumer(config config.Config) (consumer *Consumer, err error)
  • func (consumer *Consumer) Close()
  • func (consumer *Consumer) Connect() (err error)
  • func (consumer *Consumer) Consume(queue string, callback queue.ConsumeCallback) (err error)
  • func (consumer *Consumer) Start()
  • func (consumer *Consumer) Unconsume(queue string)
• type ConsumerOptions
• type MsgBody
  • func (m *MsgBody) Body() map[string]interface{}
  • func (m *MsgBody) Header() map[string]string
  • func (m *MsgBody) String() string
• type Producer
  • func NewProducer(config config.Config) (m *Producer, err error)
  • func (c *Producer) Close() error
  • func (c *Producer) Count(key string) (int64, error)
  • func (c *Producer) Push(key string, msg queue.Message) error
• type ProductOptions
• type QueueItem

Constants

const (
	Proto = "streamredis"
)

Types

type Consumer

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

Consumer Consumer

func NewConsumer

func NewConsumer(config config.Config) (consumer *Consumer, err error)

NewConsumerByConfig 创建新的Consumer

func (*Consumer) Close

func (consumer *Consumer) Close()

Close 关闭当前连接

func (*Consumer) Connect

func (consumer *Consumer) Connect() (err error)

Connect 连接服务器

func (*Consumer) Consume

func (consumer *Consumer) Consume(queue string, callback queue.ConsumeCallback) (err error)

Consume 注册消费信息

func (*Consumer) Start

func (consumer *Consumer) Start()

func (*Consumer) Unconsume

func (consumer *Consumer) Unconsume(queue string)

UnConsume 取消注册消费

type ConsumerOptions

type ConsumerOptions struct {
	Concurrency     int `json:"concurrency" yaml:"concurrency"`
	BufferSize      int `json:"buffer_size" yaml:"buffer_size"`
	BlockingTimeout int `json:"blocking_timeout" yaml:"blocking_timeout"`
}

type MsgBody

type MsgBody struct {
	HeaderMap xtypes.SMap `json:"header"`
	BodyMap   xtypes.XMap `json:"body"`
	// contains filtered or unexported fields
}

func (*MsgBody) Body

func (m *MsgBody) Body() map[string]interface{}

func (*MsgBody) Header

func (m *MsgBody) Header() map[string]string

func (*MsgBody) String

func (m *MsgBody) String() string

type Producer

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

Producer memcache配置文件

func NewProducer

func NewProducer(config config.Config) (m *Producer, err error)

NewProducerByConfig 根据配置文件创建一个redis连接

func (*Producer) Close

func (c *Producer) Close() error

Close 释放资源

func (*Producer) Count

func (c *Producer) Count(key string) (int64, error)

Count 获取列表中的元素个数

func (*Producer) Push

func (c *Producer) Push(key string, msg queue.Message) error

Push 向存于 key 的列表的尾部插入所有指定的值

type ProductOptions

type ProductOptions struct {
	StreamMaxLength int64 `json:"stream_max_len" yaml:"stream_max_len"`
}

type QueueItem

type QueueItem struct {
}