stream

Documentation

Index

• Variables
• type AuthOptions
• type BearerOptions
• type ClientHandler
• type FromS
• type Handler
  • func NewClientHandler(...) Handler[*protos.ServerToClient, *protos.ClientToServer]
  • func NewSamplerHandler(name, resource string, tags []control.Tag, ...) Handler[*protos.ServerToSampler, *protos.SamplerToServer]
• type Options
  • func NewOptionsDefault() *Options
  • func (o Options) String() string
• type SamplerHandler
• type State
  • func (s State) String() string
• type Stream
  • func New[F FromS, T ToS](uid string, opts *Options, h Handler[F, T], logger logging.Logger) *Stream[F, T]
  • func (s *Stream[FromS, ToS]) Close(timeout time.Duration) error
  • func (s *Stream[FromS, ToS]) Connect(serverAddr string) error
  • func (s *Stream[FromS, ToS]) Errors() chan error
  • func (s *Stream[FromS, ToS]) SendReqToS(ctx context.Context, toS ToS) (FromS, error)
  • func (s *Stream[FromS, ToS]) SendToS(_ context.Context, toS ToS) error
  • func (s *Stream[FromS, ToS]) ServerUID() string
  • func (s *Stream[FromS, ToS]) State() State
  • func (s *Stream[FromS, ToS]) StateChanges() chan State
  • func (s *Stream[FromS, ToS]) ToServerMsg() ToS
• type TLSOptions
• type ToS

Variables

var ErrConnectionFailure = errors.New("connection failure")

var ErrRegistrationFailure = errors.New("registration failure")

Types

type AuthOptions

type AuthOptions struct {
	// Type specifies the authentication type, supported values: 'bearer'
	// If empty, no authentication enabled.
	Type string
	// Bearer configures the bearer authentication type..
	Bearer BearerOptions
}

type BearerOptions

type BearerOptions struct {
	// Bearer token
	Token string
}

type ClientHandler

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

type FromS

type FromS interface {
	*protos.ServerToClient | *protos.ServerToSampler
}

type Handler

type Handler[F FromS, T ToS] interface {
	// contains filtered or unexported methods
}

Handler implements all the logic to handle a stream implementation (to a sampler or a client)

func NewClientHandler

func NewClientHandler(recvServerReqCb func(*protos.ServerToClient) (bool, *protos.ClientToServer, error)) Handler[*protos.ServerToClient, *protos.ClientToServer]

func NewSamplerHandler

func NewSamplerHandler(name, resource string, tags []control.Tag, recvServerReqCb func(*protos.ServerToSampler) (bool, *protos.SamplerToServer, error), initialConfig *protos.ClientSamplerConfigUpdate) Handler[*protos.ServerToSampler, *protos.SamplerToServer]

type Options

type Options struct {
	// Block determines if the initial connection request will block until it
	// successfully connects.
	Block bool
	// ConnTimeout is a duration for the maximum amount of time to wait
	// during the initial connection request. Only applicable if Block is true.
	ConnTimeout time.Duration
	// ResponseTimeout is a duration for the maximum amount of time to wait
	// for a response to a request send to the server.
	ResponseTimeout time.Duration
	// KeepAliveMaxPeriod is a duration that defines the maximum period to
	// send keep alive messages. Minimum value is 10s. A larger period reduces
	// network load but increases the time required to detect when the stream
	// disconnects.
	KeepAliveMaxPeriod time.Duration
	// ServerReqsQueueLen defines how many server requests are allowed to be queued.
	ServerReqsQueueLen int
	// TLS related options
	TLS TLSOptions
	// Auth related options
	Auth AuthOptions
}

func NewOptionsDefault

func NewOptionsDefault() *Options

func (Options) String

func (o Options) String() string

type SamplerHandler

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

type State

type State int

const (
	Unknown State = iota
	Unregistered
	Registering
	Registered
)

func (State) String

func (s State) String() string

type Stream

type Stream[F FromS, T ToS] struct {
	// contains filtered or unexported fields
}

func New

func New[F FromS, T ToS](uid string, opts *Options, h Handler[F, T], logger logging.Logger) *Stream[F, T]

func (*Stream[FromS, ToS]) Close

func (s *Stream[FromS, ToS]) Close(timeout time.Duration) error

func (*Stream[FromS, ToS]) Connect

func (s *Stream[FromS, ToS]) Connect(serverAddr string) error

func (*Stream[FromS, ToS]) Errors

func (s *Stream[FromS, ToS]) Errors() chan error

func (*Stream[FromS, ToS]) SendReqToS

func (s *Stream[FromS, ToS]) SendReqToS(ctx context.Context, toS ToS) (FromS, error)

func (*Stream[FromS, ToS]) SendToS

func (s *Stream[FromS, ToS]) SendToS(_ context.Context, toS ToS) error

func (*Stream[FromS, ToS]) ServerUID

func (s *Stream[FromS, ToS]) ServerUID() string

func (*Stream[FromS, ToS]) State

func (s *Stream[FromS, ToS]) State() State

func (*Stream[FromS, ToS]) StateChanges

func (s *Stream[FromS, ToS]) StateChanges() chan State

func (*Stream[FromS, ToS]) ToServerMsg

func (s *Stream[FromS, ToS]) ToServerMsg() ToS

type TLSOptions

type TLSOptions struct {
	// Enable determines if the server implements TLS. If TLSCert is not set,
	// it t will use the system CA certificates to validate the server certificate.
	// By default, it is disabled.
	Enable bool
	// CACertPath defines te path to the certificate CA used to validate the
	// server certificate.
	CACertPath string
}

type ToS

type ToS interface {
	*protos.ClientToServer | *protos.SamplerToServer
}