ovsdb: use ovsdb/internal/jsonrpc

Author: mdlayher

ovsdb/client.go

@@ -18,50 +18,82 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
-	"net/rpc"
-	"net/rpc/jsonrpc"
+	"log"
+	"net"
+
+	"github.com/digitalocean/go-openvswitch/ovsdb/internal/jsonrpc"
 )
 
 // A Client is an OVSDB client.
 type Client struct {
-	rc *rpc.Client
+	c  *jsonrpc.Conn
+	ll *log.Logger
+}
+
+// An OptionFunc is a function which can configure a Client.
+type OptionFunc func(c *Client) error
+
+// Debug enables debug logging for a Client.
+func Debug(ll *log.Logger) OptionFunc {
+	return func(c *Client) error {
+		c.ll = ll
+		return nil
+	}
 }
 
 // Dial dials a connection to an OVSDB server and returns a Client.
-func Dial(network, addr string) (*Client, error) {
-	c, err := jsonrpc.Dial(network, addr)
+func Dial(network, addr string, options ...OptionFunc) (*Client, error) {
+	conn, err := net.Dial(network, addr)
 	if err != nil {
 		return nil, err
 	}
 
-	return &Client{
-		rc: c,
-	}, nil
+	return New(conn, options...)
+}
+
+// New wraps an existing connection to an OVSDB server and returns a Client.
+func New(conn net.Conn, options ...OptionFunc) (*Client, error) {
+	client := &Client{}
+	for _, o := range options {
+		if err := o(client); err != nil {
+			return nil, err
+		}
+	}
+
+	client.c = jsonrpc.NewConn(conn, client.ll)
+
+	return client, nil
 }
 
 // Close closes a Client's connection.
 func (c *Client) Close() error {
-	return c.rc.Close()
+	return c.c.Close()
 }
 
 // ListDatabases returns the name of all databases known to the OVSDB server.
 func (c *Client) ListDatabases() ([]string, error) {
 	var dbs []string
-	if err := c.rpc("list_dbs", nil, &dbs); err != nil {
+	if err := c.rpc("list_dbs", &dbs); err != nil {
 		return nil, err
 	}
 
 	return dbs, nil
 }
 
 // rpc performs a single RPC request, and checks the response for errors.
-func (c *Client) rpc(method string, args, reply interface{}) error {
-	// Captures any JSON-RPC errors.
+func (c *Client) rpc(method string, out interface{}, args ...interface{}) error {
+	// Captures any OVSDB errors.
 	r := result{
-		Reply: reply,
+		Reply: out,
+	}
+
+	req := jsonrpc.Request{
+		Method: method,
+		Params: args,
+		// Let the client handle the request ID.
 	}
 
-	if err := c.rc.Call(method, args, &r); err != nil {
+	if err := c.c.Execute(req, &r); err != nil {
 		return err
 	}
 

ovsdb/client_test.go

@@ -15,24 +15,25 @@
 package ovsdb_test
 
 import (
-	"encoding/json"
 	"fmt"
-	"net"
-	"sync"
 	"testing"
 
 	"github.com/digitalocean/go-openvswitch/ovsdb"
+	"github.com/digitalocean/go-openvswitch/ovsdb/internal/jsonrpc"
 	"github.com/google/go-cmp/cmp"
 )
 
 func TestClientError(t *testing.T) {
 	const str = "some error"
 
-	c, done := testClient(t, func(_ string, _ []interface{}) interface{} {
-		return &ovsdb.Error{
-			Err:     str,
-			Details: "malformed",
-			Syntax:  "{}",
+	c, done := testClient(t, func(_ jsonrpc.Request) jsonrpc.Response {
+		return jsonrpc.Response{
+			ID: 1,
+			Result: &ovsdb.Error{
+				Err:     str,
+				Details: "malformed",
+				Syntax:  "{}",
+			},
 		}
 	})
 	defer done()
@@ -51,19 +52,23 @@ func TestClientError(t *testing.T) {
 		t.Fatalf("unexpected error (-want +got):\n%s", diff)
 	}
 }
+
 func TestClientListDatabases(t *testing.T) {
 	want := []string{"Open_vSwitch", "test"}
 
-	c, done := testClient(t, func(method string, params []interface{}) interface{} {
-		if diff := cmp.Diff("list_dbs", method); diff != "" {
-			t.Fatalf("unexpected RPC method (-want +got):\n%s", diff)
+	c, done := testClient(t, func(req jsonrpc.Request) jsonrpc.Response {
+		if diff := cmp.Diff("list_dbs", req.Method); diff != "" {
+			panicf("unexpected RPC method (-want +got):\n%s", diff)
 		}
 
-		if diff := cmp.Diff(1, len(params)); diff != "" {
-			t.Fatalf("unexpected number of RPC parameters (-want +got):\n%s", diff)
+		if diff := cmp.Diff(0, len(req.Params)); diff != "" {
+			panicf("unexpected number of RPC parameters (-want +got):\n%s", diff)
 		}
 
-		return want
+		return jsonrpc.Response{
+			ID:     1,
+			Result: want,
+		}
 	})
 	defer done()
 
@@ -77,70 +82,17 @@ func TestClientListDatabases(t *testing.T) {
 	}
 }
 
-type rpcFunc func(method string, params []interface{}) interface{}
-
-func testClient(t *testing.T, fn rpcFunc) (*ovsdb.Client, func()) {
+func testClient(t *testing.T, fn jsonrpc.TestFunc) (*ovsdb.Client, func()) {
 	t.Helper()
 
-	l, err := net.Listen("tcp", ":0")
-	if err != nil {
-		t.Fatalf("failed to listen: %v", err)
-	}
-
-	var wg sync.WaitGroup
-	wg.Add(1)
-
-	go func() {
-		defer wg.Done()
-
-		// Accept a single connection.
-		c, err := l.Accept()
-		if err != nil {
-			panicf("failed to accept: %v", err)
-		}
-		defer c.Close()
-		_ = l.Close()
-
-		if err := handleConn(c, fn); err != nil {
-			panicf("failed to handle connection: %v", err)
-		}
-	}()
+	conn, done := jsonrpc.TestNetConn(t, fn)
 
-	c, err := ovsdb.Dial("tcp", l.Addr().String())
+	c, err := ovsdb.New(conn)
 	if err != nil {
 		t.Fatalf("failed to dial: %v", err)
 	}
 
-	return c, func() {
-		// Ensure types are cleaned up, and ensure goroutine stops.
-		_ = l.Close()
-		_ = c.Close()
-		wg.Wait()
-	}
-}
-
-func handleConn(c net.Conn, fn rpcFunc) error {
-	var req struct {
-		Method string        `json:"method"`
-		Params []interface{} `json:"params"`
-		ID     int           `json:"id"`
-	}
-
-	var res struct {
-		Result interface{} `json:"result"`
-		ID     int         `json:"id"`
-	}
-
-	if err := json.NewDecoder(c).Decode(&req); err != nil {
-		return err
-	}
-
-	result := fn(req.Method, req.Params)
-
-	res.ID = req.ID
-	res.Result = result
-
-	return json.NewEncoder(c).Encode(res)
+	return c, done
 }
 
 func panicf(format string, a ...interface{}) {