func (c *mConnConnection) onReceive(chID ChannelID, payload []byte) {
	select {
	case c.receiveCh <- mConnMessage{channelID: chID, payload: payload}:
	case <-c.closeCh:
	}
}
onReceive is a callback for MConnection received messages.
package p2p
import (
	"context"
	"errors"
	"fmt"
	"io"
	"math"
	"net"
	"strconv"
	"sync"
	"golang.org/x/net/netutil"
	"github.com/tendermint/tendermint/crypto"
	"github.com/tendermint/tendermint/internal/libs/protoio"
	"github.com/tendermint/tendermint/internal/p2p/conn"
	"github.com/tendermint/tendermint/libs/log"
	p2pproto "github.com/tendermint/tendermint/proto/tendermint/p2p"
	"github.com/tendermint/tendermint/types"
)
const (
	MConnProtocol Protocol = "mconn"
	TCPProtocol   Protocol = "tcp"
)
type MConnTransportOptions struct {
	MaxAcceptedConnections uint32
}
type MConnTransport struct {
	logger       log.Logger
	options      MConnTransportOptions
	mConnConfig  conn.MConnConfig
	channelDescs []*ChannelDescriptor
	closeCh      chan struct{}
	closeOnce    sync.Once
	listener net.Listener
}
func NewMConnTransport(
	logger log.Logger,
	mConnConfig conn.MConnConfig,
	channelDescs []*ChannelDescriptor,
	options MConnTransportOptions,
) *MConnTransport {
	return &MConnTransport{
		logger:       logger,
		options:      options,
		mConnConfig:  mConnConfig,
		closeCh:      make(chan struct{}),
		channelDescs: channelDescs,
	}
}
func (m *MConnTransport) String() string {
	return string(MConnProtocol)
}
func (m *MConnTransport) Protocols() []Protocol {
	return []Protocol{MConnProtocol, TCPProtocol}
}
func (m *MConnTransport) Endpoints() []Endpoint {
	if m.listener == nil {
		return []Endpoint{}
	}
	select {
	case <-m.closeCh:
		return []Endpoint{}
	default:
	}
	endpoint := Endpoint{
		Protocol: MConnProtocol,
	}
	if addr, ok := m.listener.Addr().(*net.TCPAddr); ok {
		endpoint.IP = addr.IP
		endpoint.Port = uint16(addr.Port)
	}
	return []Endpoint{endpoint}
}
func (m *MConnTransport) Listen(endpoint Endpoint) error {
	if m.listener != nil {
		return errors.New("transport is already listening")
	}
	if err := m.validateEndpoint(endpoint); err != nil {
		return err
	}
	listener, err := net.Listen("tcp", net.JoinHostPort(
		endpoint.IP.String(), strconv.Itoa(int(endpoint.Port))))
	if err != nil {
		return err
	}
	if m.options.MaxAcceptedConnections > 0 {
		listener = netutil.LimitListener(listener, int(m.options.MaxAcceptedConnections))
	}
	m.listener = listener
	return nil
}
func (m *MConnTransport) Accept() (Connection, error) {
	if m.listener == nil {
		return nil, errors.New("transport is not listening")
	}
	tcpConn, err := m.listener.Accept()
	if err != nil {
		select {
		case <-m.closeCh:
			return nil, io.EOF
		default:
			return nil, err
		}
	}
	return newMConnConnection(m.logger, tcpConn, m.mConnConfig, m.channelDescs), nil
}
func (m *MConnTransport) Dial(ctx context.Context, endpoint Endpoint) (Connection, error) {
	if err := m.validateEndpoint(endpoint); err != nil {
		return nil, err
	}
	if endpoint.Port == 0 {
		endpoint.Port = 26657
	}
	dialer := net.Dialer{}
	tcpConn, err := dialer.DialContext(ctx, "tcp", net.JoinHostPort(
		endpoint.IP.String(), strconv.Itoa(int(endpoint.Port))))
	if err != nil {
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		default:
			return nil, err
		}
	}
	return newMConnConnection(m.logger, tcpConn, m.mConnConfig, m.channelDescs), nil
}
func (m *MConnTransport) Close() error {
	var err error
	m.closeOnce.Do(func() {
		close(m.closeCh)  
		if m.listener != nil {
			err = m.listener.Close()
		}
	})
	return err
}
func (m *MConnTransport) AddChannelDescriptors(channelDesc []*ChannelDescriptor) {
	m.channelDescs = append(m.channelDescs, channelDesc...)
}
func (m *MConnTransport) validateEndpoint(endpoint Endpoint) error {
	if err := endpoint.Validate(); err != nil {
		return err
	}
	if endpoint.Protocol != MConnProtocol && endpoint.Protocol != TCPProtocol {
		return fmt.Errorf("unsupported protocol %q", endpoint.Protocol)
	}
	if len(endpoint.IP) == 0 {
		return errors.New("endpoint has no IP address")
	}
	if endpoint.Path != "" {
		return fmt.Errorf("endpoints with path not supported (got %q)", endpoint.Path)
	}
	return nil
}
type mConnConnection struct {
	logger       log.Logger
	conn         net.Conn
	mConnConfig  conn.MConnConfig
	channelDescs []*ChannelDescriptor
	receiveCh    chan mConnMessage
	errorCh      chan error
	closeCh      chan struct{}
	closeOnce    sync.Once
	mconn *conn.MConnection  
}
type mConnMessage struct {
	channelID ChannelID
	payload   []byte
}
func newMConnConnection(
	logger log.Logger,
	conn net.Conn,
	mConnConfig conn.MConnConfig,
	channelDescs []*ChannelDescriptor,
) *mConnConnection {
	return &mConnConnection{
		logger:       logger,
		conn:         conn,
		mConnConfig:  mConnConfig,
		channelDescs: channelDescs,
		receiveCh:    make(chan mConnMessage),
		errorCh:      make(chan error, 1),  
		closeCh:      make(chan struct{}),
	}
}
func (c *mConnConnection) Handshake(
	ctx context.Context,
	nodeInfo types.NodeInfo,
	privKey crypto.PrivKey,
) (types.NodeInfo, crypto.PubKey, error) {
	var (
		mconn    *conn.MConnection
		peerInfo types.NodeInfo
		peerKey  crypto.PubKey
		errCh    = make(chan error, 1)
	)
	go func() {
		defer func() {
			if r := recover(); r != nil {
				errCh <- fmt.Errorf("recovered from panic: %v", r)
			}
		}()
		var err error
		mconn, peerInfo, peerKey, err = c.handshake(ctx, nodeInfo, privKey)
		errCh <- err
	}()
	select {
	case <-ctx.Done():
		_ = c.Close()
		return types.NodeInfo{}, nil, ctx.Err()
	case err := <-errCh:
		if err != nil {
			return types.NodeInfo{}, nil, err
		}
		c.mconn = mconn
		c.logger = mconn.Logger
		if err = c.mconn.Start(); err != nil {
			return types.NodeInfo{}, nil, err
		}
		return peerInfo, peerKey, nil
	}
}
func (c *mConnConnection) handshake(
	ctx context.Context,
	nodeInfo types.NodeInfo,
	privKey crypto.PrivKey,
) (*conn.MConnection, types.NodeInfo, crypto.PubKey, error) {
	if c.mconn != nil {
		return nil, types.NodeInfo{}, nil, errors.New("connection is already handshaked")
	}
	secretConn, err := conn.MakeSecretConnection(c.conn, privKey)
	if err != nil {
		return nil, types.NodeInfo{}, nil, err
	}
	var pbPeerInfo p2pproto.NodeInfo
	errCh := make(chan error, 2)
	go func() {
		_, err := protoio.NewDelimitedWriter(secretConn).WriteMsg(nodeInfo.ToProto())
		errCh <- err
	}()
	go func() {
		_, err := protoio.NewDelimitedReader(secretConn, types.MaxNodeInfoSize()).ReadMsg(&pbPeerInfo)
		errCh <- err
	}()
	for i := 0; i < cap(errCh); i++ {
		if err = <-errCh; err != nil {
			return nil, types.NodeInfo{}, nil, err
		}
	}
	peerInfo, err := types.NodeInfoFromProto(&pbPeerInfo)
	if err != nil {
		return nil, types.NodeInfo{}, nil, err
	}
	mconn := conn.NewMConnectionWithConfig(
		secretConn,
		c.channelDescs,
		c.onReceive,
		c.onError,
		c.mConnConfig,
	)
	mconn.SetLogger(c.logger.With("peer", c.RemoteEndpoint().NodeAddress(peerInfo.NodeID)))
	return mconn, peerInfo, secretConn.RemotePubKey(), nil
}