connection.go
func (c *MConnection) recvRoutine() {
	defer c._recover()

FOR_LOOP:
	for {
		// Block until .recvMonitor says we can read.
		c.recvMonitor.Limit(maxMsgPacketTotalSize, atomic.LoadInt64(&c.config.RecvRate), true)

		/*
			// Peek into bufReader for debugging
			if numBytes := c.bufReader.Buffered(); numBytes > 0 {
				log.Info("Peek connection buffer", "numBytes", numBytes, "bytes", log15.Lazy{func() []byte {
					bytes, err := c.bufReader.Peek(MinInt(numBytes, 100))
					if err == nil {
						return bytes
					} else {
						log.Warn("Error peeking connection buffer", "error", err)
						return nil
					}
				}})
			}
		*/

		// Read packet type
		var n int
		var err error
		pktType := wire.ReadByte(c.bufReader, &n, &err)
		c.recvMonitor.Update(int(n))
		if err != nil {
			if c.IsRunning() {
				log.Warn("Connection failed @ recvRoutine (reading byte)", "conn", c, "error", err)
				c.stopForError(err)
			}
			break FOR_LOOP
		}

		// Read more depending on packet type.
		switch pktType {
		case packetTypePing:
			// TODO: prevent abuse, as they cause flush()'s.
			log.Debug("Receive Ping")
			c.pong <- struct{}{}
		case packetTypePong:
			// do nothing
			log.Debug("Receive Pong")
		case packetTypeMsg:
			pkt, n, err := msgPacket{}, int(0), error(nil)
			wire.ReadBinaryPtr(&pkt, c.bufReader, maxMsgPacketTotalSize, &n, &err)
			c.recvMonitor.Update(int(n))
			if err != nil {
				if c.IsRunning() {
					log.Warn("Connection failed @ recvRoutine", "conn", c, "error", err)
					c.stopForError(err)
				}
				break FOR_LOOP
			}
			channel, ok := c.channelsIdx[pkt.ChannelID]
			if !ok || channel == nil {
				cmn.PanicQ(cmn.Fmt("Unknown channel %X", pkt.ChannelID))
			}
			msgBytes, err := channel.recvMsgPacket(pkt)
			if err != nil {
				if c.IsRunning() {
					log.Warn("Connection failed @ recvRoutine", "conn", c, "error", err)
					c.stopForError(err)
				}
				break FOR_LOOP
			}
			if msgBytes != nil {
				log.Debug("Received bytes", "chID", pkt.ChannelID, "msgBytes", msgBytes)
				c.onReceive(pkt.ChannelID, msgBytes)
			}
		default:
			cmn.PanicSanity(cmn.Fmt("Unknown message type %X", pktType))
		}

		// TODO: shouldn't this go in the sendRoutine?
		// Better to send a ping packet when *we* haven't sent anything for a while.
		c.pingTimer.Reset()
	}

	// Cleanup
	close(c.pong)
	for _ = range c.pong {
		// Drain
	}
}
func (c *MConnection) recvRoutine() {
	defer c._recover()

FOR_LOOP:
	for {
		// Block until .recvMonitor says we can read.
		c.recvMonitor.Limit(maxMsgPacketTotalSize, atomic.LoadInt64(&c.config.RecvRate), true)

		/*
			// Peek into bufReader for debugging
			if numBytes := c.bufReader.Buffered(); numBytes > 0 {
				log.Info("Peek connection buffer", "numBytes", numBytes, "bytes", log15.Lazy{func() []byte {
					bytes, err := c.bufReader.Peek(MinInt(numBytes, 100))
					if err == nil {
						return bytes
					} else {
						log.Warn("Error peeking connection buffer", "error", err)
						return nil
					}
				}})
			}
		*/

		// Read packet type
		var n int
		var err error
		pktType := wire.ReadByte(c.bufReader, &n, &err)
		c.recvMonitor.Update(int(n))
		if err != nil {
			if c.IsRunning() {
				log.Warn("Connection failed @ recvRoutine (reading byte)", "conn", c, "error", err)
				c.stopForError(err)
			}
			break FOR_LOOP
		}

		// Read more depending on packet type.
		switch pktType {
		case packetTypePing:
			// TODO: prevent abuse, as they cause flush()'s.
			log.Debug("Receive Ping")
			c.pong <- struct{}{}
		case packetTypePong:
			// do nothing
			log.Debug("Receive Pong")
		case packetTypeMsg:
			pkt, n, err := msgPacket{}, int(0), error(nil)
			wire.ReadBinaryPtr(&pkt, c.bufReader, maxMsgPacketTotalSize, &n, &err)
			c.recvMonitor.Update(int(n))
			if err != nil {
				if c.IsRunning() {
					log.Warn("Connection failed @ recvRoutine", "conn", c, "error", err)
					c.stopForError(err)
				}
				break FOR_LOOP
			}
			channel, ok := c.channelsIdx[pkt.ChannelID]
			if !ok || channel == nil {
				cmn.PanicQ(cmn.Fmt("Unknown channel %X", pkt.ChannelID))
			}
			msgBytes, err := channel.recvMsgPacket(pkt)
			if err != nil {
				if c.IsRunning() {
					log.Warn("Connection failed @ recvRoutine", "conn", c, "error", err)
					c.stopForError(err)
				}
				break FOR_LOOP
			}
			if msgBytes != nil {
				log.Debug("Received bytes", "chID", pkt.ChannelID, "msgBytes", msgBytes)
				c.onReceive(pkt.ChannelID, msgBytes)
			}
		default:
			cmn.PanicSanity(cmn.Fmt("Unknown message type %X", pktType))
		}

		// TODO: shouldn't this go in the sendRoutine?
		// Better to send a ping packet when *we* haven't sent anything for a while.
		c.pingTimer.Reset()
	}

	// Cleanup
	close(c.pong)
	for _ = range c.pong {
		// Drain
	}
}
func ( c * MConnection ) recvRoutine ( ) { defer c . _recover ( ) FOR_LOOP : for { // Block until .recvMonitor says we can read. c . recvMonitor . Limit ( maxMsgPacketTotalSize , atomic . LoadInt64 ( & c . config . RecvRate ) , true ) /*
			// Peek into bufReader for debugging
			if numBytes := c.bufReader.Buffered(); numBytes > 0 {
				log.Info("Peek connection buffer", "numBytes", numBytes, "bytes", log15.Lazy{func() []byte {
					bytes, err := c.bufReader.Peek(MinInt(numBytes, 100))
					if err == nil {
						return bytes
					} else {
						log.Warn("Error peeking connection buffer", "error", err)
						return nil
					}
				}})
			}
		*/ // Read packet type var n int var err error pktType := wire . ReadByte ( c . bufReader , & n , & err ) c . recvMonitor . Update ( int ( n ) ) if err != nil { if c . IsRunning ( ) { log . Warn ( " " , " " , c , " " , err ) c . stopForError ( err ) } break FOR_LOOP } // Read more depending on packet type. switch pktType { case packetTypePing : // TODO: prevent abuse, as they cause flush()'s. log . Debug ( " " ) c . pong <- struct { } { } case packetTypePong : // do nothing log . Debug ( " " ) case packetTypeMsg : pkt , n , err := msgPacket { } , int ( 0 ) , error ( nil ) wire . ReadBinaryPtr ( & pkt , c . bufReader , maxMsgPacketTotalSize , & n , & err ) c . recvMonitor . Update ( int ( n ) ) if err != nil { if c . IsRunning ( ) { log . Warn ( " " , " " , c , " " , err ) c . stopForError ( err ) } break FOR_LOOP } channel , ok := c . channelsIdx [ pkt . ChannelID ] if ! ok || channel == nil { cmn . PanicQ ( cmn . Fmt ( " " , pkt . ChannelID ) ) } msgBytes , err := channel . recvMsgPacket ( pkt ) if err != nil { if c . IsRunning ( ) { log . Warn ( " " , " " , c , " " , err ) c . stopForError ( err ) } break FOR_LOOP } if msgBytes != nil { log . Debug ( " " , " " , pkt . ChannelID , " " , msgBytes ) c . onReceive ( pkt . ChannelID , msgBytes ) } default : cmn . PanicSanity ( cmn . Fmt ( " " , pktType ) ) } // TODO: shouldn't this go in the sendRoutine? // Better to send a ping packet when *we* haven't sent anything for a while. c . pingTimer . Reset ( ) } // Cleanup close ( c . pong ) for _ = range c . pong { // Drain } }
// recvRoutine reads msgPackets and reconstructs the message using the channels' "recving" buffer.
// After a whole message has been assembled, it's pushed to onReceive().
// Blocks depending on how the connection is throttled.
recvRoutine reads msgPackets and reconstructs the message using the channels recving buffer . After a whole message has been assembled it s pushed to onReceive () . Blocks depending on how the connection is throttled .