func (p *MessageServer) readLoop() {
	timer := time.NewTicker(p.requestLifetime + time.Millisecond*100)
	defer timer.Stop()
	for {
		select {
		case <-p.exit:
			p.Debug("shutting down protocol ", p.name)
			close(p.exit)
			return
		case <-timer.C:
			go p.cleanStaleMessages()
		case msg, ok := <-p.ingressChannel:
			p.Debug("new msg received from channel")
			if !ok {
				p.Error("read loop channel was closed")
				return
			}
			p.workerCount.Add(1)
			p.workerLimiter <- struct{}{}
			go func(msg Message) {
				p.handleMessage(msg)
				<-p.workerLimiter
				p.workerCount.Done()
			}(msg.(Message))
		}
	}
}
readLoop reads incoming messages and matches them to requests or responses.
package server
import (
	"container/list"
	"runtime"
	"sync"
	"sync/atomic"
	"time"
	"github.com/libonomy/libonomy-p2p/log"
	"github.com/libonomy/libonomy-p2p/p2p/p2pcrypto"
	"github.com/libonomy/libonomy-p2p/p2p/service"
)
type MessageType uint32
type Message interface {
	service.DirectMessage
	Data() service.Data
}
func extractPayload(m Message) []byte {
	data := m.Data().(*service.DataMsgWrapper)
	return data.Payload
}
type Item struct {
	id        uint64
	timestamp time.Time
}
type MessageServer struct {
	log.Log
	ReqID              uint64  
	name               string  
	network            Service
	pendMutex          sync.RWMutex
	pendingQueue       *list.List                                    
	resHandlers        map[uint64]func(msg []byte)                   
	msgRequestHandlers map[MessageType]func(message Message) []byte  
	ingressChannel     chan service.DirectMessage                    
	requestLifetime    time.Duration                                 
	workerCount        sync.WaitGroup
	workerLimiter      chan struct{}
	exit               chan struct{}
}
type Service interface {
	RegisterDirectProtocolWithChannel(protocol string, ingressChannel chan service.DirectMessage) chan service.DirectMessage
	SendWrappedMessage(nodeID p2pcrypto.PublicKey, protocol string, payload *service.DataMsgWrapper) error
}
func NewMsgServer(network Service, name string, requestLifetime time.Duration, c chan service.DirectMessage, logger log.Log) *MessageServer {
	p := &MessageServer{
		Log:                logger,
		name:               name,
		resHandlers:        make(map[uint64]func(msg []byte)),
		pendingQueue:       list.New(),
		network:            network,
		ingressChannel:     network.RegisterDirectProtocolWithChannel(name, c),
		msgRequestHandlers: make(map[MessageType]func(message Message) []byte),
		requestLifetime:    requestLifetime,
		exit:               make(chan struct{}),
		workerLimiter:      make(chan struct{}, runtime.NumCPU()),
	}
	go p.readLoop()
	return p
}
func (p *MessageServer) Close() {
	p.exit <- struct{}{}
	<-p.exit
	p.workerCount.Wait()
}