natsbroker

Documentation

Overview

Package natsbroker defines custom Nats Broker for Centrifuge library.

Index

• type Config
• type NatsBroker
  • func New(n *centrifuge.Node, conf Config) (*NatsBroker, error)
  • func (b *NatsBroker) Close(_ context.Context) error
  • func (b *NatsBroker) History(_ string, _ centrifuge.HistoryOptions) ([]*centrifuge.Publication, centrifuge.StreamPosition, error)
  • func (b *NatsBroker) IsSupportedPublishChannel(ch string) bool
  • func (b *NatsBroker) IsSupportedSubscribeChannel(ch string) bool
  • func (b *NatsBroker) Publish(ch string, data []byte, opts centrifuge.PublishOptions) (centrifuge.StreamPosition, bool, error)
  • func (b *NatsBroker) PublishControl(data []byte, nodeID, _ string) error
  • func (b *NatsBroker) PublishJoin(ch string, info *centrifuge.ClientInfo) error
  • func (b *NatsBroker) PublishLeave(ch string, info *centrifuge.ClientInfo) error
  • func (b *NatsBroker) PublishWithStreamPosition(ch string, data []byte, opts centrifuge.PublishOptions, ...) error
  • func (b *NatsBroker) RemoveHistory(_ string) error
  • func (b *NatsBroker) Run(h centrifuge.BrokerEventHandler) error
  • func (b *NatsBroker) Subscribe(ch string) error
  • func (b *NatsBroker) Unsubscribe(ch string) error
• type RawModeConfig

Types

type Config

type Config struct {
	// URL is a Nats server URL.
	URL string
	// Prefix allows customizing channel prefix in Nats to work with a single Nats from different
	// unrelated Centrifugo setups.
	Prefix string
	// DialTimeout is a timeout for establishing connection to Nats.
	DialTimeout time.Duration
	// WriteTimeout is a timeout for write operation to Nats.
	WriteTimeout time.Duration
	// TLS for the Nats connection. TLS is not used if nil.
	TLS *tls.Config

	// AllowWildcards allows to enable wildcard subscriptions. By default, wildcard subscriptions
	// are not allowed. Using wildcard subscriptions can't be combined with join/leave events and presence
	// because subscriptions do not belong to a concrete channel after with wildcards, while join/leave events
	// require concrete channel to be published. And presence does not make a lot of sense for wildcard
	// subscriptions - there could be subscribers which use different mask, but still receive subset of updates.
	// It's required to use channels without wildcards to for mentioned features to work properly. When
	// using wildcard subscriptions a special care is needed regarding security - pay additional
	// attention to a proper permission management.
	AllowWildcards bool

	// RawMode allows enabling raw communication with Nats. When on, Centrifugo subscribes to channels
	// without adding any prefixes to channel name. Proper prefixes must be managed by the application in this
	// case. Data consumed from Nats is sent directly to subscribers without any processing. When publishing
	// to Nats Centrifugo does not add any prefixes to channel names also. Centrifugo features like Publication
	// tags, Publication ClientInfo, join/leave events are not supported in raw mode.
	RawMode RawModeConfig
}

Config of NatsBroker.

type NatsBroker

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

NatsBroker is a broker on top of Nats messaging system.

func New

func New(n *centrifuge.Node, conf Config) (*NatsBroker, error)

New creates NatsBroker.

func (*NatsBroker) Close

func (b *NatsBroker) Close(_ context.Context) error

Close ...

func (*NatsBroker) History

func (b *NatsBroker) History(_ string, _ centrifuge.HistoryOptions) ([]*centrifuge.Publication, centrifuge.StreamPosition, error)

History ...

func (*NatsBroker) IsSupportedPublishChannel

func (b *NatsBroker) IsSupportedPublishChannel(ch string) bool

func (*NatsBroker) IsSupportedSubscribeChannel

func (b *NatsBroker) IsSupportedSubscribeChannel(ch string) bool

func (*NatsBroker) Publish

func (b *NatsBroker) Publish(ch string, data []byte, opts centrifuge.PublishOptions) (centrifuge.StreamPosition, bool, error)

Publish - see Broker interface description.

func (*NatsBroker) PublishControl

func (b *NatsBroker) PublishControl(data []byte, nodeID, _ string) error

PublishControl - see Broker interface description.

func (*NatsBroker) PublishJoin

func (b *NatsBroker) PublishJoin(ch string, info *centrifuge.ClientInfo) error

PublishJoin - see Broker interface description.

func (*NatsBroker) PublishLeave

func (b *NatsBroker) PublishLeave(ch string, info *centrifuge.ClientInfo) error

PublishLeave - see Broker interface description.

func (*NatsBroker) PublishWithStreamPosition

func (b *NatsBroker) PublishWithStreamPosition(ch string, data []byte, opts centrifuge.PublishOptions, sp centrifuge.StreamPosition) error

func (*NatsBroker) RemoveHistory

func (b *NatsBroker) RemoveHistory(_ string) error

RemoveHistory ...

func (*NatsBroker) Run

func (b *NatsBroker) Run(h centrifuge.BrokerEventHandler) error

Run runs engine after node initialized.

func (*NatsBroker) Subscribe

func (b *NatsBroker) Subscribe(ch string) error

Subscribe - see Broker interface description.

func (*NatsBroker) Unsubscribe

func (b *NatsBroker) Unsubscribe(ch string) error

Unsubscribe - see Broker interface description.

type RawModeConfig

type RawModeConfig struct {
	// Enabled enables raw mode when true.
	Enabled bool

	// ChannelReplacements is a map where keys are strings to replace and values are replacements.
	// For example, you have Centrifugo namespace "chat" and using channel "chat:index", but you want to
	// use channel "chat.index" in Nats. Then you can define SymbolReplacements map like this: {":": "."}.
	// In this case Centrifugo will replace all ":" symbols in channel name with "." before sending to Nats.
	// Broker keeps reverse mapping to the original channel to broadcast to proper channels when processing
	// messages received from Nats.
	ChannelReplacements map[string]string

	// Prefix is a string that will be added to all channels when publishing messages to Nats, subscribing
	// to channels in Nats. It's also stripped from channel name when processing messages received from Nats.
	// By default, no prefix is used.
	Prefix string
}