redispub

Documentation

Index

• Constants
• Variables
• type Channel
  • func (r *Channel) Close()
• type Config
• type MessageChannel
• type RedisMessageBus
  • func NewRedisMessageBus(config Config, client *redis.Client, channel MessageChannel) *RedisMessageBus
  • func PubSub(config Config) (*RedisMessageBus, error)
  • func Stream(config Config) (*RedisMessageBus, error)
  • func (r *RedisMessageBus) Listen(topic string, grp string, handler sabuhp.TransportResponse) sabuhp.Channel
  • func (r *RedisMessageBus) ListenPubSub(topic string, grp string, handler sabuhp.TransportResponse) sabuhp.Channel
  • func (r *RedisMessageBus) ListenStream(streamTopic string, grp string, handler sabuhp.TransportResponse) sabuhp.Channel
  • func (r *RedisMessageBus) Send(data ...sabuhp.Message)
  • func (r *RedisMessageBus) SendForReply(tm time.Duration, fromTopic sabuhp.Topic, replyGroup string, ...) *nthen.Future
  • func (r *RedisMessageBus) Start()
  • func (r *RedisMessageBus) Stop()
  • func (r *RedisMessageBus) Wait()

Constants

const (
	GroupExistErrorMsg        = "BUSYGROUP Consumer Group name already exists"
	SubscriptionExistsAlready = "String is already subscribed to"
)

Variables

var (
	DefaultMessageBatchCount = 200
	DefaultMessageBatchWait  = 700 * time.Millisecond
)

Types

type Channel

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

Channel implements the sabuhp.Channel interface.

func (*Channel) Close

func (r *Channel) Close()

type Config

type Config struct {
	Logger                    sabuhp.Logger
	Ctx                       context.Context
	Codec                     sabuhp.Codec
	Redis                     redis.Options
	MaxWaitForSubConfirmation time.Duration
	StreamMessageInterval     time.Duration
	MaxWaitForSubRetry        int
	MaxMessageBatch           int
	MaxMessageBatchWait       time.Duration
}

type MessageChannel

type MessageChannel int

const (
	RedisPubSub MessageChannel = iota
	RedisStreams
)

type RedisMessageBus

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

func NewRedisMessageBus

func NewRedisMessageBus(config Config, client *redis.Client, channel MessageChannel) *RedisMessageBus

func PubSub

func PubSub(config Config) (*RedisMessageBus, error)

func Stream

func Stream(config Config) (*RedisMessageBus, error)

func (*RedisMessageBus) Listen

func (r *RedisMessageBus) Listen(topic string, grp string, handler sabuhp.TransportResponse) sabuhp.Channel

func (*RedisMessageBus) ListenPubSub

func (r *RedisMessageBus) ListenPubSub(topic string, grp string, handler sabuhp.TransportResponse) sabuhp.Channel

func (*RedisMessageBus) ListenStream

func (r *RedisMessageBus) ListenStream(streamTopic string, grp string, handler sabuhp.TransportResponse) sabuhp.Channel

func (*RedisMessageBus) Send

func (r *RedisMessageBus) Send(data ...sabuhp.Message)

func (*RedisMessageBus) SendForReply

func (r *RedisMessageBus) SendForReply(tm time.Duration, fromTopic sabuhp.Topic, replyGroup string, data ...sabuhp.Message) *nthen.Future

func (*RedisMessageBus) Start

func (r *RedisMessageBus) Start()

func (*RedisMessageBus) Stop

func (r *RedisMessageBus) Stop()

func (*RedisMessageBus) Wait

func (r *RedisMessageBus) Wait()