redispub

Documentation

Index

• Constants
• Variables
• type Channel
  • func (r *Channel) Close()
• type PubSub
  • func NewRedisPubSub(config PubSubConfig) (*PubSub, error)
  • func (r *PubSub) Conn() sabuhp.Conn
  • func (r *PubSub) Listen(topic string, handler sabuhp.TransportResponse) sabuhp.Channel
  • func (r *PubSub) SendToAll(data *sabuhp.Message, timeout time.Duration) error
  • func (r *PubSub) SendToOne(data *sabuhp.Message, timeout time.Duration) error
  • func (r *PubSub) Start()
  • func (r *PubSub) Stop()
  • func (r *PubSub) UnListen(topic string)
  • func (r *PubSub) Wait()
• type PubSubConfig

Constants

const (
	GroupExistErrorMsg        = "BUSYGROUP Consumer Group name already exists"
	SubscriptionExistsAlready = "Topic 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 PubSub

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

func NewRedisPubSub

func NewRedisPubSub(config PubSubConfig) (*PubSub, error)

func (*PubSub) Conn

func (r *PubSub) Conn() sabuhp.Conn

func (*PubSub) Listen

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

func (*PubSub) SendToAll

func (r *PubSub) SendToAll(data *sabuhp.Message, timeout time.Duration) error

func (*PubSub) SendToOne

func (r *PubSub) SendToOne(data *sabuhp.Message, timeout time.Duration) error

func (*PubSub) Start

func (r *PubSub) Start()

func (*PubSub) Stop

func (r *PubSub) Stop()

func (*PubSub) UnListen

func (r *PubSub) UnListen(topic string)

func (*PubSub) Wait

func (r *PubSub) Wait()

type PubSubConfig

type PubSubConfig 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
}