rpc

Documentation

Index

• Variables
• type Call
• type Client
  • func NewClient(mqtt bus.Bus, codec ClientCodec) *Client
  • func (client *Client) Call(topic string, serviceMethod string, args interface{}) error
  • func (client *Client) CallWithTimeout(topic string, serviceMethod string, args interface{}, reply interface{}, ...) error
• type ClientCodec
• type Codec
• type CodecRequest
• type ExportedService
  • func (s *ExportedService) SendEvent(event string, payload ...interface{}) error
• type Message
• type Server
  • func NewServer(client bus.Bus, codec Codec) *Server
  • func (s *Server) HasMethod(topic string, method string) bool
  • func (s *Server) RegisterService(receiver interface{}, topic string, schema string) (service *ExportedService, err error)
  • func (s *Server) SendNotification(topic string, params ...interface{}) error

Variables

var RedisPool *redis.Pool

XXX: This is ugly...

Types

type Call

type Call struct {
	Topic         string      // The MQTT topic this call will be sent to
	ServiceMethod string      // The name of the service and method to call.
	Args          interface{} // The argument to the function (*struct).
	Reply         interface{} // The reply from the function (*struct).
	Error         error       // After completion, the error status.
	Done          chan *Call  // Strobes when call is complete.
	ID            uint32      // Used to map responses
}

Call represents an active RPC.

type Client

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

Client represents an RPC Client. There may be multiple outstanding Calls associated with a single Client, and a Client may be used by multiple goroutines simultaneously.

func NewClient

func NewClient(mqtt bus.Bus, codec ClientCodec) *Client

NewClient creates a new rpc client using the provided MQTT connection

func (*Client) Call

func (client *Client) Call(topic string, serviceMethod string, args interface{}) error

Call invokes a function asynchronously.

func (*Client) CallWithTimeout

func (client *Client) CallWithTimeout(topic string, serviceMethod string, args interface{}, reply interface{}, timeout time.Duration) error

CallWithTimeout invokes a function synchronously.

type ClientCodec

type ClientCodec interface {
	EncodeClientRequest(call *Call) ([]byte, error)
	DecodeIdAndError(msg []byte) (*uint32, error)
	DecodeClientResponse(msg []byte, reply interface{}) error
}

ClientCodec encodes and decodes the calls and replies (currently, to json)

type Codec

type Codec interface {
	NewRequest(topic string, payload []byte) (CodecRequest, error)
	SendNotification(c bus.Bus, topic string, payload ...interface{}) error
}

Codec creates a CodecRequest to process each request.

type CodecRequest

type CodecRequest interface {
	// Reads the request and returns the RPC method name.
	Method() (string, error)
	// Reads the request filling the RPC method args.
	ReadRequest(interface{}) error
	// Writes the response using the RPC method reply.
	WriteResponse(c bus.Bus, response interface{})
	// Writes an error produced by the server.
	WriteError(c bus.Bus, err error)
}

CodecRequest decodes a request and encodes a response using a specific serialization scheme.

type ExportedService

type ExportedService struct {
	Methods []string
	// contains filtered or unexported fields
}

func (*ExportedService) SendEvent

func (s *ExportedService) SendEvent(event string, payload ...interface{}) error

type Message

type Message struct {
	Payload []byte
	Topic   string
}

type Server

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

Server serves registered RPC services using registered codecs.

func NewServer

func NewServer(client bus.Bus, codec Codec) *Server

NewServer returns a new RPC server.

func (*Server) HasMethod

func (s *Server) HasMethod(topic string, method string) bool

HasMethod returns true if the given method is registered on a topic

func (*Server) RegisterService

func (s *Server) RegisterService(receiver interface{}, topic string, schema string) (service *ExportedService, err error)

RegisterService adds a new service to the server.

The name parameter is optional: if empty it will be inferred from the receiver type name.

Methods from the receiver will be extracted if these rules are satisfied:

• The receiver is exported (begins with an upper case letter) or local (defined in the package registering the service).
• The method name is exported.
• The method's first argument is *mqtt.Message
• If there is a second argument (the RPC params value) it must be exported and a pointer
• If there is a return value, it must be first, exported and a pointer
• The method's last return value is an error

All other methods are ignored.

func (*Server) SendNotification

func (s *Server) SendNotification(topic string, params ...interface{}) error

SendNotification sends a one-way notification. Perhaps shouldn't be in Server....