func (t *tun) manageLink(link *link) {
	keepalive := time.NewTicker(KeepAliveTime)
	defer keepalive.Stop()
	discover := time.NewTicker(DiscoverTime)
	defer discover.Stop()
	wait := func(d time.Duration) {
		j := rand.Int63n(int64(d.Seconds() / 2.0))
		time.Sleep(time.Duration(j) * time.Second)
	}
	for {
		select {
		case <-t.closed:
			return
		case <-link.closed:
			return
		case <-discover.C:
			wait(DiscoverTime)
			if logger.V(logger.DebugLevel, log) {
				log.Debugf("Tunnel sending discover to link: %v", link.Remote())
			}
			if err := t.sendMsg("discover", link); err != nil {
				if logger.V(logger.DebugLevel, log) {
					log.Debugf("Tunnel failed to send discover to link %s: %v", link.Remote(), err)
				}
			}
		case <-keepalive.C:
			wait(KeepAliveTime)
			if logger.V(logger.DebugLevel, log) {
				log.Debugf("Tunnel sending keepalive to link: %v", link.Remote())
			}
			if err := t.sendMsg("keepalive", link); err != nil {
				if logger.V(logger.DebugLevel, log) {
					log.Debugf("Tunnel error sending keepalive to link %v: %v", link.Remote(), err)
				}
				t.delLink(link.Remote())
				return
			}
		}
	}
}
manageLink sends channel discover requests periodically and
keepalive messages to link
package mucp
import (
	"errors"
	"math/rand"
	"strings"
	"sync"
	"time"
	"github.com/google/uuid"
	"github.com/micro/micro/v3/service/logger"
	"github.com/micro/micro/v3/service/network/transport"
	"github.com/micro/micro/v3/service/network/tunnel"
)
var (
	DiscoverTime = 30 * time.Second
	KeepAliveTime = 30 * time.Second
	ReconnectTime = 5 * time.Second
	log = logger.NewHelper(logger.DefaultLogger).WithFields(map[string]interface{}{"package": "tunnel"})
)
type tun struct {
	options tunnel.Options
	sync.RWMutex
	id string
	token string
	connected bool
	send chan *message
	closed chan bool
	sessions map[string]*session
	links map[string]*link
	listener transport.Listener
}
func NewTunnel(opts ...tunnel.Option) *tun {
	rand.Seed(time.Now().UnixNano())
	options := tunnel.DefaultOptions()
	for _, o := range opts {
		o(&options)
	}
	return &tun{
		options:  options,
		id:       options.Id,
		token:    options.Token,
		send:     make(chan *message, 128),
		closed:   make(chan bool),
		sessions: make(map[string]*session),
		links:    make(map[string]*link),
	}
}
func (t *tun) Init(opts ...tunnel.Option) error {
	t.Lock()
	for _, o := range opts {
		o(&t.options)
	}
	t.Unlock()
	return nil
}
func (t *tun) getSession(channel, session string) (*session, bool) {
	t.RLock()
	s, ok := t.sessions[channel+session]
	t.RUnlock()
	return s, ok
}
func (t *tun) delSession(channel, session string) {
	t.Lock()
	if s, ok := t.sessions[channel+session]; ok {
		s.Close()
	}
	delete(t.sessions, channel+session)
	t.Unlock()
}
func (t *tun) listChannels() []string {
	t.RLock()
	var channels []string
	for _, session := range t.sessions {
		if session.session != "listener" {
			continue
		}
		channels = append(channels, session.channel)
	}
	t.RUnlock()
	return channels
}
func (t *tun) newSession(channel, sessionId string) (*session, bool, error) {
	s := &session{
		tunnel:  t.id,
		channel: channel,
		session: sessionId,
		token:   t.token,
		closed:  make(chan bool),
		recv:    make(chan *message, 128),
		send:    t.send,
		errChan: make(chan error, 1),
		key:     []byte(t.token + channel + sessionId),
	}
	gcm, err := newCipher(s.key)
	if err != nil {
		return nil, false, err
	}
	s.gcm = gcm
	t.Lock()
	_, ok := t.sessions[channel+sessionId]
	if ok {
		t.Unlock()
		return nil, false, nil
	}
	t.sessions[channel+sessionId] = s
	t.Unlock()
	return s, true, nil
}
func (t *tun) newSessionId() string {
	return uuid.New().String()
}
func (t *tun) announce(channel, session string, link *link) {
	msg := &transport.Message{
		Header: map[string]string{
			"Micro-Tunnel":         "announce",
			"Micro-Tunnel-Id":      t.id,
			"Micro-Tunnel-Channel": channel,
			"Micro-Tunnel-Session": session,
			"Micro-Tunnel-Link":    link.id,
		},
	}
	if len(channel) == 0 {
		channels := t.listChannels()
		if len(channels) == 0 {
			return
		}
		channel = strings.Join(channels, ",")
		msg.Header["Micro-Tunnel-Channel"] = channel
	} else {
		_, exists := t.getSession(channel, "listener")
		if !exists {
			return
		}
	}
	if logger.V(logger.TraceLevel, log) {
		log.Debugf("Tunnel sending announce for discovery of channel(s) %s", channel)
	}
	if err := link.Send(msg); err != nil {
		if logger.V(logger.DebugLevel, log) {
			log.Debugf("Tunnel failed to send announcement for channel(s) %s message: %v", channel, err)
		}
	}
}
func (t *tun) manage(reconnect time.Duration) {
	r := time.NewTicker(reconnect)
	defer r.Stop()
	for {
		select {
		case <-t.closed:
			return
		case <-r.C:
			t.manageLinks()
		}
	}
}