redis_stream

Documentation

Index

• type ConsumerCallback
• type ConsumerGroup
  • func (c *ConsumerGroup) Callback(callback ConsumerCallback)
  • func (c *ConsumerGroup) Listen() error
• type ConsumerOption
• type Payload
  • func (p *Payload) MarshalBinary() (data []byte, err error)
  • func (p *Payload) UnmarshalBinary(data []byte) error
• type RedisStream
  • func New(client2 *redis.Client) (*RedisStream, error)
  • func (r *RedisStream) Add(stream string, payload any) error
  • func (r *RedisStream) CreateConsumerGroup(stream string, option *ConsumerOption) *ConsumerGroup
  • func (r *RedisStream) Delay(stream string, payload any, duration time.Duration) error

Types

type ConsumerCallback

type ConsumerCallback func(message *Payload) error

type ConsumerGroup

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

func (*ConsumerGroup) Callback

func (c *ConsumerGroup) Callback(callback ConsumerCallback)

func (*ConsumerGroup) Listen

func (c *ConsumerGroup) Listen() error

type ConsumerOption

type ConsumerOption struct {
	ConsumerCount int
	FetchCount    int64
	StreamMaxLen  int64
	IsDelay       bool
}

type Payload

type Payload struct {
	ID      string `json:"id"`
	Payload any    `json:"payload"`
	Stream  string `json:"stream"`
	// contains filtered or unexported fields
}

func (*Payload) MarshalBinary

func (p *Payload) MarshalBinary() (data []byte, err error)

func (*Payload) UnmarshalBinary

func (p *Payload) UnmarshalBinary(data []byte) error

type RedisStream

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

func New

func New(client2 *redis.Client) (*RedisStream, error)

func (*RedisStream) Add

func (r *RedisStream) Add(stream string, payload any) error

func (*RedisStream) CreateConsumerGroup

func (r *RedisStream) CreateConsumerGroup(stream string, option *ConsumerOption) *ConsumerGroup

func (*RedisStream) Delay

func (r *RedisStream) Delay(stream string, payload any, duration time.Duration) error