~rogpeppe/juju-core/438-local-instance-Addresses

// Copyright 2012, 2013 Canonical Ltd.

// Licensed under the AGPLv3, see LICENCE file for details.

package rpc_test

import (

	"encoding/json"

	"fmt"

	"net"

	"reflect"

	"sync"

	stdtesting "testing"

	"time"

	gc "launchpad.net/gocheck"

	"launchpad.net/juju-core/log"

	"launchpad.net/juju-core/rpc"

	"launchpad.net/juju-core/rpc/jsoncodec"

	"launchpad.net/juju-core/testing/testbase"

)

type rpcSuite struct {

	testbase.LoggingSuite

}

var _ = gc.Suite(&rpcSuite{})

func TestAll(t *stdtesting.T) {

	gc.TestingT(t)

}

type callInfo struct {

	rcvr   interface{}

	method string

	arg    interface{}

}

type callError callInfo

func (e *callError) Error() string {

	return fmt.Sprintf("error calling %s", e.method)

}

type stringVal struct {

	Val string

}

type Root struct {

	mu        sync.Mutex

	conn      *rpc.Conn

	calls     []*callInfo

	returnErr bool

	simple    map[string]*SimpleMethods

	delayed   map[string]*DelayedMethods

	errorInst *ErrorMethods

}

func (r *Root) callError(rcvr interface{}, name string, arg interface{}) error {

	if r.returnErr {

		return &callError{rcvr, name, arg}

	}

	return nil

}

func (r *Root) SimpleMethods(id string) (*SimpleMethods, error) {

	r.mu.Lock()

	defer r.mu.Unlock()

	if a := r.simple[id]; a != nil {

		return a, nil

	}

	return nil, fmt.Errorf("unknown SimpleMethods id")

}

func (r *Root) DelayedMethods(id string) (*DelayedMethods, error) {

	r.mu.Lock()

	defer r.mu.Unlock()

	if a := r.delayed[id]; a != nil {

		return a, nil

	}

	return nil, fmt.Errorf("unknown DelayedMethods id")

}

func (r *Root) ErrorMethods(id string) (*ErrorMethods, error) {

	if r.errorInst == nil {

		return nil, fmt.Errorf("no error methods")

	}

	return r.errorInst, nil

}

func (r *Root) Discard1() {}

func (r *Root) Discard2(id string) error { return nil }

func (r *Root) Discard3(id string) int { return 0 }

func (r *Root) CallbackMethods(string) (*CallbackMethods, error) {

	return &CallbackMethods{r}, nil

}

func (r *Root) InterfaceMethods(id string) (InterfaceMethods, error) {

	log.Infof("interface methods called")

	m, err := r.SimpleMethods(id)

	if err != nil {

		return nil, err

	}

	return m, nil

}

type InterfaceMethods interface {

	Call1r1e(s stringVal) (stringVal, error)

}

type ChangeAPIMethods struct {

	r *Root

}

func (r *Root) ChangeAPIMethods(string) (*ChangeAPIMethods, error) {

	return &ChangeAPIMethods{r}, nil

}

func (t *Root) called(rcvr interface{}, method string, arg interface{}) {

	t.mu.Lock()

	t.calls = append(t.calls, &callInfo{rcvr, method, arg})

	t.mu.Unlock()

}

type SimpleMethods struct {

	root *Root

	id   string

}

// Each Call method is named in this standard form:

//

//     Call<narg>r<nret><e>

//

// where narg is the number of arguments, nret is the number of returned

// values (not including the error) and e is the letter 'e' if the

// method returns an error.

func (a *SimpleMethods) Call0r0() {

	a.root.called(a, "Call0r0", nil)

}

func (a *SimpleMethods) Call0r1() stringVal {

	a.root.called(a, "Call0r1", nil)

	return stringVal{"Call0r1 ret"}

}

func (a *SimpleMethods) Call0r1e() (stringVal, error) {

	a.root.called(a, "Call0r1e", nil)

	return stringVal{"Call0r1e ret"}, a.root.callError(a, "Call0r1e", nil)

}

func (a *SimpleMethods) Call0r0e() error {

	a.root.called(a, "Call0r0e", nil)

	return a.root.callError(a, "Call0r0e", nil)

}

func (a *SimpleMethods) Call1r0(s stringVal) {

	a.root.called(a, "Call1r0", s)

}

func (a *SimpleMethods) Call1r1(s stringVal) stringVal {

	a.root.called(a, "Call1r1", s)

	return stringVal{"Call1r1 ret"}

}

func (a *SimpleMethods) Call1r1e(s stringVal) (stringVal, error) {

	a.root.called(a, "Call1r1e", s)

	return stringVal{"Call1r1e ret"}, a.root.callError(a, "Call1r1e", s)

}

func (a *SimpleMethods) Call1r0e(s stringVal) error {

	a.root.called(a, "Call1r0e", s)

	return a.root.callError(a, "Call1r0e", s)

}

func (a *SimpleMethods) SliceArg(struct{ X []string }) stringVal {

	return stringVal{"SliceArg ret"}

}

func (a *SimpleMethods) Discard1(int) {}

func (a *SimpleMethods) Discard2(struct{}, struct{}) {}

func (a *SimpleMethods) Discard3() int { return 0 }

func (a *SimpleMethods) Discard4() (_, _ struct{}) { return }

type DelayedMethods struct {

	ready     chan struct{}

	done      chan string

	doneError chan error

}

func (a *DelayedMethods) Delay() (stringVal, error) {

	if a.ready != nil {

		a.ready <- struct{}{}

	}

	select {

	case s := <-a.done:

		return stringVal{s}, nil

	case err := <-a.doneError:

		return stringVal{}, err

	}

}

type ErrorMethods struct {

	err error

}

func (e *ErrorMethods) Call() error {

	return e.err

}

type CallbackMethods struct {

	root *Root

}

type int64val struct {

	I int64

}

func (a *CallbackMethods) Factorial(x int64val) (int64val, error) {

	if x.I <= 1 {

		return int64val{1}, nil

	}

	var r int64val

	err := a.root.conn.Call("CallbackMethods", "", "Factorial", int64val{x.I - 1}, &r)

	if err != nil {

		return int64val{}, err

	}

	return int64val{x.I * r.I}, nil

}

func (a *ChangeAPIMethods) ChangeAPI() {

	a.r.conn.Serve(&changedAPIRoot{}, nil)

}

func (a *ChangeAPIMethods) RemoveAPI() {

	a.r.conn.Serve(nil, nil)

}

type changedAPIRoot struct{}

func (r *changedAPIRoot) NewlyAvailable(string) (newlyAvailableMethods, error) {

	return newlyAvailableMethods{}, nil

}

type newlyAvailableMethods struct{}

func (newlyAvailableMethods) NewMethod() stringVal {

	return stringVal{"new method result"}

}

func (*rpcSuite) TestRPC(c *gc.C) {

	root := &Root{

		simple: make(map[string]*SimpleMethods),

	}

	root.simple["a99"] = &SimpleMethods{root: root, id: "a99"}

	client, srvDone := newRPCClientServer(c, root, nil, false)

	defer closeClient(c, client, srvDone)

	for narg := 0; narg < 2; narg++ {

		for nret := 0; nret < 2; nret++ {

			for nerr := 0; nerr < 2; nerr++ {

				retErr := nerr != 0

				root.testCall(c, client, "SimpleMethods", narg, nret, retErr, false)

				if retErr {

					root.testCall(c, client, "SimpleMethods", narg, nret, retErr, true)

				}

			}

		}

	}

}

func callName(narg, nret int, retErr bool) string {

	e := ""

	if retErr {

		e = "e"

	}

	return fmt.Sprintf("Call%dr%d%s", narg, nret, e)

}

func (root *Root) testCall(c *gc.C, conn *rpc.Conn, entry string, narg, nret int, retErr, testErr bool) {

	root.calls = nil

	root.returnErr = testErr

	method := callName(narg, nret, retErr)

	c.Logf("test call %s", method)

	var r stringVal

	err := conn.Call(entry, "a99", method, stringVal{"arg"}, &r)

	root.mu.Lock()

	defer root.mu.Unlock()

	expectCall := callInfo{

		rcvr:   root.simple["a99"],

		method: method,

	}

	if narg > 0 {

		expectCall.arg = stringVal{"arg"}

	}

	c.Assert(root.calls, gc.HasLen, 1)

	c.Assert(*root.calls[0], gc.Equals, expectCall)

	switch {

	case retErr && testErr:

		c.Assert(err, gc.DeepEquals, &rpc.RequestError{

			Message: fmt.Sprintf("error calling %s", method),

		})

		c.Assert(r, gc.Equals, stringVal{})

	case nret > 0:

		c.Assert(r, gc.Equals, stringVal{method + " ret"})

	}

}

func (*rpcSuite) TestInterfaceMethods(c *gc.C) {

	root := &Root{

		simple: make(map[string]*SimpleMethods),

	}

	root.simple["a99"] = &SimpleMethods{root: root, id: "a99"}

	client, srvDone := newRPCClientServer(c, root, nil, false)

	defer closeClient(c, client, srvDone)

	root.testCall(c, client, "InterfaceMethods", 1, 1, true, false)

	root.testCall(c, client, "InterfaceMethods", 1, 1, true, true)

}

func (*rpcSuite) TestConcurrentCalls(c *gc.C) {

	start1 := make(chan string)

	start2 := make(chan string)

	ready1 := make(chan struct{})

	ready2 := make(chan struct{})

	root := &Root{

		delayed: map[string]*DelayedMethods{

			"1": {ready: ready1, done: start1},

			"2": {ready: ready2, done: start2},

		},

	}

	client, srvDone := newRPCClientServer(c, root, nil, false)

	defer closeClient(c, client, srvDone)

	call := func(id string, done chan<- struct{}) {

		var r stringVal

		err := client.Call("DelayedMethods", id, "Delay", nil, &r)

		c.Check(err, gc.IsNil)

		c.Check(r.Val, gc.Equals, "return "+id)

		done <- struct{}{}

	}

	done1 := make(chan struct{})

	done2 := make(chan struct{})

	go call("1", done1)

	go call("2", done2)

	// Check that both calls are running concurrently.

	chanRead(c, ready1, "method 1 ready")

	chanRead(c, ready2, "method 2 ready")

	// Let the requests complete.

	start1 <- "return 1"

	start2 <- "return 2"

	chanRead(c, done1, "method 1 done")

	chanRead(c, done2, "method 2 done")

}

type codedError struct {

	m    string

	code string

}

func (e *codedError) Error() string {

	return e.m

}

func (e *codedError) ErrorCode() string {

	return e.code

}

func (*rpcSuite) TestErrorCode(c *gc.C) {

	root := &Root{

		errorInst: &ErrorMethods{&codedError{"message", "code"}},

	}

	client, srvDone := newRPCClientServer(c, root, nil, false)

	defer closeClient(c, client, srvDone)

	err := client.Call("ErrorMethods", "", "Call", nil, nil)

	c.Assert(err, gc.ErrorMatches, `request error: message \(code\)`)

	c.Assert(err.(rpc.ErrorCoder).ErrorCode(), gc.Equals, "code")

}

func (*rpcSuite) TestTransformErrors(c *gc.C) {

	root := &Root{

		errorInst: &ErrorMethods{&codedError{"message", "code"}},

	}

	tfErr := func(err error) error {

		c.Check(err, gc.NotNil)

		if e, ok := err.(*codedError); ok {

			return &codedError{

				m:    "transformed: " + e.m,

				code: "transformed: " + e.code,

			}

		}

		return fmt.Errorf("transformed: %v", err)

	}

	client, srvDone := newRPCClientServer(c, root, tfErr, false)

	defer closeClient(c, client, srvDone)

	err := client.Call("ErrorMethods", "", "Call", nil, nil)

	c.Assert(err, gc.DeepEquals, &rpc.RequestError{

		Message: "transformed: message",

		Code:    "transformed: code",

	})

	root.errorInst.err = nil

	err = client.Call("ErrorMethods", "", "Call", nil, nil)

	c.Assert(err, gc.IsNil)

	root.errorInst = nil

	err = client.Call("ErrorMethods", "", "Call", nil, nil)

	c.Assert(err, gc.DeepEquals, &rpc.RequestError{

		Message: "transformed: no error methods",

	})

}

func (*rpcSuite) TestServerWaitsForOutstandingCalls(c *gc.C) {

	ready := make(chan struct{})

	start := make(chan string)

	root := &Root{

		delayed: map[string]*DelayedMethods{

			"1": {

				ready: ready,

				done:  start,

			},

		},

	}

	client, srvDone := newRPCClientServer(c, root, nil, false)

	defer closeClient(c, client, srvDone)

	done := make(chan struct{})

	go func() {

		var r stringVal

		err := client.Call("DelayedMethods", "1", "Delay", nil, &r)

		c.Check(err, gc.Equals, rpc.ErrShutdown)

		done <- struct{}{}

	}()

	chanRead(c, ready, "DelayedMethods.Delay ready")

	client.Close()

	select {

	case err := <-srvDone:

		c.Fatalf("server returned while outstanding operation in progress: %v", err)

		<-done

	case <-time.After(25 * time.Millisecond):

	}

	start <- "xxx"

}

func chanRead(c *gc.C, ch <-chan struct{}, what string) {

	select {

	case <-ch:

		return

	case <-time.After(3 * time.Second):

		c.Fatalf("timeout on channel read %s", what)

	}

}

func (*rpcSuite) TestCompatibility(c *gc.C) {

	root := &Root{

		simple: make(map[string]*SimpleMethods),

	}

	a0 := &SimpleMethods{root: root, id: "a0"}

	root.simple["a0"] = a0

	client, srvDone := newRPCClientServer(c, root, nil, false)

	defer closeClient(c, client, srvDone)

	call := func(method string, arg, ret interface{}) (passedArg interface{}) {

		root.calls = nil

		err := client.Call("SimpleMethods", "a0", method, arg, ret)

		c.Assert(err, gc.IsNil)

		c.Assert(root.calls, gc.HasLen, 1)

		info := root.calls[0]

		c.Assert(info.rcvr, gc.Equals, a0)

		c.Assert(info.method, gc.Equals, method)

		return info.arg

	}

	type extra struct {

		Val   string

		Extra string

	}

	// Extra fields in request and response.

	var r extra

	arg := call("Call1r1", extra{"x", "y"}, &r)

	c.Assert(arg, gc.Equals, stringVal{"x"})

	// Nil argument as request.

	r = extra{}

	arg = call("Call1r1", nil, &r)

	c.Assert(arg, gc.Equals, stringVal{})

	// Nil argument as response.

	arg = call("Call1r1", stringVal{"x"}, nil)

	c.Assert(arg, gc.Equals, stringVal{"x"})

	// Non-nil argument for no response.

	r = extra{}

	arg = call("Call1r0", stringVal{"x"}, &r)

	c.Assert(arg, gc.Equals, stringVal{"x"})

	c.Assert(r, gc.Equals, extra{})

}

func (*rpcSuite) TestBadCall(c *gc.C) {

	root := &Root{

		simple: make(map[string]*SimpleMethods),

	}

	a0 := &SimpleMethods{root: root, id: "a0"}

	root.simple["a0"] = a0

	client, srvDone := newRPCClientServer(c, root, nil, false)

	defer closeClient(c, client, srvDone)

	err := client.Call("BadSomething", "a0", "No", nil, nil)

	c.Assert(err, gc.ErrorMatches, `request error: unknown object type "BadSomething"`)

	err = client.Call("SimpleMethods", "xx", "No", nil, nil)

	c.Assert(err, gc.ErrorMatches, `request error: no such request "No" on SimpleMethods`)

	err = client.Call("SimpleMethods", "xx", "Call0r0", nil, nil)

	c.Assert(err, gc.ErrorMatches, "request error: unknown SimpleMethods id")

}

func (*rpcSuite) TestContinueAfterReadBodyError(c *gc.C) {

	root := &Root{

		simple: make(map[string]*SimpleMethods),

	}

	a0 := &SimpleMethods{root: root, id: "a0"}

	root.simple["a0"] = a0

	client, srvDone := newRPCClientServer(c, root, nil, false)

	defer closeClient(c, client, srvDone)

	var ret stringVal

	arg0 := struct {

		X map[string]int

	}{

		X: map[string]int{"hello": 65},

	}

	err := client.Call("SimpleMethods", "a0", "SliceArg", arg0, &ret)

	c.Assert(err, gc.ErrorMatches, `request error: json: cannot unmarshal object into Go value of type \[\]string`)

	err = client.Call("SimpleMethods", "a0", "SliceArg", arg0, &ret)

	c.Assert(err, gc.ErrorMatches, `request error: json: cannot unmarshal object into Go value of type \[\]string`)

	arg1 := struct {

		X []string

	}{

		X: []string{"one"},

	}

	err = client.Call("SimpleMethods", "a0", "SliceArg", arg1, &ret)

	c.Assert(err, gc.IsNil)

	c.Assert(ret.Val, gc.Equals, "SliceArg ret")

}

func (*rpcSuite) TestErrorAfterClientClose(c *gc.C) {

	client, srvDone := newRPCClientServer(c, &Root{}, nil, false)

	err := client.Close()

	c.Assert(err, gc.IsNil)

	err = client.Call("Foo", "", "Bar", nil, nil)

	c.Assert(err, gc.Equals, rpc.ErrShutdown)

	err = chanReadError(c, srvDone, "server done")

	c.Assert(err, gc.IsNil)