rpc2

README

rpc2

rpc2 is a fork of net/rpc package in the standard library. The main goal is to add bi-directional support to calls. That means server can call the methods of client. This is not possible with net/rpc package. In order to do this it adds a *Client argument to method signatures.

Install

go get github.com/cenkalti/rpc2

Example server

package main

import (
	"fmt"
	"net"

	"github.com/cenkalti/rpc2"
)

type Args struct{ A, B int }
type Reply int

func main() {
	srv := rpc2.NewServer()
	srv.Handle("add", func(client *rpc2.Client, args *Args, reply *Reply) error {

		// Reversed call (server to client)
		var rep Reply
		client.Call("mult", Args{2, 3}, &rep)
		fmt.Println("mult result:", rep)

		*reply = Reply(args.A + args.B)
		return nil
	})

	lis, _ := net.Listen("tcp", "127.0.0.1:5000")
	srv.Accept(lis)
}

Example Client

package main

import (
	"fmt"
	"net"

	"github.com/cenkalti/rpc2"
)

type Args struct{ A, B int }
type Reply int

func main() {
	conn, _ := net.Dial("tcp", "127.0.0.1:5000")

	clt := rpc2.NewClient(conn)
	clt.Handle("mult", func(client *rpc2.Client, args *Args, reply *Reply) error {
		*reply = Reply(args.A * args.B)
		return nil
	})
	go clt.Run()

	var rep Reply
	clt.Call("add", Args{1, 2}, &rep)
	fmt.Println("add result:", rep)
}

Documentation

Overview

Package rpc2 provides bi-directional RPC client and server similar to net/rpc.

Index

• Variables
• type Call
• type Client
  • func NewClient(conn io.ReadWriteCloser) *Client
  • func NewClientWithCodec(codec Codec) *Client
  • func (c *Client) Call(method string, args interface{}, reply interface{}) error
  • func (c *Client) CallWithContext(ctx context.Context, method string, args interface{}, reply interface{}) error
  • func (c *Client) Close() error
  • func (c *Client) DisconnectNotify() chan struct{}
  • func (c *Client) Go(method string, args interface{}, reply interface{}, done chan *Call) *Call
  • func (c *Client) Handle(method string, handlerFunc interface{})
  • func (c *Client) Notify(method string, args interface{}) error
  • func (c *Client) Run()
  • func (c *Client) SetBlocking(blocking bool)
• type Codec
  • func NewGobCodec(conn io.ReadWriteCloser) Codec
• type Request
• type Response
• type Server
  • func NewServer() *Server
  • func (s *Server) Accept(lis net.Listener)
  • func (s *Server) Handle(method string, handlerFunc interface{})
  • func (s *Server) OnConnect(f func(*Client))
  • func (s *Server) OnDisconnect(f func(*Client))
  • func (s *Server) ServeCodec(codec Codec)
  • func (s *Server) ServeCodecWithState(codec Codec, state *State)
  • func (s *Server) ServeConn(conn io.ReadWriteCloser)
• type ServerError
  • func (e ServerError) Error() string
• type State
  • func NewState() *State
  • func (s *State) Get(key string) (value interface{}, ok bool)
  • func (s *State) Set(key string, value interface{})

Variables

var DebugLog = false

DebugLog controls the printing of internal and I/O errors.

var ErrShutdown = errors.New("connection is shut down")

ErrShutdown is returned when the connection is closing or closed.

Types

type Call

type Call struct {
	Method 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.
}

Call represents an active RPC.

type Client

type Client struct {
	State *State // additional information to associate with client
	// 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(conn io.ReadWriteCloser) *Client

NewClient returns a new Client to handle requests to the set of services at the other end of the connection. It adds a buffer to the write side of the connection so the header and payload are sent as a unit.

func NewClientWithCodec

func NewClientWithCodec(codec Codec) *Client

NewClientWithCodec is like NewClient but uses the specified codec to encode requests and decode responses.

func (*Client) Call

func (c *Client) Call(method string, args interface{}, reply interface{}) error

Call invokes the named function, waits for it to complete, and returns its error status.

func (*Client) CallWithContext

func (c *Client) CallWithContext(ctx context.Context, method string, args interface{}, reply interface{}) error

CallWithContext invokes the named function, waits for it to complete, and returns its error status, or an error from Context timeout.

func (*Client) Close

func (c *Client) Close() error

Close waits for active calls to finish and closes the codec.

func (*Client) DisconnectNotify

func (c *Client) DisconnectNotify() chan struct{}

DisconnectNotify returns a channel that is closed when the client connection has gone away.

func (*Client) Go

func (c *Client) Go(method string, args interface{}, reply interface{}, done chan *Call) *Call

Go invokes the function asynchronously. It returns the Call structure representing the invocation. The done channel will signal when the call is complete by returning the same Call object. If done is nil, Go will allocate a new channel. If non-nil, done must be buffered or Go will deliberately crash.

func (*Client) Handle

func (c *Client) Handle(method string, handlerFunc interface{})

Handle registers the handler function for the given method. If a handler already exists for method, Handle panics.

func (*Client) Notify

func (c *Client) Notify(method string, args interface{}) error

Notify sends a request to the receiver but does not wait for a return value.

func (*Client) Run

func (c *Client) Run()

Run the client's read loop. You must run this method before calling any methods on the server.

func (*Client) SetBlocking

func (c *Client) SetBlocking(blocking bool)

SetBlocking puts the client in blocking mode. In blocking mode, received requests are processes synchronously. If you have methods that may take a long time, other subsequent requests may time out.

type Codec

type Codec interface {
	// ReadHeader must read a message and populate either the request
	// or the response by inspecting the incoming message.
	ReadHeader(*Request, *Response) error

	// ReadRequestBody into args argument of handler function.
	ReadRequestBody(interface{}) error

	// ReadResponseBody into reply argument of handler function.
	ReadResponseBody(interface{}) error

	// WriteRequest must be safe for concurrent use by multiple goroutines.
	WriteRequest(*Request, interface{}) error

	// WriteResponse must be safe for concurrent use by multiple goroutines.
	WriteResponse(*Response, interface{}) error

	// Close is called when client/server finished with the connection.
	Close() error
}

A Codec implements reading and writing of RPC requests and responses. The client calls ReadHeader to read a message header. The implementation must populate either Request or Response argument. Depending on which argument is populated, ReadRequestBody or ReadResponseBody is called right after ReadHeader. ReadRequestBody and ReadResponseBody may be called with a nil argument to force the body to be read and then discarded.

func NewGobCodec

func NewGobCodec(conn io.ReadWriteCloser) Codec

NewGobCodec returns a new rpc2.Codec using gob encoding/decoding on conn.

type Request

type Request struct {
	Seq    uint64 // sequence number chosen by client
	Method string
}

Request is a header written before every RPC call.

type Response

type Response struct {
	Seq   uint64 // echoes that of the request
	Error string // error, if any.
}

Response is a header written before every RPC return.

type Server

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

Server responds to RPC requests made by Client.

func NewServer

func NewServer() *Server

NewServer returns a new Server.

func (*Server) Accept

func (s *Server) Accept(lis net.Listener)

Accept accepts connections on the listener and serves requests for each incoming connection. Accept blocks; the caller typically invokes it in a go statement.

func (*Server) Handle

func (s *Server) Handle(method string, handlerFunc interface{})

Handle registers the handler function for the given method. If a handler already exists for method, Handle panics.

func (*Server) OnConnect

func (s *Server) OnConnect(f func(*Client))

OnConnect registers a function to run when a client connects.

func (*Server) OnDisconnect

func (s *Server) OnDisconnect(f func(*Client))

OnDisconnect registers a function to run when a client disconnects.

func (*Server) ServeCodec

func (s *Server) ServeCodec(codec Codec)

ServeCodec is like ServeConn but uses the specified codec to decode requests and encode responses.

func (*Server) ServeCodecWithState

func (s *Server) ServeCodecWithState(codec Codec, state *State)

ServeCodecWithState is like ServeCodec but also gives the ability to associate a state variable with the client that persists across RPC calls.

func (*Server) ServeConn

func (s *Server) ServeConn(conn io.ReadWriteCloser)

ServeConn runs the server on a single connection. ServeConn blocks, serving the connection until the client hangs up. The caller typically invokes ServeConn in a go statement. ServeConn uses the gob wire format (see package gob) on the connection. To use an alternate codec, use ServeCodec.

type ServerError

type ServerError string

ServerError represents an error that has been returned from the remote side of the RPC connection.

func (ServerError) Error

func (e ServerError) Error() string

type State

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

func NewState

func NewState() *State

func (*State) Get

func (s *State) Get(key string) (value interface{}, ok bool)

func (*State) Set

func (s *State) Set(key string, value interface{})