Function TestRPCComponents

rpc/pool_test.go:119–224

(t *testing.T)

}

func TestRPCComponents(t *testing.T) {
	Convey("Setup a single server for pool test", t, func(c C) {
		nodes, stop, err := setupEnvironment(1, AcceptNAConn)
		So(err, ShouldBeNil)
		defer stop()

		target := nodes[0].ID
		pool := &ClientPool{}
		defer func() { _ = pool.Close() }()

		So(pool.Len(), ShouldEqual, 0)
		cli1, err := pool.GetEx(target, true)
		So(err, ShouldBeNil)
		_ = cli1.Close()
		So(pool.Len(), ShouldEqual, 0)

		cli2, err := pool.GetEx(target, false)
		So(err, ShouldBeNil)
		_ = cli2.Close()
		So(pool.Len(), ShouldEqual, 1)

		So(pool.Len(), ShouldEqual, 1)
		pool.Remove(target)
		So(pool.Len(), ShouldEqual, 0)

		Convey("Test pool functionality", func() {
			var resp AddResp
			// Test broken pipe
			cli3, err := pool.GetEx(target, false)
			So(err, ShouldBeNil)
			err = cli3.Call("Count.Add", &AddReq{Delta: 1}, &resp)
			So(err, ShouldBeNil)
			_ = cli3.(*pooledClient).Client.Close() // close underlying client, should produce an error upon any following Call
			err = cli3.Call("Count.Add", &AddReq{Delta: 1}, &resp)
			So(err, ShouldNotBeNil)
			_ = cli3.Close() // should not return a broken client to pool
			So(pool.Len(), ShouldEqual, 0)
		})

		Convey("Test caller functionality", func() {
			pool := &ClientPool{}
			defer func() { _ = pool.Close() }()
			caller := NewPersistentCallerWithPool(pool, target)
			defer caller.Close()

			// Set a bad client with raw tcp
			rawconn, err := net.Dial("tcp", nodes[0].Addr)
			So(err, ShouldBeNil)
			caller.client = NewClient(rawconn)
			err = caller.Call("Count.Add", &AddReq{Delta: 1}, &AddResp{}) // should try reconnect, and return err
			So(err, ShouldNotBeNil)
			err = caller.Call("Count.Add", &AddReq{Delta: 1}, &AddResp{}) // should be no err
			So(err, ShouldBeNil)
		})
	})

	Convey("Setup servers", t, func(c C) {
		nodes, stop, err := setupEnvironment(10, AcceptNAConn)