Add test for vrpc

Author: vaastav

example/RpcClientServer/RpcClientServer.go

@@ -54,13 +54,15 @@ func rpcserver() {
 		log.Fatal("listen error:", e)
 	}
 
-	vrpc.ServeRPCConn(server, l, logger)
+	options := govec.GetDefaultLogOptions()
+	vrpc.ServeRPCConn(server, l, logger, options)
 }
 
 func rpcclient() {
 	fmt.Println("Starting client")
 	logger := govec.InitGoVector("client", "clientlogfile", govec.GetDefaultConfig())
-	client, err := vrpc.RPCDial("tcp", "127.0.0.1:8080", logger)
+	options := govec.GetDefaultLogOptions()
+	client, err := vrpc.RPCDial("tcp", "127.0.0.1:8080", logger, options)
 	if err != nil {
 		log.Fatal(err)
 	}

govec/vrpc/vrpc.go

@@ -16,26 +16,26 @@ import (
 //RPCDial connects to a RPC server at the specified network address. The
 //logger is provided to be used by the RPCClientCodec for message
 //capture.
-func RPCDial(network, address string, logger *govec.GoLog) (*rpc.Client, error) {
+func RPCDial(network, address string, logger *govec.GoLog, options govec.GoLogOptions) (*rpc.Client, error) {
 	conn, err := net.Dial(network, address)
 	if err != nil {
 		return nil, err
 	}
-	return rpc.NewClientWithCodec(newClientCodec(conn, logger)), err
+	return rpc.NewClientWithCodec(newClientCodec(conn, logger, options)), err
 }
 
 //ServerRPCCon is a convenience function that accepts connections for a
 //given listener and starts a new goroutine for the server to serve a
 //new connection. The logger is provided to be used by the
 //RPCServerCodec for message capture.
-func ServeRPCConn(server *rpc.Server, l net.Listener, logger *govec.GoLog) {
+func ServeRPCConn(server *rpc.Server, l net.Listener, logger *govec.GoLog, options govec.GoLogOptions) {
 	for {
 		conn, err := l.Accept()
 		if err != nil {
 			log.Fatal(err)
 		}
 
-		go server.ServeCodec(newServerCodec(conn, logger))
+		go server.ServeCodec(newServerCodec(conn, logger, options))
 	}
 }
 
@@ -48,6 +48,7 @@ type RPCClientCodec struct {
 	Enc    *gob.Encoder
 	EncBuf *bufio.Writer
 	Logger *govec.GoLog
+	Options govec.GoLogOptions
 }
 
 //RPCServerCodec is an extension of the default rpc codec which uses a
@@ -59,17 +60,18 @@ type RPCServerCodec struct {
 	Enc    *gob.Encoder
 	EncBuf *bufio.Writer
 	Logger *govec.GoLog
+	Options govec.GoLogOptions
 	Closed bool
 }
 
 //NewClient returs an rpc.Client insturmented with vector clocks.
-func NewClient(conn io.ReadWriteCloser, logger *govec.GoLog) *rpc.Client {
-	return rpc.NewClientWithCodec(newClientCodec(conn, logger))
+func NewClient(conn io.ReadWriteCloser, logger *govec.GoLog, options govec.GoLogOptions) *rpc.Client {
+	return rpc.NewClientWithCodec(newClientCodec(conn, logger, options))
 }
 
-func newClientCodec(conn io.ReadWriteCloser, logger *govec.GoLog) rpc.ClientCodec {
+func newClientCodec(conn io.ReadWriteCloser, logger *govec.GoLog, options govec.GoLogOptions) rpc.ClientCodec {
 	encBuf := bufio.NewWriter(conn)
-	return &RPCClientCodec{conn, gob.NewDecoder(conn), gob.NewEncoder(encBuf), encBuf, logger}
+	return &RPCClientCodec{conn, gob.NewDecoder(conn), gob.NewEncoder(encBuf), encBuf, logger, options}
 }
 
 //WriteRequest marshalls and sends an rpc request, and it's associated
@@ -78,8 +80,8 @@ func (c *RPCClientCodec) WriteRequest(req *rpc.Request, param interface{}) (err
 	if err = c.Enc.Encode(req); err != nil {
 		return
 	}
-	opts := govec.GetDefaultLogOptions()
-	buf := c.Logger.PrepareSend("Making RPC call", param, opts)
+
+	buf := c.Logger.PrepareSend("Making RPC call", param, c.Options)
 	if err = c.Enc.Encode(buf); err != nil {
 		return
 	}
@@ -99,8 +101,7 @@ func (c *RPCClientCodec) ReadResponseBody(body interface{}) (err error) {
 	if err = c.Dec.Decode(&buf); err != nil {
 		return
 	}
-	opts := govec.GetDefaultLogOptions()
-	c.Logger.UnpackReceive("Received RPC Call response from server", buf, body, opts)
+	c.Logger.UnpackReceive("Received RPC Call response from server", buf, body, c.Options)
 	return nil
 }
 
@@ -109,14 +110,15 @@ func (c *RPCClientCodec) Close() error {
 	return c.C.Close()
 }
 
-func newServerCodec(conn io.ReadWriteCloser, logger *govec.GoLog) rpc.ServerCodec {
+func newServerCodec(conn io.ReadWriteCloser, logger *govec.GoLog, options govec.GoLogOptions) rpc.ServerCodec {
 	buf := bufio.NewWriter(conn)
 	srv := &RPCServerCodec{
 		Rwc:    conn,
 		Dec:    gob.NewDecoder(conn),
 		Enc:    gob.NewEncoder(buf),
 		EncBuf: buf,
 		Logger: logger,
+		Options: options,
 	}
 	return srv
 }
@@ -133,17 +135,15 @@ func (c *RPCServerCodec) ReadRequestBody(body interface{}) (err error) {
 	if err = c.Dec.Decode(&buf); err != nil {
 		return
 	}
-	opts := govec.GetDefaultLogOptions()
-	c.Logger.UnpackReceive("Received RPC request", buf, body, opts)
+	c.Logger.UnpackReceive("Received RPC request", buf, body, c.Options)
 	return nil
 }
 
 //WriteResponse sends an rpc response, and it's associated result back
 //to the client
 func (c *RPCServerCodec) WriteResponse(r *rpc.Response, body interface{}) (err error) {
 	Encode(c, r)
-	opts := govec.GetDefaultLogOptions()
-	buf := c.Logger.PrepareSend("Sending response to RPC request", body, opts)
+	buf := c.Logger.PrepareSend("Sending response to RPC request", body, c.Options)
 	Encode(c, buf)
 	return c.EncBuf.Flush()
 }

govec/vrpc/vrpc_test.go

@@ -0,0 +1,101 @@
+package vrpc
+
+import (
+	"errors"
+	"fmt"
+	"log"
+	"net"
+	"net/rpc"
+	"time"
+	"testing"
+
+	"github.com/DistributedClocks/GoVector/govec"
+)
+
+var done chan int = make(chan int, 1)
+
+//Args are matematical arguments for the rpc operations
+type Args struct {
+	A, B int
+}
+
+//Quotient is the result of a Divide RPC
+type Quotient struct {
+	Quo, Rem int
+}
+
+//Arith is an RPC math server type
+type Arith int
+
+//Multiply performs multiplication on two integers
+func (t *Arith) Multiply(args *Args, reply *int) error {
+	*reply = args.A * args.B
+	return nil
+}
+
+//Divide divides a by b and returns a quotient with a remainder
+func (t *Arith) Divide(args *Args, quo *Quotient) error {
+	if args.B == 0 {
+		return errors.New("divide by zero")
+	}
+	quo.Quo = args.A / args.B
+	quo.Rem = args.A % args.B
+	return nil
+}
+
+func rpcserver(logger *govec.GoLog) {
+	fmt.Println("Starting server")
+	arith := new(Arith)
+	server := rpc.NewServer()
+	server.Register(arith)
+	l, e := net.Listen("tcp", ":8080")
+	if e != nil {
+		log.Fatal("listen error:", e)
+	}
+
+	options := govec.GetDefaultLogOptions()
+	ServeRPCConn(server, l, logger, options)
+}
+
+func rpcclient(logger *govec.GoLog) {
+	fmt.Println("Starting client")
+	options := govec.GetDefaultLogOptions()
+	client, err := RPCDial("tcp", "127.0.0.1:8080", logger, options)
+	if err != nil {
+		log.Fatal(err)
+	}
+	var result int
+	err = client.Call("Arith.Multiply", Args{5, 6}, &result)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	var qresult Quotient
+	err = client.Call("Arith.Divide", Args{4, 2}, &qresult)
+	if err != nil {
+		log.Fatal(err)
+	}
+	done <- 1
+}
+
+func TestRPC(t *testing.T) {
+	serverlogger := govec.InitGoVector("server", "serverlogfile", govec.GetDefaultConfig())
+	clientlogger := govec.InitGoVector("client", "clientlogfile", govec.GetDefaultConfig())
+	go rpcserver(serverlogger)
+	time.Sleep(time.Millisecond)
+	go rpcclient(clientlogger)
+	<-done
+	server_vc := serverlogger.GetCurrentVC()
+	server_ticks, _ := server_vc.FindTicks("server")
+	client_vc := clientlogger.GetCurrentVC()
+	client_ticks, _ := client_vc.FindTicks("client")
+
+	AssertEquals(t, uint64(5), server_ticks, "Server Clock value not incremented")
+	AssertEquals(t, uint64(5), client_ticks, "Client Clock value not incremented")
+}
+
+func AssertEquals(t *testing.T, expected interface{}, actual interface{}, message string) {
+	if expected != actual {
+		t.Fatalf(message+"Expected: %s, Actual: %s", expected, actual)
+	}
+}