func (b *Broker) PublishJoin(ch string, info *centrifuge.ClientInfo) error {
	s := consistentShard(ch, b.shards)
	pr := pubRequest{
		MsgType: "j",
		Channel: ch,
		Info:    b.clientInfoString(info),
	}
	_, err := s.Exec(tarantool.Call("centrifuge.publish", pr))
	return err
}
PublishJoin - see centrifuge.Broker interface description.
package tntengine
import (
	"context"
	"errors"
	"fmt"
	"runtime"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/centrifugal/centrifugo/v3/internal/tools"
	"github.com/FZambia/tarantool"
	"github.com/centrifugal/centrifuge"
	"github.com/centrifugal/protocol"
	"github.com/google/uuid"
	"github.com/vmihailenco/msgpack/v5"
)
const internalChannelPrefix = "__"
const (
	tarantoolControlChannel = internalChannelPrefix + "control"
	tarantoolNodeChannelPrefix = internalChannelPrefix + "node."
)
type Broker struct {
	controlRound uint64  
	node         *centrifuge.Node
	sharding     bool
	config       BrokerConfig
	shards       []*Shard
	nodeChannel  string
}
var _ centrifuge.Broker = (*Broker)(nil)
type BrokerConfig struct {
	HistoryMetaTTL time.Duration
	UsePolling bool
	Shards []*Shard
}
func NewBroker(n *centrifuge.Node, config BrokerConfig) (*Broker, error) {
	if len(config.Shards) == 0 {
		return nil, errors.New("no Tarantool shards provided in configuration")
	}
	if len(config.Shards) > 1 {
		n.Log(centrifuge.NewLogEntry(centrifuge.LogLevelInfo, fmt.Sprintf("Tarantool sharding enabled: %d shards", len(config.Shards))))
	}
	e := &Broker{
		node:        n,
		shards:      config.Shards,
		config:      config,
		sharding:    len(config.Shards) > 1,
		nodeChannel: nodeChannel(n.ID()),
	}
	return e, nil
}
func (b *Broker) Run(h centrifuge.BrokerEventHandler) error {
	for _, shard := range b.shards {
		if err := b.runShard(shard, h); err != nil {
			return err
		}
	}
	return nil
}
func (b *Broker) runForever(fn func(), minDelay time.Duration) {
	for {
		started := time.Now()
		fn()
		elapsed := time.Since(started)
		if elapsed < minDelay {
			time.Sleep(minDelay - elapsed)
		}
	}
}
const pubSubRoutineMinDelay = 300 * time.Millisecond
func (b *Broker) runShard(s *Shard, h centrifuge.BrokerEventHandler) error {
	b.node.Log(centrifuge.NewLogEntry(centrifuge.LogLevelInfo, fmt.Sprintf("start Tarantool shard: %v", tools.GetLogAddresses(s.config.Addresses))))
	go b.runForever(func() {
		b.runPubSub(s, h)
	}, pubSubRoutineMinDelay)
	go b.runForever(func() {
		b.runControlPubSub(s, h)
	}, pubSubRoutineMinDelay)
	return nil
}
type pubRequest struct {
	MsgType        string
	Channel        string
	Data           string
	Info           string
	HistoryTTL     int
	HistorySize    int
	HistoryMetaTTL int
}
type pubResponse struct {
	Offset uint64
	Epoch  string
}
func (m *pubResponse) DecodeMsgpack(d *msgpack.Decoder) error {
	var err error
	var l int
	if l, err = d.DecodeArrayLen(); err != nil {
		return err
	}
	if l != 2 {
		return fmt.Errorf("malformed array len: %d", l)
	}
	if m.Offset, err = d.DecodeUint64(); err != nil {
		return err
	}
	if m.Epoch, err = d.DecodeString(); err != nil {
		return err
	}
	return nil
}
func (b *Broker) Publish(ch string, data []byte, opts centrifuge.PublishOptions) (centrifuge.StreamPosition, error) {
	s := consistentShard(ch, b.shards)
	pr := &pubRequest{
		MsgType:        "p",
		Channel:        ch,
		Data:           string(data),
		Info:           b.clientInfoString(opts.ClientInfo),
		HistoryTTL:     int(opts.HistoryTTL.Seconds()),
		HistorySize:    opts.HistorySize,
		HistoryMetaTTL: int(b.config.HistoryMetaTTL.Seconds()),
	}
	var resp pubResponse
	err := s.ExecTyped(tarantool.Call("centrifuge.publish", pr), &resp)
	if err != nil {
		return centrifuge.StreamPosition{}, err
	}
	return centrifuge.StreamPosition{Offset: resp.Offset, Epoch: resp.Epoch}, err
}