muxrpc

README

go-muxrpc

golang.org implementation of ssbc/muxrpc

main purpose: efficient interface to scuttlebot

Documentation

Index

• Constants
• Variables
• func IsSinkClosed(err error) bool
• func NewSinkWriter(sink luigi.Sink) io.WriteCloser
• func NewSourceReader(src luigi.Source) io.Reader
• type CallError
  • func (e *CallError) Error() string
• type CallType
  • func (t CallType) Flags() codec.Flag
• type Endpoint
  • func Handle(pkr Packer, handler Handler) Endpoint
  • func HandleWithLogger(pkr Packer, handler Handler, logger log.Logger) Endpoint
  • func HandleWithRemote(pkr Packer, handler Handler, addr net.Addr) Endpoint
• type Handler
  • func ApplyHandlerWrappers(h Handler, hws ...HandlerWrapper) Handler
• type HandlerMux
  • func (hm *HandlerMux) HandleCall(ctx context.Context, req *Request, edp Endpoint)
  • func (hm *HandlerMux) HandleConnect(ctx context.Context, edp Endpoint)
  • func (hm *HandlerMux) Register(m Method, h Handler)
  • func (hm *HandlerMux) RegisterAll(handlers ...NamedHandler)
• type HandlerWrapper
• type Method
  • func (m Method) String() string
• type NamedHandler
• type Packer
  • func NewPacker(rwc io.ReadWriteCloser) Packer
• type Request
  • func (req *Request) Args() []interface{}
  • func (req *Request) Close() error
  • func (req *Request) CloseWithError(cerr error) error
  • func (req *Request) Return(ctx context.Context, v interface{}) error
• type Server
• type Stream

Constants

const ChunkSize = 65536

Variables

var (
	ErrStreamNotReadable = errors.New("muxrpc: this stream can not be read from")
	ErrStreamNotWritable = errors.New("muxrpc: this stream can not be written to")
	ErrStreamNotClosable = errors.New("muxrpc: this stream can not be closed")
)

var ErrSessionTerminated = errors.New("muxrpc: session terminated")

Functions

func IsSinkClosed

func IsSinkClosed(err error) bool

IsSinkClosed should be moved to luigi to gether with the error

func NewSinkWriter

func NewSinkWriter(sink luigi.Sink) io.WriteCloser

func NewSourceReader

func NewSourceReader(src luigi.Source) io.Reader

Types

type CallError

type CallError struct {
	Name    string `json:"name"`
	Message string `json:"message"`
	Stack   string `json:"stack"`
}

CallError is returned when a call fails

func (*CallError) Error

func (e *CallError) Error() string

type CallType

type CallType string

CallType is the type of a call

func (CallType) Flags

func (t CallType) Flags() codec.Flag

Flags returns the packet flags of the respective call type

type Endpoint

type Endpoint interface {
	// The different call types:
	Async(ctx context.Context, tipe interface{}, method Method, args ...interface{}) (interface{}, error)
	Source(ctx context.Context, tipe interface{}, method Method, args ...interface{}) (luigi.Source, error)
	Sink(ctx context.Context, method Method, args ...interface{}) (luigi.Sink, error)
	Duplex(ctx context.Context, tipe interface{}, method Method, args ...interface{}) (luigi.Source, luigi.Sink, error)

	// Do allows general calls
	Do(ctx context.Context, req *Request) error

	// Terminate wraps up the RPC session
	Terminate() error

	// Remote returns the network address of the remote
	Remote() net.Addr
}

Endpoint allows calling functions on the RPC peer.

func Handle

func Handle(pkr Packer, handler Handler) Endpoint

Handle handles the connection of the packer using the specified handler.

func HandleWithLogger

func HandleWithLogger(pkr Packer, handler Handler, logger log.Logger) Endpoint

HandleWithLogger same as Handle but let's you overwrite the stderr logger

func HandleWithRemote

func HandleWithRemote(pkr Packer, handler Handler, addr net.Addr) Endpoint

HandleWithRemote also sets the remote address the endpoint is connected to TODO: better passing through packer maybe?!

type Handler

type Handler interface {
	HandleCall(ctx context.Context, req *Request, edp Endpoint)
	HandleConnect(ctx context.Context, edp Endpoint)
}

Handler allows handling connections. When we are being called, HandleCall is called. When a connection is established, HandleConnect is called. TODO: let HandleCall return an error

func ApplyHandlerWrappers

func ApplyHandlerWrappers(h Handler, hws ...HandlerWrapper) Handler

type HandlerMux

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

func (*HandlerMux) HandleCall

func (hm *HandlerMux) HandleCall(ctx context.Context, req *Request, edp Endpoint)

func (*HandlerMux) HandleConnect

func (hm *HandlerMux) HandleConnect(ctx context.Context, edp Endpoint)

func (*HandlerMux) Register

func (hm *HandlerMux) Register(m Method, h Handler)

func (*HandlerMux) RegisterAll

func (hm *HandlerMux) RegisterAll(handlers ...NamedHandler)

type HandlerWrapper

type HandlerWrapper func(Handler) Handler

type Method

type Method []string

func (Method) String

func (m Method) String() string

type NamedHandler

type NamedHandler struct {
	Method  Method
	Handler Handler
}

type Packer

type Packer interface {
	luigi.Source
	luigi.Sink
}

Packer is a duplex stream that sends and receives *codec.Packet values. Usually wraps a network connection or stdio.

func NewPacker

func NewPacker(rwc io.ReadWriteCloser) Packer

NewPacker takes an io.ReadWriteCloser and returns a Packer.

type Request

type Request struct {
	// Stream allows sending and receiving packets
	Stream Stream `json:"-"`

	// Method is the name of the called function
	Method Method `json:"name"`
	// Args contains the call arguments
	RawArgs json.RawMessage `json:"args"`
	// Type is the type of the call, i.e. async, sink, source or duplex
	Type CallType `json:"type"`
	// contains filtered or unexported fields
}

Request assembles the state of an RPC call

func (*Request) Args

func (req *Request) Args() []interface{}

Legacy

func (*Request) Close

func (req *Request) Close() error

func (*Request) CloseWithError

func (req *Request) CloseWithError(cerr error) error

func (*Request) Return

func (req *Request) Return(ctx context.Context, v interface{}) error

Return is a helper that returns on an async call

type Server

type Server interface {
	Remote() net.Addr
	Serve(context.Context) error
}

Server can handle packets to and from a remote party

type Stream

type Stream interface {
	luigi.Source
	luigi.Sink
	luigi.ErrorCloser

	// WithType tells the stream in what type JSON data should be unmarshalled into
	WithType(tipe interface{})

	// WithReq tells the stream what request number should be used for sent messages
	WithReq(req int32)
}

Stream is a muxrpc stream for the general duplex case.