tntengine

Documentation

Index

• Constants
• Variables
• func AcquireTimer(d time.Duration) *time.Timer
• func ReleaseTimer(tm *time.Timer)
• type Broker
  • func NewBroker(n *centrifuge.Node, config BrokerConfig) (*Broker, error)
  • func (b *Broker) History(ch string, opts centrifuge.HistoryOptions) ([]*centrifuge.Publication, centrifuge.StreamPosition, error)
  • func (b *Broker) Publish(ch string, data []byte, opts centrifuge.PublishOptions) (centrifuge.StreamPosition, bool, error)
  • func (b *Broker) PublishControl(data []byte, nodeID, _ string) error
  • func (b *Broker) PublishJoin(ch string, info *centrifuge.ClientInfo) error
  • func (b *Broker) PublishLeave(ch string, info *centrifuge.ClientInfo) error
  • func (b *Broker) RemoveHistory(ch string) error
  • func (b *Broker) Run(h centrifuge.BrokerEventHandler) error
  • func (b *Broker) Subscribe(ch string) error
  • func (b *Broker) Unsubscribe(ch string) error
• type BrokerConfig
• type ConnectionMode
• type MultiConnection
  • func Connect(addrs []string, opts tarantool.Opts, multiOpts MultiOpts) (*MultiConnection, error)
  • func (c *MultiConnection) Close() error
  • func (c *MultiConnection) IsLeader(conn *tarantool.Connection) (bool, error)
  • func (c *MultiConnection) LeaderChanged()
  • func (c *MultiConnection) LeaderConn() (*tarantool.Connection, error)
  • func (c *MultiConnection) NewLeaderConn(opts tarantool.Opts) (*tarantool.Connection, error)
• type MultiOpts
• type PresenceManager
  • func NewPresenceManager(n *centrifuge.Node, config PresenceManagerConfig) (*PresenceManager, error)
  • func (m PresenceManager) AddPresence(ch string, clientID string, info *centrifuge.ClientInfo) error
  • func (m PresenceManager) Presence(ch string) (map[string]*centrifuge.ClientInfo, error)
  • func (m PresenceManager) PresenceStats(ch string) (centrifuge.PresenceStats, error)
  • func (m PresenceManager) RemovePresence(ch string, clientID string, _ string) error
• type PresenceManagerConfig
• type Shard
  • func NewShard(c ShardConfig) (*Shard, error)
  • func (s *Shard) Exec(request *tarantool.Request) (*tarantool.Response, error)
  • func (s *Shard) ExecTyped(request *tarantool.Request, result any) error
• type ShardConfig

Constants

const DefaultPresenceTTL = 60 * time.Second

DefaultPresenceTTL is a default value for presence TTL in Tarantool.

Variables

var ErrNoLeader = errors.New("no leader")

Functions

func AcquireTimer

func AcquireTimer(d time.Duration) *time.Timer

AcquireTimer from pool.

func ReleaseTimer

func ReleaseTimer(tm *time.Timer)

ReleaseTimer to pool.

Types

type Broker

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

Broker uses Tarantool to implement centrifuge.Broker functionality.

func NewBroker

func NewBroker(n *centrifuge.Node, config BrokerConfig) (*Broker, error)

NewBroker initializes Tarantool Broker.

func (*Broker) History

func (b *Broker) History(ch string, opts centrifuge.HistoryOptions) ([]*centrifuge.Publication, centrifuge.StreamPosition, error)

History - see centrifuge.Broker interface description.

func (*Broker) Publish

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

Publish - see centrifuge.Broker interface description.

func (*Broker) PublishControl

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

PublishControl - see centrifuge.Broker interface description.

func (*Broker) PublishJoin

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

PublishJoin - see centrifuge.Broker interface description.

func (*Broker) PublishLeave

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

PublishLeave - see centrifuge.Broker interface description.

func (*Broker) RemoveHistory

func (b *Broker) RemoveHistory(ch string) error

RemoveHistory - see centrifuge.Broker interface description.

func (*Broker) Run

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

Run runs broker after node initialized.

func (*Broker) Subscribe

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

Subscribe - see centrifuge.Broker interface description.

func (*Broker) Unsubscribe

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

Unsubscribe - see centrifuge.Broker interface description.

type BrokerConfig

type BrokerConfig struct {
	// HistoryMetaTTL sets a time of stream meta key expiration in Tarantool. Stream
	// meta key is a Tarantool HASH that contains top offset in channel and epoch value.
	// By default, stream meta keys do not expire.
	HistoryMetaTTL time.Duration

	// UsePolling allows turning on polling mode instead of push.
	UsePolling bool

	// Shards is a list of Tarantool instances to shard data by channel.
	Shards []*Shard
}

BrokerConfig is a config for Tarantool Broker.

type ConnectionMode

type ConnectionMode int

const (
	ConnectionModeSingleInstance     ConnectionMode = 0
	ConnectionModeLeaderFollower     ConnectionMode = 1
	ConnectionModeLeaderFollowerRaft ConnectionMode = 2
)

type MultiConnection

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

func Connect

func Connect(addrs []string, opts tarantool.Opts, multiOpts MultiOpts) (*MultiConnection, error)

func (*MultiConnection) Close

func (c *MultiConnection) Close() error

func (*MultiConnection) IsLeader

func (c *MultiConnection) IsLeader(conn *tarantool.Connection) (bool, error)

func (*MultiConnection) LeaderChanged

func (c *MultiConnection) LeaderChanged()

func (*MultiConnection) LeaderConn

func (c *MultiConnection) LeaderConn() (*tarantool.Connection, error)

func (*MultiConnection) NewLeaderConn

func (c *MultiConnection) NewLeaderConn(opts tarantool.Opts) (*tarantool.Connection, error)

type MultiOpts

type MultiOpts struct {
	ConnectionMode      ConnectionMode
	LeaderCheckInterval time.Duration
}

type PresenceManager

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

PresenceManager uses Tarantool to implement centrifuge.PresenceManager functionality.

func NewPresenceManager

func NewPresenceManager(n *centrifuge.Node, config PresenceManagerConfig) (*PresenceManager, error)

NewPresenceManager initializes Tarantool-based centrifuge.PresenceManager.

func (PresenceManager) AddPresence

func (m PresenceManager) AddPresence(ch string, clientID string, info *centrifuge.ClientInfo) error

func (PresenceManager) Presence

func (m PresenceManager) Presence(ch string) (map[string]*centrifuge.ClientInfo, error)

func (PresenceManager) PresenceStats

func (m PresenceManager) PresenceStats(ch string) (centrifuge.PresenceStats, error)

func (PresenceManager) RemovePresence

func (m PresenceManager) RemovePresence(ch string, clientID string, _ string) error

type PresenceManagerConfig

type PresenceManagerConfig struct {
	// PresenceTTL is an interval how long to consider presence info
	// valid after receiving presence update. This allows to automatically
	// clean up unnecessary presence entries after TTL passed.
	PresenceTTL time.Duration

	// Shards is a list of Tarantool instances to shard data by channel.
	Shards []*Shard
}

PresenceManagerConfig is a config for Tarantool-based PresenceManager.

type Shard

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

Shard represents single Tarantool instance.

func NewShard

func NewShard(c ShardConfig) (*Shard, error)

func (*Shard) Exec

func (s *Shard) Exec(request *tarantool.Request) (*tarantool.Response, error)

func (*Shard) ExecTyped

func (s *Shard) ExecTyped(request *tarantool.Request, result any) error

type ShardConfig

type ShardConfig struct {
	// Addresses of Tarantool instances.
	Addresses []string
	// User for auth.
	User string
	// Password for auth.
	Password string
	// ConnectionMode for shard.
	ConnectionMode ConnectionMode
}

ShardConfig allows providing options to connect to Tarantool.