diff options
Diffstat (limited to '')
| -rw-r--r-- | src/net/rpc/jsonrpc/all_test.go | 353 | ||||
| -rw-r--r-- | src/net/rpc/jsonrpc/client.go | 124 | ||||
| -rw-r--r-- | src/net/rpc/jsonrpc/server.go | 134 |
3 files changed, 611 insertions, 0 deletions
diff --git a/src/net/rpc/jsonrpc/all_test.go b/src/net/rpc/jsonrpc/all_test.go new file mode 100644 index 0000000..667f839 --- /dev/null +++ b/src/net/rpc/jsonrpc/all_test.go @@ -0,0 +1,353 @@ +// 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 ( + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "net/rpc" + "reflect" + "strings" + "testing" +) + +type Args struct { + A, B int +} + +type Reply struct { + C int +} + +type Arith int + +type ArithAddResp struct { + Id interface{} `json:"id"` + Result Reply `json:"result"` + Error interface{} `json:"error"` +} + +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") +} + +type BuiltinTypes struct{} + +func (BuiltinTypes) Map(i int, reply *map[int]int) error { + (*reply)[i] = i + return nil +} + +func (BuiltinTypes) Slice(i int, reply *[]int) error { + *reply = append(*reply, i) + return nil +} + +func (BuiltinTypes) Array(i int, reply *[1]int) error { + (*reply)[0] = i + return nil +} + +func init() { + rpc.Register(new(Arith)) + rpc.Register(BuiltinTypes{}) +} + +func TestServerNoParams(t *testing.T) { + cli, srv := net.Pipe() + defer cli.Close() + go ServeConn(srv) + dec := json.NewDecoder(cli) + + fmt.Fprintf(cli, `{"method": "Arith.Add", "id": "123"}`) + var resp ArithAddResp + if err := dec.Decode(&resp); err != nil { + t.Fatalf("Decode after no params: %s", err) + } + if resp.Error == nil { + t.Fatalf("Expected error, got nil") + } +} + +func TestServerEmptyMessage(t *testing.T) { + cli, srv := net.Pipe() + defer cli.Close() + go ServeConn(srv) + dec := json.NewDecoder(cli) + + fmt.Fprintf(cli, "{}") + var resp ArithAddResp + 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 TestServer(t *testing.T) { + 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 ArithAddResp + 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(rune(i)) { + t.Fatalf("resp: bad id %q want %q", resp.Id.(string), string(rune(i))) + } + if resp.Result.C != 2*i+1 { + t.Fatalf("resp: bad result: %d+%d=%d", i, i+1, resp.Result.C) + } + } +} + +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: got %d expected %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: got %d expected %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: got %d expected %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: got %d expected %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 TestBuiltinTypes(t *testing.T) { + cli, srv := net.Pipe() + go ServeConn(srv) + + client := NewClient(cli) + defer client.Close() + + // Map + arg := 7 + replyMap := map[int]int{} + err := client.Call("BuiltinTypes.Map", arg, &replyMap) + if err != nil { + t.Errorf("Map: expected no error but got string %q", err.Error()) + } + if replyMap[arg] != arg { + t.Errorf("Map: expected %d got %d", arg, replyMap[arg]) + } + + // Slice + replySlice := []int{} + err = client.Call("BuiltinTypes.Slice", arg, &replySlice) + if err != nil { + t.Errorf("Slice: expected no error but got string %q", err.Error()) + } + if e := []int{arg}; !reflect.DeepEqual(replySlice, e) { + t.Errorf("Slice: expected %v got %v", e, replySlice) + } + + // Array + replyArray := [1]int{} + err = client.Call("BuiltinTypes.Array", arg, &replyArray) + if err != nil { + t.Errorf("Array: expected no error but got string %q", err.Error()) + } + if e := [1]int{arg}; !reflect.DeepEqual(replyArray, e) { + t.Errorf("Array: expected %v got %v", e, replyArray) + } +} + +func TestMalformedInput(t *testing.T) { + cli, srv := net.Pipe() + go cli.Write([]byte(`{id:1}`)) // invalid json + ServeConn(srv) // must return, not loop +} + +func TestMalformedOutput(t *testing.T) { + cli, srv := net.Pipe() + go srv.Write([]byte(`{"id":0,"result":null,"error":null}`)) + go io.ReadAll(srv) + + client := NewClient(cli) + defer client.Close() + + args := &Args{7, 8} + reply := new(Reply) + err := client.Call("Arith.Add", args, reply) + if err == nil { + t.Error("expected error") + } +} + +func TestServerErrorHasNullResult(t *testing.T) { + var out bytes.Buffer + sc := NewServerCodec(struct { + io.Reader + io.Writer + io.Closer + }{ + Reader: strings.NewReader(`{"method": "Arith.Add", "id": "123", "params": []}`), + Writer: &out, + Closer: io.NopCloser(nil), + }) + r := new(rpc.Request) + if err := sc.ReadRequestHeader(r); err != nil { + t.Fatal(err) + } + const valueText = "the value we don't want to see" + const errorText = "some error" + err := sc.WriteResponse(&rpc.Response{ + ServiceMethod: "Method", + Seq: 1, + Error: errorText, + }, valueText) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(out.String(), errorText) { + t.Fatalf("Response didn't contain expected error %q: %s", errorText, &out) + } + if strings.Contains(out.String(), valueText) { + t.Errorf("Response contains both an error and value: %s", &out) + } +} + +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/src/net/rpc/jsonrpc/client.go b/src/net/rpc/jsonrpc/client.go new file mode 100644 index 0000000..e6359be --- /dev/null +++ b/src/net/rpc/jsonrpc/client.go @@ -0,0 +1,124 @@ +// 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 1.0 ClientCodec and ServerCodec +// for the rpc package. +// For JSON-RPC 2.0 support, see https://godoc.org/?q=json-rpc+2.0 +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 || c.resp.Result == 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/src/net/rpc/jsonrpc/server.go b/src/net/rpc/jsonrpc/server.go new file mode 100644 index 0000000..40e4e6f --- /dev/null +++ b/src/net/rpc/jsonrpc/server.go @@ -0,0 +1,134 @@ +// 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" +) + +var errMissingParams = errors.New("jsonrpc: request body missing params") + +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 + + // 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 = "" + r.Params = nil + r.Id = nil +} + +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 + } + if c.req.Params == nil { + return errMissingParams + } + // 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, ¶ms) +} + +var null = json.RawMessage([]byte("null")) + +func (c *serverCodec) WriteResponse(r *rpc.Response, x interface{}) error { + 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 := serverResponse{Id: b} + if r.Error == "" { + resp.Result = x + } 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)) +} |