stream_srpc_server

Documentation

Index

• Variables
• type Config
  • func (c *Config) ApplyDefaults(protocolIds []protocol.ID) *Config
  • func (c *Config) BuildServer(b bus.Bus, le *logrus.Entry, info *controller.Info, registerFns []RegisterFn) (*Server, error)
  • func (m *Config) CloneMessageVT() proto.Message
  • func (m *Config) CloneVT() *Config
  • func (*Config) Descriptor() ([]byte, []int)
  • func (this *Config) EqualMessageVT(thatMsg proto.Message) bool
  • func (this *Config) EqualVT(that *Config) bool
  • func (c *Config) GetDebugVals() config.DebugValues
  • func (x *Config) GetDisableEstablishLink() bool
  • func (x *Config) GetPeerIds() []string
  • func (x *Config) GetProtocolIds() []string
  • func (m *Config) MarshalToSizedBufferVT(dAtA []byte) (int, error)
  • func (m *Config) MarshalToVT(dAtA []byte) (int, error)
  • func (m *Config) MarshalVT() (dAtA []byte, err error)
  • func (c *Config) ParsePeerIDs() ([]peer.ID, error)
  • func (c *Config) ParseProtocolIDs() ([]protocol.ID, error)
  • func (*Config) ProtoMessage()
  • func (x *Config) ProtoReflect() protoreflect.Message
  • func (x *Config) Reset()
  • func (m *Config) SizeVT() (n int)
  • func (x *Config) String() string
  • func (m *Config) UnmarshalVT(dAtA []byte) error
  • func (c *Config) Validate() error
• type RegisterFn
• type Server
  • func NewServer(b bus.Bus, le *logrus.Entry, info *controller.Info, registerFns []RegisterFn, ...) (*Server, error)
  • func (s *Server) Close() error
  • func (s *Server) Execute(ctx context.Context) error
  • func (s *Server) GetControllerInfo() *controller.Info
  • func (s *Server) HandleDirective(ctx context.Context, di directive.Instance) ([]directive.Resolver, error)
  • func (s *Server) HandleMountedStream(ctx context.Context, ms link.MountedStream) error
  • func (s *Server) ResolveHandleMountedStream(ctx context.Context, di directive.Instance, dir link.HandleMountedStream) ([]directive.Resolver, error)

Variables

var (
	ErrInvalidLength        = fmt.Errorf("proto: negative length found during unmarshaling")
	ErrIntOverflow          = fmt.Errorf("proto: integer overflow")
	ErrUnexpectedEndOfGroup = fmt.Errorf("proto: unexpected end of group")
)

var File_github_com_aperturerobotics_bifrost_stream_srpc_server_server_proto protoreflect.FileDescriptor

Types

type Config

type Config struct {

	// PeerIds are the list of peer IDs to listen on.
	// If empty, allows any incoming peer id w/ the protocol id(s).
	PeerIds []string `protobuf:"bytes,1,rep,name=peer_ids,json=peerIds,proto3" json:"peer_ids,omitempty"`
	// ProtocolIds is the list of protocol ids to listen on.
	// If empty, no incoming streams will be accepted.
	ProtocolIds []string `protobuf:"bytes,2,rep,name=protocol_ids,json=protocolIds,proto3" json:"protocol_ids,omitempty"`
	// DisableEstablishLink disables adding an EstablishLink directive for each incoming peer.
	DisableEstablishLink bool `protobuf:"varint,3,opt,name=disable_establish_link,json=disableEstablishLink,proto3" json:"disable_establish_link,omitempty"`
	// contains filtered or unexported fields
}

Config configures the server for the srpc service.

func (*Config) ApplyDefaults

func (c *Config) ApplyDefaults(protocolIds []protocol.ID) *Config

ApplyDefaults applies defaults to the config, returning a clone.

If the fields are set, skips the defaults.

func (*Config) BuildServer

func (c *Config) BuildServer(b bus.Bus, le *logrus.Entry, info *controller.Info, registerFns []RegisterFn) (*Server, error)

BuildServer constructs the server from the args.

func (*Config) CloneMessageVT

func (m *Config) CloneMessageVT() proto.Message

func (*Config) CloneVT

func (m *Config) CloneVT() *Config

func (*Config) Descriptor

func (*Config) Descriptor() ([]byte, []int)

Deprecated: Use Config.ProtoReflect.Descriptor instead.

func (*Config) EqualMessageVT

func (this *Config) EqualMessageVT(thatMsg proto.Message) bool

func (*Config) EqualVT

func (this *Config) EqualVT(that *Config) bool

func (*Config) GetDebugVals

func (c *Config) GetDebugVals() config.DebugValues

GetDebugVals returns the directive arguments as key/value pairs. This should be something like param1="test", param2="test". This is not necessarily unique, and is primarily intended for display.

func (*Config) GetDisableEstablishLink

func (x *Config) GetDisableEstablishLink() bool

func (*Config) GetPeerIds

func (x *Config) GetPeerIds() []string

func (*Config) GetProtocolIds

func (x *Config) GetProtocolIds() []string

func (*Config) MarshalToSizedBufferVT

func (m *Config) MarshalToSizedBufferVT(dAtA []byte) (int, error)

func (*Config) MarshalToVT

func (m *Config) MarshalToVT(dAtA []byte) (int, error)

func (*Config) MarshalVT

func (m *Config) MarshalVT() (dAtA []byte, err error)

func (*Config) ParsePeerIDs

func (c *Config) ParsePeerIDs() ([]peer.ID, error)

ParsePeerIDs parses the peer ids field.

func (*Config) ParseProtocolIDs

func (c *Config) ParseProtocolIDs() ([]protocol.ID, error)

ParseProtocolIDs parses the protocol ids field.

func (*Config) ProtoMessage

func (*Config) ProtoMessage()

func (*Config) ProtoReflect

func (x *Config) ProtoReflect() protoreflect.Message

func (*Config) Reset

func (x *Config) Reset()

func (*Config) SizeVT

func (m *Config) SizeVT() (n int)

func (*Config) String

func (x *Config) String() string

func (*Config) UnmarshalVT

func (m *Config) UnmarshalVT(dAtA []byte) error

func (*Config) Validate

func (c *Config) Validate() error

Validate checks the config.

type RegisterFn

type RegisterFn func(mux srpc.Mux) error

RegisterFn is a callback to register apis to a mux.

type Server

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

Server handles incoming streams for a peer id.

func NewServer

func NewServer(
	b bus.Bus,
	le *logrus.Entry,
	info *controller.Info,
	registerFns []RegisterFn,
	protocolIDs []protocol.ID,
	peerIDs []string,
	disableEstablishLink bool,
) (*Server, error)

NewServer constructs a common srpc controller. If peerIDs and/or domainIDs are empty, matches any.

func (*Server) Close

func (s *Server) Close() error

Close releases any resources used by the controller. Error indicates any issue encountered releasing.

func (*Server) Execute

func (s *Server) Execute(ctx context.Context) error

Execute executes the given controller. Returning nil ends execution. Returning an error triggers a retry with backoff.

func (*Server) GetControllerInfo

func (s *Server) GetControllerInfo() *controller.Info

GetControllerInfo returns information about the controller.

func (*Server) HandleDirective

func (s *Server) HandleDirective(ctx context.Context, di directive.Instance) ([]directive.Resolver, error)

HandleDirective asks if the handler can resolve the directive. If it can, it returns a resolver. If not, returns nil.

func (*Server) HandleMountedStream

func (s *Server) HandleMountedStream(ctx context.Context, ms link.MountedStream) error

HandleMountedStream handles an incoming mounted stream. Any returned error indicates the stream should be closed. This function should return as soon as possible, and start additional goroutines to manage the lifecycle of the stream. Typically EstablishLink is asserted in HandleMountedStream.

func (*Server) ResolveHandleMountedStream

func (s *Server) ResolveHandleMountedStream(
	ctx context.Context,
	di directive.Instance,
	dir link.HandleMountedStream,
) ([]directive.Resolver, error)

ResolveHandleMountedStream resolves a HandleMountedStream directive.