libgo: Update to weekly 2011-11-09.

From-SVN: r182073

3 files changed, 480 insertions, 0 deletions

diff --git a/libgo/go/net/rpc/jsonrpc/all_test.go b/libgo/go/net/rpc/jsonrpc/all_test.go
new file mode 100644
index 0000000..e6c7441
--- /dev/null
+++ b/libgo/go/net/rpc/jsonrpc/all_test.go
@@ -0,0 +1,221 @@
+// Copyright 2010 The Go Authors.  All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package jsonrpc
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/rpc"
+	"testing"
+)
+
+type Args struct {
+	A, B int
+}
+
+type Reply struct {
+	C int
+}
+
+type Arith int
+
+func (t *Arith) Add(args *Args, reply *Reply) error {
+	reply.C = args.A + args.B
+	return nil
+}
+
+func (t *Arith) Mul(args *Args, reply *Reply) error {
+	reply.C = args.A * args.B
+	return nil
+}
+
+func (t *Arith) Div(args *Args, reply *Reply) error {
+	if args.B == 0 {
+		return errors.New("divide by zero")
+	}
+	reply.C = args.A / args.B
+	return nil
+}
+
+func (t *Arith) Error(args *Args, reply *Reply) error {
+	panic("ERROR")
+}
+
+func init() {
+	rpc.Register(new(Arith))
+}
+
+func TestServer(t *testing.T) {
+	type addResp struct {
+		Id     interface{} `json:"id"`
+		Result Reply       `json:"result"`
+		Error  interface{} `json:"error"`
+	}
+
+	cli, srv := net.Pipe()
+	defer cli.Close()
+	go ServeConn(srv)
+	dec := json.NewDecoder(cli)
+
+	// Send hand-coded requests to server, parse responses.
+	for i := 0; i < 10; i++ {
+		fmt.Fprintf(cli, `{"method": "Arith.Add", "id": "\u%04d", "params": [{"A": %d, "B": %d}]}`, i, i, i+1)
+		var resp addResp
+		err := dec.Decode(&resp)
+		if err != nil {
+			t.Fatalf("Decode: %s", err)
+		}
+		if resp.Error != nil {
+			t.Fatalf("resp.Error: %s", resp.Error)
+		}
+		if resp.Id.(string) != string(i) {
+			t.Fatalf("resp: bad id %q want %q", resp.Id.(string), string(i))
+		}
+		if resp.Result.C != 2*i+1 {
+			t.Fatalf("resp: bad result: %d+%d=%d", i, i+1, resp.Result.C)
+		}
+	}
+
+	fmt.Fprintf(cli, "{}\n")
+	var resp addResp
+	if err := dec.Decode(&resp); err != nil {
+		t.Fatalf("Decode after empty: %s", err)
+	}
+	if resp.Error == nil {
+		t.Fatalf("Expected error, got nil")
+	}
+}
+
+func TestClient(t *testing.T) {
+	// Assume server is okay (TestServer is above).
+	// Test client against server.
+	cli, srv := net.Pipe()
+	go ServeConn(srv)
+
+	client := NewClient(cli)
+	defer client.Close()
+
+	// Synchronous calls
+	args := &Args{7, 8}
+	reply := new(Reply)
+	err := client.Call("Arith.Add", args, reply)
+	if err != nil {
+		t.Errorf("Add: expected no error but got string %q", err.Error())
+	}
+	if reply.C != args.A+args.B {
+		t.Errorf("Add: expected %d got %d", reply.C, args.A+args.B)
+	}
+
+	args = &Args{7, 8}
+	reply = new(Reply)
+	err = client.Call("Arith.Mul", args, reply)
+	if err != nil {
+		t.Errorf("Mul: expected no error but got string %q", err.Error())
+	}
+	if reply.C != args.A*args.B {
+		t.Errorf("Mul: expected %d got %d", reply.C, args.A*args.B)
+	}
+
+	// Out of order.
+	args = &Args{7, 8}
+	mulReply := new(Reply)
+	mulCall := client.Go("Arith.Mul", args, mulReply, nil)
+	addReply := new(Reply)
+	addCall := client.Go("Arith.Add", args, addReply, nil)
+
+	addCall = <-addCall.Done
+	if addCall.Error != nil {
+		t.Errorf("Add: expected no error but got string %q", addCall.Error.Error())
+	}
+	if addReply.C != args.A+args.B {
+		t.Errorf("Add: expected %d got %d", addReply.C, args.A+args.B)
+	}
+
+	mulCall = <-mulCall.Done
+	if mulCall.Error != nil {
+		t.Errorf("Mul: expected no error but got string %q", mulCall.Error.Error())
+	}
+	if mulReply.C != args.A*args.B {
+		t.Errorf("Mul: expected %d got %d", mulReply.C, args.A*args.B)
+	}
+
+	// Error test
+	args = &Args{7, 0}
+	reply = new(Reply)
+	err = client.Call("Arith.Div", args, reply)
+	// expect an error: zero divide
+	if err == nil {
+		t.Error("Div: expected error")
+	} else if err.Error() != "divide by zero" {
+		t.Error("Div: expected divide by zero error; got", err)
+	}
+}
+
+func TestMalformedInput(t *testing.T) {
+	cli, srv := net.Pipe()
+	go cli.Write([]byte(`{id:1}`)) // invalid json
+	ServeConn(srv)                 // must return, not loop
+}
+
+func TestUnexpectedError(t *testing.T) {
+	cli, srv := myPipe()
+	go cli.PipeWriter.CloseWithError(errors.New("unexpected error!")) // reader will get this error
+	ServeConn(srv)                                                    // must return, not loop
+}
+
+// Copied from package net.
+func myPipe() (*pipe, *pipe) {
+	r1, w1 := io.Pipe()
+	r2, w2 := io.Pipe()
+
+	return &pipe{r1, w2}, &pipe{r2, w1}
+}
+
+type pipe struct {
+	*io.PipeReader
+	*io.PipeWriter
+}
+
+type pipeAddr int
+
+func (pipeAddr) Network() string {
+	return "pipe"
+}
+
+func (pipeAddr) String() string {
+	return "pipe"
+}
+
+func (p *pipe) Close() error {
+	err := p.PipeReader.Close()
+	err1 := p.PipeWriter.Close()
+	if err == nil {
+		err = err1
+	}
+	return err
+}
+
+func (p *pipe) LocalAddr() net.Addr {
+	return pipeAddr(0)
+}
+
+func (p *pipe) RemoteAddr() net.Addr {
+	return pipeAddr(0)
+}
+
+func (p *pipe) SetTimeout(nsec int64) error {
+	return errors.New("net.Pipe does not support timeouts")
+}
+
+func (p *pipe) SetReadTimeout(nsec int64) error {
+	return errors.New("net.Pipe does not support timeouts")
+}
+
+func (p *pipe) SetWriteTimeout(nsec int64) error {
+	return errors.New("net.Pipe does not support timeouts")
+}
diff --git a/libgo/go/net/rpc/jsonrpc/client.go b/libgo/go/net/rpc/jsonrpc/client.go
new file mode 100644
index 0000000..3fa8cbf
--- /dev/null
+++ b/libgo/go/net/rpc/jsonrpc/client.go
@@ -0,0 +1,123 @@
+// Copyright 2010 The Go Authors.  All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// Package jsonrpc implements a JSON-RPC ClientCodec and ServerCodec
+// for the rpc package.
+package jsonrpc
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"net/rpc"
+	"sync"
+)
+
+type clientCodec struct {
+	dec *json.Decoder // for reading JSON values
+	enc *json.Encoder // for writing JSON values
+	c   io.Closer
+
+	// temporary work space
+	req  clientRequest
+	resp clientResponse
+
+	// JSON-RPC responses include the request id but not the request method.
+	// Package rpc expects both.
+	// We save the request method in pending when sending a request
+	// and then look it up by request ID when filling out the rpc Response.
+	mutex   sync.Mutex        // protects pending
+	pending map[uint64]string // map request id to method name
+}
+
+// NewClientCodec returns a new rpc.ClientCodec using JSON-RPC on conn.
+func NewClientCodec(conn io.ReadWriteCloser) rpc.ClientCodec {
+	return &clientCodec{
+		dec:     json.NewDecoder(conn),
+		enc:     json.NewEncoder(conn),
+		c:       conn,
+		pending: make(map[uint64]string),
+	}
+}
+
+type clientRequest struct {
+	Method string         `json:"method"`
+	Params [1]interface{} `json:"params"`
+	Id     uint64         `json:"id"`
+}
+
+func (c *clientCodec) WriteRequest(r *rpc.Request, param interface{}) error {
+	c.mutex.Lock()
+	c.pending[r.Seq] = r.ServiceMethod
+	c.mutex.Unlock()
+	c.req.Method = r.ServiceMethod
+	c.req.Params[0] = param
+	c.req.Id = r.Seq
+	return c.enc.Encode(&c.req)
+}
+
+type clientResponse struct {
+	Id     uint64           `json:"id"`
+	Result *json.RawMessage `json:"result"`
+	Error  interface{}      `json:"error"`
+}
+
+func (r *clientResponse) reset() {
+	r.Id = 0
+	r.Result = nil
+	r.Error = nil
+}
+
+func (c *clientCodec) ReadResponseHeader(r *rpc.Response) error {
+	c.resp.reset()
+	if err := c.dec.Decode(&c.resp); err != nil {
+		return err
+	}
+
+	c.mutex.Lock()
+	r.ServiceMethod = c.pending[c.resp.Id]
+	delete(c.pending, c.resp.Id)
+	c.mutex.Unlock()
+
+	r.Error = ""
+	r.Seq = c.resp.Id
+	if c.resp.Error != nil {
+		x, ok := c.resp.Error.(string)
+		if !ok {
+			return fmt.Errorf("invalid error %v", c.resp.Error)
+		}
+		if x == "" {
+			x = "unspecified error"
+		}
+		r.Error = x
+	}
+	return nil
+}
+
+func (c *clientCodec) ReadResponseBody(x interface{}) error {
+	if x == nil {
+		return nil
+	}
+	return json.Unmarshal(*c.resp.Result, x)
+}
+
+func (c *clientCodec) Close() error {
+	return c.c.Close()
+}
+
+// NewClient returns a new rpc.Client to handle requests to the
+// set of services at the other end of the connection.
+func NewClient(conn io.ReadWriteCloser) *rpc.Client {
+	return rpc.NewClientWithCodec(NewClientCodec(conn))
+}
+
+// Dial connects to a JSON-RPC server at the specified network address.
+func Dial(network, address string) (*rpc.Client, error) {
+	conn, err := net.Dial(network, address)
+	if err != nil {
+		return nil, err
+	}
+	return NewClient(conn), err
+}
diff --git a/libgo/go/net/rpc/jsonrpc/server.go b/libgo/go/net/rpc/jsonrpc/server.go
new file mode 100644
index 0000000..4c54553
--- /dev/null
+++ b/libgo/go/net/rpc/jsonrpc/server.go
@@ -0,0 +1,136 @@
+// Copyright 2010 The Go Authors.  All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package jsonrpc
+
+import (
+	"encoding/json"
+	"errors"
+	"io"
+	"net/rpc"
+	"sync"
+)
+
+type serverCodec struct {
+	dec *json.Decoder // for reading JSON values
+	enc *json.Encoder // for writing JSON values
+	c   io.Closer
+
+	// temporary work space
+	req  serverRequest
+	resp serverResponse
+
+	// JSON-RPC clients can use arbitrary json values as request IDs.
+	// Package rpc expects uint64 request IDs.
+	// We assign uint64 sequence numbers to incoming requests
+	// but save the original request ID in the pending map.
+	// When rpc responds, we use the sequence number in
+	// the response to find the original request ID.
+	mutex   sync.Mutex // protects seq, pending
+	seq     uint64
+	pending map[uint64]*json.RawMessage
+}
+
+// NewServerCodec returns a new rpc.ServerCodec using JSON-RPC on conn.
+func NewServerCodec(conn io.ReadWriteCloser) rpc.ServerCodec {
+	return &serverCodec{
+		dec:     json.NewDecoder(conn),
+		enc:     json.NewEncoder(conn),
+		c:       conn,
+		pending: make(map[uint64]*json.RawMessage),
+	}
+}
+
+type serverRequest struct {
+	Method string           `json:"method"`
+	Params *json.RawMessage `json:"params"`
+	Id     *json.RawMessage `json:"id"`
+}
+
+func (r *serverRequest) reset() {
+	r.Method = ""
+	if r.Params != nil {
+		*r.Params = (*r.Params)[0:0]
+	}
+	if r.Id != nil {
+		*r.Id = (*r.Id)[0:0]
+	}
+}
+
+type serverResponse struct {
+	Id     *json.RawMessage `json:"id"`
+	Result interface{}      `json:"result"`
+	Error  interface{}      `json:"error"`
+}
+
+func (c *serverCodec) ReadRequestHeader(r *rpc.Request) error {
+	c.req.reset()
+	if err := c.dec.Decode(&c.req); err != nil {
+		return err
+	}
+	r.ServiceMethod = c.req.Method
+
+	// JSON request id can be any JSON value;
+	// RPC package expects uint64.  Translate to
+	// internal uint64 and save JSON on the side.
+	c.mutex.Lock()
+	c.seq++
+	c.pending[c.seq] = c.req.Id
+	c.req.Id = nil
+	r.Seq = c.seq
+	c.mutex.Unlock()
+
+	return nil
+}
+
+func (c *serverCodec) ReadRequestBody(x interface{}) error {
+	if x == nil {
+		return nil
+	}
+	// JSON params is array value.
+	// RPC params is struct.
+	// Unmarshal into array containing struct for now.
+	// Should think about making RPC more general.
+	var params [1]interface{}
+	params[0] = x
+	return json.Unmarshal(*c.req.Params, &params)
+}
+
+var null = json.RawMessage([]byte("null"))
+
+func (c *serverCodec) WriteResponse(r *rpc.Response, x interface{}) error {
+	var resp serverResponse
+	c.mutex.Lock()
+	b, ok := c.pending[r.Seq]
+	if !ok {
+		c.mutex.Unlock()
+		return errors.New("invalid sequence number in response")
+	}
+	delete(c.pending, r.Seq)
+	c.mutex.Unlock()
+
+	if b == nil {
+		// Invalid request so no id.  Use JSON null.
+		b = &null
+	}
+	resp.Id = b
+	resp.Result = x
+	if r.Error == "" {
+		resp.Error = nil
+	} else {
+		resp.Error = r.Error
+	}
+	return c.enc.Encode(resp)
+}
+
+func (c *serverCodec) Close() error {
+	return c.c.Close()
+}
+
+// ServeConn runs the JSON-RPC server on a single connection.
+// ServeConn blocks, serving the connection until the client hangs up.
+// The caller typically invokes ServeConn in a go statement.
+func ServeConn(conn io.ReadWriteCloser) {
+	rpc.ServeCodec(NewServerCodec(conn))
+}