56 lines
1.8 KiB
Go
56 lines
1.8 KiB
Go
package redisnatsbroker
|
|
|
|
import (
|
|
"github.com/centrifugal/centrifugo/v6/internal/natsbroker"
|
|
|
|
"github.com/centrifugal/centrifuge"
|
|
)
|
|
|
|
// Broker is a combination of centrifuge.RedisBroker and NatsBroker. Redis is used for history
|
|
// operations and/or idempotent result cache. Nats is used for PUB/SUB part. The important limitation
|
|
// is that publications to this Broker must be sequential for the same channel. Otherwise, we
|
|
// can't guarantee message ordering and stable behavior of clients with auto recovery on. The
|
|
// benefit is more efficient fan-in in Nats PUB/SUB. Also, this allows scaling Redis Cluster
|
|
// without PUB/SUB scalability restrictions.
|
|
// This is EXPERIMENTAL.
|
|
type Broker struct {
|
|
*natsbroker.NatsBroker
|
|
redis *centrifuge.RedisBroker
|
|
}
|
|
|
|
func New(nats *natsbroker.NatsBroker, redis *centrifuge.RedisBroker) (*Broker, error) {
|
|
return &Broker{
|
|
NatsBroker: nats,
|
|
redis: redis,
|
|
}, nil
|
|
}
|
|
|
|
func (b *Broker) Publish(ch string, data []byte, opts centrifuge.PublishOptions) (centrifuge.PublishResult, error) {
|
|
if !b.NatsBroker.IsSupportedPublishChannel(ch) {
|
|
// Do not support wildcard subscriptions just like natsbroker.NatsBroker.
|
|
return centrifuge.PublishResult{}, centrifuge.ErrorBadRequest
|
|
}
|
|
if opts.IdempotencyKey != "" || (opts.HistorySize > 0 && opts.HistoryTTL > 0) {
|
|
res, err := b.redis.Publish(ch, data, opts)
|
|
if err != nil {
|
|
return res, err
|
|
}
|
|
if res.Suppressed {
|
|
return res, nil
|
|
}
|
|
_ = b.NatsBroker.PublishWithStreamPosition(ch, data, opts, res.StreamPosition)
|
|
return res, nil
|
|
}
|
|
return b.NatsBroker.Publish(ch, data, opts)
|
|
}
|
|
|
|
// History ...
|
|
func (b *Broker) History(ch string, opts centrifuge.HistoryOptions) ([]*centrifuge.Publication, centrifuge.StreamPosition, error) {
|
|
return b.redis.History(ch, opts)
|
|
}
|
|
|
|
// RemoveHistory ...
|
|
func (b *Broker) RemoveHistory(ch string) error {
|
|
return b.redis.RemoveHistory(ch)
|
|
}
|