func (s *Switch) SendWrappedMessage(nodeID p2pcrypto.PublicKey, protocol string, payload *service.DataMsgWrapper) error {
	return s.sendMessageImpl(nodeID, protocol, payload)
}
SendWrappedMessage sends a wrapped message in order to differentiate between request response and sub protocol messages.
It is used by `MessageServer`.
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
}