Merge pull request #11 from digitalocean/mdl-ovsdb-jsonrpc

ovsdb/internal/jsonrpc: initial commit

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{}) {

ovsdb/internal/jsonrpc/conn.go

@@ -0,0 +1,142 @@
+// Copyright 2017 DigitalOcean.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package jsonrpc
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"log"
+	"sync"
+)
+
+// A Request is a JSON-RPC request.
+type Request struct {
+	ID     int           `json:"id"`
+	Method string        `json:"method"`
+	Params []interface{} `json:"params"`
+}
+
+// A Response is a JSON-RPC response.
+type Response struct {
+	ID     int         `json:"id"`
+	Result interface{} `json:"result"`
+	Error  interface{} `json:"error"`
+}
+
+// NewConn creates a new Conn with the input io.ReadWriteCloser.
+// If a logger is specified, it is used for debug logs.
+func NewConn(rwc io.ReadWriteCloser, ll *log.Logger) *Conn {
+	if ll != nil {
+		rwc = &debugReadWriteCloser{
+			rwc: rwc,
+			ll:  ll,
+		}
+	}
+
+	return &Conn{
+		enc:    json.NewEncoder(rwc),
+		dec:    json.NewDecoder(rwc),
+		closer: rwc,
+	}
+}
+
+// A Conn is a JSON-RPC connection.
+type Conn struct {
+	mu     sync.RWMutex
+	enc    *json.Encoder
+	dec    *json.Decoder
+	closer io.Closer
+	seq    int
+}
+
+// Close closes the connection.
+func (c *Conn) Close() error {
+	return c.closer.Close()
+}
+
+// Execute executes a single request and unmarshals its results into out.
+func (c *Conn) Execute(req Request, out interface{}) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.seq++
+
+	// Use auto-increment sequence, or user-defined if requested.
+	seq := c.seq
+	if req.ID != 0 {
+		seq = req.ID
+	} else {
+		req.ID = seq
+	}
+
+	// Non-nil array required for ovsdb-server to reply.
+	if req.Params == nil {
+		req.Params = []interface{}{}
+	}
+
+	if err := c.enc.Encode(req); err != nil {
+		return fmt.Errorf("failed to encode JSON-RPC request: %v", err)
+	}
+
+	res := Response{
+		Result: out,
+	}
+
+	if err := c.dec.Decode(&res); err != nil {
+		return fmt.Errorf("failed to decode JSON-RPC request: %v", err)
+	}
+
+	if res.ID != seq {
+		return fmt.Errorf("bad JSON-RPC sequence: %d, want: %d", res.ID, seq)
+	}
+
+	// TODO(mdlayher): better errors.
+	if res.Error != nil {
+		return fmt.Errorf("received JSON-RPC error: %#v", res.Error)
+	}
+
+	return nil
+}
+
+type debugReadWriteCloser struct {
+	rwc io.ReadWriteCloser
+	ll  *log.Logger
+}
+
+func (rwc *debugReadWriteCloser) Read(b []byte) (int, error) {
+	n, err := rwc.rwc.Read(b)
+	if err != nil {
+		return n, err
+	}
+
+	rwc.ll.Printf(" read: %s", string(b[:n]))
+	return n, nil
+}
+
+func (rwc *debugReadWriteCloser) Write(b []byte) (int, error) {
+	n, err := rwc.rwc.Write(b)
+	if err != nil {
+		return n, err
+	}
+
+	rwc.ll.Printf("write: %s", string(b[:n]))
+	return n, nil
+}
+
+func (rwc *debugReadWriteCloser) Close() error {
+	err := rwc.rwc.Close()
+	rwc.ll.Println("close:", err)
+	return err
+}