func (b *NatsBroker) PublishJoin(ch string, info *centrifuge.ClientInfo) error {
	data, err := infoToProto(info).MarshalVT()
	if err != nil {
		return err
	}
	push := &protocol.Push{
		Type:    protocol.Push_JOIN,
		Channel: ch,
		Data:    data,
	}
	byteMessage, err := push.MarshalVT()
	if err != nil {
		return err
	}
	return b.nc.Publish(string(b.clientChannel(ch)), byteMessage)
}
PublishJoin - see Broker interface description.
package natsbroker
import (
	"context"
	"fmt"
	"strings"
	"sync"
	"time"
	"github.com/centrifugal/centrifuge"
	"github.com/centrifugal/protocol"
	"github.com/nats-io/nats.go"
)
type (
	channelID string
)
type Config struct {
	URL          string
	Prefix       string
	DialTimeout  time.Duration
	WriteTimeout time.Duration
}
type NatsBroker struct {
	node   *centrifuge.Node
	config Config
	nc           *nats.Conn
	subsMu       sync.Mutex
	subs         map[channelID]*nats.Subscription
	eventHandler centrifuge.BrokerEventHandler
}
var _ centrifuge.Broker = (*NatsBroker)(nil)
func New(n *centrifuge.Node, conf Config) (*NatsBroker, error) {
	b := &NatsBroker{
		node:   n,
		config: conf,
		subs:   make(map[channelID]*nats.Subscription),
	}
	return b, nil
}
func (b *NatsBroker) controlChannel() channelID {
	return channelID(b.config.Prefix + ".control")
}
func (b *NatsBroker) nodeChannel(nodeID string) channelID {
	return channelID(b.config.Prefix + ".node." + nodeID)
}
func (b *NatsBroker) clientChannel(ch string) channelID {
	return channelID(b.config.Prefix + ".client." + ch)
}
func (b *NatsBroker) Run(h centrifuge.BrokerEventHandler) error {
	b.eventHandler = h
	url := b.config.URL
	if url == "" {
		url = nats.DefaultURL
	}
	nc, err := nats.Connect(
		url,
		nats.ReconnectBufSize(-1),
		nats.MaxReconnects(-1),
		nats.Timeout(b.config.DialTimeout),
		nats.FlusherTimeout(b.config.WriteTimeout),
	)
	if err != nil {
		return fmt.Errorf("error connecting to %s: %w", url, err)
	}
	_, err = nc.Subscribe(string(b.controlChannel()), b.handleControl)
	if err != nil {
		return err
	}
	_, err = nc.Subscribe(string(b.nodeChannel(b.node.ID())), b.handleControl)
	if err != nil {
		return err
	}
	b.nc = nc
	b.node.Log(centrifuge.NewLogEntry(centrifuge.LogLevelInfo, fmt.Sprintf("Nats Broker connected to: %s", url)))
	return nil
}
func (b *NatsBroker) Close(_ context.Context) error {
	return nil
}
func isUnsupportedChannel(ch string) bool {
	return strings.Contains(ch, "*") || strings.Contains(ch, ">")
}
func (b *NatsBroker) Publish(ch string, data []byte, opts centrifuge.PublishOptions) (centrifuge.StreamPosition, error) {
	if isUnsupportedChannel(ch) {
		return centrifuge.StreamPosition{}, centrifuge.ErrorBadRequest
	}
	protoPub := &protocol.Publication{
		Data: data,
		Info: infoToProto(opts.ClientInfo),
	}
	data, err := protoPub.MarshalVT()
	if err != nil {
		return centrifuge.StreamPosition{}, err
	}
	push := &protocol.Push{
		Type:    protocol.Push_PUBLICATION,
		Channel: ch,
		Data:    data,
	}
	byteMessage, err := push.MarshalVT()
	if err != nil {
		return centrifuge.StreamPosition{}, err
	}
	return centrifuge.StreamPosition{}, b.nc.Publish(string(b.clientChannel(ch)), byteMessage)
}