fix para test

plugin/dapp/paracross/testnode/cfg.go

 package testnode
 
-var paraconfig = `
+//default config for testnode
+var DefaultConfig = `
 Title="user.p.guodun."
 # TestNet=true
 
@@ -71,6 +72,9 @@ poolCacheSize=10240
 minTxFee=100000
 maxTxNumPerAccount=10000
 
+[mempool.sub.para]
+poolCacheSize=102400
+
 [consensus]
 name="para"
 genesisBlockTime=1514533394

plugin/dapp/paracross/testnode/paranode.go

@@ -23,7 +23,7 @@ func NewParaNode(main *testnode.Chain33Mock, para *testnode.Chain33Mock) *ParaNo
 		main.Listen()
 	}
 	if para == nil {
-		cfg, sub := types.InitCfgString(paraconfig)
+		cfg, sub := types.InitCfgString(DefaultConfig)
 		testnode.ModifyParaClient(sub, main.GetCfg().RPC.GrpcBindAddr)
 		para = testnode.NewWithConfig(cfg, sub, nil)
 		para.Listen()

plugin/mempool/para/base.go

@@ -46,7 +46,6 @@ func NewMempool(cfg *types.Mempool) *Mempool {
 func (mem *Mempool) SetQueueClient(client queue.Client) {
 	mem.client = client
 	mem.client.Sub(mem.key)
-
 	mem.wg.Add(1)
 	go func() {
 		for msg := range client.Recv() {
@@ -69,8 +68,7 @@ func (mem *Mempool) SetQueueClient(client queue.Client) {
 					}
 				}
 			default:
-				msg.Reply(client.NewMessage(mem.key, types.EventReply, &types.Reply{IsOk: false,
-					Msg: []byte(fmt.Sprintf("para %v doesn't handle message %v", mem.key, msg.Ty))}))
+				msg.Reply(client.NewMessage(mem.key, types.EventReply, types.ErrActionNotSupport))
 			}
 		}
 	}()
@@ -81,10 +79,9 @@ func (mem *Mempool) Wait() {}
 
 // Close method
 func (mem *Mempool) Close() {
-	if mem.isClose() {
+	if !atomic.CompareAndSwapInt32(&mem.isclose, 0, 1) {
 		return
 	}
-	atomic.StoreInt32(&mem.isclose, 1)
 	if mem.client != nil {
 		mem.client.Close()
 	}
@@ -93,7 +90,3 @@ func (mem *Mempool) Close() {
 	mem.wg.Wait()
 	mlog.Info("para mempool module closed")
 }
-
-func (mem *Mempool) isClose() bool {
-	return atomic.LoadInt32(&mem.isclose) == 1
-}

plugin/mempool/para/para_test.go

+package para_test
+
+import (
+	"testing"
+
+	"github.com/33cn/chain33/types"
+	"github.com/33cn/chain33/util"
+	"github.com/33cn/chain33/util/testnode"
+	paratest "github.com/33cn/plugin/plugin/dapp/paracross/testnode"
+	"github.com/33cn/plugin/plugin/mempool/para"
+	"github.com/stretchr/testify/assert"
+
+	_ "github.com/33cn/chain33/system"
+	_ "github.com/33cn/plugin/plugin"
+)
+
+func TestClose(t *testing.T) {
+	mem := para.NewMempool(nil)
+	n := 1000
+	done := make(chan struct{}, n)
+	for i := 0; i < n; i++ {
+		go func() {
+			mem.Close()
+			done <- struct{}{}
+		}()
+	}
+	for i := 0; i < n; i++ {
+		<-done
+	}
+}
+
+func TestParaNodeMempool(t *testing.T) {
+	main := testnode.New("", nil)
+	main.Listen()
+
+	cfg, sub := types.InitCfgString(paratest.DefaultConfig)
+	testnode.ModifyParaClient(sub, main.GetCfg().RPC.GrpcBindAddr)
+	cfg.Mempool.Name = "para"
+	para := testnode.NewWithConfig(cfg, sub, nil)
+	para.Listen()
+	mockpara := paratest.NewParaNode(main, para)
+	tx := util.CreateTxWithExecer(mockpara.Para.GetGenesisKey(), "user.p.guodun.none")
+	hash := mockpara.Para.SendTx(tx)
+	assert.Equal(t, tx.Hash(), hash)
+
+	_, err := mockpara.Para.GetAPI().GetMempool()
+	assert.Equal(t, err, types.ErrActionNotSupport)
+	t.Log(err)
+}