sources

Documentation

Index

• type RedisStream
  • func NewRedisStream(config *redis.Options, name, step string) (*RedisStream, error)
  • func (r *RedisStream) Consume(i int, batchSize int64, channel types.Channel) error
  • func (r *RedisStream) Push(record interface{}, shard *types.Shard) error
  • func (r *RedisStream) Read(options *types.ReadOptions, channel types.Channel)

Types

type RedisStream

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

RedisStream implements Redis Streams

func NewRedisStream

func NewRedisStream(config *redis.Options, name, step string) (*RedisStream, error)

NewRedisStream implements a new RedisStream instance

func (*RedisStream) Consume

func (r *RedisStream) Consume(i int, batchSize int64, channel types.Channel) error

Consume will consume messages from the stream

func (*RedisStream) Push

func (r *RedisStream) Push(record interface{}, shard *types.Shard) error

Push will push a record to the stream

func (*RedisStream) Read

func (r *RedisStream) Read(options *types.ReadOptions, channel types.Channel)

Read starts to read from ConsumerCount at rate of BatchSize