ovsdb/internal/jsonrpc: initial commit

Author: mdlayher

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
+}

ovsdb/internal/jsonrpc/conn_test.go

@@ -0,0 +1,105 @@
+// 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_test
+
+import (
+	"fmt"
+	"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,
+		}
+	})
+	defer done()
+
+	if err := c.Execute(req, nil); err == nil {
+		t.Fatal("expected an error, but none occurred")
+	}
+}
+
+func TestConnExecuteError(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 {
+		return jsonrpc.Response{
+			ID: 10,
+			Error: rpcError{
+				Details: "some error",
+			},
+		}
+	})
+	defer done()
+
+	if err := c.Execute(jsonrpc.Request{ID: 10}, nil); err == nil {
+		t.Fatal("expected an error, but none occurred")
+	}
+}
+
+func TestConnExecuteOK(t *testing.T) {
+	req := jsonrpc.Request{
+		Method: "hello",
+		Params: []interface{}{"world"},
+	}
+
+	type message struct {
+		Message string `json:"message"`
+	}
+
+	want := message{
+		Message: "hello world",
+	}
+
+	c, done := jsonrpc.TestConn(t, func(got jsonrpc.Request) jsonrpc.Response {
+		req.ID = 1
+
+		if diff := cmp.Diff(req, got); diff != "" {
+			panicf("unexpected request (-want +got):\n%s", diff)
+		}
+
+		return jsonrpc.Response{
+			ID:     1,
+			Result: want,
+		}
+	})
+	defer done()
+
+	var out message
+	if err := c.Execute(req, &out); err != nil {
+		t.Fatalf("failed to execute: %v", err)
+	}
+
+	if diff := cmp.Diff(want, out); diff != "" {
+		t.Fatalf("unexpected response (-want +got):\n%s", diff)
+	}
+}
+
+func panicf(format string, a ...interface{}) {
+	panic(fmt.Sprintf(format, a...))
+}

ovsdb/internal/jsonrpc/doc.go

@@ -0,0 +1,16 @@
+// 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 is a minimal JSON-RPC 1.0 implementation.
+package jsonrpc

ovsdb/internal/jsonrpc/testconn.go

@@ -0,0 +1,98 @@
+// 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"
+	"net"
+	"strings"
+	"sync"
+	"testing"
+)
+
+// A TestFunc is used to create RPC responses in TestConn.
+type TestFunc func(req Request) Response
+
+// TestConn creates a Conn backed by a server that calls a TestFunc.
+// Invoke the returned closure to clean up its resources.
+func TestConn(t *testing.T, fn TestFunc) (*Conn, func()) {
+	t.Helper()
+
+	conn, done := TestNetConn(t, fn)
+
+	c := NewConn(conn, nil)
+
+	return c, func() {
+		_ = c.Close()
+		done()
+	}
+}
+
+// TestNetConn creates a net.Conn backed by a server that calls a TestFunc.
+// Invoke the returned closure to clean up its resources.
+func TestNetConn(t *testing.T, fn TestFunc) (net.Conn, 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()
+
+		for {
+			c, err := l.Accept()
+			if err != nil {
+				if strings.Contains(err.Error(), "use of closed network") {
+					return
+				}
+
+				panicf("failed to accept: %v", err)
+			}
+
+			var req Request
+			if err := json.NewDecoder(c).Decode(&req); err != nil {
+				panicf("failed to decode request: %v", err)
+			}
+
+			res := fn(req)
+			if err := json.NewEncoder(c).Encode(res); err != nil {
+				panicf("failed to encode response: %v", err)
+			}
+			_ = c.Close()
+		}
+	}()
+
+	c, err := net.Dial("tcp", l.Addr().String())
+	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 panicf(format string, a ...interface{}) {
+	panic(fmt.Sprintf(format, a...))
+}