func (q *QueueProtocol) ExecWalletFunc(driver string, funcname string, param types.Message) (types.Message, error) {
	if types.IsNilP(param) {
		err := types.ErrInvalidParam
		log.Error("ExecWalletFunc", "Error", err)
		return nil, err
	}
	query := &types.ChainExecutor{Driver: driver, FuncName: funcname, Param: types.Encode(param)}
	return q.ExecWallet(query)
}
ExecWalletFunc exec wallet function interface
package client
import (
	"errors"
	"fmt"
	"time"
	"github.com/33cn/chain33/common"
	"github.com/33cn/chain33/common/crypto"
	"github.com/33cn/chain33/common/log/log15"
	"github.com/33cn/chain33/common/version"
	"github.com/33cn/chain33/queue"
	"github.com/33cn/chain33/types"
)
const (
	mempoolKey = "mempool"  
	p2pKey     = "p2p"      
	consensusKey = "consensus"  
	executorKey   = "execs"       
	walletKey     = "wallet"      
	blockchainKey = "blockchain"  
	storeKey      = "store"
)
var log = log15.New("module", "client")
type QueueProtocolOption struct {
	SendTimeout time.Duration
	WaitTimeout time.Duration
}
type QueueProtocol struct {
	client queue.Client
	option QueueProtocolOption
}
func New(client queue.Client, option *QueueProtocolOption) (QueueProtocolAPI, error) {
	if client == nil {
		return nil, types.ErrInvalidParam
	}
	q := &QueueProtocol{}
	q.client = client
	if option != nil {
		q.option = *option
	} else {
		q.option.SendTimeout = time.Duration(-1)
		q.option.WaitTimeout = time.Duration(-1)
	}
	return q, nil
}
func (q *QueueProtocol) send(topic string, ty int64, data interface{}) (*queue.Message, error) {
	client := q.client
	msg := client.NewMessage(topic, ty, data)
	err := client.SendTimeout(msg, true, q.option.SendTimeout)
	if err != nil {
		return &queue.Message{}, err
	}
	reply, err := client.WaitTimeout(msg, q.option.WaitTimeout)
	if err != nil {
		return nil, err
	}
	client.FreeMessage(msg)
	return reply, nil
}
func (q *QueueProtocol) notify(topic string, ty int64, data interface{}) (*queue.Message, error) {
	client := q.client
	msg := client.NewMessage(topic, ty, data)
	err := client.SendTimeout(msg, false, q.option.SendTimeout)
	if err != nil {
		return &queue.Message{}, err
	}
	return msg, err
}
func (q *QueueProtocol) Notify(topic string, ty int64, data interface{}) (*queue.Message, error) {
	return q.notify(topic, ty, data)
}
func (q *QueueProtocol) Close() {
	q.client.Close()
}
func (q *QueueProtocol) NewMessage(topic string, msgid int64, data interface{}) *queue.Message {
	return q.client.NewMessage(topic, msgid, data)
}
func (q *QueueProtocol) setOption(option *QueueProtocolOption) {
	if option != nil {
		q.option = *option
	}
}
func (q *QueueProtocol) SendTx(param *types.Transaction) (*types.Reply, error) {
	if param == nil {
		err := types.ErrInvalidParam
		log.Error("SendTx", "Error", err)
		return nil, err
	}
	msg, err := q.send(mempoolKey, types.EventTx, param)
	if err != nil {
		return nil, err
	}
	reply, ok := msg.GetData().(*types.Reply)
	if ok {
		if reply.GetIsOk() {
			reply.Msg = param.Hash()
		} else {
			msg := string(reply.Msg)
			err = fmt.Errorf(msg)
			reply = nil
		}
	} else {
		err = types.ErrTypeAsset
	}
	q.client.FreeMessage(msg)
	return reply, err
}
func (q *QueueProtocol) GetTxList(param *types.TxHashList) (*types.ReplyTxList, error) {
	if param == nil {
		err := types.ErrInvalidParam
		log.Error("GetTxList", "Error", err)
		return nil, err
	}
	msg, err := q.send(mempoolKey, types.EventTxList, param)
	if err != nil {
		log.Error("GetTxList", "Error", err.Error())
		return nil, err
	}
	if reply, ok := msg.GetData().(*types.ReplyTxList); ok {
		return reply, nil
	}
	return nil, types.ErrTypeAsset
}
func (q *QueueProtocol) GetBlocks(param *types.ReqBlocks) (*types.BlockDetails, error) {
	if param == nil {
		err := types.ErrInvalidParam
		log.Error("GetBlocks", "Error", err)
		return nil, err
	}
	msg, err := q.send(blockchainKey, types.EventGetBlocks, param)
	if err != nil {
		log.Error("GetBlocks", "Error", err.Error())
		return nil, err
	}
	if reply, ok := msg.GetData().(*types.BlockDetails); ok {
		return reply, nil
	}
	err = types.ErrTypeAsset
	log.Error("GetBlocks", "Error", err.Error())
	return nil, err
}
func (q *QueueProtocol) QueryTx(param *types.ReqHash) (*types.TransactionDetail, error) {
	if param == nil {
		err := types.ErrInvalidParam
		log.Debug("QueryTx", "Error", err)
		return nil, err
	}
	msg, err := q.send(blockchainKey, types.EventQueryTx, param)
	if err != nil {
		log.Debug("QueryTx", "Error", err.Error())
		return nil, err
	}
	if reply, ok := msg.GetData().(*types.TransactionDetail); ok {
		return reply, nil
	}
	return nil, types.ErrTypeAsset
}
func (q *QueueProtocol) GetTransactionByAddr(param *types.ReqAddr) (*types.ReplyTxInfos, error) {
	if param == nil {
		err := types.ErrInvalidParam
		log.Error("GetTransactionByAddr", "Error", err)
		return nil, err
	}
	msg, err := q.send(blockchainKey, types.EventGetTransactionByAddr, param)
	if err != nil {
		log.Error("GetTransactionByAddr", "Error", err.Error())
		return nil, err
	}
	if reply, ok := msg.GetData().(*types.ReplyTxInfos); ok {
		return reply, nil
	}
	err = types.ErrTypeAsset
	log.Error("GetTransactionByAddr", "Error", err)
	return nil, types.ErrTypeAsset
}
func (q *QueueProtocol) GetTransactionByHash(param *types.ReqHashes) (*types.TransactionDetails, error) {
	if param == nil {
		err := types.ErrInvalidParam
		log.Error("GetTransactionByHash", "Error", err)
		return nil, err
	}
	msg, err := q.send(blockchainKey, types.EventGetTransactionByHash, param)
	if err != nil {
		log.Error("GetTransactionByHash", "Error", err.Error())
		return nil, err
	}
	if reply, ok := msg.GetData().(*types.TransactionDetails); ok {
		return reply, nil
	}
	return nil, types.ErrTypeAsset
}
func (q *QueueProtocol) GetMempool(req *types.ReqGetMempool) (*types.ReplyTxList, error) {
	msg, err := q.send(mempoolKey, types.EventGetMempool, req)
	if err != nil {
		log.Error("GetMempool", "Error", err.Error())
		return nil, err
	}
	if reply, ok := msg.GetData().(*types.ReplyTxList); ok {
		return reply, nil
	}
	return nil, types.ErrTypeAsset
}
func (q *QueueProtocol) PeerInfo(req *types.P2PGetPeerReq) (*types.PeerList, error) {
	msg, err := q.send(p2pKey, types.EventPeerInfo, req)
	if err != nil {
		log.Error("PeerInfo", "Error", err.Error())
		return nil, err
	}
	if reply, ok := msg.GetData().(*types.PeerList); ok {
		return reply, nil
	}
	return nil, types.ErrTypeAsset
}
func (q *QueueProtocol) NetProtocols(req *types.ReqNil) (*types.NetProtocolInfos, error) {
	msg, err := q.send(p2pKey, types.EventNetProtocols, req)
	if err != nil {
		log.Error("PeerInfo", "Error", err.Error())
		return nil, err
	}
	if reply, ok := msg.GetData().(*types.NetProtocolInfos); ok {
		return reply, nil
	}
	return nil, types.ErrTypeAsset
}
func (q *QueueProtocol) GetHeaders(param *types.ReqBlocks) (*types.Headers, error) {
	if param == nil {
		err := types.ErrInvalidParam
		log.Error("GetHeaders", "Error", err)
		return nil, err
	}
	msg, err := q.send(blockchainKey, types.EventGetHeaders, param)
	if err != nil {
		log.Error("GetHeaders", "Error", err.Error())
		return nil, err
	}
	if reply, ok := msg.GetData().(*types.Headers); ok {
		return reply, nil
	}
	return nil, types.ErrTypeAsset
}
func (q *QueueProtocol) GetLastMempool() (*types.ReplyTxList, error) {
	msg, err := q.send(mempoolKey, types.EventGetLastMempool, &types.ReqNil{})
	if err != nil {
		log.Error("GetLastMempool", "Error", err.Error())
		return nil, err
	}
	if reply, ok := msg.GetData().(*types.ReplyTxList); ok {
		return reply, nil
	}
	return nil, types.ErrTypeAsset
}
func (q *QueueProtocol) GetProperFee(req *types.ReqProperFee) (*types.ReplyProperFee, error) {
	msg, err := q.send(mempoolKey, types.EventGetProperFee, req)
	if err != nil {
		log.Error("GetProperFee", "Error", err.Error())
		return nil, err
	}
	if reply, ok := msg.GetData().(*types.ReplyProperFee); ok {
		return reply, nil
	}
	return nil, types.ErrTypeAsset
}
func (q *QueueProtocol) GetBlockOverview(param *types.ReqHash) (*types.BlockOverview, error) {
	if param == nil {
		err := types.ErrInvalidParam
		log.Error("GetBlockOverview", "Error", err)
		return nil, err
	}
	msg, err := q.send(blockchainKey, types.EventGetBlockOverview, param)
	if err != nil {
		log.Error("GetBlockOverview", "Error", err.Error())
		return nil, err
	}
	if reply, ok := msg.GetData().(*types.BlockOverview); ok {
		return reply, nil
	}
	return nil, types.ErrTypeAsset
}
func (q *QueueProtocol) GetAddrOverview(param *types.ReqAddr) (*types.AddrOverview, error) {
	if param == nil {
		err := types.ErrInvalidParam
		log.Error("GetAddrOverview", "Error", err)
		return nil, err
	}
	msg, err := q.send(blockchainKey, types.EventGetAddrOverview, param)
	if err != nil {
		log.Error("GetAddrOverview", "Error", err.Error())
		return nil, err
	}
	if reply, ok := msg.GetData().(*types.AddrOverview); ok {
		return reply, nil
	}
	return nil, types.ErrTypeAsset
}
func (q *QueueProtocol) GetBlockHash(param *types.ReqInt) (*types.ReplyHash, error) {
	if param == nil {
		err := types.ErrInvalidParam
		log.Error("GetBlockHash", "Error", err)
		return nil, err
	}
	msg, err := q.send(blockchainKey, types.EventGetBlockHash, param)
	if err != nil {
		log.Error("GetBlockHash", "Error", err.Error())
		return nil, err
	}
	if reply, ok := msg.GetData().(*types.ReplyHash); ok {
		return reply, nil
	}
	return nil, types.ErrTypeAsset
}
func (q *QueueProtocol) Query(driver, funcname string, param types.Message) (types.Message, error) {
	if types.IsNilP(param) {
		err := types.ErrInvalidParam
		log.Error("Query", "Error", err)
		return nil, err
	}
	query := &types.ChainExecutor{Driver: driver, FuncName: funcname, Param: types.Encode(param)}
	return q.QueryChain(query)
}
func (q *QueueProtocol) QueryConsensus(param *types.ChainExecutor) (types.Message, error) {
	if param == nil {
		err := types.ErrInvalidParam
		log.Error("ExecWallet", "Error", err)
		return nil, err
	}
	msg, err := q.send(consensusKey, types.EventConsensusQuery, param)
	if err != nil {
		log.Error("query QueryConsensus", "Error", err.Error())
		return nil, err
	}
	if reply, ok := msg.GetData().(types.Message); ok {
		return reply, nil
	}
	return nil, types.ErrTypeAsset
}