func (ch *Channel) isSendPending() bool {
	if len(ch.sending) == 0 {
		if len(ch.sendQueue) == 0 {
			return false
		}
		ch.sending = <-ch.sendQueue
	}
	return true
}
Returns true if any msgPackets are pending to be sent.
Call before calling nextMsgPacket()
Goroutine-safe
package p2p
import (
	"bufio"
	"errors"
	"fmt"
	"io"
	"math"
	"net"
	"runtime/debug"
	"strings"
	"sync/atomic"
	"time"
	"github.com/spf13/viper"
	"go.uber.org/zap"
	. "github.com/annchain/annchain/module/lib/go-common"
	flow "github.com/annchain/annchain/module/lib/go-flowrate/flowrate"
	"github.com/annchain/annchain/module/xlib"
)
const (
	numBatchMsgPackets = 10
	minReadBufferSize  = 1024
	minWriteBufferSize = 65536
	idleTimeoutMinutes = 5
	updateStatsSeconds = 2
	pingTimeoutSeconds = 40
	flushThrottleMS    = 100
	defaultSendQueueCapacity   = 1
	defaultRecvBufferCapacity  = 4096
	defaultRecvMessageCapacity = 22020096  
	defaultSendTimeoutSeconds  = 10
)
type receiveCbFunc func(chID byte, msgBytes []byte)
type errorCbFunc func(interface{})
type MConnection struct {
	BaseService
	conn        net.Conn
	bufReader   *bufio.Reader
	bufWriter   *bufio.Writer
	sendMonitor *flow.Monitor
	recvMonitor *flow.Monitor
	sendRate    int64
	recvRate    int64
	send        chan struct{}
	pong        chan struct{}
	channels    []*Channel
	channelsIdx map[byte]*Channel
	onReceive   receiveCbFunc
	onError     errorCbFunc
	errored     uint32
	quit         chan struct{}
	flushTimer   *ThrottleTimer  
	pingTimer    *RepeatTimer    
	chStatsTimer *RepeatTimer    
	LocalAddress  *NetAddress
	RemoteAddress *NetAddress
	connectResetWait time.Duration
	logger  *zap.Logger
	slogger *zap.SugaredLogger
}
func NewMConnection(logger *zap.Logger, config *viper.Viper, conn net.Conn, chDescs []*ChannelDescriptor, onReceive receiveCbFunc, onError errorCbFunc) *MConnection {
	mconn := &MConnection{
		conn:        conn,
		bufReader:   bufio.NewReaderSize(conn, minReadBufferSize),
		bufWriter:   bufio.NewWriterSize(conn, minWriteBufferSize),
		sendMonitor: flow.New(0, 0),
		recvMonitor: flow.New(0, 0),
		sendRate:    config.GetInt64(configKeySendRate),
		recvRate:    config.GetInt64(configKeyRecvRate),
		send:        make(chan struct{}, 1),
		pong:        make(chan struct{}),
		onReceive:   onReceive,
		onError:     onError,
		quit:         nil,
		flushTimer:   nil,
		pingTimer:    nil,
		chStatsTimer: nil,
		LocalAddress:  NewNetAddress(logger, conn.LocalAddr()),
		RemoteAddress: NewNetAddress(logger, conn.RemoteAddr()),
		connectResetWait: time.Duration(config.GetInt("connection_reset_wait")) * time.Millisecond,
		logger:  logger,
		slogger: logger.Sugar(),
	}
	var channelsIdx = map[byte]*Channel{}
	var channels = []*Channel{}
	for _, desc := range chDescs {
		descCopy := *desc  
		channel := newChannel(mconn, &descCopy)
		channelsIdx[channel.id] = channel
		channels = append(channels, channel)
	}
	mconn.channels = channels
	mconn.channelsIdx = channelsIdx
	mconn.BaseService = *NewBaseService(logger, "MConnection", mconn)
	return mconn
}
func (c *MConnection) OnStart() error {
	c.BaseService.OnStart()
	c.quit = make(chan struct{})
	c.flushTimer = NewThrottleTimer("flush", flushThrottleMS*time.Millisecond)
	c.pingTimer = NewRepeatTimer("ping", pingTimeoutSeconds*time.Second)
	c.chStatsTimer = NewRepeatTimer("chStats", updateStatsSeconds*time.Second)
	go c.sendRoutine()
	go c.recvRoutine()
	return nil
}
func (c *MConnection) OnStop() {
	c.BaseService.OnStop()
	c.flushTimer.Stop()
	c.pingTimer.Stop()
	c.chStatsTimer.Stop()
	if c.quit != nil {
		close(c.quit)
	}
	c.conn.Close()
}
func (c *MConnection) String() string {
	return fmt.Sprintf("MConn{%v}", c.conn.RemoteAddr())
}
func (c *MConnection) flush() {
	err := c.bufWriter.Flush()
	if err != nil {
		c.logger.Warn("MConnection flush failed", zap.Error(err))
	}
}
func (c *MConnection) _recover() {
	if r := recover(); r != nil {
		stack := debug.Stack()
		c.stopForError(StackError{r, stack})
	}
}
func (c *MConnection) stopForError(r interface{}) {
	c.Stop()
	if atomic.CompareAndSwapUint32(&c.errored, 0, 1) {
		if c.onError != nil {
			c.onError(r)
		}
	}
}
func (c *MConnection) Send(chID byte, msg []byte) bool {
	if !c.IsRunning() {
		return false
	}
	channel, ok := c.channelsIdx[chID]
	if !ok {
		c.logger.Error("Cannot send bytes, unknown channel", zap.ByteString("chID", []byte{chID}))
		return false
	}
	success := channel.sendBytes(msg)
	if success {
		select {
		case c.send <- struct{}{}:
		default:
		}
	} else {
		c.slogger.Warnw("Send failed", "channel", chID, "conn", c, "msg", msg)
	}
	return success
}
func (c *MConnection) TrySend(chID byte, msg []byte) bool {
	if !c.IsRunning() {
		return false
	}
	channel, ok := c.channelsIdx[chID]
	if !ok {
		c.logger.Error("Cannot send bytes, unknown channels", zap.ByteString("chID", []byte{chID}))
		return false
	}
	ok = channel.trySendBytes(msg)
	if ok {
		select {
		case c.send <- struct{}{}:
		default:
		}
	}
	return ok
}
func (c *MConnection) CanSend(chID byte) bool {
	if !c.IsRunning() {
		return false
	}
	channel, ok := c.channelsIdx[chID]
	if !ok {
		c.slogger.Errorf("Unknown channel %X", chID)
		return false
	}
	return channel.canSend()
}
func (c *MConnection) sendRoutine() {
	defer c._recover()
FOR_LOOP:
	for {
		var err error
		select {
		case <-c.flushTimer.Ch:
			c.flush()
		case <-c.chStatsTimer.Ch:
			for _, channel := range c.channels {
				channel.updateStats()
			}
		case <-c.pingTimer.Ch:
			xlib.BinWrite(c.bufWriter, packetTypePing)
			c.sendMonitor.Update(1)
			c.flush()
		case <-c.pong:
			xlib.BinWrite(c.bufWriter, packetTypePong)
			c.sendMonitor.Update(1)
			c.flush()
		case <-c.quit:
			break FOR_LOOP
		case <-c.send:
			eof := c.sendSomeMsgPackets()
			if !eof {
				select {
				case c.send <- struct{}{}:
				default:
				}
			}
		}
		if !c.IsRunning() {
			break FOR_LOOP
		}
		if err != nil {
			if strings.Contains(err.Error(), "connection reset by peer") {
				time.Sleep(c.connectResetWait)
			} else {
				c.slogger.Warnw("Connection failed @ sendRoutine", "conn", c, "error", err)
				c.stopForError(err)
				break FOR_LOOP
			}
		}
	}
}
func (c *MConnection) sendSomeMsgPackets() bool {
	c.sendMonitor.Limit(maxMsgPacketTotalSize, atomic.LoadInt64(&c.sendRate), true)
	for i := 0; i < numBatchMsgPackets; i++ {
		if c.sendMsgPacket() {
			return true
		}
	}
	return false
}
func (c *MConnection) sendMsgPacket() bool {
	var leastRatio float32 = math.MaxFloat32
	var leastChannel *Channel
	for _, channel := range c.channels {
		if !channel.isSendPending() {
			continue
		}
		ratio := float32(channel.recentlySent) / float32(channel.priority)
		if ratio < leastRatio {
			leastRatio = ratio
			leastChannel = channel
		}
	}
	if leastChannel == nil {
		return true
	}
	n, err := leastChannel.writeMsgPacketTo(c.bufWriter)
	if err != nil {
		c.logger.Warn("Failed to write msgPacket", zap.Error(err))
		c.stopForError(err)
		return true
	}
	c.sendMonitor.Update(int(n))
	c.flushTimer.Set()
	return false
}
func (c *MConnection) recvRoutine() {
	defer c._recover()
FOR_LOOP:
	for {
		c.recvMonitor.Limit(maxMsgPacketTotalSize, atomic.LoadInt64(&c.recvRate), true)
		var err error
		var pktType byte
		err = xlib.BinRead(c.bufReader, &pktType)
		c.recvMonitor.Update(1)
		if err != nil {
			if c.IsRunning() {
				c.slogger.Warnw("Connection failed @ recvRoutine (reading byte)", "conn", c, "error", err)
				c.stopForError(err)
			}
			break FOR_LOOP
		}
		switch pktType {
		case packetTypePing:
			c.pong <- struct{}{}
		case packetTypePong:
		case packetTypeMsg:
			pkt := &msgPacket{}
			n, err := pkt.ReadBytes(c.bufReader)
			c.recvMonitor.Update(n)
			if err != nil {
				if c.IsRunning() {
					c.slogger.Warnw("Connection failed @ recvRoutine", "conn", c, "error", err)
					c.stopForError(err)
				}
				break FOR_LOOP
			}
			channel, ok := c.channelsIdx[pkt.ChannelID]
			if !ok || channel == nil {
				c.logger.Error("Unknown channel", zap.ByteString("ChannelID", []byte{pkt.ChannelID}))
				continue FOR_LOOP
			}
			msgBytes, err := channel.recvMsgPacket(pkt)
			if err != nil {
				if c.IsRunning() {
					c.slogger.Warnw("Connection failed @ recvRoutine", "conn", c, "error", err)
					c.stopForError(err)
				}
				break FOR_LOOP
			}
			if msgBytes != nil {
				c.onReceive(pkt.ChannelID, msgBytes)
			}
		default:
			c.logger.Error("Unknown message type", zap.ByteString("message type", []byte{pktType}))
		}
		c.pingTimer.Reset()
	}
	close(c.pong)
	for _ = range c.pong {
	}
}
type ConnectionStatus struct {
	SendMonitor flow.Status
	RecvMonitor flow.Status
	Channels    []ChannelStatus
}
type ChannelStatus struct {
	ID                byte
	SendQueueCapacity int
	SendQueueSize     int
	Priority          int
	RecentlySent      int64
}
func (c *MConnection) Status() ConnectionStatus {
	var status ConnectionStatus
	status.SendMonitor = c.sendMonitor.Status()
	status.RecvMonitor = c.recvMonitor.Status()
	status.Channels = make([]ChannelStatus, len(c.channels))
	for i, channel := range c.channels {
		status.Channels[i] = ChannelStatus{
			ID:                channel.id,
			SendQueueCapacity: cap(channel.sendQueue),
			SendQueueSize:     int(channel.sendQueueSize),  
			Priority:          channel.priority,
			RecentlySent:      channel.recentlySent,
		}
	}
	return status
}
type ChannelDescriptor struct {
	ID                  byte
	Priority            int
	SendQueueCapacity   int
	RecvBufferCapacity  int
	RecvMessageCapacity int
}
func (chDesc *ChannelDescriptor) FillDefaults() {
	if chDesc.SendQueueCapacity == 0 {
		chDesc.SendQueueCapacity = defaultSendQueueCapacity
	}
	if chDesc.RecvBufferCapacity == 0 {
		chDesc.RecvBufferCapacity = defaultRecvBufferCapacity
	}
	if chDesc.RecvMessageCapacity == 0 {
		chDesc.RecvMessageCapacity = defaultRecvMessageCapacity
	}
}
type Channel struct {
	conn          *MConnection
	desc          *ChannelDescriptor
	id            byte
	sendQueue     chan []byte
	sendQueueSize int32  
	recving       []byte
	sending       []byte
	priority      int
	recentlySent  int64  
	slogger *zap.SugaredLogger
}
func newChannel(conn *MConnection, desc *ChannelDescriptor) *Channel {
	desc.FillDefaults()
	if desc.Priority <= 0 {
		PanicSanity("Channel default priority must be a postive integer")
	}
	return &Channel{
		conn:      conn,
		desc:      desc,
		id:        desc.ID,
		sendQueue: make(chan []byte, desc.SendQueueCapacity),
		recving:   make([]byte, 0, desc.RecvBufferCapacity),
		priority:  desc.Priority,
	}
}
func (ch *Channel) sendBytes(bytes []byte) bool {
	timeout := time.NewTimer(defaultSendTimeoutSeconds * time.Second)
	select {
	case <-timeout.C:
		return false
	case ch.sendQueue <- bytes:
		atomic.AddInt32(&ch.sendQueueSize, 1)
		return true
	}
}
func (ch *Channel) trySendBytes(bytes []byte) bool {
	select {
	case ch.sendQueue <- bytes:
		atomic.AddInt32(&ch.sendQueueSize, 1)
		return true
	default:
		return false
	}
}
func (ch *Channel) loadSendQueueSize() (size int) {
	return int(atomic.LoadInt32(&ch.sendQueueSize))
}
func (ch *Channel) canSend() bool {
	return ch.loadSendQueueSize() < defaultSendQueueCapacity
}