Function TestNewSessionPool

rpc/mux/pool_test.go:106–191

(t *testing.T)

}

func TestNewSessionPool(t *testing.T) {
	Convey("session pool", t, func(c C) {
		log.SetLevel(log.FatalLevel)

		// setup pool
		p := &SessionPool{
			sessions: make(map[proto.NodeID]*Session),
		}
		defer withTCPDialer()()

		// setup server
		l, err := net.Listen("tcp", ":0")
		if err != nil {
			log.Fatal(err)
		}
		server, err := NewServerWithService(ServiceMap{"Test": NewTestService()})
		server.SetListener(l)
		go server.WithAcceptConnFunc(rpc.AcceptRawConn).Serve()

		wg := &sync.WaitGroup{}
		wg.Add(concurrency)
		for i := 0; i < concurrency; i++ {
			go func(c2 C, n int) {
				defer wg.Done()
				client, err := p.Get(proto.NodeID(l.Addr().String()))
				defer func() { _ = client.Close() }()
				if err != nil {
					log.Errorf("get session failed: %s", err)
					return
				}
				c2.So(err, ShouldBeNil)
				err = client.Call("Test.IncCounter", &TestReq{Step: 1}, &TestRep{})
				c2.So(err, ShouldBeNil)
			}(c, packetCount)
		}

		wg.Wait()
		So(p.Len(), ShouldEqual, conf.MaxRPCMuxPoolPhysicalConnection)

		// setup server
		l2, err := net.Listen("tcp", ":0")
		if err != nil {
			log.Fatal(err)
		}
		server2, err := NewServerWithService(ServiceMap{"Test": NewTestService()})
		server2.SetListener(l2)
		go server2.WithAcceptConnFunc(rpc.AcceptRawConn).Serve()

		_, err = p.Get(proto.NodeID(l2.Addr().String()))
		So(err, ShouldBeNil)
		So(p.Len(), ShouldEqual, conf.MaxRPCMuxPoolPhysicalConnection+1)

		wg2 := &sync.WaitGroup{}
		wg2.Add(concurrency)
		for i := 0; i < concurrency; i++ {
			go func(c2 C, n int) {
				// Open a new stream
				// Stream implements net.Conn