add unsupported msg reply

plugin/mempool/para/base.go

@@ -17,14 +17,14 @@ var RETRY_TIMES = 3
 
 //Mempool mempool 基础类
 type Mempool struct {
-	Key         string
+	key         string
 	mainGrpcCli types.Chain33Client
 }
 
 //NewMempool 新建mempool 实例
 func NewMempool(cfg *types.Mempool) *Mempool {
 	pool := &Mempool{}
-	pool.Key = topic
+	pool.key = topic
 	if types.IsPara() {
 		grpcCli, err := grpcclient.NewMainChainClient("")
 		if err != nil {
@@ -39,7 +39,7 @@ func NewMempool(cfg *types.Mempool) *Mempool {
 //SetQueueClient 初始化mempool模块
 func (mem *Mempool) SetQueueClient(client queue.Client) {
 	go func() {
-		client.Sub(mem.Key)
+		client.Sub(mem.key)
 		for msg := range client.Recv() {
 			switch msg.Ty {
 			case types.EventTx:
@@ -59,14 +59,16 @@ func (mem *Mempool) SetQueueClient(client queue.Client) {
 							}
 						}
 						if err != nil {
-							msg.Reply(client.NewMessage(topic, types.EventReply, &types.Reply{IsOk: false,
+							msg.Reply(client.NewMessage(mem.key, types.EventReply, &types.Reply{IsOk: false,
 								Msg: []byte(fmt.Sprintf("Send transaction to main chain failed, %v", err))}))
 							break
 						}
 					}
-					msg.Reply(client.NewMessage(topic, types.EventReply, &types.Reply{IsOk: true, Msg: []byte(reply.GetMsg())}))
+					msg.Reply(client.NewMessage(mem.key, types.EventReply, &types.Reply{IsOk: true, Msg: []byte(reply.GetMsg())}))
 				}
 			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))}))
 			}
 		}
 	}()