去除连接池的支持

code/client/app/app.go

@@ -2,11 +2,10 @@ package app
 
 import (
 	rt "runtime"
-	"time"
 
+	"github.com/jkstack/natpass/code/client/conn"
 	"github.com/jkstack/natpass/code/client/dashboard"
 	"github.com/jkstack/natpass/code/client/global"
-	"github.com/jkstack/natpass/code/client/pool"
 	"github.com/jkstack/natpass/code/client/rule"
 	"github.com/jkstack/natpass/code/client/rule/bench"
 	"github.com/jkstack/natpass/code/client/rule/shell"
@@ -22,6 +21,7 @@ type App struct {
 	confDir string
 	cfg     *global.Configure
 	version string
+	conn    *conn.Conn
 }
 
 // New create application
@@ -52,67 +52,53 @@ func (a *App) run() {
 	logging.SetSizeRotate(a.cfg.LogDir, "np-cli", int(a.cfg.LogSize.Bytes()), a.cfg.LogRotate, stdout)
 	defer logging.Flush()
 
-	pl := pool.New(a.cfg)
+	a.conn = conn.New(a.cfg)
 	mgr := rule.New()
 
 	for _, t := range a.cfg.Rules {
 		switch t.Type {
 		case "shell":
-			sh := shell.New(t)
+			sh := shell.New(t, a.cfg.ReadTimeout, a.cfg.WriteTimeout)
 			mgr.Add(sh)
-			go sh.Handle(pl)
+			go sh.Handle(a.conn)
 		case "vnc":
-			v := vnc.New(t)
+			v := vnc.New(t, a.cfg.ReadTimeout, a.cfg.WriteTimeout)
 			mgr.Add(v)
-			go v.Handle(pl)
+			go v.Handle(a.conn)
 		case "bench":
 			b := bench.New(t)
 			mgr.Add(b)
-			go b.Handle(pl)
+			go b.Handle(a.conn)
 		}
 	}
 
-	for i := 0; i < a.cfg.Links-pl.Size(); i++ {
-		go func() {
-			for {
-				conn := pl.Get()
-				if conn == nil {
-					time.Sleep(time.Second)
-					continue
+	go func() {
+		for {
+			msg := <-a.conn.ChanUnknown()
+			var linkID string
+			switch msg.GetXType() {
+			case network.Msg_connect_req:
+				switch msg.GetCreq().GetXType() {
+				case network.ConnectRequest_shell:
+					a.shellCreate(mgr, a.conn, msg)
+				case network.ConnectRequest_vnc:
+					a.vncCreate(a.confDir, mgr, a.conn, msg)
+				case network.ConnectRequest_bench:
+					a.benchCreate(a.confDir, mgr, a.conn, msg)
 				}
-				for {
-					msg := <-conn.ChanUnknown()
-					if msg == nil {
-						break
-					}
-					var linkID string
-					switch msg.GetXType() {
-					case network.Msg_connect_req:
-						switch msg.GetCreq().GetXType() {
-						case network.ConnectRequest_shell:
-							a.shellCreate(mgr, conn, msg)
-						case network.ConnectRequest_vnc:
-							a.vncCreate(a.confDir, mgr, conn, msg)
-						case network.ConnectRequest_bench:
-							a.benchCreate(a.confDir, mgr, conn, msg)
-						}
-					default:
-						linkID = msg.GetLinkId()
-					}
-					if len(linkID) > 0 {
-						logging.Error("link of %s on connection %d not found, type=%s",
-							linkID, conn.Idx, msg.GetXType().String())
-						continue
-					}
-				}
-				logging.Info("connection %s-%d exited", a.cfg.ID, conn.Idx)
-				time.Sleep(time.Second)
+			default:
+				linkID = msg.GetLinkId()
 			}
-		}()
-	}
+			if len(linkID) > 0 {
+				logging.Error("link of %s not found, type=%s",
+					linkID, msg.GetXType().String())
+				continue
+			}
+		}
+	}()
 
 	if a.cfg.DashboardEnabled {
-		db := dashboard.New(a.cfg, pl, mgr, a.version)
+		db := dashboard.New(a.cfg, a.conn, mgr, a.version)
 		runtime.Assert(db.ListenAndServe(a.cfg.DashboardListen, a.cfg.DashboardPort))
 	} else {
 		select {}

code/client/app/callback.go

@@ -0,0 +1,73 @@
+package app
+
+import (
+	"github.com/jkstack/natpass/code/client/conn"
+	"github.com/jkstack/natpass/code/client/global"
+	"github.com/jkstack/natpass/code/client/rule"
+	"github.com/jkstack/natpass/code/client/rule/shell"
+	"github.com/jkstack/natpass/code/client/rule/vnc"
+	"github.com/jkstack/natpass/code/network"
+	"github.com/lwch/logging"
+)
+
+func (a *App) shellCreate(mgr *rule.Mgr, conn *conn.Conn, msg *network.Msg) {
+	create := msg.GetCreq()
+	tn := mgr.Get(create.GetName(), msg.GetFrom())
+	if tn == nil {
+		tn = shell.New(global.Rule{
+			Name:   create.GetName(),
+			Target: msg.GetFrom(),
+			Type:   "shell",
+			Exec:   create.GetCshell().GetExec(),
+			Env:    create.GetCshell().GetEnv(),
+		}, a.cfg.ReadTimeout, a.cfg.WriteTimeout)
+		mgr.Add(tn)
+	}
+	lk := tn.NewLink(msg.GetLinkId(), msg.GetFrom(), nil, conn).(*shell.Link)
+	logging.Info("create link %s for shell rule [%s] from %s to %s",
+		msg.GetLinkId(), create.GetName(),
+		msg.GetFrom(), a.cfg.ID)
+	err := lk.Exec()
+	if err != nil {
+		logging.Error("create shell failed: %v", err)
+		conn.SendConnectError(msg.GetFrom(), msg.GetLinkId(), err.Error())
+		return
+	}
+	conn.SendConnectOK(msg.GetFrom(), msg.GetLinkId())
+	lk.Forward()
+}
+
+func (a *App) vncCreate(confDir string, mgr *rule.Mgr, conn *conn.Conn, msg *network.Msg) {
+	create := msg.GetCreq()
+	tn := mgr.Get(create.GetName(), msg.GetFrom())
+	if tn == nil {
+		tn = vnc.New(global.Rule{
+			Name:   create.GetName(),
+			Target: msg.GetFrom(),
+			Type:   "vnc",
+			Fps:    create.GetCvnc().GetFps(),
+		}, a.cfg.ReadTimeout, a.cfg.WriteTimeout)
+		mgr.Add(tn)
+	}
+	lk := tn.NewLink(msg.GetLinkId(), msg.GetFrom(), nil, conn).(*vnc.Link)
+	logging.Info("create link %s for vnc rule [%s] from %s to %s",
+		msg.GetLinkId(), create.GetName(),
+		msg.GetFrom(), a.cfg.ID)
+	lk.SetQuality(create.GetCvnc().GetQuality())
+	err := lk.Fork(confDir)
+	if err != nil {
+		logging.Error("create vnc failed: %v", err)
+		conn.SendConnectError(msg.GetFrom(), msg.GetLinkId(), err.Error())
+		return
+	}
+	conn.SendConnectOK(msg.GetFrom(), msg.GetLinkId())
+	lk.Forward()
+}
+
+func (a *App) benchCreate(confDir string, mgr *rule.Mgr, conn *conn.Conn, msg *network.Msg) {
+	create := msg.GetCreq()
+	logging.Info("create link %s for bench rule [%s] from %s to %s",
+		msg.GetLinkId(), create.GetName(),
+		msg.GetFrom(), a.cfg.ID)
+	conn.SendConnectOK(msg.GetFrom(), msg.GetLinkId())
+}

code/client/app/connect.go

@@ -1,76 +0,0 @@
-package app
-
-import (
-	"github.com/jkstack/natpass/code/client/global"
-	"github.com/jkstack/natpass/code/client/pool"
-	"github.com/jkstack/natpass/code/client/rule"
-	"github.com/jkstack/natpass/code/client/rule/shell"
-	"github.com/jkstack/natpass/code/client/rule/vnc"
-	"github.com/jkstack/natpass/code/network"
-	"github.com/lwch/logging"
-)
-
-func (a *App) shellCreate(mgr *rule.Mgr, conn *pool.Conn, msg *network.Msg) {
-	create := msg.GetCreq()
-	tn := mgr.Get(create.GetName(), msg.GetFrom())
-	if tn == nil {
-		tn = shell.New(global.Rule{
-			Name:   create.GetName(),
-			Target: msg.GetFrom(),
-			Type:   "shell",
-			Exec:   create.GetCshell().GetExec(),
-			Env:    create.GetCshell().GetEnv(),
-		})
-		mgr.Add(tn)
-	}
-	lk := tn.NewLink(msg.GetLinkId(), msg.GetFrom(), msg.GetFromIdx(), nil, conn).(*shell.Link)
-	logging.Info("create link %s for shell rule [%s] from %s-%d to %s-%d",
-		msg.GetLinkId(), create.GetName(),
-		msg.GetFrom(), msg.GetFromIdx(),
-		a.cfg.ID, conn.GetIdx())
-	err := lk.Exec()
-	if err != nil {
-		logging.Error("create shell failed: %v", err)
-		conn.SendConnectError(msg.GetFrom(), msg.GetFromIdx(), msg.GetLinkId(), err.Error())
-		return
-	}
-	conn.SendConnectOK(msg.GetFrom(), msg.GetFromIdx(), msg.GetLinkId())
-	lk.Forward()
-}
-
-func (a *App) vncCreate(confDir string, mgr *rule.Mgr, conn *pool.Conn, msg *network.Msg) {
-	create := msg.GetCreq()
-	tn := mgr.Get(create.GetName(), msg.GetFrom())
-	if tn == nil {
-		tn = vnc.New(global.Rule{
-			Name:   create.GetName(),
-			Target: msg.GetFrom(),
-			Type:   "vnc",
-			Fps:    create.GetCvnc().GetFps(),
-		})
-		mgr.Add(tn)
-	}
-	lk := tn.NewLink(msg.GetLinkId(), msg.GetFrom(), msg.GetFromIdx(), nil, conn).(*vnc.Link)
-	logging.Info("create link %s for vnc rule [%s] from %s-%d to %s-%d",
-		msg.GetLinkId(), create.GetName(),
-		msg.GetFrom(), msg.GetFromIdx(),
-		a.cfg.ID, conn.GetIdx())
-	lk.SetQuality(create.GetCvnc().GetQuality())
-	err := lk.Fork(confDir)
-	if err != nil {
-		logging.Error("create vnc failed: %v", err)
-		conn.SendConnectError(msg.GetFrom(), msg.GetFromIdx(), msg.GetLinkId(), err.Error())
-		return
-	}
-	conn.SendConnectOK(msg.GetFrom(), msg.GetFromIdx(), msg.GetLinkId())
-	lk.Forward()
-}
-
-func (a *App) benchCreate(confDir string, mgr *rule.Mgr, conn *pool.Conn, msg *network.Msg) {
-	create := msg.GetCreq()
-	logging.Info("create link %s for bench rule [%s] from %s-%d to %s-%d",
-		msg.GetLinkId(), create.GetName(),
-		msg.GetFrom(), msg.GetFromIdx(),
-		a.cfg.ID, conn.GetIdx())
-	conn.SendConnectOK(msg.GetFrom(), msg.GetFromIdx(), msg.GetLinkId())
-}

code/client/conn/conn.go

@@ -0,0 +1,167 @@
+package conn
+
+import (
+	"crypto/tls"
+	"net"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/jkstack/natpass/code/client/global"
+	"github.com/jkstack/natpass/code/network"
+	"github.com/jkstack/natpass/code/utils"
+	"github.com/lwch/logging"
+	"github.com/lwch/runtime"
+)
+
+// Conn connection
+type Conn struct {
+	sync.RWMutex
+	cfg         *global.Configure
+	conn        *network.Conn
+	read        map[string]chan *network.Msg // link id => channel
+	unknownRead chan *network.Msg            // read message without link
+	write       chan *network.Msg
+}
+
+// New new connection
+func New(cfg *global.Configure) *Conn {
+	conn := &Conn{
+		cfg:         cfg,
+		read:        make(map[string]chan *network.Msg),
+		unknownRead: make(chan *network.Msg, 1024),
+		write:       make(chan *network.Msg, 1024),
+	}
+	conn.conn = conn.connect()
+	go conn.loopRead()
+	go conn.loopWrite()
+	go conn.keepalive()
+	return conn
+}
+
+func (conn *Conn) connect() *network.Conn {
+	defer func() {
+		if err := recover(); err != nil {
+			logging.Error("connect error: %v", err)
+		}
+	}()
+	var dial net.Conn
+	var err error
+	if conn.cfg.UseSSL {
+		dial, err = tls.Dial("tcp", conn.cfg.Server, nil)
+	} else {
+		dial, err = net.Dial("tcp", conn.cfg.Server)
+	}
+	runtime.Assert(err)
+	cn := network.NewConn(dial)
+	err = writeHandshake(cn, conn.cfg)
+	runtime.Assert(err)
+	logging.Info("%s connected", conn.cfg.Server)
+	return cn
+}
+
+func writeHandshake(conn *network.Conn, cfg *global.Configure) error {
+	var msg network.Msg
+	msg.XType = network.Msg_handshake
+	msg.From = cfg.ID
+	msg.To = "server"
+	msg.Payload = &network.Msg_Hsp{
+		Hsp: &network.HandshakePayload{
+			Enc: cfg.Enc[:],
+		},
+	}
+	return conn.WriteMessage(&msg, 5*time.Second)
+}
+
+func (conn *Conn) loopRead() {
+	defer utils.Recover("loopRead")
+	var timeout int
+	for {
+		msg, _, err := conn.conn.ReadMessage(conn.cfg.ReadTimeout)
+		if err != nil {
+			if strings.Contains(err.Error(), "i/o timeout") {
+				timeout++
+				if timeout >= 60 {
+					logging.Error("too many timeout times")
+					conn.conn = conn.connect()
+					continue
+				}
+				continue
+			}
+			logging.Error("read message: %v", err)
+			conn.conn = conn.connect()
+			continue
+		}
+		timeout = 0
+		if msg.GetXType() == network.Msg_keepalive {
+			continue
+		}
+		logging.Debug("read message %s(%s) from %s",
+			msg.GetXType().String(), msg.GetLinkId(), msg.GetFrom())
+		linkID := msg.GetLinkId()
+		conn.RLock()
+		ch := conn.read[linkID]
+		conn.RUnlock()
+		if ch == nil {
+			ch = conn.unknownRead
+		}
+		select {
+		case ch <- msg:
+		case <-time.After(conn.cfg.ReadTimeout):
+			logging.Error("drop message: %s", msg.GetXType().String())
+		}
+	}
+}
+
+func (conn *Conn) loopWrite() {
+	defer utils.Recover("loopWrite")
+	for {
+		msg := <-conn.write
+		msg.From = conn.cfg.ID
+		err := conn.conn.WriteMessage(msg, conn.cfg.WriteTimeout)
+		if err != nil {
+			logging.Error("write message error on %s: %v",
+				conn.cfg.ID, err)
+			conn.conn = conn.connect()
+			continue
+		}
+	}
+}
+
+func (conn *Conn) keepalive() {
+	defer utils.Recover("keepalive")
+	for {
+		time.Sleep(10 * time.Second)
+		conn.SendKeepalive()
+	}
+}
+
+// AddLink attach read message
+func (conn *Conn) AddLink(id string) {
+	logging.Info("add link %s", id)
+	conn.Lock()
+	if _, ok := conn.read[id]; !ok {
+		conn.read[id] = make(chan *network.Msg, 10)
+	}
+	conn.Unlock()
+}
+
+// Reset reset message next read
+func (conn *Conn) Reset(id string, msg *network.Msg) {
+	conn.RLock()
+	ch := conn.read[id]
+	conn.RUnlock()
+	ch <- msg
+}
+
+// ChanRead get read channel from link id
+func (conn *Conn) ChanRead(id string) <-chan *network.Msg {
+	conn.RLock()
+	defer conn.RUnlock()
+	return conn.read[id]
+}
+
+// ChanUnknown get channel of unknown link id
+func (conn *Conn) ChanUnknown() <-chan *network.Msg {
+	return conn.unknownRead
+}

code/client/conn/send.go

@@ -0,0 +1,18 @@
+package conn
+
+import (
+	"time"
+
+	"github.com/jkstack/natpass/code/network"
+)
+
+// SendKeepalive send keepalive message
+func (conn *Conn) SendKeepalive() {
+	var msg network.Msg
+	msg.To = "server"
+	msg.XType = network.Msg_keepalive
+	select {
+	case conn.write <- &msg:
+	case <-time.After(conn.cfg.WriteTimeout):
+	}
+}

code/client/conn/send_conn.go

@@ -1,4 +1,4 @@
-package pool
+package conn
 
 import (
 	"time"
@@ -56,7 +56,7 @@ func (conn *Conn) SendConnectReq(id string, cfg global.Rule) {
 	}
 	select {
 	case conn.write <- &msg:
-	case <-time.After(conn.parent.cfg.WriteTimeout):
+	case <-time.After(conn.cfg.WriteTimeout):
 	}
 }
 
@@ -87,15 +87,29 @@ func (conn *Conn) SendConnectVnc(id string, cfg global.Rule, quality uint64, sho
 	}
 	select {
 	case conn.write <- &msg:
-	case <-time.After(conn.parent.cfg.WriteTimeout):
+	case <-time.After(conn.cfg.WriteTimeout):
+	}
+}
+
+// SendDisconnect send disconnect message
+func (conn *Conn) SendDisconnect(to string, id string) uint64 {
+	var msg network.Msg
+	msg.To = to
+	msg.XType = network.Msg_disconnect
+	msg.LinkId = id
+	select {
+	case conn.write <- &msg:
+		data, _ := proto.Marshal(&msg)
+		return uint64(len(data))
+	case <-time.After(conn.cfg.WriteTimeout):
+		return 0
 	}
 }
 
 // SendConnectError send connect error response message
-func (conn *Conn) SendConnectError(to string, toIdx uint32, id, info string) {
+func (conn *Conn) SendConnectError(to string, id, info string) {
 	var msg network.Msg
 	msg.To = to
-	msg.ToIdx = toIdx
 	msg.XType = network.Msg_connect_rep
 	msg.LinkId = id
 	msg.Payload = &network.Msg_Crep{
@@ -106,15 +120,14 @@ func (conn *Conn) SendConnectError(to string, toIdx uint32, id, info string) {
 	}
 	select {
 	case conn.write <- &msg:
-	case <-time.After(conn.parent.cfg.WriteTimeout):
+	case <-time.After(conn.cfg.WriteTimeout):
 	}
 }
 
 // SendConnectOK send connect success response message
-func (conn *Conn) SendConnectOK(to string, toIdx uint32, id string) {
+func (conn *Conn) SendConnectOK(to string, id string) {
 	var msg network.Msg
 	msg.To = to
-	msg.ToIdx = toIdx
 	msg.XType = network.Msg_connect_rep
 	msg.LinkId = id
 	msg.Payload = &network.Msg_Crep{
@@ -124,22 +137,6 @@ func (conn *Conn) SendConnectOK(to string, toIdx uint32, id string) {
 	}
 	select {
 	case conn.write <- &msg:
-	case <-time.After(conn.parent.cfg.WriteTimeout):
-	}
-}
-
-// SendDisconnect send disconnect message
-func (conn *Conn) SendDisconnect(to string, toIdx uint32, id string) uint64 {
-	var msg network.Msg
-	msg.To = to
-	msg.ToIdx = toIdx
-	msg.XType = network.Msg_disconnect
-	msg.LinkId = id
-	select {
-	case conn.write <- &msg:
-		data, _ := proto.Marshal(&msg)
-		return uint64(len(data))
-	case <-time.After(conn.parent.cfg.WriteTimeout):
-		return 0
+	case <-time.After(conn.cfg.WriteTimeout):
 	}
 }

code/client/conn/send_shell.go

@@ -1,4 +1,4 @@
-package pool
+package conn
 
 import (
 	"time"
@@ -8,7 +8,7 @@ import (
 )
 
 // SendShellData send shell data
-func (conn *Conn) SendShellData(to string, toIdx uint32, id string, data []byte) uint64 {
+func (conn *Conn) SendShellData(to string, id string, data []byte) uint64 {
 	dup := func(data []byte) []byte {
 		ret := make([]byte, len(data))
 		copy(ret, data)
@@ -16,7 +16,6 @@ func (conn *Conn) SendShellData(to string, toIdx uint32, id string, data []byte)
 	}
 	var msg network.Msg
 	msg.To = to
-	msg.ToIdx = toIdx
 	msg.XType = network.Msg_shell_data
 	msg.LinkId = id
 	msg.Payload = &network.Msg_Sdata{
@@ -28,16 +27,15 @@ func (conn *Conn) SendShellData(to string, toIdx uint32, id string, data []byte)
 	case conn.write <- &msg:
 		data, _ := proto.Marshal(&msg)
 		return uint64(len(data))
-	case <-time.After(conn.parent.cfg.WriteTimeout):
+	case <-time.After(conn.cfg.WriteTimeout):
 		return 0
 	}
 }
 
 // SendShellResize send shell resize
-func (conn *Conn) SendShellResize(to string, toIdx uint32, id string, rows, cols uint32) {
+func (conn *Conn) SendShellResize(to string, id string, rows, cols uint32) {
 	var msg network.Msg
 	msg.To = to
-	msg.ToIdx = toIdx
 	msg.XType = network.Msg_shell_resize
 	msg.LinkId = id
 	msg.Payload = &network.Msg_Sresize{
@@ -48,6 +46,6 @@ func (conn *Conn) SendShellResize(to string, toIdx uint32, id string, rows, cols
 	}
 	select {
 	case conn.write <- &msg:
-	case <-time.After(conn.parent.cfg.WriteTimeout):
+	case <-time.After(conn.cfg.WriteTimeout):
 	}
 }

code/client/conn/send_vnc.go

@@ -1,4 +1,4 @@
-package pool
+package conn
 
 import (
 	"image"
@@ -8,7 +8,7 @@ import (
 )
 
 // SendVNCImage send vnc image data
-func (conn *Conn) SendVNCImage(to string, toIdx uint32, id string, screen, rect image.Rectangle,
+func (conn *Conn) SendVNCImage(to string, id string, screen, rect image.Rectangle,
 	encode network.VncImageEncoding, data []byte) {
 	dup := func(data []byte) []byte {
 		ret := make([]byte, len(data))
@@ -17,7 +17,6 @@ func (conn *Conn) SendVNCImage(to string, toIdx uint32, id string, screen, rect
 	}
 	var msg network.Msg
 	msg.To = to
-	msg.ToIdx = toIdx
 	msg.XType = network.Msg_vnc_image
 	msg.LinkId = id
 	msg.Payload = &network.Msg_Vimg{
@@ -36,15 +35,14 @@ func (conn *Conn) SendVNCImage(to string, toIdx uint32, id string, screen, rect
 	}
 	select {
 	case conn.write <- &msg:
-	case <-time.After(conn.parent.cfg.WriteTimeout):
+	case <-time.After(conn.cfg.WriteTimeout):
 	}
 }
 
 // SendVNCCtrl send vnc config
-func (conn *Conn) SendVNCCtrl(to string, toIdx uint32, id string, quality uint64, showCursor bool) {
+func (conn *Conn) SendVNCCtrl(to string, id string, quality uint64, showCursor bool) {
 	var msg network.Msg
 	msg.To = to
-	msg.ToIdx = toIdx
 	msg.XType = network.Msg_vnc_ctrl
 	msg.LinkId = id
 	msg.Payload = &network.Msg_Vctrl{
@@ -55,12 +53,12 @@ func (conn *Conn) SendVNCCtrl(to string, toIdx uint32, id string, quality uint64
 	}
 	select {
 	case conn.write <- &msg:
-	case <-time.After(conn.parent.cfg.WriteTimeout):
+	case <-time.After(conn.cfg.WriteTimeout):
 	}
 }
 
 // SendVNCMouse send vnc mouse event
-func (conn *Conn) SendVNCMouse(to string, toIdx uint32, id string,
+func (conn *Conn) SendVNCMouse(to string, id string,
 	button, status string, x, y int) {
 	t := network.VncStatus_unset_st
 	switch status {
@@ -80,7 +78,6 @@ func (conn *Conn) SendVNCMouse(to string, toIdx uint32, id string,
 	}
 	var msg network.Msg
 	msg.To = to
-	msg.ToIdx = toIdx
 	msg.XType = network.Msg_vnc_mouse
 	msg.LinkId = id
 	msg.Payload = &network.Msg_Vmouse{
@@ -93,12 +90,12 @@ func (conn *Conn) SendVNCMouse(to string, toIdx uint32, id string,
 	}
 	select {
 	case conn.write <- &msg:
-	case <-time.After(conn.parent.cfg.WriteTimeout):
+	case <-time.After(conn.cfg.WriteTimeout):
 	}
 }
 
 // SendVNCKeyboard send vnc keyboard event
-func (conn *Conn) SendVNCKeyboard(to string, toIdx uint32, id string,
+func (conn *Conn) SendVNCKeyboard(to string, id string,
 	status, key string) {
 	t := network.VncStatus_unset_st
 	switch status {
@@ -109,7 +106,6 @@ func (conn *Conn) SendVNCKeyboard(to string, toIdx uint32, id string,
 	}
 	var msg network.Msg
 	msg.To = to
-	msg.ToIdx = toIdx
 	msg.XType = network.Msg_vnc_keyboard
 	msg.LinkId = id
 	msg.Payload = &network.Msg_Vkbd{
@@ -120,28 +116,26 @@ func (conn *Conn) SendVNCKeyboard(to string, toIdx uint32, id string,
 	}
 	select {
 	case conn.write <- &msg:
-	case <-time.After(conn.parent.cfg.WriteTimeout):
+	case <-time.After(conn.cfg.WriteTimeout):
 	}
 }
 
 // SendVNCCADEvent send vnc keyboard event
-func (conn *Conn) SendVNCCADEvent(to string, toIdx uint32, id string) {
+func (conn *Conn) SendVNCCADEvent(to string, id string) {
 	var msg network.Msg
 	msg.To = to
-	msg.ToIdx = toIdx
 	msg.XType = network.Msg_vnc_cad
 	msg.LinkId = id
 	select {
 	case conn.write <- &msg:
-	case <-time.After(conn.parent.cfg.WriteTimeout):
+	case <-time.After(conn.cfg.WriteTimeout):
 	}
 }
 
 // SendVNCScroll send vnc scroll event
-func (conn *Conn) SendVNCScroll(to string, toIdx uint32, id string, x, y int32) {
+func (conn *Conn) SendVNCScroll(to string, id string, x, y int32) {
 	var msg network.Msg
 	msg.To = to
-	msg.ToIdx = toIdx
 	msg.XType = network.Msg_vnc_scroll
 	msg.LinkId = id
 	msg.Payload = &network.Msg_Vscroll{
@@ -152,15 +146,14 @@ func (conn *Conn) SendVNCScroll(to string, toIdx uint32, id string, x, y int32)
 	}
 	select {
 	case conn.write <- &msg:
-	case <-time.After(conn.parent.cfg.WriteTimeout):
+	case <-time.After(conn.cfg.WriteTimeout):
 	}
 }
 
 // SendVNCClipboardData send vnc clipboard data
-func (conn *Conn) SendVNCClipboardData(to string, toIdx uint32, id string, set bool, data string) {
+func (conn *Conn) SendVNCClipboardData(to string, id string, set bool, data string) {
 	var msg network.Msg
 	msg.To = to
-	msg.ToIdx = toIdx
 	msg.XType = network.Msg_vnc_clipboard
 	msg.LinkId = id
 	msg.Payload = &network.Msg_Vclipboard{
@@ -174,6 +167,6 @@ func (conn *Conn) SendVNCClipboardData(to string, toIdx uint32, id string, set b
 	}
 	select {
 	case conn.write <- &msg:
-	case <-time.After(conn.parent.cfg.WriteTimeout):
+	case <-time.After(conn.cfg.WriteTimeout):
 	}
 }

code/client/dashboard/dashboard.go

@@ -5,24 +5,24 @@ import (
 	"net/http"
 	"net/http/pprof"
 
+	"github.com/jkstack/natpass/code/client/conn"
 	"github.com/jkstack/natpass/code/client/global"
-	"github.com/jkstack/natpass/code/client/pool"
 	"github.com/jkstack/natpass/code/client/rule"
 )
 
 // Dashboard dashboard object
 type Dashboard struct {
 	cfg     *global.Configure
-	pl      *pool.Pool
+	conn    *conn.Conn
 	mgr     *rule.Mgr
 	Version string
 }
 
 // New create dashboard object
-func New(cfg *global.Configure, pl *pool.Pool, mgr *rule.Mgr, version string) *Dashboard {
+func New(cfg *global.Configure, conn *conn.Conn, mgr *rule.Mgr, version string) *Dashboard {
 	return &Dashboard{
 		cfg:     cfg,
-		pl:      pl,
+		conn:    conn,
 		mgr:     mgr,
 		Version: version,
 	}

code/client/dashboard/h_info.go

@@ -16,7 +16,7 @@ func (db *Dashboard) Info(w http.ResponseWriter, r *http.Request) {
 		Session       int `json:"sessions"`
 	}
 	ret.Rules = len(db.cfg.Rules)
-	ret.PhysicalLinks = db.pl.Size()
+	ret.PhysicalLinks = 0 // TODO
 	db.mgr.Range(func(t rule.Rule) {
 		n := len(t.GetLinks())
 		ret.VirtualLinks += n

code/client/global/conf.go

@@ -52,7 +52,6 @@ func LoadConf(dir string) *Configure {
 		Secret string `yaml:"secret"`
 		SSL    bool   `yaml:"ssl"`
 		Link   struct {
-			Connections  int           `yaml:"connections"`
 			ReadTimeout  time.Duration `yaml:"read_timeout"`
 			WriteTimeout time.Duration `yaml:"write_timeout"`
 		} `yaml:"link"`
@@ -77,9 +76,6 @@ func LoadConf(dir string) *Configure {
 		}
 		cfg.Rules[i] = t
 	}
-	if cfg.Link.Connections <= 0 {
-		cfg.Link.Connections = 3
-	}
 	if cfg.Link.ReadTimeout <= 0 {
 		cfg.Link.ReadTimeout = 5 * time.Second
 	}
@@ -96,7 +92,6 @@ func LoadConf(dir string) *Configure {
 		Server:           cfg.Server,
 		UseSSL:           cfg.SSL,
 		Enc:              md5.Sum([]byte(cfg.Secret)),
-		Links:            cfg.Link.Connections,
 		ReadTimeout:      cfg.Link.ReadTimeout,
 		WriteTimeout:     cfg.Link.WriteTimeout,
 		LogDir:           cfg.Log.Dir,

code/client/pool/conn.go

@@ -1,190 +0,0 @@
-package pool
-
-import (
-	"context"
-	"strings"
-	"sync"
-	"time"
-
-	"github.com/jkstack/natpass/code/network"
-	"github.com/jkstack/natpass/code/utils"
-	"github.com/lwch/logging"
-)
-
-// Conn pool connection
-type Conn struct {
-	sync.RWMutex
-	Idx          uint32
-	ReadTimeout  time.Duration
-	WriteTimeout time.Duration
-	parent       *Pool
-	conn         *network.Conn
-	read         map[string]chan *network.Msg // link id => channel
-	unknownRead  chan *network.Msg            // read message without link
-	write        chan *network.Msg            // link id => channel
-}
-
-func newConn(parent *Pool, conn *network.Conn, idx uint32) *Conn {
-	ret := &Conn{
-		Idx:          idx,
-		ReadTimeout:  parent.cfg.ReadTimeout,
-		WriteTimeout: parent.cfg.WriteTimeout,
-		parent:       parent,
-		conn:         conn,
-		read:         make(map[string]chan *network.Msg),
-		unknownRead:  make(chan *network.Msg),
-		write:        make(chan *network.Msg),
-	}
-	logging.Info("new connection: %s-%d", ret.parent.cfg.ID, ret.Idx)
-	ctx, cancel := context.WithCancel(context.Background())
-	go ret.loopRead(cancel)
-	go ret.loopWrite(cancel)
-	go ret.keepalive(ctx)
-	return ret
-}
-
-func (conn *Conn) hasLink(id string) bool {
-	conn.RLock()
-	defer conn.RUnlock()
-	_, ok := conn.read[id]
-	return ok
-}
-
-// AddLink attach read message
-func (conn *Conn) AddLink(id string) {
-	logging.Info("add link %s from %d", id, conn.Idx)
-	conn.Lock()
-	if _, ok := conn.read[id]; !ok {
-		conn.read[id] = make(chan *network.Msg, 10)
-	}
-	conn.Unlock()
-}
-
-// RemoveLink detach read message
-func (conn *Conn) RemoveLink(id string) {
-	logging.Info("remove link %s from %d", id, conn.Idx)
-	conn.Lock()
-	ch := conn.read[id]
-	if ch != nil {
-		close(ch)
-	}
-	delete(conn.read, id)
-	conn.Unlock()
-}
-
-// Close close connection
-func (conn *Conn) Close() {
-	conn.conn.Close()
-	conn.Lock()
-	for id, ch := range conn.read {
-		close(ch)
-		delete(conn.read, id)
-	}
-	conn.Unlock()
-	if conn.unknownRead != nil {
-		close(conn.unknownRead)
-		conn.unknownRead = nil
-	}
-	if conn.write != nil {
-		close(conn.write)
-		conn.write = nil
-	}
-	conn.parent.onClose(conn.Idx)
-	logging.Error("connection %s-%d closed", conn.parent.cfg.ID, conn.Idx)
-}
-
-func (conn *Conn) loopRead(cancel context.CancelFunc) {
-	defer utils.Recover("loopRead")
-	defer conn.Close()
-	defer cancel()
-	var timeout int
-	for {
-		msg, _, err := conn.conn.ReadMessage(conn.parent.cfg.ReadTimeout)
-		if err != nil {
-			if strings.Contains(err.Error(), "i/o timeout") {
-				timeout++
-				if timeout >= 60 {
-					logging.Error("too many timeout times")
-					return
-				}
-				continue
-			}
-			logging.Error("read message: %v", err)
-			return
-		}
-		timeout = 0
-		if msg.GetXType() == network.Msg_keepalive {
-			continue
-		}
-		logging.Debug("read message %s(%s) from %s-%d",
-			msg.GetXType().String(), msg.GetLinkId(), msg.GetFrom(), msg.GetFromIdx())
-		linkID := msg.GetLinkId()
-		conn.RLock()
-		ch := conn.read[linkID]
-		conn.RUnlock()
-		if ch == nil {
-			ch = conn.unknownRead
-		}
-		select {
-		case ch <- msg:
-		case <-time.After(conn.WriteTimeout):
-		}
-	}
-}
-
-func (conn *Conn) loopWrite(cancel context.CancelFunc) {
-	defer utils.Recover("loopWrite")
-	defer conn.Close()
-	defer cancel()
-	for {
-		msg := <-conn.write
-		if msg == nil {
-			return
-		}
-		msg.From = conn.parent.cfg.ID
-		msg.FromIdx = conn.Idx
-		err := conn.conn.WriteMessage(msg, conn.parent.cfg.WriteTimeout)
-		if err != nil {
-			logging.Error("write message error on %s-%d: %v",
-				conn.parent.cfg.ID, conn.Idx, err)
-			return
-		}
-	}
-}
-
-// ChanRead get read channel from link id
-func (conn *Conn) ChanRead(id string) <-chan *network.Msg {
-	conn.RLock()
-	defer conn.RUnlock()
-	return conn.read[id]
-}
-
-// Reset reset message next read
-func (conn *Conn) Reset(id string, msg *network.Msg) {
-	conn.RLock()
-	ch := conn.read[id]
-	conn.RUnlock()
-	ch <- msg
-}
-
-// ChanUnknown get channel of unknown link id
-func (conn *Conn) ChanUnknown() <-chan *network.Msg {
-	return conn.unknownRead
-}
-
-func (conn *Conn) keepalive(ctx context.Context) {
-	defer utils.Recover("keepalive")
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-time.After(10 * time.Second):
-			conn.SendKeepalive()
-		}
-	}
-}
-
-// GetIdx get connection index
-func (conn *Conn) GetIdx() uint32 {
-	return conn.Idx
-}

code/client/pool/pool.go

@@ -1,120 +0,0 @@
-package pool
-
-import (
-	"crypto/tls"
-	"net"
-	"sync"
-	"sync/atomic"
-	"time"
-
-	"github.com/jkstack/natpass/code/client/global"
-	"github.com/jkstack/natpass/code/network"
-	"github.com/lwch/logging"
-	"github.com/lwch/runtime"
-)
-
-// Pool connection pool
-type Pool struct {
-	sync.RWMutex
-	cfg   *global.Configure
-	conns map[uint32]*Conn
-	count int
-	idx   uint32
-}
-
-// New create connection pool
-func New(cfg *global.Configure) *Pool {
-	return &Pool{
-		cfg:   cfg,
-		conns: make(map[uint32]*Conn, cfg.Links),
-		count: cfg.Links,
-		idx:   0,
-	}
-}
-
-func (p *Pool) getConns() []*Conn {
-	ret := make([]*Conn, 0, len(p.conns))
-	p.RLock()
-	for _, conn := range p.conns {
-		ret = append(ret, conn)
-	}
-	p.RUnlock()
-	return ret
-}
-
-// Get get connection
-func (p *Pool) Get(id ...string) *Conn {
-	conns := p.getConns()
-	if len(id) > 0 {
-		for _, conn := range conns {
-			if conn.hasLink(id[0]) {
-				return conn
-			}
-		}
-	}
-	if len(conns) >= p.count {
-		p.Lock()
-		conn := conns[int(p.idx)%len(conns)]
-		p.idx++
-		p.Unlock()
-		return conn
-	}
-
-	idx := atomic.AddUint32(&p.idx, 1)
-	conn := p.connect(idx)
-	if conn == nil {
-		return nil
-	}
-	c := newConn(p, conn, idx)
-
-	p.Lock()
-	p.conns[c.Idx] = c
-	p.Unlock()
-	return c
-}
-
-func (p *Pool) connect(idx uint32) *network.Conn {
-	defer func() {
-		if err := recover(); err != nil {
-			logging.Error("connect error: %v", err)
-		}
-	}()
-	var conn net.Conn
-	var err error
-	if p.cfg.UseSSL {
-		conn, err = tls.Dial("tcp", p.cfg.Server, nil)
-	} else {
-		conn, err = net.Dial("tcp", p.cfg.Server)
-	}
-	runtime.Assert(err)
-	c := network.NewConn(conn)
-	err = p.writeHandshake(c, p.cfg, idx)
-	runtime.Assert(err)
-	logging.Info("%s connected", p.cfg.Server)
-	return c
-}
-
-func (p *Pool) writeHandshake(conn *network.Conn, cfg *global.Configure, idx uint32) error {
-	var msg network.Msg
-	msg.XType = network.Msg_handshake
-	msg.From = p.cfg.ID
-	msg.FromIdx = idx
-	msg.To = "server"
-	msg.Payload = &network.Msg_Hsp{
-		Hsp: &network.HandshakePayload{
-			Enc: cfg.Enc[:],
-		},
-	}
-	return conn.WriteMessage(&msg, 5*time.Second)
-}
-
-func (p *Pool) onClose(idx uint32) {
-	p.Lock()
-	delete(p.conns, idx)
-	p.Unlock()
-}
-
-// Size get pool size
-func (p *Pool) Size() int {
-	return len(p.conns)
-}

code/client/pool/send.go

@@ -1,45 +0,0 @@
-package pool
-
-import (
-	"time"
-
-	"github.com/jkstack/natpass/code/network"
-	"google.golang.org/protobuf/proto"
-)
-
-// SendData send forward data
-func (conn *Conn) SendData(to string, toIdx uint32, id string, data []byte) uint64 {
-	dup := func(data []byte) []byte {
-		ret := make([]byte, len(data))
-		copy(ret, data)
-		return ret
-	}
-	var msg network.Msg
-	msg.To = to
-	msg.ToIdx = toIdx
-	msg.XType = network.Msg_forward
-	msg.LinkId = id
-	msg.Payload = &network.Msg_XData{
-		XData: &network.Data{
-			Data: dup(data),
-		},
-	}
-	select {
-	case conn.write <- &msg:
-		data, _ := proto.Marshal(&msg)
-		return uint64(len(data))
-	case <-time.After(conn.parent.cfg.WriteTimeout):
-		return 0
-	}
-}
-
-// SendKeepalive send keepalive message
-func (conn *Conn) SendKeepalive() {
-	var msg network.Msg
-	msg.To = "server"
-	msg.XType = network.Msg_keepalive
-	select {
-	case conn.write <- &msg:
-	case <-time.After(conn.parent.cfg.WriteTimeout):
-	}
-}

code/client/rule/bench/bench.go

@@ -5,8 +5,8 @@ import (
 	"net"
 	"net/http"
 
+	"github.com/jkstack/natpass/code/client/conn"
 	"github.com/jkstack/natpass/code/client/global"
-	"github.com/jkstack/natpass/code/client/pool"
 	"github.com/jkstack/natpass/code/client/rule"
 	"github.com/lwch/logging"
 	"github.com/lwch/runtime"
@@ -47,7 +47,7 @@ func New(cfg global.Rule) *Bench {
 }
 
 // NewLink new link
-func (bench *Bench) NewLink(id, remote string, remoteIdx uint32, localConn net.Conn, remoteConn *pool.Conn) rule.Link {
+func (bench *Bench) NewLink(id, remote string, localConn net.Conn, remoteConn *conn.Conn) rule.Link {
 	return &Link{id: id}
 }
 
@@ -82,7 +82,7 @@ func (bench *Bench) GetPort() uint16 {
 }
 
 // Handle handle shell
-func (bench *Bench) Handle(pl *pool.Pool) {
+func (bench *Bench) Handle(conn *conn.Conn) {
 	defer func() {
 		if err := recover(); err != nil {
 			logging.Error("close shell: %s, err=%v", bench.Name, err)
@@ -90,7 +90,7 @@ func (bench *Bench) Handle(pl *pool.Pool) {
 	}()
 	mux := http.NewServeMux()
 	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
-		bench.http(pl, w, r)
+		bench.http(conn, w, r)
 	})
 	svr := &http.Server{
 		Addr:    fmt.Sprintf("%s:%d", bench.cfg.LocalAddr, bench.cfg.LocalPort),

code/client/rule/bench/h_http.go

@@ -4,12 +4,12 @@ import (
 	"fmt"
 	"net/http"
 
-	"github.com/jkstack/natpass/code/client/pool"
+	"github.com/jkstack/natpass/code/client/conn"
 	"github.com/lwch/logging"
 	"github.com/lwch/runtime"
 )
 
-func (bench *Bench) http(pool *pool.Pool, w http.ResponseWriter, r *http.Request) {
+func (bench *Bench) http(conn *conn.Conn, w http.ResponseWriter, r *http.Request) {
 	id, err := runtime.UUID(16, "0123456789abcdef")
 	if err != nil {
 		logging.Error("failed to generate link_id for bench: %s, err=%v",
@@ -17,11 +17,9 @@ func (bench *Bench) http(pool *pool.Pool, w http.ResponseWriter, r *http.Request
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
-	conn := pool.Get(id)
 	conn.AddLink(id)
 	conn.SendConnectReq(id, bench.cfg)
 	ch := conn.ChanRead(id)
 	<-ch
-	conn.RemoveLink(id)
 	fmt.Fprint(w, id)
 }

code/client/rule/mgr.go

@@ -4,7 +4,7 @@ import (
 	"net"
 	"sync"
 
-	"github.com/jkstack/natpass/code/client/pool"
+	"github.com/jkstack/natpass/code/client/conn"
 )
 
 // Link link interface
@@ -18,7 +18,7 @@ type Link interface {
 
 // Rule rule interface
 type Rule interface {
-	NewLink(id, remote string, remoteIdx uint32, localConn net.Conn, remoteConn *pool.Conn) Link
+	NewLink(id, remote string, localConn net.Conn, remoteConn *conn.Conn) Link
 	GetName() string
 	GetRemote() string
 	GetPort() uint16

code/client/rule/shell/h_new.go

@@ -5,14 +5,14 @@ import (
 	"net/http"
 	"time"
 
-	"github.com/jkstack/natpass/code/client/pool"
+	"github.com/jkstack/natpass/code/client/conn"
 	"github.com/jkstack/natpass/code/network"
 	"github.com/lwch/logging"
 	"github.com/lwch/runtime"
 )
 
 // New new shell
-func (shell *Shell) New(pool *pool.Pool, w http.ResponseWriter, r *http.Request) {
+func (shell *Shell) New(conn *conn.Conn, w http.ResponseWriter, r *http.Request) {
 	id, err := runtime.UUID(16, "0123456789abcdef")
 	if err != nil {
 		logging.Error("failed to generate link_id for shell: %s, err=%v",
@@ -20,11 +20,10 @@ func (shell *Shell) New(pool *pool.Pool, w http.ResponseWriter, r *http.Request)
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
-	conn := pool.Get(id)
-	link := shell.NewLink(id, shell.cfg.Target, 0, nil, conn).(*Link)
+	link := shell.NewLink(id, shell.cfg.Target, nil, conn).(*Link)
 	conn.SendConnectReq(id, shell.cfg)
 	ch := conn.ChanRead(id)
-	timeout := time.After(conn.ReadTimeout)
+	timeout := time.After(shell.readTimeout)
 	var repMsg *network.Msg
 	for {
 		var msg *network.Msg
@@ -35,10 +34,9 @@ func (shell *Shell) New(pool *pool.Pool, w http.ResponseWriter, r *http.Request)
 			http.Error(w, "timeout", http.StatusBadGateway)
 			return
 		}
-		link.SetTargetIdx(msg.GetFromIdx())
 		if msg.GetXType() != network.Msg_connect_rep {
 			conn.Reset(id, msg)
-			time.Sleep(conn.ReadTimeout / 10)
+			time.Sleep(shell.readTimeout / 10)
 			continue
 		}
 		rep := msg.GetCrep()
@@ -51,9 +49,8 @@ func (shell *Shell) New(pool *pool.Pool, w http.ResponseWriter, r *http.Request)
 		repMsg = msg
 		break
 	}
-	logging.Info("create link %s for shell rule [%s] from %s-%d to %s-%d",
+	logging.Info("create link %s for shell rule [%s] from %s to %s",
 		link.GetID(), shell.cfg.Name,
-		repMsg.GetTo(), repMsg.GetToIdx(),
-		repMsg.GetFrom(), repMsg.GetFromIdx())
+		repMsg.GetTo(), repMsg.GetFrom())
 	fmt.Fprint(w, id)
 }

code/client/rule/shell/h_resize.go

@@ -5,11 +5,11 @@ import (
 	"net/http"
 	"strconv"
 
-	"github.com/jkstack/natpass/code/client/pool"
+	"github.com/jkstack/natpass/code/client/conn"
 )
 
 // Resize resize terminal
-func (shell *Shell) Resize(pool *pool.Pool, w http.ResponseWriter, r *http.Request) {
+func (shell *Shell) Resize(conn *conn.Conn, w http.ResponseWriter, r *http.Request) {
 	id := r.FormValue("id")
 	rows := r.FormValue("rows")
 	cols := r.FormValue("cols")

code/client/rule/shell/h_ws.go

@@ -6,7 +6,7 @@ import (
 	"sync"
 
 	"github.com/gorilla/websocket"
-	"github.com/jkstack/natpass/code/client/pool"
+	"github.com/jkstack/natpass/code/client/conn"
 	"github.com/jkstack/natpass/code/network"
 	"github.com/jkstack/natpass/code/utils"
 	"github.com/lwch/logging"
@@ -16,7 +16,7 @@ import (
 var upgrader = websocket.Upgrader{}
 
 // WS websocket for forward data
-func (shell *Shell) WS(pool *pool.Pool, w http.ResponseWriter, r *http.Request) {
+func (shell *Shell) WS(conn *conn.Conn, w http.ResponseWriter, r *http.Request) {
 	id := strings.TrimPrefix(r.URL.Path, "/ws/")
 
 	local, err := upgrader.Upgrade(w, r, nil)

code/client/rule/shell/link.go

@@ -4,7 +4,7 @@ import (
 	"io"
 	"os"
 
-	"github.com/jkstack/natpass/code/client/pool"
+	"github.com/jkstack/natpass/code/client/conn"
 	"github.com/jkstack/natpass/code/network"
 	"github.com/jkstack/natpass/code/utils"
 	"github.com/lwch/logging"
@@ -14,11 +14,10 @@ import (
 
 // Link shell link
 type Link struct {
-	parent    *Shell
-	id        string // link id
-	target    string // target id
-	targetIdx uint32 // target idx
-	remote    *pool.Conn
+	parent *Shell
+	id     string // link id
+	target string // target id
+	remote *conn.Conn
 	// in remote
 	pid    int
 	stdin  io.WriteCloser
@@ -45,11 +44,6 @@ func (link *Link) GetPackets() (uint64, uint64) {
 	return link.recvPacket, link.sendPacket
 }
 
-// SetTargetIdx set link remote index
-func (link *Link) SetTargetIdx(idx uint32) {
-	link.targetIdx = idx
-}
-
 // Close close link
 func (link *Link) Close() {
 	link.onClose()
@@ -57,7 +51,7 @@ func (link *Link) Close() {
 	if err == nil {
 		p.Kill()
 	}
-	link.remote.SendDisconnect(link.target, link.targetIdx, link.id)
+	link.remote.SendDisconnect(link.target, link.id)
 	link.parent.remove(link.id)
 }
 
@@ -79,7 +73,6 @@ func (link *Link) remoteRead() {
 		data, _ := proto.Marshal(msg)
 		link.recvBytes += uint64(len(data))
 		link.recvPacket++
-		link.targetIdx = msg.GetFromIdx()
 		switch msg.GetXType() {
 		case network.Msg_shell_resize:
 			size := msg.GetSresize()
@@ -125,7 +118,7 @@ func (link *Link) localRead() {
 		}
 		logging.Debug("link %s on shell %s read from local %d bytes",
 			link.id, link.parent.Name, n)
-		send := link.remote.SendShellData(link.target, link.targetIdx, link.id, data)
+		send := link.remote.SendShellData(link.target, link.id, data)
 		link.sendBytes += send
 		link.sendPacket++
 	}
@@ -133,12 +126,12 @@ func (link *Link) localRead() {
 
 // SendData send data
 func (link *Link) SendData(data []byte) {
-	send := link.remote.SendShellData(link.target, link.targetIdx, link.id, data)
+	send := link.remote.SendShellData(link.target, link.id, data)
 	link.sendBytes += send
 	link.sendPacket++
 }
 
 // SendResize send resize message
 func (link *Link) SendResize(rows, cols uint32) {
-	link.remote.SendShellResize(link.target, link.targetIdx, link.id, rows, cols)
+	link.remote.SendShellResize(link.target, link.id, rows, cols)
 }

code/client/rule/shell/shell.go

@@ -5,9 +5,10 @@ import (
 	"net"
 	"net/http"
 	"sync"
+	"time"
 
+	"github.com/jkstack/natpass/code/client/conn"
 	"github.com/jkstack/natpass/code/client/global"
-	"github.com/jkstack/natpass/code/client/pool"
 	"github.com/jkstack/natpass/code/client/rule"
 	"github.com/lwch/logging"
 	"github.com/lwch/runtime"
@@ -16,29 +17,32 @@ import (
 // Shell shell handler
 type Shell struct {
 	sync.RWMutex
-	Name  string
-	cfg   global.Rule
-	links map[string]*Link
+	Name         string
+	cfg          global.Rule
+	links        map[string]*Link
+	readTimeout  time.Duration
+	writeTimeout time.Duration
 }
 
 // New new shell
-func New(cfg global.Rule) *Shell {
+func New(cfg global.Rule, readTimeout, writeTimeout time.Duration) *Shell {
 	return &Shell{
-		Name:  cfg.Name,
-		cfg:   cfg,
-		links: make(map[string]*Link),
+		Name:         cfg.Name,
+		cfg:          cfg,
+		links:        make(map[string]*Link),
+		readTimeout:  readTimeout,
+		writeTimeout: writeTimeout,
 	}
 }
 
 // NewLink new link
-func (shell *Shell) NewLink(id, remote string, remoteIdx uint32, localConn net.Conn, remoteConn *pool.Conn) rule.Link {
+func (shell *Shell) NewLink(id, remote string, localConn net.Conn, remoteConn *conn.Conn) rule.Link {
 	remoteConn.AddLink(id)
 	link := &Link{
-		parent:    shell,
-		id:        id,
-		target:    remote,
-		targetIdx: remoteIdx,
-		remote:    remoteConn,
+		parent: shell,
+		id:     id,
+		target: remote,
+		remote: remoteConn,
 	}
 	shell.Lock()
 	shell.links[link.id] = link
@@ -83,15 +87,15 @@ func (shell *Shell) GetPort() uint16 {
 }
 
 // Handle handle shell
-func (shell *Shell) Handle(pl *pool.Pool) {
+func (shell *Shell) Handle(c *conn.Conn) {
 	defer func() {
 		if err := recover(); err != nil {
 			logging.Error("close shell: %s, err=%v", shell.Name, err)
 		}
 	}()
-	pf := func(cb func(*pool.Pool, http.ResponseWriter, *http.Request)) http.HandlerFunc {
+	pf := func(cb func(*conn.Conn, http.ResponseWriter, *http.Request)) http.HandlerFunc {
 		return func(w http.ResponseWriter, r *http.Request) {
-			cb(pl, w, r)
+			cb(c, w, r)
 		}
 	}
 	mux := http.NewServeMux()

code/client/rule/vnc/events.go

@@ -3,11 +3,11 @@ package vnc
 import (
 	"encoding/json"
 
-	"github.com/jkstack/natpass/code/client/pool"
+	"github.com/jkstack/natpass/code/client/conn"
 	"github.com/lwch/logging"
 )
 
-func (v *VNC) mouseEvent(remote *pool.Conn, data []byte) {
+func (v *VNC) mouseEvent(remote *conn.Conn, data []byte) {
 	var payload struct {
 		Payload struct {
 			Button string `json:"button"`
@@ -21,11 +21,11 @@ func (v *VNC) mouseEvent(remote *pool.Conn, data []byte) {
 		logging.Error("unmarshal: %v", err)
 		return
 	}
-	remote.SendVNCMouse(v.link.target, v.link.targetIdx, v.link.id,
+	remote.SendVNCMouse(v.link.target, v.link.id,
 		payload.Payload.Button, payload.Payload.Status, payload.Payload.X, payload.Payload.Y)
 }
 
-func (v *VNC) keyboardEvent(remote *pool.Conn, data []byte) {
+func (v *VNC) keyboardEvent(remote *conn.Conn, data []byte) {
 	var payload struct {
 		Payload struct {
 			Status string `json:"status"`
@@ -37,15 +37,15 @@ func (v *VNC) keyboardEvent(remote *pool.Conn, data []byte) {
 		logging.Error("unmarshal: %v", err)
 		return
 	}
-	remote.SendVNCKeyboard(v.link.target, v.link.targetIdx, v.link.id,
+	remote.SendVNCKeyboard(v.link.target, v.link.id,
 		payload.Payload.Status, payload.Payload.Key)
 }
 
-func (v *VNC) cadEvent(remote *pool.Conn) {
-	remote.SendVNCCADEvent(v.link.target, v.link.targetIdx, v.link.id)
+func (v *VNC) cadEvent(remote *conn.Conn) {
+	remote.SendVNCCADEvent(v.link.target, v.link.id)
 }
 
-func (v *VNC) scrollEvent(remote *pool.Conn, data []byte) {
+func (v *VNC) scrollEvent(remote *conn.Conn, data []byte) {
 	var payload struct {
 		Payload struct {
 			X int32 `json:"x"`
@@ -57,6 +57,6 @@ func (v *VNC) scrollEvent(remote *pool.Conn, data []byte) {
 		logging.Error("unmarshal: %v", err)
 		return
 	}
-	remote.SendVNCScroll(v.link.target, v.link.targetIdx, v.link.id,
+	remote.SendVNCScroll(v.link.target, v.link.id,
 		payload.Payload.X, payload.Payload.Y)
 }

code/client/rule/vnc/h_clipboard.go

@@ -4,36 +4,34 @@ import (
 	"fmt"
 	"net/http"
 
-	"github.com/jkstack/natpass/code/client/pool"
+	"github.com/jkstack/natpass/code/client/conn"
 )
 
 // Clipboard get/set clipboard
-func (v *VNC) Clipboard(pool *pool.Pool, w http.ResponseWriter, r *http.Request) {
+func (v *VNC) Clipboard(conn *conn.Conn, w http.ResponseWriter, r *http.Request) {
 	if r.Method == http.MethodGet {
-		v.getClipboard(pool, w, r)
+		v.getClipboard(conn, w, r)
 		return
 	}
-	v.setClipboard(pool, w, r)
+	v.setClipboard(conn, w, r)
 }
 
-func (v *VNC) getClipboard(pool *pool.Pool, w http.ResponseWriter, r *http.Request) {
+func (v *VNC) getClipboard(conn *conn.Conn, w http.ResponseWriter, r *http.Request) {
 	if v.link == nil {
 		http.NotFound(w, r)
 		return
 	}
-	conn := pool.Get(v.link.id)
-	conn.SendVNCClipboardData(v.link.target, v.link.targetIdx, v.link.id, false, "")
+	conn.SendVNCClipboardData(v.link.target, v.link.id, false, "")
 	data := <-v.chClipboard
 	fmt.Fprint(w, data.GetData())
 }
 
-func (v *VNC) setClipboard(pool *pool.Pool, w http.ResponseWriter, r *http.Request) {
+func (v *VNC) setClipboard(conn *conn.Conn, w http.ResponseWriter, r *http.Request) {
 	data := r.FormValue("data")
 	if v.link == nil {
 		http.NotFound(w, r)
 		return
 	}
-	conn := pool.Get(v.link.id)
-	conn.SendVNCClipboardData(v.link.target, v.link.targetIdx, v.link.id, true, data)
+	conn.SendVNCClipboardData(v.link.target, v.link.id, true, data)
 	fmt.Fprint(w, "ok")
 }

code/client/rule/vnc/h_ctrl.go

@@ -5,11 +5,11 @@ import (
 	"net/http"
 	"strconv"
 
-	"github.com/jkstack/natpass/code/client/pool"
+	"github.com/jkstack/natpass/code/client/conn"
 )
 
 // Ctrl change vnc rule config
-func (v *VNC) Ctrl(pool *pool.Pool, w http.ResponseWriter, r *http.Request) {
+func (v *VNC) Ctrl(conn *conn.Conn, w http.ResponseWriter, r *http.Request) {
 	q := r.FormValue("quality")
 	s := r.FormValue("show_cursor")
 	quality, err := strconv.ParseUint(q, 10, 32)
@@ -24,7 +24,6 @@ func (v *VNC) Ctrl(pool *pool.Pool, w http.ResponseWriter, r *http.Request) {
 		http.NotFound(w, r)
 		return
 	}
-	conn := pool.Get(v.link.id)
-	conn.SendVNCCtrl(v.link.target, v.link.targetIdx, v.link.id, quality, showCursor)
+	conn.SendVNCCtrl(v.link.target, v.link.id, quality, showCursor)
 	fmt.Fprint(w, "ok")
 }

code/client/rule/vnc/h_new.go

@@ -6,14 +6,14 @@ import (
 	"strconv"
 	"time"
 
-	"github.com/jkstack/natpass/code/client/pool"
+	"github.com/jkstack/natpass/code/client/conn"
 	"github.com/jkstack/natpass/code/network"
 	"github.com/lwch/logging"
 	"github.com/lwch/runtime"
 )
 
 // New new vnc
-func (v *VNC) New(pool *pool.Pool, w http.ResponseWriter, r *http.Request) {
+func (v *VNC) New(conn *conn.Conn, w http.ResponseWriter, r *http.Request) {
 	if v.link != nil {
 		v.link.close()
 	}
@@ -34,17 +34,13 @@ func (v *VNC) New(pool *pool.Pool, w http.ResponseWriter, r *http.Request) {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
-	conn := pool.Get(id)
 	if v.link != nil {
-		old := pool.Get(v.link.id)
-		if old != nil {
-			old.SendDisconnect(v.link.target, v.link.targetIdx, v.link.id)
-		}
+		conn.SendDisconnect(v.link.target, v.link.id)
 	}
 	conn.SendConnectVnc(id, v.cfg, quality, showCursor)
-	v.link = v.NewLink(id, v.cfg.Target, 0, nil, conn).(*Link)
+	v.link = v.NewLink(id, v.cfg.Target, nil, conn).(*Link)
 	ch := conn.ChanRead(id)
-	timeout := time.After(conn.ReadTimeout)
+	timeout := time.After(v.readTimeout)
 	for {
 		var msg *network.Msg
 		select {
@@ -54,10 +50,9 @@ func (v *VNC) New(pool *pool.Pool, w http.ResponseWriter, r *http.Request) {
 			http.Error(w, "timeout", http.StatusBadGateway)
 			return
 		}
-		v.link.SetTargetIdx(msg.GetFromIdx())
 		if msg.GetXType() != network.Msg_connect_rep {
 			conn.Reset(id, msg)
-			time.Sleep(conn.ReadTimeout / 10)
+			time.Sleep(v.readTimeout / 10)
 			continue
 		}
 		rep := msg.GetCrep()

code/client/rule/vnc/h_ws.go

@@ -15,7 +15,7 @@ import (
 	"sync"
 
 	"github.com/gorilla/websocket"
-	"github.com/jkstack/natpass/code/client/pool"
+	"github.com/jkstack/natpass/code/client/conn"
 	"github.com/jkstack/natpass/code/network"
 	"github.com/jkstack/natpass/code/utils"
 	"github.com/lwch/logging"
@@ -25,13 +25,8 @@ import (
 var upgrader = websocket.Upgrader{}
 
 // WS websocket handler
-func (v *VNC) WS(pool *pool.Pool, w http.ResponseWriter, r *http.Request) {
+func (v *VNC) WS(conn *conn.Conn, w http.ResponseWriter, r *http.Request) {
 	id := strings.TrimPrefix(r.URL.Path, "/ws/")
-	conn := pool.Get(id)
-	if conn == nil {
-		http.NotFound(w, r)
-		return
-	}
 	local, err := upgrader.Upgrade(w, r, nil)
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
@@ -39,7 +34,7 @@ func (v *VNC) WS(pool *pool.Pool, w http.ResponseWriter, r *http.Request) {
 	}
 	defer local.Close()
 	ch := conn.ChanRead(id)
-	defer conn.SendDisconnect(v.link.target, v.link.targetIdx, v.link.id)
+	defer conn.SendDisconnect(v.link.target, v.link.id)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	var wg sync.WaitGroup
@@ -80,7 +75,7 @@ func (v *VNC) remoteRead(ctx context.Context, ch <-chan *network.Msg, local *web
 	}
 }
 
-func (v *VNC) localRead(ctx context.Context, local *websocket.Conn, remote *pool.Conn) {
+func (v *VNC) localRead(ctx context.Context, local *websocket.Conn, remote *conn.Conn) {
 	defer utils.Recover("localRead")
 	for {
 		select {

code/client/rule/vnc/link.go

@@ -6,7 +6,7 @@ import (
 	"image/jpeg"
 	"time"
 
-	"github.com/jkstack/natpass/code/client/pool"
+	"github.com/jkstack/natpass/code/client/conn"
 	"github.com/jkstack/natpass/code/client/rule/vnc/process"
 	"github.com/jkstack/natpass/code/network"
 	"github.com/jkstack/natpass/code/utils"
@@ -20,11 +20,10 @@ const (
 
 // Link vnc link
 type Link struct {
-	parent    *VNC
-	id        string // link id
-	target    string // target id
-	targetIdx uint32 // target idx
-	remote    *pool.Conn
+	parent *VNC
+	id     string // link id
+	target string // target id
+	remote *conn.Conn
 	// vnc
 	ps         *process.Process
 	quality    uint32
@@ -54,11 +53,6 @@ func (link *Link) GetPackets() (uint64, uint64) {
 	return link.recvPacket, link.sendPacket
 }
 
-// SetTargetIdx set link remote index
-func (link *Link) SetTargetIdx(idx uint32) {
-	link.targetIdx = idx
-}
-
 // SetQuality transfer quality
 func (link *Link) SetQuality(q uint32) {
 	link.quality = q
@@ -114,7 +108,7 @@ func (link *Link) remoteRead() {
 				link.ps.SetClipboard(msg.GetVclipboard())
 			} else {
 				data := link.ps.GetClipboard()
-				link.remote.SendVNCClipboardData(link.target, link.targetIdx, link.id, true, data)
+				link.remote.SendVNCClipboardData(link.target, link.id, true, data)
 			}
 		case network.Msg_disconnect:
 			logging.Info("link %s disconnected", link.id)
@@ -162,7 +156,7 @@ func (link *Link) close() {
 	if link.ps != nil {
 		link.ps.Close()
 	}
-	link.remote.SendDisconnect(link.target, link.targetIdx, link.id)
+	link.remote.SendDisconnect(link.target, link.id)
 }
 
 func cut(src *image.RGBA, rect image.Rectangle) *image.RGBA {
@@ -195,17 +189,17 @@ func (link *Link) sendAll(img *image.RGBA) {
 			rect := image.Rect(x, y, x+width, y+height)
 			next := cut(img, rect)
 			if link.quality == 100 {
-				link.remote.SendVNCImage(link.target, link.targetIdx, link.id,
+				link.remote.SendVNCImage(link.target, link.id,
 					screen, rect, network.VncImage_raw, next.Pix)
 				continue
 			}
 			buf.Reset()
 			err := jpeg.Encode(&buf, next, &jpeg.Options{Quality: int(link.quality)})
 			if err == nil {
-				link.remote.SendVNCImage(link.target, link.targetIdx, link.id,
+				link.remote.SendVNCImage(link.target, link.id,
 					screen, rect, network.VncImage_jpeg, buf.Bytes())
 			} else {
-				link.remote.SendVNCImage(link.target, link.targetIdx, link.id,
+				link.remote.SendVNCImage(link.target, link.id,
 					screen, rect, network.VncImage_raw, next.Pix)
 			}
 		}
@@ -219,17 +213,17 @@ func (link *Link) sendDiff(img *image.RGBA) {
 	for _, block := range blocks {
 		next := cut(img, block)
 		if link.quality == 100 {
-			link.remote.SendVNCImage(link.target, link.targetIdx, link.id,
+			link.remote.SendVNCImage(link.target, link.id,
 				screen, block, network.VncImage_raw, next.Pix)
 			continue
 		}
 		buf.Reset()
 		err := jpeg.Encode(&buf, next, &jpeg.Options{Quality: int(link.quality)})
 		if err == nil {
-			link.remote.SendVNCImage(link.target, link.targetIdx, link.id,
+			link.remote.SendVNCImage(link.target, link.id,
 				screen, block, network.VncImage_jpeg, buf.Bytes())
 		} else {
-			link.remote.SendVNCImage(link.target, link.targetIdx, link.id,
+			link.remote.SendVNCImage(link.target, link.id,
 				screen, block, network.VncImage_raw, next.Pix)
 		}
 	}

code/client/rule/vnc/vnc.go

@@ -5,9 +5,10 @@ import (
 	"net"
 	"net/http"
 	"sync"
+	"time"
 
+	"github.com/jkstack/natpass/code/client/conn"
 	"github.com/jkstack/natpass/code/client/global"
-	"github.com/jkstack/natpass/code/client/pool"
 	"github.com/jkstack/natpass/code/client/rule"
 	"github.com/jkstack/natpass/code/network"
 	"github.com/lwch/logging"
@@ -17,30 +18,33 @@ import (
 // VNC vnc handler
 type VNC struct {
 	sync.RWMutex
-	Name        string
-	cfg         global.Rule
-	link        *Link
-	chClipboard chan *network.VncClipboard
+	Name         string
+	cfg          global.Rule
+	link         *Link
+	readTimeout  time.Duration
+	writeTimeout time.Duration
+	chClipboard  chan *network.VncClipboard
 }
 
 // New new vnc
-func New(cfg global.Rule) *VNC {
+func New(cfg global.Rule, readTimeout, writeTimeout time.Duration) *VNC {
 	return &VNC{
-		Name:        cfg.Name,
-		cfg:         cfg,
-		chClipboard: make(chan *network.VncClipboard),
+		Name:         cfg.Name,
+		cfg:          cfg,
+		readTimeout:  readTimeout,
+		writeTimeout: writeTimeout,
+		chClipboard:  make(chan *network.VncClipboard),
 	}
 }
 
 // NewLink new link
-func (v *VNC) NewLink(id, remote string, remoteIdx uint32, localConn net.Conn, remoteConn *pool.Conn) rule.Link {
+func (v *VNC) NewLink(id, remote string, localConn net.Conn, remoteConn *conn.Conn) rule.Link {
 	remoteConn.AddLink(id)
 	link := &Link{
-		parent:    v,
-		id:        id,
-		target:    remote,
-		targetIdx: remoteIdx,
-		remote:    remoteConn,
+		parent: v,
+		id:     id,
+		target: remote,
+		remote: remoteConn,
 	}
 	if v.link != nil {
 		v.link.close()
@@ -83,15 +87,15 @@ func (v *VNC) GetPort() uint16 {
 }
 
 // Handle handle shell
-func (v *VNC) Handle(pl *pool.Pool) {
+func (v *VNC) Handle(c *conn.Conn) {
 	defer func() {
 		if err := recover(); err != nil {
 			logging.Error("close shell: %s, err=%v", v.Name, err)
 		}
 	}()
-	pf := func(cb func(*pool.Pool, http.ResponseWriter, *http.Request)) http.HandlerFunc {
+	pf := func(cb func(*conn.Conn, http.ResponseWriter, *http.Request)) http.HandlerFunc {
 		return func(w http.ResponseWriter, r *http.Request) {
-			cb(pl, w, r)
+			cb(c, w, r)
 		}
 	}
 	mux := http.NewServeMux()

code/network/msg.pb.go

@@ -162,12 +162,10 @@ type Msg struct {
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
-	XType   MsgType `protobuf:"varint,1,opt,name=_type,json=Type,proto3,enum=network.MsgType" json:"_type,omitempty"`
-	From    string  `protobuf:"bytes,2,opt,name=from,proto3" json:"from,omitempty"`
-	FromIdx uint32  `protobuf:"varint,3,opt,name=from_idx,json=fromIdx,proto3" json:"from_idx,omitempty"`
-	To      string  `protobuf:"bytes,4,opt,name=to,proto3" json:"to,omitempty"`
-	ToIdx   uint32  `protobuf:"varint,5,opt,name=to_idx,json=toIdx,proto3" json:"to_idx,omitempty"`
-	LinkId  string  `protobuf:"bytes,6,opt,name=link_id,json=linkId,proto3" json:"link_id,omitempty"`
+	XType  MsgType `protobuf:"varint,1,opt,name=_type,json=Type,proto3,enum=network.MsgType" json:"_type,omitempty"`
+	From   string  `protobuf:"bytes,2,opt,name=from,proto3" json:"from,omitempty"`
+	To     string  `protobuf:"bytes,4,opt,name=to,proto3" json:"to,omitempty"`
+	LinkId string  `protobuf:"bytes,6,opt,name=link_id,json=linkId,proto3" json:"link_id,omitempty"`
 	// Types that are assignable to Payload:
 	//	*Msg_Hsp
 	//	*Msg_Creq
@@ -230,13 +228,6 @@ func (x *Msg) GetFrom() string {
 	return ""
 }
 
-func (x *Msg) GetFromIdx() uint32 {
-	if x != nil {
-		return x.FromIdx
-	}
-	return 0
-}
-
 func (x *Msg) GetTo() string {
 	if x != nil {
 		return x.To
@@ -244,13 +235,6 @@ func (x *Msg) GetTo() string {
 	return ""
 }
 
-func (x *Msg) GetToIdx() uint32 {
-	if x != nil {
-		return x.ToIdx
-	}
-	return 0
-}
-
 func (x *Msg) GetLinkId() string {
 	if x != nil {
 		return x.LinkId
@@ -437,70 +421,67 @@ var file_msg_proto_rawDesc = []byte{
 	0x09, 0x76, 0x6e, 0x63, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x25, 0x0a, 0x11, 0x68, 0x61,
 	0x6e, 0x64, 0x73, 0x68, 0x61, 0x6b, 0x65, 0x5f, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x12,
 	0x10, 0x0a, 0x03, 0x65, 0x6e, 0x63, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x03, 0x65, 0x6e,
-	0x63, 0x22, 0xdf, 0x07, 0x0a, 0x03, 0x6d, 0x73, 0x67, 0x12, 0x26, 0x0a, 0x05, 0x5f, 0x74, 0x79,
+	0x63, 0x22, 0xad, 0x07, 0x0a, 0x03, 0x6d, 0x73, 0x67, 0x12, 0x26, 0x0a, 0x05, 0x5f, 0x74, 0x79,
 	0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x11, 0x2e, 0x6e, 0x65, 0x74, 0x77, 0x6f,
 	0x72, 0x6b, 0x2e, 0x6d, 0x73, 0x67, 0x2e, 0x74, 0x79, 0x70, 0x65, 0x52, 0x04, 0x54, 0x79, 0x70,
 	0x65, 0x12, 0x12, 0x0a, 0x04, 0x66, 0x72, 0x6f, 0x6d, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52,
-	0x04, 0x66, 0x72, 0x6f, 0x6d, 0x12, 0x19, 0x0a, 0x08, 0x66, 0x72, 0x6f, 0x6d, 0x5f, 0x69, 0x64,
-	0x78, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x07, 0x66, 0x72, 0x6f, 0x6d, 0x49, 0x64, 0x78,
-	0x12, 0x0e, 0x0a, 0x02, 0x74, 0x6f, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x74, 0x6f,
-	0x12, 0x15, 0x0a, 0x06, 0x74, 0x6f, 0x5f, 0x69, 0x64, 0x78, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0d,
-	0x52, 0x05, 0x74, 0x6f, 0x49, 0x64, 0x78, 0x12, 0x17, 0x0a, 0x07, 0x6c, 0x69, 0x6e, 0x6b, 0x5f,
-	0x69, 0x64, 0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x6c, 0x69, 0x6e, 0x6b, 0x49, 0x64,
-	0x12, 0x2e, 0x0a, 0x03, 0x68, 0x73, 0x70, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e,
-	0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, 0x2e, 0x68, 0x61, 0x6e, 0x64, 0x73, 0x68, 0x61, 0x6b,
-	0x65, 0x5f, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x48, 0x00, 0x52, 0x03, 0x68, 0x73, 0x70,
-	0x12, 0x2e, 0x0a, 0x04, 0x63, 0x72, 0x65, 0x71, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x18,
-	0x2e, 0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, 0x2e, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74,
-	0x5f, 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x48, 0x00, 0x52, 0x04, 0x63, 0x72, 0x65, 0x71,
-	0x12, 0x2f, 0x0a, 0x04, 0x63, 0x72, 0x65, 0x70, 0x18, 0x0c, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x19,
-	0x2e, 0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, 0x2e, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74,
-	0x5f, 0x72, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x48, 0x00, 0x52, 0x04, 0x63, 0x72, 0x65,
-	0x70, 0x12, 0x24, 0x0a, 0x05, 0x5f, 0x64, 0x61, 0x74, 0x61, 0x18, 0x0d, 0x20, 0x01, 0x28, 0x0b,
-	0x32, 0x0d, 0x2e, 0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, 0x2e, 0x64, 0x61, 0x74, 0x61, 0x48,
-	0x00, 0x52, 0x04, 0x44, 0x61, 0x74, 0x61, 0x12, 0x31, 0x0a, 0x07, 0x73, 0x72, 0x65, 0x73, 0x69,
-	0x7a, 0x65, 0x18, 0x14, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15, 0x2e, 0x6e, 0x65, 0x74, 0x77, 0x6f,
-	0x72, 0x6b, 0x2e, 0x73, 0x68, 0x65, 0x6c, 0x6c, 0x5f, 0x72, 0x65, 0x73, 0x69, 0x7a, 0x65, 0x48,
-	0x00, 0x52, 0x07, 0x73, 0x72, 0x65, 0x73, 0x69, 0x7a, 0x65, 0x12, 0x2b, 0x0a, 0x05, 0x73, 0x64,
-	0x61, 0x74, 0x61, 0x18, 0x15, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x6e, 0x65, 0x74, 0x77,
-	0x6f, 0x72, 0x6b, 0x2e, 0x73, 0x68, 0x65, 0x6c, 0x6c, 0x5f, 0x64, 0x61, 0x74, 0x61, 0x48, 0x00,
-	0x52, 0x05, 0x73, 0x64, 0x61, 0x74, 0x61, 0x12, 0x2c, 0x0a, 0x05, 0x76, 0x63, 0x74, 0x72, 0x6c,
-	0x18, 0x1e, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b,
-	0x2e, 0x76, 0x6e, 0x63, 0x5f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x48, 0x00, 0x52, 0x05,
-	0x76, 0x63, 0x74, 0x72, 0x6c, 0x12, 0x28, 0x0a, 0x04, 0x76, 0x69, 0x6d, 0x67, 0x18, 0x1f, 0x20,
-	0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, 0x2e, 0x76, 0x6e,
-	0x63, 0x5f, 0x69, 0x6d, 0x61, 0x67, 0x65, 0x48, 0x00, 0x52, 0x04, 0x76, 0x69, 0x6d, 0x67, 0x12,
-	0x2c, 0x0a, 0x06, 0x76, 0x6d, 0x6f, 0x75, 0x73, 0x65, 0x18, 0x20, 0x20, 0x01, 0x28, 0x0b, 0x32,
-	0x12, 0x2e, 0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, 0x2e, 0x76, 0x6e, 0x63, 0x5f, 0x6d, 0x6f,
-	0x75, 0x73, 0x65, 0x48, 0x00, 0x52, 0x06, 0x76, 0x6d, 0x6f, 0x75, 0x73, 0x65, 0x12, 0x2b, 0x0a,
-	0x04, 0x76, 0x6b, 0x62, 0x64, 0x18, 0x21, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15, 0x2e, 0x6e, 0x65,
-	0x74, 0x77, 0x6f, 0x72, 0x6b, 0x2e, 0x76, 0x6e, 0x63, 0x5f, 0x6b, 0x65, 0x79, 0x62, 0x6f, 0x61,
-	0x72, 0x64, 0x48, 0x00, 0x52, 0x04, 0x76, 0x6b, 0x62, 0x64, 0x12, 0x2f, 0x0a, 0x07, 0x76, 0x73,
-	0x63, 0x72, 0x6f, 0x6c, 0x6c, 0x18, 0x22, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x6e, 0x65,
-	0x74, 0x77, 0x6f, 0x72, 0x6b, 0x2e, 0x76, 0x6e, 0x63, 0x5f, 0x73, 0x63, 0x72, 0x6f, 0x6c, 0x6c,
-	0x48, 0x00, 0x52, 0x07, 0x76, 0x73, 0x63, 0x72, 0x6f, 0x6c, 0x6c, 0x12, 0x38, 0x0a, 0x0a, 0x76,
-	0x63, 0x6c, 0x69, 0x70, 0x62, 0x6f, 0x61, 0x72, 0x64, 0x18, 0x23, 0x20, 0x01, 0x28, 0x0b, 0x32,
-	0x16, 0x2e, 0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, 0x2e, 0x76, 0x6e, 0x63, 0x5f, 0x63, 0x6c,
-	0x69, 0x70, 0x62, 0x6f, 0x61, 0x72, 0x64, 0x48, 0x00, 0x52, 0x0a, 0x76, 0x63, 0x6c, 0x69, 0x70,
-	0x62, 0x6f, 0x61, 0x72, 0x64, 0x22, 0x80, 0x02, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x0b,
-	0x0a, 0x07, 0x75, 0x6e, 0x6b, 0x6e, 0x6f, 0x77, 0x6e, 0x10, 0x00, 0x12, 0x0d, 0x0a, 0x09, 0x68,
-	0x61, 0x6e, 0x64, 0x73, 0x68, 0x61, 0x6b, 0x65, 0x10, 0x01, 0x12, 0x0d, 0x0a, 0x09, 0x6b, 0x65,
-	0x65, 0x70, 0x61, 0x6c, 0x69, 0x76, 0x65, 0x10, 0x02, 0x12, 0x0f, 0x0a, 0x0b, 0x63, 0x6f, 0x6e,
-	0x6e, 0x65, 0x63, 0x74, 0x5f, 0x72, 0x65, 0x71, 0x10, 0x03, 0x12, 0x0f, 0x0a, 0x0b, 0x63, 0x6f,
-	0x6e, 0x6e, 0x65, 0x63, 0x74, 0x5f, 0x72, 0x65, 0x70, 0x10, 0x04, 0x12, 0x0e, 0x0a, 0x0a, 0x64,
-	0x69, 0x73, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x10, 0x05, 0x12, 0x0b, 0x0a, 0x07, 0x66,
-	0x6f, 0x72, 0x77, 0x61, 0x72, 0x64, 0x10, 0x06, 0x12, 0x10, 0x0a, 0x0c, 0x73, 0x68, 0x65, 0x6c,
-	0x6c, 0x5f, 0x72, 0x65, 0x73, 0x69, 0x7a, 0x65, 0x10, 0x0a, 0x12, 0x0e, 0x0a, 0x0a, 0x73, 0x68,
-	0x65, 0x6c, 0x6c, 0x5f, 0x64, 0x61, 0x74, 0x61, 0x10, 0x0b, 0x12, 0x0c, 0x0a, 0x08, 0x76, 0x6e,
-	0x63, 0x5f, 0x63, 0x74, 0x72, 0x6c, 0x10, 0x14, 0x12, 0x0d, 0x0a, 0x09, 0x76, 0x6e, 0x63, 0x5f,
-	0x69, 0x6d, 0x61, 0x67, 0x65, 0x10, 0x15, 0x12, 0x0d, 0x0a, 0x09, 0x76, 0x6e, 0x63, 0x5f, 0x6d,
-	0x6f, 0x75, 0x73, 0x65, 0x10, 0x16, 0x12, 0x10, 0x0a, 0x0c, 0x76, 0x6e, 0x63, 0x5f, 0x6b, 0x65,
-	0x79, 0x62, 0x6f, 0x61, 0x72, 0x64, 0x10, 0x17, 0x12, 0x0b, 0x0a, 0x07, 0x76, 0x6e, 0x63, 0x5f,
-	0x63, 0x61, 0x64, 0x10, 0x18, 0x12, 0x0e, 0x0a, 0x0a, 0x76, 0x6e, 0x63, 0x5f, 0x73, 0x63, 0x72,
-	0x6f, 0x6c, 0x6c, 0x10, 0x19, 0x12, 0x11, 0x0a, 0x0d, 0x76, 0x6e, 0x63, 0x5f, 0x63, 0x6c, 0x69,
-	0x70, 0x62, 0x6f, 0x61, 0x72, 0x64, 0x10, 0x1a, 0x42, 0x09, 0x0a, 0x07, 0x70, 0x61, 0x79, 0x6c,
-	0x6f, 0x61, 0x64, 0x42, 0x0c, 0x5a, 0x0a, 0x2e, 0x2f, 0x3b, 0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72,
-	0x6b, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x04, 0x66, 0x72, 0x6f, 0x6d, 0x12, 0x0e, 0x0a, 0x02, 0x74, 0x6f, 0x18, 0x04, 0x20, 0x01, 0x28,
+	0x09, 0x52, 0x02, 0x74, 0x6f, 0x12, 0x17, 0x0a, 0x07, 0x6c, 0x69, 0x6e, 0x6b, 0x5f, 0x69, 0x64,
+	0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x6c, 0x69, 0x6e, 0x6b, 0x49, 0x64, 0x12, 0x2e,
+	0x0a, 0x03, 0x68, 0x73, 0x70, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x6e, 0x65,
+	0x74, 0x77, 0x6f, 0x72, 0x6b, 0x2e, 0x68, 0x61, 0x6e, 0x64, 0x73, 0x68, 0x61, 0x6b, 0x65, 0x5f,
+	0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x48, 0x00, 0x52, 0x03, 0x68, 0x73, 0x70, 0x12, 0x2e,
+	0x0a, 0x04, 0x63, 0x72, 0x65, 0x71, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x18, 0x2e, 0x6e,
+	0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, 0x2e, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x5f, 0x72,
+	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x48, 0x00, 0x52, 0x04, 0x63, 0x72, 0x65, 0x71, 0x12, 0x2f,
+	0x0a, 0x04, 0x63, 0x72, 0x65, 0x70, 0x18, 0x0c, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x19, 0x2e, 0x6e,
+	0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, 0x2e, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x5f, 0x72,
+	0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x48, 0x00, 0x52, 0x04, 0x63, 0x72, 0x65, 0x70, 0x12,
+	0x24, 0x0a, 0x05, 0x5f, 0x64, 0x61, 0x74, 0x61, 0x18, 0x0d, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0d,
+	0x2e, 0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, 0x2e, 0x64, 0x61, 0x74, 0x61, 0x48, 0x00, 0x52,
+	0x04, 0x44, 0x61, 0x74, 0x61, 0x12, 0x31, 0x0a, 0x07, 0x73, 0x72, 0x65, 0x73, 0x69, 0x7a, 0x65,
+	0x18, 0x14, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15, 0x2e, 0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b,
+	0x2e, 0x73, 0x68, 0x65, 0x6c, 0x6c, 0x5f, 0x72, 0x65, 0x73, 0x69, 0x7a, 0x65, 0x48, 0x00, 0x52,
+	0x07, 0x73, 0x72, 0x65, 0x73, 0x69, 0x7a, 0x65, 0x12, 0x2b, 0x0a, 0x05, 0x73, 0x64, 0x61, 0x74,
+	0x61, 0x18, 0x15, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72,
+	0x6b, 0x2e, 0x73, 0x68, 0x65, 0x6c, 0x6c, 0x5f, 0x64, 0x61, 0x74, 0x61, 0x48, 0x00, 0x52, 0x05,
+	0x73, 0x64, 0x61, 0x74, 0x61, 0x12, 0x2c, 0x0a, 0x05, 0x76, 0x63, 0x74, 0x72, 0x6c, 0x18, 0x1e,
+	0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, 0x2e, 0x76,
+	0x6e, 0x63, 0x5f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x48, 0x00, 0x52, 0x05, 0x76, 0x63,
+	0x74, 0x72, 0x6c, 0x12, 0x28, 0x0a, 0x04, 0x76, 0x69, 0x6d, 0x67, 0x18, 0x1f, 0x20, 0x01, 0x28,
+	0x0b, 0x32, 0x12, 0x2e, 0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, 0x2e, 0x76, 0x6e, 0x63, 0x5f,
+	0x69, 0x6d, 0x61, 0x67, 0x65, 0x48, 0x00, 0x52, 0x04, 0x76, 0x69, 0x6d, 0x67, 0x12, 0x2c, 0x0a,
+	0x06, 0x76, 0x6d, 0x6f, 0x75, 0x73, 0x65, 0x18, 0x20, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e,
+	0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, 0x2e, 0x76, 0x6e, 0x63, 0x5f, 0x6d, 0x6f, 0x75, 0x73,
+	0x65, 0x48, 0x00, 0x52, 0x06, 0x76, 0x6d, 0x6f, 0x75, 0x73, 0x65, 0x12, 0x2b, 0x0a, 0x04, 0x76,
+	0x6b, 0x62, 0x64, 0x18, 0x21, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15, 0x2e, 0x6e, 0x65, 0x74, 0x77,
+	0x6f, 0x72, 0x6b, 0x2e, 0x76, 0x6e, 0x63, 0x5f, 0x6b, 0x65, 0x79, 0x62, 0x6f, 0x61, 0x72, 0x64,
+	0x48, 0x00, 0x52, 0x04, 0x76, 0x6b, 0x62, 0x64, 0x12, 0x2f, 0x0a, 0x07, 0x76, 0x73, 0x63, 0x72,
+	0x6f, 0x6c, 0x6c, 0x18, 0x22, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x6e, 0x65, 0x74, 0x77,
+	0x6f, 0x72, 0x6b, 0x2e, 0x76, 0x6e, 0x63, 0x5f, 0x73, 0x63, 0x72, 0x6f, 0x6c, 0x6c, 0x48, 0x00,
+	0x52, 0x07, 0x76, 0x73, 0x63, 0x72, 0x6f, 0x6c, 0x6c, 0x12, 0x38, 0x0a, 0x0a, 0x76, 0x63, 0x6c,
+	0x69, 0x70, 0x62, 0x6f, 0x61, 0x72, 0x64, 0x18, 0x23, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x16, 0x2e,
+	0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, 0x2e, 0x76, 0x6e, 0x63, 0x5f, 0x63, 0x6c, 0x69, 0x70,
+	0x62, 0x6f, 0x61, 0x72, 0x64, 0x48, 0x00, 0x52, 0x0a, 0x76, 0x63, 0x6c, 0x69, 0x70, 0x62, 0x6f,
+	0x61, 0x72, 0x64, 0x22, 0x80, 0x02, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x0b, 0x0a, 0x07,
+	0x75, 0x6e, 0x6b, 0x6e, 0x6f, 0x77, 0x6e, 0x10, 0x00, 0x12, 0x0d, 0x0a, 0x09, 0x68, 0x61, 0x6e,
+	0x64, 0x73, 0x68, 0x61, 0x6b, 0x65, 0x10, 0x01, 0x12, 0x0d, 0x0a, 0x09, 0x6b, 0x65, 0x65, 0x70,
+	0x61, 0x6c, 0x69, 0x76, 0x65, 0x10, 0x02, 0x12, 0x0f, 0x0a, 0x0b, 0x63, 0x6f, 0x6e, 0x6e, 0x65,
+	0x63, 0x74, 0x5f, 0x72, 0x65, 0x71, 0x10, 0x03, 0x12, 0x0f, 0x0a, 0x0b, 0x63, 0x6f, 0x6e, 0x6e,
+	0x65, 0x63, 0x74, 0x5f, 0x72, 0x65, 0x70, 0x10, 0x04, 0x12, 0x0e, 0x0a, 0x0a, 0x64, 0x69, 0x73,
+	0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x10, 0x05, 0x12, 0x0b, 0x0a, 0x07, 0x66, 0x6f, 0x72,
+	0x77, 0x61, 0x72, 0x64, 0x10, 0x06, 0x12, 0x10, 0x0a, 0x0c, 0x73, 0x68, 0x65, 0x6c, 0x6c, 0x5f,
+	0x72, 0x65, 0x73, 0x69, 0x7a, 0x65, 0x10, 0x0a, 0x12, 0x0e, 0x0a, 0x0a, 0x73, 0x68, 0x65, 0x6c,
+	0x6c, 0x5f, 0x64, 0x61, 0x74, 0x61, 0x10, 0x0b, 0x12, 0x0c, 0x0a, 0x08, 0x76, 0x6e, 0x63, 0x5f,
+	0x63, 0x74, 0x72, 0x6c, 0x10, 0x14, 0x12, 0x0d, 0x0a, 0x09, 0x76, 0x6e, 0x63, 0x5f, 0x69, 0x6d,
+	0x61, 0x67, 0x65, 0x10, 0x15, 0x12, 0x0d, 0x0a, 0x09, 0x76, 0x6e, 0x63, 0x5f, 0x6d, 0x6f, 0x75,
+	0x73, 0x65, 0x10, 0x16, 0x12, 0x10, 0x0a, 0x0c, 0x76, 0x6e, 0x63, 0x5f, 0x6b, 0x65, 0x79, 0x62,
+	0x6f, 0x61, 0x72, 0x64, 0x10, 0x17, 0x12, 0x0b, 0x0a, 0x07, 0x76, 0x6e, 0x63, 0x5f, 0x63, 0x61,
+	0x64, 0x10, 0x18, 0x12, 0x0e, 0x0a, 0x0a, 0x76, 0x6e, 0x63, 0x5f, 0x73, 0x63, 0x72, 0x6f, 0x6c,
+	0x6c, 0x10, 0x19, 0x12, 0x11, 0x0a, 0x0d, 0x76, 0x6e, 0x63, 0x5f, 0x63, 0x6c, 0x69, 0x70, 0x62,
+	0x6f, 0x61, 0x72, 0x64, 0x10, 0x1a, 0x42, 0x09, 0x0a, 0x07, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61,
+	0x64, 0x42, 0x0c, 0x5a, 0x0a, 0x2e, 0x2f, 0x3b, 0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, 0x62,
+	0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (

code/network/msg.proto

@@ -35,9 +35,7 @@ message msg {
     }
     type      _type = 1;
     string     from = 2;
-    uint32 from_idx = 3;
     string       to = 4;
-    uint32   to_idx = 5;
     string  link_id = 6;
     oneof payload {
         handshake_payload  hsp = 10;

code/server/handler/client.go

@@ -11,15 +11,27 @@ import (
 
 type client struct {
 	sync.RWMutex
+	id      string
 	parent  *clients
-	idx     uint32
 	conn    *network.Conn
 	updated time.Time
 	links   map[string]struct{} // link id => struct{}
 }
 
+func (c *client) close() {
+	for _, link := range c.getLinks() {
+		c.parent.parent.closeLink(link)
+		c.Lock()
+		delete(c.links, link)
+		c.Unlock()
+	}
+	c.conn.Close()
+	c.parent.remove(c.id)
+	logging.Info("client %s connection closed", c.id)
+}
+
 func (c *client) run() {
-	defer c.parent.parent.closeClient(c)
+	defer c.close()
 	for {
 		if time.Since(c.updated).Seconds() > 600 {
 			links := make([]string, 0, len(c.links))
@@ -28,7 +40,7 @@ func (c *client) run() {
 				links = append(links, id)
 			}
 			c.RUnlock()
-			logging.Info("%s-%d is not keepalived, links: %v", c.parent.id, c.idx, links)
+			logging.Info("%s is not keepalived, links: %v", c.id, links)
 			return
 		}
 		msg, size, err := c.conn.ReadMessage(c.parent.parent.cfg.ReadTimeout)
@@ -36,7 +48,7 @@ func (c *client) run() {
 			if strings.Contains(err.Error(), "i/o timeout") {
 				continue
 			}
-			logging.Error("read message from %s-%d: %v", c.parent.id, c.idx, err)
+			logging.Error("read message from %s: %v", c.id, err)
 			return
 		}
 		c.updated = time.Now()
@@ -70,11 +82,10 @@ func (c *client) getLinks() []string {
 	return ret
 }
 
-func (c *client) closeLink(id string) {
+func (c *client) sendClose(id string) {
 	var msg network.Msg
 	msg.From = "server"
-	msg.To = c.parent.id
-	msg.ToIdx = c.idx
+	msg.To = c.id
 	msg.XType = network.Msg_disconnect
 	msg.LinkId = id
 	c.conn.WriteMessage(&msg, c.parent.parent.cfg.WriteTimeout)
@@ -83,15 +94,10 @@ func (c *client) closeLink(id string) {
 	c.Unlock()
 }
 
-func (c *client) is(id string, idx uint32) bool {
-	return c.parent.id == id && c.idx == idx
-}
-
 func (c *client) keepalive() {
 	var msg network.Msg
 	msg.From = "server"
-	msg.To = c.parent.id
-	msg.ToIdx = c.idx
+	msg.To = c.id
 	msg.XType = network.Msg_keepalive
 	for {
 		time.Sleep(10 * time.Second)

code/server/handler/clients.go

@@ -2,65 +2,49 @@ package handler
 
 import (
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"github.com/jkstack/natpass/code/network"
-	"github.com/lwch/logging"
 )
 
 type clients struct {
 	sync.RWMutex
 	parent *Handler
-	id     string
-	data   map[uint32]*client // idx => client
-	idx    uint32
+	data   map[string]*client // id => client
 }
 
-func newClients(parent *Handler, id string) *clients {
-	logging.Info("new clients: %s", id)
+func newClients(parent *Handler) *clients {
 	return &clients{
 		parent: parent,
-		id:     id,
-		data:   make(map[uint32]*client),
+		data:   make(map[string]*client),
 	}
 }
 
-func (cs *clients) new(idx uint32, conn *network.Conn) *client {
-	logging.Info("new client: %s-%d", cs.id, idx)
+func (cs *clients) new(id string, conn *network.Conn) *client {
 	cli := &client{
+		id:      id,
 		parent:  cs,
-		idx:     idx,
 		conn:    conn,
 		updated: time.Now(),
 		links:   make(map[string]struct{}),
 	}
 	cs.Lock()
-	cs.data[idx] = cli
+	if c, ok := cs.data[id]; ok {
+		c.close()
+	}
+	cs.data[id] = cli
 	cs.Unlock()
 	return cli
 }
 
-func (cs *clients) next() *client {
-	list := make([]*client, 0, len(cs.data))
-	cs.RLock()
-	for _, cli := range cs.data {
-		list = append(list, cli)
-	}
-	cs.RUnlock()
-	if len(list) > 0 {
-		idx := atomic.AddUint32(&cs.idx, 1)
-		cli := list[int(idx)%len(list)]
-		return cli
-	}
-	return nil
+func (cs *clients) remove(id string) {
+	cs.Lock()
+	delete(cs.data, id)
+	cs.Unlock()
 }
 
-func (cs *clients) close(idx uint32) {
-	cs.Lock()
-	delete(cs.data, idx)
-	cs.Unlock()
-	if len(cs.data) == 0 {
-		cs.parent.removeClients(cs.id)
-	}
+func (cs *clients) lookup(id string) *client {
+	cs.RLock()
+	defer cs.RUnlock()
+	return cs.data[id]
 }

code/server/handler/handler.go

@@ -22,7 +22,7 @@ func (link *link) close() {
 		if cli == nil {
 			return
 		}
-		cli.closeLink(link.id)
+		cli.sendClose(link.id)
 	}
 	close(link.endPoints[0])
 	close(link.endPoints[1])
@@ -30,20 +30,20 @@ func (link *link) close() {
 
 // Handler handler
 type Handler struct {
-	cfg         *global.Configure
-	lockClients sync.RWMutex
-	clients     map[string]*clients // client id => client
-	lockLinks   sync.RWMutex
-	links       map[string]link // link id => endpoints
+	cfg       *global.Configure
+	clis      *clients
+	lockLinks sync.RWMutex
+	links     map[string]link // link id => endpoints
 }
 
 // New create handler
 func New(cfg *global.Configure) *Handler {
-	return &Handler{
-		cfg:     cfg,
-		clients: make(map[string]*clients),
-		links:   make(map[string]link),
+	h := &Handler{
+		cfg:   cfg,
+		links: make(map[string]link),
 	}
+	h.clis = newClients(h)
+	return h
 }
 
 // Handle main loop
@@ -59,7 +59,7 @@ func (h *Handler) Handle(conn net.Conn) {
 	}()
 	var err error
 	for i := 0; i < 10; i++ {
-		id, idx, err = h.readHandshake(c)
+		id, err = h.readHandshake(c)
 		if err != nil {
 			if err == errInvalidHandshake {
 				logging.Error("invalid handshake from %s", c.RemoteAddr().String())
@@ -75,83 +75,61 @@ func (h *Handler) Handle(conn net.Conn) {
 	}
 	logging.Info("%s-%d connected", id, idx)
 
-	clients := h.tryGetClients(id)
-	cli := clients.new(idx, c)
+	cli := h.clis.new(id, c)
 
-	defer h.closeClient(cli)
+	defer cli.close()
 	go cli.keepalive()
 
 	cli.run()
 }
 
-func (h *Handler) tryGetClients(id string) *clients {
-	h.lockClients.Lock()
-	defer h.lockClients.Unlock()
-	clients := h.clients[id]
-	if clients != nil {
-		return clients
-	}
-	clients = newClients(h, id)
-	h.clients[id] = clients
-	return clients
-}
-
 // readHandshake read handshake message and compare secret encoded from md5
-func (h *Handler) readHandshake(c *network.Conn) (string, uint32, error) {
+func (h *Handler) readHandshake(c *network.Conn) (string, error) {
 	msg, _, err := c.ReadMessage(5 * time.Second)
 	if err != nil {
-		return "", 0, err
+		return "", err
 	}
 	if msg.GetXType() != network.Msg_handshake {
-		return "", 0, errNotHandshake
+		return "", errNotHandshake
 	}
 	n := bytes.Compare(msg.GetHsp().GetEnc(), h.cfg.Enc[:])
 	if n != 0 {
-		return "", 0, errInvalidHandshake
+		return "", errInvalidHandshake
 	}
-	return msg.GetFrom(), msg.GetFromIdx(), nil
+	return msg.GetFrom(), nil
 }
 
-func (h *Handler) getClient(linkID, to string, toIdx uint32) *client {
+func (h *Handler) getClient(linkID, to string) *client {
 	h.lockLinks.RLock()
 	link := h.links[linkID]
 	h.lockLinks.RUnlock()
 
-	if link.endPoints[0] != nil && link.endPoints[0].is(to, toIdx) {
+	if link.endPoints[0] != nil && link.endPoints[0].id == to {
 		return link.endPoints[0]
 	}
-	if link.endPoints[1] != nil && link.endPoints[1].is(to, toIdx) {
+	if link.endPoints[1] != nil && link.endPoints[1].id == to {
 		return link.endPoints[1]
 	}
 
-	h.lockClients.RLock()
-	clients := h.clients[to]
-	h.lockClients.RUnlock()
-
-	if clients == nil {
-		return nil
-	}
-	return clients.next()
+	return h.clis.lookup(to)
 }
 
 func (h *Handler) onMessage(from *client, conn *network.Conn, msg *network.Msg, size uint16) {
 	to := msg.GetTo()
-	toIdx := msg.GetToIdx()
 	if msg.GetXType() == network.Msg_keepalive {
 		return
 	}
-	cli := h.getClient(msg.GetLinkId(), to, toIdx)
+	cli := h.getClient(msg.GetLinkId(), to)
 	if cli == nil {
-		logging.Error("client %s-%d not found", to, toIdx)
+		logging.Error("client %s not found", to)
 		return
 	}
 	h.msgHook(msg, from, cli, size)
 	err := cli.writeMessage(msg)
 	if err != nil {
-		logging.Error("write message %s from %s-%d to %s-%d: %v",
+		logging.Error("write message %s from %s to %s: %v",
 			msg.GetXType().String(),
-			msg.GetFrom(), msg.GetFromIdx(),
-			msg.GetTo(), msg.GetToIdx(),
+			msg.GetFrom(), msg.GetTo(),
 			err)
 	}
 }
@@ -171,8 +149,8 @@ func (h *Handler) addLink(name, id string, t network.ConnectRequestType, from, t
 	h.lockLinks.Lock()
 	h.links[id] = link
 	h.lockLinks.Unlock()
-	logging.Info("add link %s name %s from %s-%d to %s-%d",
-		id, name, from.parent.id, from.idx, to.parent.id, to.idx)
+	logging.Info("add link %s name %s from %s to %s",
+		id, name, from.id, to.id)
 }
 
 func (h *Handler) removeLink(id string, from, to *client) {
@@ -185,17 +163,17 @@ func (h *Handler) removeLink(id string, from, to *client) {
 	h.lockLinks.Lock()
 	delete(h.links, id)
 	h.lockLinks.Unlock()
-	logging.Info("remove link %s from %s-%d to %s-%d",
-		id, from.parent.id, from.idx, to.parent.id, to.idx)
+	logging.Info("remove link %s from %s to %s",
+		id, from.id, to.id)
 }
 
 func (h *Handler) responseLink(id string, ok bool, msg string, from, to *client) {
 	if ok {
-		logging.Info("link %s from %s-%d to %s-%d connect successed",
-			id, from.parent.id, from.idx, to.parent.id, to.idx)
+		logging.Info("link %s from %s to %s connect successed",
+			id, from.id, to.id)
 	} else {
-		logging.Info("link %s from %s-%d to %s-%d connect failed, %s",
-			id, from.parent.id, from.idx, to.parent.id, to.idx, msg)
+		logging.Info("link %s from %s to %s connect failed, %s",
+			id, from.id, to.id, msg)
 		// TODO: remove link?
 	}
 }
@@ -216,48 +194,31 @@ func (h *Handler) msgHook(msg *network.Msg, from, to *client, size uint16) {
 	// forward data
 	case network.Msg_forward:
 		data := msg.GetXData()
-		logging.Debug("link %s forward %d bytes from %s-%d to %s-%d",
-			msg.GetLinkId(), len(data.GetData()), from.parent.id, from.idx, to.parent.id, to.idx)
+		logging.Debug("link %s forward %d bytes from %s to %s",
+			msg.GetLinkId(), len(data.GetData()), from.id, to.id)
 	case network.Msg_shell_data:
 		data := msg.GetSdata()
-		logging.Debug("shell %s forward %d bytes from %s-%d to %s-%d",
-			msg.GetLinkId(), len(data.GetData()), from.parent.id, from.idx, to.parent.id, to.idx)
+		logging.Debug("shell %s forward %d bytes from %s to %s",
+			msg.GetLinkId(), len(data.GetData()), from.id, to.id)
 	// shell
 	case network.Msg_shell_resize:
 		data := msg.GetSresize()
-		logging.Info("shell %s from %s-%d to %s-%d resize to (%d,%d)",
-			msg.GetLinkId(), from.parent.id, from.idx, to.parent.id, to.idx,
+		logging.Info("shell %s from %s to %s resize to (%d,%d)",
+			msg.GetLinkId(), from.id, to.id,
 			data.GetRows(), data.GetCols())
 	}
-	msg.From = from.parent.id
-	msg.FromIdx = from.idx
-	msg.To = to.parent.id
-	msg.ToIdx = to.idx
-	logging.Debug("forward %d bytes on link %s from %s-%d to %s-%d", size, msg.GetLinkId(),
-		from.parent.id, from.idx, to.parent.id, to.idx)
+	msg.From = from.id
+	msg.To = to.id
+	logging.Debug("forward %d bytes on link %s from %s to %s", size, msg.GetLinkId(),
+		from.id, to.id)
 }
 
-func (h *Handler) closeClient(cli *client) {
-	links := cli.getLinks()
-	for _, t := range links {
-		h.lockLinks.RLock()
-		link := h.links[t]
-		h.lockLinks.RUnlock()
-		link.close()
-		h.lockLinks.Lock()
-		delete(h.links, t)
-		h.lockLinks.Unlock()
-	}
-	h.lockClients.RLock()
-	clients := h.clients[cli.parent.id]
-	h.lockClients.RUnlock()
-	if clients != nil {
-		clients.close(cli.idx)
-	}
-}
-
-func (h *Handler) removeClients(id string) {
-	h.lockClients.Lock()
-	delete(h.clients, id)
-	h.lockClients.Unlock()
+func (h *Handler) closeLink(id string) {
+	h.lockLinks.RLock()
+	link := h.links[id]
+	h.lockLinks.RUnlock()
+	link.close()
+	h.lockLinks.Lock()
+	delete(h.links, id)
+	h.lockLinks.Unlock()
 }

conf/common.yaml

@@ -1,6 +1,5 @@
 secret: 0123456789  # 预共享密钥，否则握手失败
 link:
-  connections:   3  # 连接数，仅client有效
   read_timeout:  1s # 读取数据包超时时间
   write_timeout: 1s # 发送数据包超时时间
 log: