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
}
Run Tarantool shard.
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