func (s *Switch) listenToNetworkMessages() {
	netqueues := s.network.IncomingMessages()
	for nq := range netqueues {  
		go func(c chan net.IncomingMessageEvent) {
			for {
				select {
				case msg := <-c:
					s.processMessage(msg)
				case <-s.shutdown:
					return
				}
			}
		}(netqueues[nq])
	}
}
listenToNetworkMessages is listening on new messages from the opened connections and processes them.
package p2p
import (
	"context"
	"errors"
	"fmt"
	"strings"
	"github.com/libonomy/libonomy-p2p/common/types"
	"github.com/libonomy/libonomy-p2p/log"
	"github.com/libonomy/libonomy-p2p/nattraversal"
	"github.com/libonomy/libonomy-p2p/p2p/config"
	"github.com/libonomy/libonomy-p2p/p2p/connectionpool"
	"github.com/libonomy/libonomy-p2p/p2p/discovery"
	"github.com/libonomy/libonomy-p2p/p2p/gossip"
	"github.com/libonomy/libonomy-p2p/p2p/metrics"
	"github.com/libonomy/libonomy-p2p/p2p/net"
	"github.com/libonomy/libonomy-p2p/p2p/node"
	"github.com/libonomy/libonomy-p2p/p2p/p2pcrypto"
	"github.com/libonomy/libonomy-p2p/p2p/service"
	"github.com/libonomy/libonomy-p2p/priorityq"
	"github.com/libonomy/libonomy-p2p/timesync"
	inet "net"
	"sync"
	"sync/atomic"
	"time"
)
const ConnectingTimeout = 20 * time.Second  
const UPNPRetries = 20
type cPool interface {
	GetConnection(address inet.Addr, pk p2pcrypto.PublicKey) (net.Connection, error)
	GetConnectionIfExists(pk p2pcrypto.PublicKey) (net.Connection, error)
	CloseConnection(key p2pcrypto.PublicKey)
	Shutdown()
}
type Switch struct {
	started   uint32
	bootErr   error
	bootChan  chan struct{}
	gossipErr error
	gossipC   chan struct{}
	config    config.Config
	logger    log.Log
	ctx        context.Context
	cancelFunc context.CancelFunc
	shutdownOnce sync.Once
	shutdown     chan struct{}  
	lNode node.LocalNode
	directProtocolHandlers map[string]chan service.DirectMessage
	gossipProtocolHandlers map[string]chan service.GossipMessage
	protocolHandlerMutex   sync.RWMutex
	network    *net.Net     
	udpnetwork *net.UDPNet  
	cPool cPool
	gossip *gossip.Protocol
	discover discovery.PeerStore  
	udpServer *UDPMux
	initOnce sync.Once
	initial  chan struct{}
	outpeersMutex sync.RWMutex
	inpeersMutex  sync.RWMutex
	outpeers      map[p2pcrypto.PublicKey]struct{}
	inpeers       map[p2pcrypto.PublicKey]struct{}
	morePeersReq      chan struct{}
	connectingTimeout time.Duration
	peerLock   sync.RWMutex
	newPeerSub []chan p2pcrypto.PublicKey
	delPeerSub []chan p2pcrypto.PublicKey
	releaseUpnp func()
}
func (s *Switch) waitForBoot() error {
	_, ok := <-s.bootChan
	if !ok {
		return s.bootErr
	}
	return nil
}
func (s *Switch) waitForGossip() error {
	_, ok := <-s.gossipC
	if !ok {
		return s.gossipErr
	}
	return nil
}
func newSwarm(ctx context.Context, config config.Config, logger log.Log, datadir string) (*Switch, error) {
	var l node.LocalNode
	var err error
	if config.NodeID != "" {
		l, err = node.LoadIdentity(datadir, config.NodeID)
	} else {
		l, err = node.ReadFirstNodeData(datadir)
		if err != nil {
			logger.Warning("failed to load p2p identity from disk at %v. creating a new identity, err:%v", datadir, err)
			l, err = node.NewNodeIdentity()
		}
	}
	if err != nil {
		return nil, err
	}
	logger.Info("Local node identity >> %v", l.PublicKey().String())
	logger = logger.WithFields(log.String("P2PID", l.PublicKey().String()))
	if datadir != "" {
		err := l.PersistData(datadir)
		if err != nil {
			logger.Warning("failed to persist p2p node data err=%v", err)
		}
	}
	n, err := net.NewNet(config, l, logger.WithName("tcpnet"))
	if err != nil {
		return nil, fmt.Errorf("can't create Switch without a network, err: %v", err)
	}
	udpnet, err := net.NewUDPNet(config, l, logger.WithName("udpnet"))
	if err != nil {
		return nil, err
	}
	localCtx, cancel := context.WithCancel(ctx)
	s := &Switch{
		ctx:        localCtx,
		cancelFunc: cancel,
		config: config,
		logger: logger,
		lNode: l,
		bootChan: make(chan struct{}),
		gossipC:  make(chan struct{}),
		shutdown: make(chan struct{}),  
		initial:           make(chan struct{}),
		morePeersReq:      make(chan struct{}, config.MaxInboundPeers+config.OutboundPeersTarget),
		inpeers:           make(map[p2pcrypto.PublicKey]struct{}),
		outpeers:          make(map[p2pcrypto.PublicKey]struct{}),
		newPeerSub:        make([]chan p2pcrypto.PublicKey, 0, 10),
		delPeerSub:        make([]chan p2pcrypto.PublicKey, 0, 10),
		connectingTimeout: ConnectingTimeout,
		directProtocolHandlers: make(map[string]chan service.DirectMessage),
		gossipProtocolHandlers: make(map[string]chan service.GossipMessage),
		network:    n,
		udpnetwork: udpnet,
	}
	mux := NewUDPMux(s.lNode, s.lookupFunc, udpnet, s.config.NetworkID, s.logger)
	s.udpServer = mux
	s.discover = discovery.New(l, config.SwarmConfig, s.udpServer, datadir, s.logger)  
	cpool := connectionpool.NewConnectionPool(s.network.Dial, l.PublicKey(), logger)
	s.network.SubscribeOnNewRemoteConnections(func(nce net.NewConnectionEvent) {
		err := cpool.OnNewConnection(nce)
		if err != nil {
			s.logger.Warning("adding incoming connection err=", err)
			return
		}
		s.onNewConnection(nce)
	})
	s.network.SubscribeClosingConnections(cpool.OnClosedConnection)
	s.network.SubscribeClosingConnections(s.onClosedConnection)
	s.cPool = cpool
	s.gossip = gossip.NewProtocol(config.SwarmConfig, s, s.LocalNode().PublicKey(), s.logger)
	s.logger.Debug("Created newSwarm with key %s", l.PublicKey())
	return s, nil
}
func (s *Switch) lookupFunc(target p2pcrypto.PublicKey) (*node.Info, error) {
	return s.discover.Lookup(target)
}
func (s *Switch) onNewConnection(nce net.NewConnectionEvent) {
	err := s.addIncomingPeer(nce.Node.PublicKey())
	if err != nil {
		s.logger.Warning("Error adding new connection %v, err: %v", nce.Node.PublicKey(), err)
		s.cPool.CloseConnection(nce.Node.PublicKey())
	}
}
func (s *Switch) onClosedConnection(cwe net.ConnectionWithErr) {
	s.Disconnect(cwe.Conn.RemotePublicKey())
}
func (s *Switch) Start() error {
	if atomic.LoadUint32(&s.started) == 1 {
		return errors.New("Switch already running")
	}
	var err error
	atomic.StoreUint32(&s.started, 1)
	s.logger.Debug("Starting the p2p layer")
	tcpListener, udpListener, err := s.getListeners(getTCPListener, getUDPListener, discoverUPnPGateway)
	if err != nil {
		return fmt.Errorf("error getting port: %v", err)
	}
	s.network.Start(tcpListener)
	s.udpnetwork.Start(udpListener)
	tcpAddress := s.network.LocalAddr().(*inet.TCPAddr)
	udpAddress := s.udpnetwork.LocalAddr().(*inet.UDPAddr)
	s.discover.SetLocalAddresses(tcpAddress.Port, udpAddress.Port)  
	err = s.udpServer.Start()
	if err != nil {
		return err
	}
	s.logger.Debug("Starting to listen for network messages")
	s.listenToNetworkMessages()  
	s.logger.Debug("starting the udp server")
	if s.config.SwarmConfig.Bootstrap {
		go func() {
			b := time.Now()
			err := s.discover.Bootstrap(s.ctx)
			if err != nil {
				s.bootErr = err
				close(s.bootChan)
				s.Shutdown()
				return
			}
			close(s.bootChan)
			size := s.discover.Size()
			s.logger.Event().Info("discovery_bootstrap", log.Bool("success", size >= s.config.SwarmConfig.RandomConnections && s.bootErr == nil),
				log.Int("size", size), log.Duration("time_elapsed", time.Since(b)))
		}()
	}
	if s.config.SwarmConfig.Gossip {
		s.gossip.Start()
		go func() {
			if s.config.SwarmConfig.Bootstrap {
				if err := s.waitForBoot(); err != nil {
					return
				}
			}
			err := s.startNeighborhood()  
			if err != nil {
				s.gossipErr = err
				close(s.gossipC)
				s.Shutdown()
				return
			}
			<-s.initial
			close(s.gossipC)
		}()  
	}
	return nil
}
func (s *Switch) LocalNode() node.LocalNode {
	return s.lNode
}
func (s *Switch) SendWrappedMessage(nodeID p2pcrypto.PublicKey, protocol string, payload *service.DataMsgWrapper) error {
	return s.sendMessageImpl(nodeID, protocol, payload)
}
func (s *Switch) SendMessage(peerPubkey p2pcrypto.PublicKey, protocol string, payload []byte) error {
	return s.sendMessageImpl(peerPubkey, protocol, service.DataBytes{Payload: payload})
}
func (s *Switch) sendMessageImpl(peerPubKey p2pcrypto.PublicKey, protocol string, payload service.Data) error {
	var err error
	var conn net.Connection
	if s.discover.IsLocalAddress(&node.Info{ID: peerPubKey.Array()}) {
		return errors.New("can't sent message to self")
	}
	conn, err = s.cPool.GetConnectionIfExists(peerPubKey)
	if err != nil {
		return errors.New("this peers isn't a neighbor or lost connection")
	}
	session := conn.Session()
	if session == nil {
		s.logger.Warning("failed to send message to %v, no valid session.", conn.RemotePublicKey().String())
		return ErrNoSession
	}
	protomessage := &ProtocolMessage{
		Metadata: &ProtocolMessageMetadata{NextProtocol: protocol, ClientVersion: config.ClientVersion,
			Timestamp: time.Now().Unix(), AuthPubkey: s.LocalNode().PublicKey().Bytes()},
		Payload: nil,
	}
	realpayload, err := CreatePayload(payload)
	if err != nil {
		return err
	}
	protomessage.Payload = realpayload
	data, err := types.InterfaceToBytes(protomessage)
	if err != nil {
		return fmt.Errorf("failed to encode signed message err: %v", err)
	}
	final := session.SealMessage(data)
	if final == nil {
		return errors.New("encryption failed")
	}
	err = conn.Send(final)
	s.logger.Debug("DirectMessage sent successfully")
	return err
}
func (s *Switch) RegisterDirectProtocol(protocol string) chan service.DirectMessage {  
	mchan := make(chan service.DirectMessage, s.config.BufferSize)
	s.protocolHandlerMutex.Lock()
	s.directProtocolHandlers[protocol] = mchan
	s.protocolHandlerMutex.Unlock()
	return mchan
}
func (s *Switch) RegisterGossipProtocol(protocol string, prio priorityq.Priority) chan service.GossipMessage {
	mchan := make(chan service.GossipMessage, s.config.BufferSize)
	s.protocolHandlerMutex.Lock()
	s.gossip.SetPriority(protocol, prio)
	s.gossipProtocolHandlers[protocol] = mchan
	s.protocolHandlerMutex.Unlock()
	return mchan
}
func (s *Switch) Shutdown() {
	s.shutdownOnce.Do(func() {
		s.cancelFunc()
		close(s.shutdown)
		s.gossip.Close()
		s.discover.Shutdown()
		s.cPool.Shutdown()
		s.network.Shutdown()
		s.udpServer.Shutdown()
		s.protocolHandlerMutex.Lock()
		for i := range s.directProtocolHandlers {
			delete(s.directProtocolHandlers, i)
		}
		for i := range s.gossipProtocolHandlers {
			delete(s.gossipProtocolHandlers, i)
		}
		s.protocolHandlerMutex.Unlock()
		s.peerLock.Lock()
		for _, ch := range s.newPeerSub {
			close(ch)
		}
		for _, ch := range s.delPeerSub {
			close(ch)
		}
		s.delPeerSub = nil
		s.newPeerSub = nil
		s.peerLock.Unlock()
		if s.releaseUpnp != nil {
			s.releaseUpnp()
		}
	})
}
func (s *Switch) processMessage(ime net.IncomingMessageEvent) {
	if s.config.MsgSizeLimit != config.UnlimitedMsgSize && len(ime.Message) > s.config.MsgSizeLimit {
		s.logger.With().Error("processMessage: message is too big",
			log.Int("limit", s.config.MsgSizeLimit), log.Int("actual", len(ime.Message)))
		return
	}
	err := s.onRemoteClientMessage(ime)
	if err != nil {
		s.logger.Error("Err reading message from %v, closing connection err=%v", ime.Conn.RemotePublicKey(), err)
		if err := ime.Conn.Close(); err == nil {
			s.cPool.CloseConnection(ime.Conn.RemotePublicKey())
			s.Disconnect(ime.Conn.RemotePublicKey())
		}
	}
}
func (s *Switch) RegisterDirectProtocolWithChannel(protocol string, ingressChannel chan service.DirectMessage) chan service.DirectMessage {
	s.protocolHandlerMutex.Lock()
	s.directProtocolHandlers[protocol] = ingressChannel
	s.protocolHandlerMutex.Unlock()
	return ingressChannel
}