centrifuge

README

Docs: link

Documentation

Overview

Package centrifuge provides the RoadRunner Centrifuge plugin.

Index

• Constants
• type Config
  • func (c *Config) InitDefaults() error
• type Configurer
• type Informer
• type Logger
• type Plugin
  • func (p *Plugin) AddWorker() error
  • func (p *Plugin) Init(cfg Configurer, log Logger, server Server) error
  • func (p *Plugin) MetricsCollector() []prometheus.Collector
  • func (p *Plugin) Name() string
  • func (p *Plugin) RPC() any
  • func (p *Plugin) Ready() (*status.Status, error)
  • func (p *Plugin) RemoveWorker(ctx context.Context) error
  • func (p *Plugin) Reset() error
  • func (p *Plugin) Serve() chan error
  • func (p *Plugin) Status() (*status.Status, error)
  • func (p *Plugin) Stop(ctx context.Context) error
  • func (p *Plugin) Workers() []*process.State
• type Pool
• type Proxy
  • func (p *Proxy) Connect(ctx context.Context, request *centrifugov1.ConnectRequest) (*centrifugov1.ConnectResponse, error)
  • func (p *Proxy) NotifyCacheEmpty(ctx context.Context, request *centrifugov1.NotifyCacheEmptyRequest) (*centrifugov1.NotifyCacheEmptyResponse, error)
  • func (p *Proxy) NotifyChannelState(ctx context.Context, request *centrifugov1.NotifyChannelStateRequest) (*centrifugov1.NotifyChannelStateResponse, error)
  • func (p *Proxy) Publish(ctx context.Context, request *centrifugov1.PublishRequest) (*centrifugov1.PublishResponse, error)
  • func (p *Proxy) RPC(ctx context.Context, request *centrifugov1.RPCRequest) (*centrifugov1.RPCResponse, error)
  • func (p *Proxy) Refresh(ctx context.Context, request *centrifugov1.RefreshRequest) (*centrifugov1.RefreshResponse, error)
  • func (p *Proxy) SubRefresh(ctx context.Context, request *centrifugov1.SubRefreshRequest) (*centrifugov1.SubRefreshResponse, error)
  • func (p *Proxy) Subscribe(ctx context.Context, request *centrifugov1.SubscribeRequest) (*centrifugov1.SubscribeResponse, error)
  • func (p *Proxy) SubscribeBidirectional(_ centrifugov1.CentrifugoProxy_SubscribeBidirectionalServer) error
  • func (p *Proxy) SubscribeUnidirectional(_ *centrifugov1.SubscribeRequest, ...) error
• type Server
• type StatsExporter
  • func (s *StatsExporter) Collect(ch chan<- prometheus.Metric)
  • func (s *StatsExporter) Describe(d chan<- *prometheus.Desc)
• type TLS

Constants

const (
	RRMode           = "RR_MODE"
	RRModeCentrifuge = "centrifuge"
)

Types

type Config

type Config struct {
	// host + port
	ProxyAddress string `mapstructure:"proxy_address"`
	// host + port
	GrpcAPIAddress string `mapstructure:"grpc_api_address"`
	UseCompressor  bool   `mapstructure:"use_compressor"`
	Version        string `mapstructure:"version"`
	Name           string `mapstructure:"name"`
	TLS            *TLS   `mapstructure:"tls"`

	Pool *pool.Config `mapstructure:"pool"`
}

func (*Config) InitDefaults

func (c *Config) InitDefaults() error

type Configurer

type Configurer interface {
	// UnmarshalKey takes a single key and unmarshal it into a Struct.
	UnmarshalKey(name string, out any) error
	// Has checks if a config section exists.
	Has(name string) bool
}

type Informer

type Informer interface {
	Workers() []*process.State
}

type Logger

type Logger interface {
	NamedLogger(name string) *slog.Logger
}

type Plugin

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

func (*Plugin) AddWorker

func (p *Plugin) AddWorker() error

func (*Plugin) Init

func (p *Plugin) Init(cfg Configurer, log Logger, server Server) error

func (*Plugin) MetricsCollector

func (p *Plugin) MetricsCollector() []prometheus.Collector

func (*Plugin) Name

func (p *Plugin) Name() string

func (*Plugin) RPC

func (p *Plugin) RPC() any

func (*Plugin) Ready

func (p *Plugin) Ready() (*status.Status, error)

Ready return readiness status of the particular plugin

func (*Plugin) RemoveWorker

func (p *Plugin) RemoveWorker(ctx context.Context) error

func (*Plugin) Reset

func (p *Plugin) Reset() error

Reset destroys the old pool and replaces it with new one, waiting for old pool to die

func (*Plugin) Serve

func (p *Plugin) Serve() chan error

func (*Plugin) Status

func (p *Plugin) Status() (*status.Status, error)

Status return status of the particular plugin

func (*Plugin) Stop

func (p *Plugin) Stop(ctx context.Context) error

func (*Plugin) Workers

func (p *Plugin) Workers() []*process.State

Workers returns slice with the process states for the workers

type Pool

type Pool interface {
	// Workers return workers' list associated with the pool.
	Workers() (workers []*worker.Process)
	// RemoveWorker removes worker from the pool.
	RemoveWorker(ctx context.Context) error
	// AddWorker adds worker to the pool.
	AddWorker() error
	// Exec payload
	Exec(ctx context.Context, p *payload.Payload, stopCh chan struct{}) (chan *staticPool.PExec, error)
	// Reset kills all workers inside the watcher and replaces with new
	Reset(ctx context.Context) error
	// Destroy the underlying stack (but let them complete the task).
	Destroy(ctx context.Context)
}

type Proxy

type Proxy struct {
	centrifugov1.UnimplementedCentrifugoProxyServer
	// contains filtered or unexported fields
}

func (*Proxy) Connect

func (p *Proxy) Connect(ctx context.Context, request *centrifugov1.ConnectRequest) (*centrifugov1.ConnectResponse, error)

func (*Proxy) NotifyCacheEmpty

func (p *Proxy) NotifyCacheEmpty(ctx context.Context, request *centrifugov1.NotifyCacheEmptyRequest) (*centrifugov1.NotifyCacheEmptyResponse, error)

func (*Proxy) NotifyChannelState

func (p *Proxy) NotifyChannelState(ctx context.Context, request *centrifugov1.NotifyChannelStateRequest) (*centrifugov1.NotifyChannelStateResponse, error)

func (*Proxy) Publish

func (p *Proxy) Publish(ctx context.Context, request *centrifugov1.PublishRequest) (*centrifugov1.PublishResponse, error)

func (*Proxy) RPC

func (p *Proxy) RPC(ctx context.Context, request *centrifugov1.RPCRequest) (*centrifugov1.RPCResponse, error)

func (*Proxy) Refresh

func (p *Proxy) Refresh(ctx context.Context, request *centrifugov1.RefreshRequest) (*centrifugov1.RefreshResponse, error)

func (*Proxy) SubRefresh

func (p *Proxy) SubRefresh(ctx context.Context, request *centrifugov1.SubRefreshRequest) (*centrifugov1.SubRefreshResponse, error)

func (*Proxy) Subscribe

func (p *Proxy) Subscribe(ctx context.Context, request *centrifugov1.SubscribeRequest) (*centrifugov1.SubscribeResponse, error)

func (*Proxy) SubscribeBidirectional

func (p *Proxy) SubscribeBidirectional(_ centrifugov1.CentrifugoProxy_SubscribeBidirectionalServer) error

func (*Proxy) SubscribeUnidirectional

func (p *Proxy) SubscribeUnidirectional(_ *centrifugov1.SubscribeRequest, _ centrifugov1.CentrifugoProxy_SubscribeUnidirectionalServer) error

type Server

type Server interface {
	NewPool(ctx context.Context, cfg *pool.Config, env map[string]string, _ *slog.Logger) (*staticPool.Pool, error)
}

Server creates workers for the application.

type StatsExporter

type StatsExporter struct {
	TotalMemoryDesc  *prometheus.Desc
	StateDesc        *prometheus.Desc
	WorkerMemoryDesc *prometheus.Desc
	TotalWorkersDesc *prometheus.Desc

	WorkersReady   *prometheus.Desc
	WorkersWorking *prometheus.Desc
	WorkersInvalid *prometheus.Desc

	Workers Informer
}

func (*StatsExporter) Collect

func (s *StatsExporter) Collect(ch chan<- prometheus.Metric)

func (*StatsExporter) Describe

func (s *StatsExporter) Describe(d chan<- *prometheus.Desc)

type TLS

type TLS struct {
	Key  string `mapstructure:"key"`
	Cert string `mapstructure:"cert"`
}