ovsdb/internal/jsonrpc: break Execute into Send and Receive, add notification ability

Author: mdlayher

ovsdb/internal/jsonrpc/conn.go

@@ -16,6 +16,7 @@ package jsonrpc
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"log"
@@ -29,11 +30,28 @@ type Request struct {
 	Params []interface{} `json:"params"`
 }
 
-// A Response is a JSON-RPC response.
+// A Response is either a JSON-RPC response, or a JSON-RPC request notification.
 type Response struct {
-	ID     int         `json:"id"`
-	Result interface{} `json:"result"`
-	Error  interface{} `json:"error"`
+	// Non-null for response; null for request notification.
+	ID *int `json:"id"`
+
+	// Response fields.
+	Result json.RawMessage `json:"result,omitempty"`
+	Error  interface{}     `json:"error"`
+
+	// Request notification fields.
+	Method string          `json:"method,omitempty"`
+	Params json.RawMessage `json:"params,omitempty"`
+}
+
+// Err returns an error, if one occurred in a Response.
+func (r *Response) Err() error {
+	// TODO(mdlayher): better errors.
+	if r.Error == nil {
+		return nil
+	}
+
+	return fmt.Errorf("received JSON-RPC error: %#v", r.Error)
 }
 
 // NewConn creates a new Conn with the input io.ReadWriteCloser.
@@ -47,67 +65,67 @@ func NewConn(rwc io.ReadWriteCloser, ll *log.Logger) *Conn {
 	}
 
 	return &Conn{
-		enc:    json.NewEncoder(rwc),
-		dec:    json.NewDecoder(rwc),
-		closer: rwc,
+		c:   rwc,
+		enc: json.NewEncoder(rwc),
+		dec: json.NewDecoder(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
+	c io.Closer
+
+	encMu sync.Mutex
+	enc   *json.Encoder
+
+	decMu sync.Mutex
+	dec   *json.Decoder
 }
 
 // Close closes the connection.
 func (c *Conn) Close() error {
-	return c.closer.Close()
+	// TODO(mdlayher): acquiring mutex will block forever if receive loop
+	// is happening elsewhere. Any way to avoid this?
+	return c.c.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
+// Send sends a single JSON-RPC request.
+func (c *Conn) Send(req Request) error {
+	if req.ID == 0 {
+		return errors.New("JSON-RPC request ID must be non-zero")
 	}
 
 	// Non-nil array required for ovsdb-server to reply.
 	if req.Params == nil {
 		req.Params = []interface{}{}
 	}
 
+	c.encMu.Lock()
+	defer c.encMu.Unlock()
+
 	if err := c.enc.Encode(req); err != nil {
 		return fmt.Errorf("failed to encode JSON-RPC request: %v", err)
 	}
 
-	res := Response{
-		Result: out,
-	}
+	return nil
+}
 
-	if err := c.dec.Decode(&res); err != nil {
-		return fmt.Errorf("failed to decode JSON-RPC request: %v", err)
-	}
+// Receive receives a single JSON-RPC response.
+func (c *Conn) Receive() (*Response, error) {
+	c.decMu.Lock()
+	defer c.decMu.Unlock()
 
-	if res.ID != seq {
-		return fmt.Errorf("bad JSON-RPC sequence: %d, want: %d", res.ID, seq)
-	}
+	var res Response
+	if err := c.dec.Decode(&res); err != nil {
+		// Don't mask EOF errors with added detail.
+		if err == io.EOF {
+			return nil, err
+		}
 
-	// TODO(mdlayher): better errors.
-	if res.Error != nil {
-		return fmt.Errorf("received JSON-RPC error: %#v", res.Error)
+		return nil, fmt.Errorf("failed to decode JSON-RPC response: %v", err)
 	}
 
-	return nil
+	return &res, nil
 }
 
 type debugReadWriteCloser struct {

ovsdb/internal/jsonrpc/conn_test.go

@@ -15,57 +15,69 @@
 package jsonrpc_test
 
 import (
+	"encoding/json"
 	"fmt"
+	"io"
 	"testing"
 
 	"github.com/digitalocean/go-openvswitch/ovsdb/internal/jsonrpc"
 	"github.com/google/go-cmp/cmp"
 )
 
-func TestConnExecuteBadSequence(t *testing.T) {
-	req := jsonrpc.Request{
-		ID:     10,
-		Method: "test",
-	}
-
-	c, done := jsonrpc.TestConn(t, func(_ jsonrpc.Request) jsonrpc.Response {
-		return jsonrpc.Response{
-			// Bad sequence.
-			ID: 1,
-		}
-	})
+func TestConnSendNoRequestID(t *testing.T) {
+	c, _, done := jsonrpc.TestConn(t, nil)
 	defer done()
 
-	if err := c.Execute(req, nil); err == nil {
+	if err := c.Send(jsonrpc.Request{}); err == nil {
 		t.Fatal("expected an error, but none occurred")
 	}
 }
 
-func TestConnExecuteError(t *testing.T) {
+func TestConnReceiveEOF(t *testing.T) {
+	c := jsonrpc.NewConn(&eofReadWriteCloser{}, nil)
+
+	// Conn should not mask io.EOF.
+	_, err := c.Receive()
+	if err != io.EOF {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
+func TestConnSendReceiveError(t *testing.T) {
 	// TODO(mdlayher): what does this actually look like?
 	type rpcError struct {
 		Details string
 	}
 
-	c, done := jsonrpc.TestConn(t, func(_ jsonrpc.Request) jsonrpc.Response {
+	c, _, done := jsonrpc.TestConn(t, func(_ jsonrpc.Request) jsonrpc.Response {
 		return jsonrpc.Response{
-			ID: 10,
+			ID: intPtr(10),
 			Error: rpcError{
 				Details: "some error",
 			},
 		}
 	})
 	defer done()
 
-	if err := c.Execute(jsonrpc.Request{ID: 10}, nil); err == nil {
+	if err := c.Send(jsonrpc.Request{ID: 10}); err != nil {
+		t.Fatalf("failed to send request: %v", err)
+	}
+
+	res, err := c.Receive()
+	if err != nil {
+		t.Fatalf("failed to receive response: %v", err)
+	}
+
+	if err := res.Err(); err == nil {
 		t.Fatal("expected an error, but none occurred")
 	}
 }
 
-func TestConnExecuteOK(t *testing.T) {
+func TestConnSendReceiveOK(t *testing.T) {
 	req := jsonrpc.Request{
 		Method: "hello",
 		Params: []interface{}{"world"},
+		ID:     1,
 	}
 
 	type message struct {
@@ -76,30 +88,128 @@ func TestConnExecuteOK(t *testing.T) {
 		Message: "hello world",
 	}
 
-	c, done := jsonrpc.TestConn(t, func(got jsonrpc.Request) jsonrpc.Response {
-		req.ID = 1
-
+	c, _, done := jsonrpc.TestConn(t, func(got jsonrpc.Request) jsonrpc.Response {
 		if diff := cmp.Diff(req, got); diff != "" {
 			panicf("unexpected request (-want +got):\n%s", diff)
 		}
 
 		return jsonrpc.Response{
-			ID:     1,
-			Result: want,
+			ID:     intPtr(1),
+			Result: mustMarshalJSON(t, want),
 		}
 	})
 	defer done()
 
+	if err := c.Send(req); err != nil {
+		t.Fatalf("failed to send request: %v", err)
+	}
+
+	res, err := c.Receive()
+	if err != nil {
+		t.Fatalf("failed to receive response: %v", err)
+	}
+
+	if err := res.Err(); err != nil {
+		t.Fatalf("request failed: %v", err)
+	}
+
 	var out message
-	if err := c.Execute(req, &out); err != nil {
-		t.Fatalf("failed to execute: %v", err)
+	if err := json.Unmarshal(res.Result, &out); err != nil {
+		t.Fatalf("failed to unmarshal JSON: %v", err)
 	}
 
 	if diff := cmp.Diff(want, out); diff != "" {
 		t.Fatalf("unexpected response (-want +got):\n%s", diff)
 	}
 }
 
+func TestConnSendReceiveNotificationsOK(t *testing.T) {
+	const id = 10
+
+	req := jsonrpc.Request{
+		ID:     id,
+		Method: "monitor",
+		Params: []interface{}{"Open_vSwitch"},
+	}
+
+	res := jsonrpc.Response{
+		ID:     intPtr(id),
+		Result: mustMarshalJSON(t, "some bytes"),
+	}
+
+	c, notifC, done := jsonrpc.TestConn(t, func(got jsonrpc.Request) jsonrpc.Response {
+		if diff := cmp.Diff(req, got); diff != "" {
+			panicf("unexpected request (-want +got):\n%s", diff)
+		}
+
+		return res
+	})
+	defer done()
+
+	note := &jsonrpc.Response{
+		Method: "notify",
+	}
+	notifC <- note
+	notifC <- note
+
+	if err := c.Send(req); err != nil {
+		t.Fatalf("failed to send request: %v", err)
+	}
+
+	var responses, notes int
+	for i := 0; i < 3; i++ {
+		res, err := c.Receive()
+		if err != nil {
+			t.Fatalf("failed to receive response: %v", err)
+		}
+
+		if res.ID != nil {
+			responses++
+			if diff := cmp.Diff(req.ID, *res.ID); diff != "" {
+				t.Fatalf("unexpected response request ID (-want +got):\n%s", diff)
+			}
+
+			continue
+		}
+
+		notes++
+		if diff := cmp.Diff(note.Method, res.Method); diff != "" {
+			t.Fatalf("unexpected notification method (-want +got):\n%s", diff)
+		}
+	}
+
+	if diff := cmp.Diff(1, responses); diff != "" {
+		t.Fatalf("unexpected number of responses (-want +got):\n%s", diff)
+	}
+
+	if diff := cmp.Diff(2, notes); diff != "" {
+		t.Fatalf("unexpected number of notifications (-want +got):\n%s", diff)
+	}
+}
+
+func mustMarshalJSON(t *testing.T, v interface{}) []byte {
+	t.Helper()
+
+	b, err := json.Marshal(v)
+	if err != nil {
+		t.Fatalf("failed to marshal JSON: %v", err)
+	}
+
+	return b
+}
+
+func intPtr(i int) *int {
+	return &i
+}
+
 func panicf(format string, a ...interface{}) {
 	panic(fmt.Sprintf(format, a...))
 }
+
+type eofReadWriteCloser struct {
+	io.ReadWriteCloser
+}
+
+func (rwc *eofReadWriteCloser) Read(b []byte) (int, error) {
+	return 0, io.EOF
+}