bilateralrpc

README

bilateralrpc - RPC across ØMQ connection

Package bilateralrpc implements RPC calls among connected system in either direction across a single ØMQ connection.

Documentation

Overview

   Package bilateralrpc implements RPC calls among connected system
   in either direction across a single ØMQ connection.

   i.e. The connection is symmetrical and either end can call
        procedures registered in the other

   The call is by default multicast with a timeout and collects as
   many of the destination results available within a timeout.

   // the procedures
   // --------------

   // sample server object
   type Arith struct{}

   // parameters
   type Args struct {
	A, B int
   }

   // reply from one server
   type Reply struct {
	S int
	D int
   }

   // result slice type from multicast call
   type Result struct {
	From  string
	Reply Reply
   }

   // example
   func (t *Arith) SumDiff(args *Args, reply *Reply) error {
	reply.S = args.A + args.B
	reply.D = args.A - args.B
	return nil
   }

   // the client/server setup code
   // ----------------------------

   // start up the communication system
   server := bilateralrpc.NewBilateral("myserver")
   defer server.Close()

   // listen on ports (multiple ListenOn allowed)
   server.ListenOn("tcp://*:5555")

   // register server objects
   arith := new(Arith)
   server.Register(arith)

   // many connections can be made like this:
   to := "yourserver"
   err := server.ConnectTo(to, "tcp://127.0.0.1:5566")
   if nil != err {
	panic(fmt.Sprintf("connect to %s err = %v\n", to, err))
   }

   // calling an RPC
   // --------------

   args := Args{
       A: 100,
       B: 250,
   }

   // this will receive slice of results
   var results []Result

   // empty string will result in calls to all active connections
   err = server.Call([]string{}, "Arith.SumDiff", args, &results, 5*time.Second)

   if nil != err {
       panic(fmt.Sprintf("err = %v", err))
   }

   if 0 == len(results) {
       // there were no active connections or nothing replied within the timeout
   }

Index

• Constants
• Variables
• func NewKeypair() (publicKey string, privateKey string, err error)
• type Bilateral
  • func NewEncrypted(networkName string, publicKey string, privateKey string) *Bilateral
  • func NewPlaintext(networkName string, serverName string) *Bilateral
  • func (twoway *Bilateral) ActiveConnections() []string
  • func (twoway *Bilateral) Call(to []string, method string, args interface{}, results interface{}, ...) error
  • func (twoway *Bilateral) Cast(to []string, method string, args interface{}) error
  • func (twoway *Bilateral) Close()
  • func (twoway *Bilateral) ConnectTo(name string, address string) error
  • func (twoway *Bilateral) ConnectionCount() int
  • func (twoway *Bilateral) DisconnectFrom(name string) error
  • func (twoway *Bilateral) ListenOn(address string) error
  • func (twoway *Bilateral) Register(receiver interface{}) error

Constants

const (
	// state machine timeouts
	FIRST_START_TIME = 200 * time.Millisecond // from program start to first connect request
	FIRST_JOIN_TIME  = 200 * time.Millisecond // time to send first join after connect
	JOIN_TIME        = 5 * time.Second        // time between join requests
	LIVE_TIME        = 30 * time.Second       // time to send ping
	RETRY_TIME       = 60 * time.Second       // starting retry time
	SERVE_TIME       = LIVE_TIME + RETRY_TIME // timeout for server side
	BASE_PART_TIME   = 5 * time.Second        // delay before retry on a PART
	RAND_PART_TIME   = 20 * time.Millisecond  // multipled by a random byte (0..255)

	// state machine counts
	MAXIMUM_BACKOFF = 7  // backoff counter for retries
	MAXIMUM_JOINS   = 10 // number of joins before disconnect then retry

	// ØMQ settings
	LINGER_TIME  = 50 * time.Millisecond // time to try and send data after a disconnect
	SEND_TIMEOUT = 0                     // 0 => fail immediately if no buffer space for send

	// default timeout for Call()
	MINIMUM_CALL_TIMEOUT = 200 * time.Millisecond
	DEFAULT_CALL_TIMEOUT = 2000 * time.Millisecond // used if no tiemout is passed is passed
	DEFAULT_RETRIES      = 3
)

Variables

var SendToAll = []string{}

sent to all connections value for Call

Functions

func NewKeypair

func NewKeypair() (publicKey string, privateKey string, err error)

create a new key pair

Types

type Bilateral

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

type to hold the client/server data

func NewEncrypted

func NewEncrypted(networkName string, publicKey string, privateKey string) *Bilateral

func NewPlaintext

func NewPlaintext(networkName string, serverName string) *Bilateral

set up the connection (can set plain or encrypted) Note that in encrypted mode the public key functions as the server name and that the netwokName must match on both sides for connection to be established

for plaintext call as: NewPlaintext(network_name, server_name)

for encrypted call as: NewEncrypted(network_name, public_key, private_key)

with values from:  pub, priv, err := NewKeypair()

func (*Bilateral) ActiveConnections

func (twoway *Bilateral) ActiveConnections() []string

get a list of acive peers names

func (*Bilateral) Call

func (twoway *Bilateral) Call(to []string, method string, args interface{}, results interface{}, optionalTimeout ...time.Duration) error

rpc call routine and receive reply, waiting if necessary to can be set to SendToAll to do a broadcast

func (*Bilateral) Cast

func (twoway *Bilateral) Cast(to []string, method string, args interface{}) error

rpc cast routine - no reply will be sent by remote to can be set to SendToAll to do a broadcast

func (*Bilateral) Close

func (twoway *Bilateral) Close()

shutdown all connections, terminate all background processing

func (*Bilateral) ConnectTo

func (twoway *Bilateral) ConnectTo(name string, address string) error

connect to a remote

func (*Bilateral) ConnectionCount

func (twoway *Bilateral) ConnectionCount() int

count active connections

func (*Bilateral) DisconnectFrom

func (twoway *Bilateral) DisconnectFrom(name string) error

disconnect from a remote

func (*Bilateral) ListenOn

func (twoway *Bilateral) ListenOn(address string) error

listen for incomming requests

func (*Bilateral) Register

func (twoway *Bilateral) Register(receiver interface{}) error

register a callback object