stream

Documentation

Index

• func NewRedisStream(config config.MQConfig, logger zerolog.Logger) (mq.MQ, error)
• type Consumer
  • func NewConsumer(queue string, processTimeout time.Duration, logger zerolog.Logger, ...) *Consumer
  • func (consumer *Consumer) Consume(delivery rmq.Delivery)
• type RedisStream
  • func (stream *RedisStream) InitSubscriber(ctx context.Context, projectID, topicID string, subIDs ...string) error
  • func (stream *RedisStream) InitTopic(ctx context.Context, projectID, topicID string) error
  • func (stream *RedisStream) Publish(distributedID int64, projectID, topicID string, message []byte) error
  • func (stream *RedisStream) Stop()
  • func (stream *RedisStream) Subscribe(projectID, subscription string, process func(context.Context, []byte) error) error

Functions

func NewRedisStream

func NewRedisStream(config config.MQConfig, logger zerolog.Logger) (mq.MQ, error)

NewRedisStream returns a mq.MQ implement with redis

Types

type Consumer

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

func NewConsumer

func NewConsumer(queue string, processTimeout time.Duration, logger zerolog.Logger, process func(context.Context, []byte) error) *Consumer

func (*Consumer) Consume

func (consumer *Consumer) Consume(delivery rmq.Delivery)

type RedisStream

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

func (*RedisStream) InitSubscriber

func (stream *RedisStream) InitSubscriber(ctx context.Context, projectID, topicID string, subIDs ...string) error

func (*RedisStream) InitTopic

func (stream *RedisStream) InitTopic(ctx context.Context, projectID, topicID string) error

func (*RedisStream) Publish

func (stream *RedisStream) Publish(distributedID int64, projectID, topicID string, message []byte) error

func (*RedisStream) Stop

func (stream *RedisStream) Stop()

func (*RedisStream) Subscribe

func (stream *RedisStream) Subscribe(projectID, subscription string, process func(context.Context, []byte) error) error