func HandlerWithRW(f func(request *types.P2PRequest, response *types.P2PResponse) error) network.StreamHandler {
	return func(stream network.Stream) {
		var req types.P2PRequest
		if err := ReadStream(&req, stream); err != nil {
			log.Error("HandlerWithRW", "read stream error", err)
			return
		}
		var res types.P2PResponse
		err := f(&req, &res)
		if err != nil {
			res.Response = nil
			res.Error = err.Error()
		}
		res.Headers = &types.P2PMessageHeaders{
			Version:   types2.Version,
			Timestamp: time.Now().Unix(),
			Id:        rand.Int63(),
		}
		if err := WriteStream(&res, stream); err != nil {
			log.Error("HandlerWithAuthAndSign", "write stream error", err)
			return
		}
	}
}
HandlerWithRW wraps handler with reading, writing, closing stream and recovering from panic.
package protocol
import (
	"bytes"
	"errors"
	"fmt"
	"io"
	"math/rand"
	"runtime"
	"time"
	"github.com/33cn/chain33/common/log/log15"
	"github.com/33cn/chain33/queue"
	types2 "github.com/33cn/chain33/system/p2p/dht/types"
	"github.com/33cn/chain33/types"
	"github.com/libp2p/go-libp2p-core/crypto"
	"github.com/libp2p/go-libp2p-core/network"
	"github.com/libp2p/go-msgio"
)
var log = log15.New("module", "p2p.protocol")
func init() {
	rand.Seed(time.Now().UnixNano())
}
const messageHeaderLen = 17
var (
	messageHeader = headerSafe([]byte("/protobuf/msgio"))
)
func ReadStream(data types.Message, stream network.Stream) error {
	var header [messageHeaderLen]byte
	_, err := io.ReadFull(stream, header[:])
	if err != nil || !bytes.Equal(header[:], messageHeader) {
		log.Error("ReadStream", "pid", stream.Conn().RemotePeer().Pretty(), "protocolID", stream.Protocol(), "read header err", err)
		return err
	}
	reader := msgio.NewReaderSize(stream, types.MaxBlockSize)
	msg, err := reader.ReadMsg()
	defer reader.ReleaseMsg(msg)
	if err != nil {
		log.Error("ReadStream", "pid", stream.Conn().RemotePeer().Pretty(), "protocolID", stream.Protocol(), "read msg err", err)
		return err
	}
	err = types.Decode(msg, data)
	if err != nil {
		log.Error("ReadStream", "pid", stream.Conn().RemotePeer().Pretty(), "protocolID", stream.Protocol(), "decode err", err)
		return err
	}
	return nil
}
func WriteStream(data types.Message, stream network.Stream) error {
	_, err := stream.Write(messageHeader)
	if err != nil {
		log.Error("WriteStream", "pid", stream.Conn().RemotePeer().Pretty(), "protocolID", stream.Protocol(), "write header err", err)
		return err
	}
	msg := types.Encode(data)
	writer := msgio.NewWriter(stream)
	err = writer.WriteMsg(msg)
	if err != nil {
		log.Error("WriteStream", "pid", stream.Conn().RemotePeer().Pretty(), "protocolID", stream.Protocol(), "write msg err", err)
		return err
	}
	return nil
}
func CloseStream(stream network.Stream) {
	if stream == nil {
		return
	}
	_ = stream.CloseWrite()
	_ = stream.CloseRead()
	go func() {
		err := AwaitEOF(stream)
		if err != nil {
			log.Debug("CloseStream", "err", err, "protocol ID", stream.Protocol())
		}
	}()
}
func AuthenticateMessage(message types.Message, stream network.Stream) bool {
	var sign, bin []byte
	switch t := message.(type) {
	case *types.P2PRequest:
		sign = t.Headers.Sign
		t.Headers.Sign = nil
		bin = types.Encode(t)
		t.Headers.Sign = sign
	case *types.P2PResponse:
		sign = t.Headers.Sign
		t.Headers.Sign = nil
		bin = types.Encode(t)
		t.Headers.Sign = sign
	default:
		return false
	}
	return verifyData(bin, sign, stream.Conn().RemotePublicKey())
}
func verifyData(data []byte, signature []byte, pubKey crypto.PubKey) bool {
	res, err := pubKey.Verify(data, signature)
	if err != nil {
		log.Error("Error authenticating data", "err", err)
		return false
	}
	return res
}
func ReadStreamAndAuthenticate(message types.Message, stream network.Stream) error {
	if err := ReadStream(message, stream); err != nil {
		return err
	}
	if !AuthenticateMessage(message, stream) {
		return types2.ErrWrongSignature
	}
	return nil
}
func signProtoMessage(message types.Message, stream network.Stream) ([]byte, error) {
	privKey := stream.Conn().LocalPrivateKey()
	return privKey.Sign(types.Encode(message))
}
func SignAndWriteStream(message types.Message, stream network.Stream) error {
	switch t := message.(type) {
	case *types.P2PRequest:
		t.Headers = &types.P2PMessageHeaders{
			Version:   types2.Version,
			Timestamp: time.Now().Unix(),
			Id:        rand.Int63(),
		}
		sign, err := signProtoMessage(t, stream)
		if err != nil {
			return err
		}
		t.Headers.Sign = sign
	case *types.P2PResponse:
		t.Headers = &types.P2PMessageHeaders{
			Version:   types2.Version,
			Timestamp: time.Now().Unix(),
			Id:        rand.Int63(),
		}
		sign, err := signProtoMessage(t, stream)
		if err != nil {
			return err
		}
		t.Headers.Sign = sign
	default:
		log.Error("SignAndWriteStream wrong message type")
		return types2.ErrInvalidMessageType
	}
	return WriteStream(message, stream)
}
func HandlerWithClose(f network.StreamHandler) network.StreamHandler {
	return func(stream network.Stream) {
		defer func() {
			if r := recover(); r != nil {
				log.Error("handle stream", "panic error", r)
				fmt.Println(string(panicTrace(4)))
				_ = stream.Reset()
			}
		}()
		f(stream)
		CloseStream(stream)
	}
}
func HandlerWithWrite(f func(resp *types.P2PResponse) error) network.StreamHandler {
	return func(stream network.Stream) {
		var res types.P2PResponse
		err := f(&res)
		if err != nil {
			res.Response = nil
			res.Error = err.Error()
		}
		res.Headers = &types.P2PMessageHeaders{
			Version:   types2.Version,
			Timestamp: time.Now().Unix(),
			Id:        rand.Int63(),
		}
		if err := WriteStream(&res, stream); err != nil {
			log.Error("HandlerWithWrite", "write stream error", err)
			return
		}
	}
}
func HandlerWithRead(f func(request *types.P2PRequest)) network.StreamHandler {
	return func(stream network.Stream) {
		var req types.P2PRequest
		if err := ReadStream(&req, stream); err != nil {
			log.Error("HandlerWithRead", "read stream error", err)
			return
		}
		f(&req)
	}
}
func HandlerWithAuth(f func(request *types.P2PRequest)) network.StreamHandler {
	return func(stream network.Stream) {
		var req types.P2PRequest
		if err := ReadStream(&req, stream); err != nil {
			log.Error("HandlerWithAuthAndSign", "read stream error", err)
			return
		}
		if !AuthenticateMessage(&req, stream) {
			return
		}
		f(&req)
	}
}