func (r *Router) OpenChannel(chDesc *ChannelDescriptor) (*Channel, error) {
	r.channelMtx.Lock()
	defer r.channelMtx.Unlock()
	id := chDesc.ID
	if _, ok := r.channelQueues[id]; ok {
		return nil, fmt.Errorf("channel %v already exists", id)
	}
	r.chDescs = append(r.chDescs, chDesc)
	messageType := chDesc.MessageType
	queue := r.queueFactory(chDesc.RecvBufferCapacity)
	outCh := make(chan Envelope, chDesc.RecvBufferCapacity)
	errCh := make(chan PeerError, chDesc.RecvBufferCapacity)
	channel := NewChannel(id, messageType, queue.dequeue(), outCh, errCh)
	var wrapper Wrapper
	if w, ok := messageType.(Wrapper); ok {
		wrapper = w
	}
	r.channelQueues[id] = queue
	r.channelMessages[id] = messageType
	r.nodeInfo.AddChannel(uint16(chDesc.ID))
	for _, t := range r.transports {
		t.AddChannelDescriptors([]*ChannelDescriptor{chDesc})
	}
	go func() {
		defer func() {
			r.channelMtx.Lock()
			delete(r.channelQueues, id)
			delete(r.channelMessages, id)
			r.channelMtx.Unlock()
			queue.close()
		}()
		r.routeChannel(id, outCh, errCh, wrapper)
	}()
	return channel, nil
}
OpenChannel opens a new channel for the given message type. The caller must
close the channel when done, before stopping the Router. messageType is the
type of message passed through the channel (used for unmarshaling), which can
implement Wrapper to automatically (un)wrap multiple message types in a
wrapper message. The caller may provide a size to make the channel buffered,
which internally makes the inbound, outbound, and error channel buffered.
package p2p
import (
	"context"
	"errors"
	"fmt"
	"io"
	"math/rand"
	"net"
	"runtime"
	"sync"
	"time"
	"github.com/gogo/protobuf/proto"
	"github.com/tendermint/tendermint/crypto"
	"github.com/tendermint/tendermint/libs/log"
	"github.com/tendermint/tendermint/libs/service"
	"github.com/tendermint/tendermint/types"
)
const queueBufferDefault = 32
const dialRandomizerIntervalMillisecond = 3000
type Envelope struct {
	From      types.NodeID   
	To        types.NodeID   
	Broadcast bool           
	Message   proto.Message  
	channelID ChannelID
}
type PeerError struct {
	NodeID types.NodeID
	Err    error
}
type Channel struct {
	ID    ChannelID
	In    <-chan Envelope   
	Out   chan<- Envelope   
	Error chan<- PeerError  
	messageType proto.Message  
	closeCh     chan struct{}
	closeOnce   sync.Once
}
func NewChannel(
	id ChannelID,
	messageType proto.Message,
	inCh <-chan Envelope,
	outCh chan<- Envelope,
	errCh chan<- PeerError,
) *Channel {
	return &Channel{
		ID:          id,
		messageType: messageType,
		In:          inCh,
		Out:         outCh,
		Error:       errCh,
		closeCh:     make(chan struct{}),
	}
}
func (c *Channel) Close() {
	c.closeOnce.Do(func() {
		close(c.closeCh)
		close(c.Out)
		close(c.Error)
	})
}
func (c *Channel) Done() <-chan struct{} {
	return c.closeCh
}
type Wrapper interface {
	proto.Message
	Wrap(proto.Message) error
	Unwrap() (proto.Message, error)
}
type RouterOptions struct {
	ResolveTimeout time.Duration
	DialTimeout time.Duration
	HandshakeTimeout time.Duration
	QueueType string
	MaxIncomingConnectionAttempts uint
	IncomingConnectionWindow time.Duration
	FilterPeerByIP func(context.Context, net.IP, uint16) error
	FilterPeerByID func(context.Context, types.NodeID) error
	DialSleep func(context.Context)
	NumConcurrentDials func() int
}
const (
	queueTypeFifo     = "fifo"
	queueTypePriority = "priority"
)
func (o *RouterOptions) Validate() error {
	switch o.QueueType {
	case "":
		o.QueueType = queueTypeFifo
	case queueTypeFifo, queueTypePriority:
	default:
		return fmt.Errorf("queue type %q is not supported", o.QueueType)
	}
	switch {
	case o.IncomingConnectionWindow == 0:
		o.IncomingConnectionWindow = 100 * time.Millisecond
	case o.IncomingConnectionWindow < time.Millisecond:
		return fmt.Errorf("incomming connection window must be grater than 1m [%s]",
			o.IncomingConnectionWindow)
	}
	if o.MaxIncomingConnectionAttempts == 0 {
		o.MaxIncomingConnectionAttempts = 100
	}
	return nil
}
type Router struct {
	*service.BaseService
	logger             log.Logger
	metrics            *Metrics
	options            RouterOptions
	nodeInfo           types.NodeInfo
	privKey            crypto.PrivKey
	peerManager        *PeerManager
	chDescs            []*ChannelDescriptor
	transports         []Transport
	connTracker        connectionTracker
	protocolTransports map[Protocol]Transport
	stopCh             chan struct{}  
	peerMtx    sync.RWMutex
	peerQueues map[types.NodeID]queue  
	peerChannels map[types.NodeID]channelIDs
	queueFactory func(int) queue
	channelMtx      sync.RWMutex
	channelQueues   map[ChannelID]queue  
	channelMessages map[ChannelID]proto.Message
}
func NewRouter(
	logger log.Logger,
	metrics *Metrics,
	nodeInfo types.NodeInfo,
	privKey crypto.PrivKey,
	peerManager *PeerManager,
	transports []Transport,
	options RouterOptions,
) (*Router, error) {
	if err := options.Validate(); err != nil {
		return nil, err
	}
	router := &Router{
		logger:   logger,
		metrics:  metrics,
		nodeInfo: nodeInfo,
		privKey:  privKey,
		connTracker: newConnTracker(
			options.MaxIncomingConnectionAttempts,
			options.IncomingConnectionWindow,
		),
		chDescs:            make([]*ChannelDescriptor, 0),
		transports:         transports,
		protocolTransports: map[Protocol]Transport{},
		peerManager:        peerManager,
		options:            options,
		stopCh:             make(chan struct{}),
		channelQueues:      map[ChannelID]queue{},
		channelMessages:    map[ChannelID]proto.Message{},
		peerQueues:         map[types.NodeID]queue{},
		peerChannels:       make(map[types.NodeID]channelIDs),
	}
	router.BaseService = service.NewBaseService(logger, "router", router)
	qf, err := router.createQueueFactory()
	if err != nil {
		return nil, err
	}
	router.queueFactory = qf
	for _, transport := range transports {
		for _, protocol := range transport.Protocols() {
			if _, ok := router.protocolTransports[protocol]; !ok {
				router.protocolTransports[protocol] = transport
			}
		}
	}
	return router, nil
}
func (r *Router) createQueueFactory() (func(int) queue, error) {
	switch r.options.QueueType {
	case queueTypeFifo:
		return newFIFOQueue, nil
	case queueTypePriority:
		return func(size int) queue {
			if size%2 != 0 {
				size++
			}
			q := newPQScheduler(r.logger, r.metrics, r.chDescs, uint(size)/2, uint(size)/2, defaultCapacity)
			q.start()
			return q
		}, nil
	default:
		return nil, fmt.Errorf("cannot construct queue of type %q", r.options.QueueType)
	}
}