totem

README

Totem is a Go library that can turn a single gRPC stream into bidirectional unary gRPC servers.

Background

Streaming RPCs enable several useful design patterns for client-server connections that can't be done with unary RPCs. For example, keeping track of long-lived client connections, and sending server-initiated requests to such clients. However, implementing bidirectional messaging over streams can quickly become very complicated for non-trivial use cases. Totem enables these design patterns and abstracts away the underlying stream, allowing you to implement your streaming RPC in terms of simpler unary RPCs.

Examples

See the examples directory for example code.

Documentation

Index

• Variables
• func CheckContext(ctx context.Context) bool
• func LoadServiceDesc(svc *grpc.ServiceDesc) (*descriptorpb.ServiceDescriptorProto, error)
• func WaitErrOrTimeout(errC <-chan error, timeout time.Duration) error
• type ClientStream
• type MethodInvoker
• type RPC
  • func (*RPC) Descriptor() ([]byte, []int)
  • func (m *RPC) GetContent() isRPC_Content
  • func (x *RPC) GetMethod() string
  • func (x *RPC) GetRequest() []byte
  • func (x *RPC) GetResponse() *Response
  • func (x *RPC) GetTag() uint64
  • func (*RPC) ProtoMessage()
  • func (x *RPC) ProtoReflect() protoreflect.Message
  • func (x *RPC) Reset()
  • func (x *RPC) String() string
• type RPC_Request
• type RPC_Response
• type Response
  • func (*Response) Descriptor() ([]byte, []int)
  • func (x *Response) GetError() []byte
  • func (x *Response) GetResponse() []byte
  • func (*Response) ProtoMessage()
  • func (x *Response) ProtoReflect() protoreflect.Message
  • func (x *Response) Reset()
  • func (x *Response) String() string
• type Server
  • func NewServer(stream Stream) *Server
  • func (r *Server) RegisterService(desc *grpc.ServiceDesc, impl interface{})
  • func (r *Server) Serve(condition ...chan struct{}) (grpc.ClientConnInterface, <-chan error)
  • func (r *Server) Splice(stream Stream, descs ...*descriptorpb.ServiceDescriptorProto)
• type ServerStream
• type Stream

Variables

var ErrKicked = errors.New("kicked")

var ErrStreamClosed = errors.New("stream closed")

var ErrTimeout = fmt.Errorf("timed out")

var File_totem_proto protoreflect.FileDescriptor

Functions

func CheckContext

func CheckContext(ctx context.Context) bool

func LoadServiceDesc

func LoadServiceDesc(svc *grpc.ServiceDesc) (*descriptorpb.ServiceDescriptorProto, error)

func WaitErrOrTimeout

func WaitErrOrTimeout(errC <-chan error, timeout time.Duration) error

Types

type ClientStream

type ClientStream interface {
	Stream
	grpc.ClientStream
}

type MethodInvoker

type MethodInvoker interface {
	Invoke(ctx context.Context, req []byte) ([]byte, error)
}

type RPC

type RPC struct {
	Tag    uint64 `protobuf:"varint,1,opt,name=Tag,proto3" json:"Tag,omitempty"`
	Method string `protobuf:"bytes,2,opt,name=Method,proto3" json:"Method,omitempty"`
	// Types that are assignable to Content:
	//	*RPC_Request
	//	*RPC_Response
	Content isRPC_Content `protobuf_oneof:"Content"`
	// contains filtered or unexported fields
}

func (*RPC) Descriptor

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

Deprecated: Use RPC.ProtoReflect.Descriptor instead.

func (*RPC) GetContent

func (m *RPC) GetContent() isRPC_Content

func (*RPC) GetMethod

func (x *RPC) GetMethod() string

func (*RPC) GetRequest

func (x *RPC) GetRequest() []byte

func (*RPC) GetResponse

func (x *RPC) GetResponse() *Response

func (*RPC) GetTag

func (x *RPC) GetTag() uint64

func (*RPC) ProtoMessage

func (*RPC) ProtoMessage()

func (*RPC) ProtoReflect

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

func (*RPC) Reset

func (x *RPC) Reset()

func (*RPC) String

func (x *RPC) String() string

type RPC_Request

type RPC_Request struct {
	Request []byte `protobuf:"bytes,3,opt,name=Request,proto3,oneof"`
}

type RPC_Response

type RPC_Response struct {
	Response *Response `protobuf:"bytes,4,opt,name=Response,proto3,oneof"`
}

type Response

type Response struct {
	Response []byte `protobuf:"bytes,1,opt,name=Response,proto3" json:"Response,omitempty"`
	Error    []byte `protobuf:"bytes,2,opt,name=Error,proto3" json:"Error,omitempty"`
	// contains filtered or unexported fields
}

func (*Response) Descriptor

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

Deprecated: Use Response.ProtoReflect.Descriptor instead.

func (*Response) GetError

func (x *Response) GetError() []byte

func (*Response) GetResponse

func (x *Response) GetResponse() []byte

func (*Response) ProtoMessage

func (*Response) ProtoMessage()

func (*Response) ProtoReflect

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

func (*Response) Reset

func (x *Response) Reset()

func (*Response) String

func (x *Response) String() string

type Server

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

func NewServer

func NewServer(stream Stream) *Server

func (*Server) RegisterService

func (r *Server) RegisterService(desc *grpc.ServiceDesc, impl interface{})

func (*Server) Serve

func (r *Server) Serve(condition ...chan struct{}) (grpc.ClientConnInterface, <-chan error)

Serve starts the totem server, which takes control of the stream and begins handling incoming and outgoing RPCs.

Optionally, if one non-nil channel is passed to this function, the server will wait until the channel is closed before starting. This can be used to prevent race conditions if you want to interact with the returned ClientConn and prevent the server from invoking any message handlers while doing so.

func (*Server) Splice

func (r *Server) Splice(stream Stream, descs ...*descriptorpb.ServiceDescriptorProto)

Splice configures this server to forward any incoming RPCs for the given service(s) to a different totem stream.

type ServerStream

type ServerStream interface {
	Stream
	grpc.ServerStream
}

type Stream

type Stream interface {
	Send(*RPC) error
	Recv() (*RPC, error)
}