rpc

Documentation

Index

• Constants
• func NewHTTPClientHelper(s *HTTPService) *httpClientHelper
• type Barrier
• type ClientHelper
• type Config
  • func NewConfig() Config
  • func (c *Config) Impl() string
  • func (c *Config) TLSConfig() (*tls.Config, error)
  • func (c *Config) TLSEnabled() bool
  • func (c Config) ToToml() string
• type Controller
  • func NewController(metrics metrics.Instrumenter, config *Config, l *slog.Logger) (*Controller, error)
  • func (c *Controller) Authenticate(sid string, env *common.SessionEnv) (*common.ConnectResult, error)
  • func (c *Controller) Disconnect(sid string, env *common.SessionEnv, id string, subscriptions []string) error
  • func (c *Controller) Perform(sid string, env *common.SessionEnv, id string, channel string, data string) (*common.CommandResult, error)
  • func (c *Controller) Shutdown() error
  • func (c *Controller) Start() error
  • func (c *Controller) Subscribe(sid string, env *common.SessionEnv, id string, channel string) (*common.CommandResult, error)
  • func (c *Controller) Unsubscribe(sid string, env *common.SessionEnv, id string, channel string) (*common.CommandResult, error)
• type Dialer
  • func NewHTTPDialer(c *Config) (Dialer, error)
  • func NewInprocessServiceDialer(service pb.RPCServer, stateHandler ClientHelper) Dialer
• type FixedSizeBarrier
  • func NewFixedSizeBarrier(capacity int) (*FixedSizeBarrier, error)
  • func (b *FixedSizeBarrier) Acquire()
  • func (b *FixedSizeBarrier) BusyCount() int
  • func (b *FixedSizeBarrier) Capacity() int
  • func (b *FixedSizeBarrier) CapacityInfo() string
  • func (FixedSizeBarrier) Exhausted()
  • func (FixedSizeBarrier) HasDynamicCapacity() (res bool)
  • func (b *FixedSizeBarrier) Release()
  • func (FixedSizeBarrier) Start()
  • func (FixedSizeBarrier) Stop()
• type HTTPService
  • func NewHTTPService(c *Config) (*HTTPService, error)
  • func (s *HTTPService) Command(ctx context.Context, r *pb.CommandMessage) (*pb.CommandResponse, error)
  • func (s *HTTPService) Connect(ctx context.Context, r *pb.ConnectionRequest) (*pb.ConnectionResponse, error)
  • func (s *HTTPService) Disconnect(ctx context.Context, r *pb.DisconnectRequest) (*pb.DisconnectResponse, error)

Constants

const (
	// ProtoVersions contains a comma-seprated list of compatible RPC protos versions
	// (we pass it as request meta to notify clients)
	ProtoVersions = "v1"
)

Functions

func NewHTTPClientHelper

func NewHTTPClientHelper(s *HTTPService) *httpClientHelper

Types

type Barrier

type Barrier interface {
	Acquire()
	Release()
	BusyCount() int
	Capacity() int
	CapacityInfo() string
	Exhausted()
	HasDynamicCapacity() bool
	Start()
	Stop()
}

type ClientHelper

type ClientHelper interface {
	Ready() error
	SupportsActiveConns() bool
	ActiveConns() int
	Close()
}

ClientHelper provides additional methods to operate gRPC client

type Config

type Config struct {
	// RPC instance host
	Host string `toml:"host"`
	// ProxyHeaders to add to RPC request env
	ProxyHeaders []string `toml:"proxy_headers"`
	// ProxyCookies to add to RPC request env
	ProxyCookies []string `toml:"proxy_cookies"`
	// The max number of simultaneous requests.
	// Should be slightly less than the RPC server concurrency to avoid
	// ResourceExhausted errors
	Concurrency int `toml:"concurrency"`
	// Enable client-side TLS on RPC connections?
	EnableTLS bool `toml:"enable_tls"`
	// Whether to verify the RPC server's certificate chain and host name
	TLSVerify bool `toml:"tls_verify"`
	// CA root TLS certificate path
	TLSRootCA string `toml:"tls_root_ca_path"`
	// Max receive msg size (bytes)
	MaxRecvSize int `toml:"max_recv_size"`
	// Max send msg size (bytes)
	MaxSendSize int `toml:"max_send_size"`
	// Underlying implementation (grpc, http, or none)
	Implementation string `toml:"implementation"`
	// Alternative dialer implementation
	DialFun Dialer
	// Secret for HTTP RPC authentication
	Secret string `toml:"secret"`
	// Timeout for HTTP RPC requests (in ms)
	RequestTimeout int `toml:"http_request_timeout"`
	// SecretBase is a secret used to generate authentication token
	SecretBase string
}

Config contains RPC controller configuration

func NewConfig

func NewConfig() Config

NewConfig builds a new config

func (*Config) Impl

func (c *Config) Impl() string

Return chosen implementation either from the user provided value or from the host scheme

func (*Config) TLSConfig

func (c *Config) TLSConfig() (*tls.Config, error)

TLSConfig builds TLS configuration for RPC client

func (*Config) TLSEnabled

func (c *Config) TLSEnabled() bool

Whether secure connection to RPC server is enabled either explicitly or implicitly

func (Config) ToToml

func (c Config) ToToml() string

type Controller

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

Controller implements node.Controller interface for gRPC

func NewController

func NewController(metrics metrics.Instrumenter, config *Config, l *slog.Logger) (*Controller, error)

NewController builds new Controller

func (*Controller) Authenticate

func (c *Controller) Authenticate(sid string, env *common.SessionEnv) (*common.ConnectResult, error)

Authenticate performs Connect RPC call

func (*Controller) Disconnect

func (c *Controller) Disconnect(sid string, env *common.SessionEnv, id string, subscriptions []string) error

Disconnect performs disconnect RPC call

func (*Controller) Perform

func (c *Controller) Perform(sid string, env *common.SessionEnv, id string, channel string, data string) (*common.CommandResult, error)

Perform performs Command RPC call with "perform" command

func (*Controller) Shutdown

func (c *Controller) Shutdown() error

Shutdown closes connections

func (*Controller) Start

func (c *Controller) Start() error

Start initializes RPC connection pool

func (*Controller) Subscribe

func (c *Controller) Subscribe(sid string, env *common.SessionEnv, id string, channel string) (*common.CommandResult, error)

Subscribe performs Command RPC call with "subscribe" command

func (*Controller) Unsubscribe

func (c *Controller) Unsubscribe(sid string, env *common.SessionEnv, id string, channel string) (*common.CommandResult, error)

Unsubscribe performs Command RPC call with "unsubscribe" command

type Dialer

type Dialer = func(c *Config, l *slog.Logger) (pb.RPCClient, ClientHelper, error)

Dialer is factory function to build a new client with its helper

func NewHTTPDialer

func NewHTTPDialer(c *Config) (Dialer, error)

func NewInprocessServiceDialer

func NewInprocessServiceDialer(service pb.RPCServer, stateHandler ClientHelper) Dialer

type FixedSizeBarrier

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

func NewFixedSizeBarrier

func NewFixedSizeBarrier(capacity int) (*FixedSizeBarrier, error)

func (*FixedSizeBarrier) Acquire

func (b *FixedSizeBarrier) Acquire()

func (*FixedSizeBarrier) BusyCount

func (b *FixedSizeBarrier) BusyCount() int

func (*FixedSizeBarrier) Capacity

func (b *FixedSizeBarrier) Capacity() int

func (*FixedSizeBarrier) CapacityInfo

func (b *FixedSizeBarrier) CapacityInfo() string

func (FixedSizeBarrier) Exhausted

func (FixedSizeBarrier) Exhausted()

func (FixedSizeBarrier) HasDynamicCapacity

func (FixedSizeBarrier) HasDynamicCapacity() (res bool)

func (*FixedSizeBarrier) Release

func (b *FixedSizeBarrier) Release()

func (FixedSizeBarrier) Start

func (FixedSizeBarrier) Start()

func (FixedSizeBarrier) Stop

func (FixedSizeBarrier) Stop()

type HTTPService

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

func NewHTTPService

func NewHTTPService(c *Config) (*HTTPService, error)

func (*HTTPService) Command

func (s *HTTPService) Command(ctx context.Context, r *pb.CommandMessage) (*pb.CommandResponse, error)

func (*HTTPService) Connect

func (s *HTTPService) Connect(ctx context.Context, r *pb.ConnectionRequest) (*pb.ConnectionResponse, error)

func (*HTTPService) Disconnect

func (s *HTTPService) Disconnect(ctx context.Context, r *pb.DisconnectRequest) (*pb.DisconnectResponse, error)