From 8a60567799225a6cdbbf8f28ea517cdf1ba34c23 Mon Sep 17 00:00:00 2001 From: lwch Date: Fri, 21 Jan 2022 18:45:39 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8E=BB=E9=99=A4=E8=BF=9E=E6=8E=A5=E6=B1=A0?= =?UTF-8?q?=E7=9A=84=E6=94=AF=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- code/client/app/app.go | 76 ++++----- code/client/app/callback.go | 73 +++++++++ code/client/app/connect.go | 76 --------- code/client/conn/conn.go | 167 ++++++++++++++++++++ code/client/conn/send.go | 18 +++ code/client/{pool => conn}/send_conn.go | 47 +++--- code/client/{pool => conn}/send_shell.go | 12 +- code/client/{pool => conn}/send_vnc.go | 37 ++--- code/client/dashboard/dashboard.go | 8 +- code/client/dashboard/h_info.go | 2 +- code/client/global/conf.go | 5 - code/client/pool/conn.go | 190 ----------------------- code/client/pool/pool.go | 120 -------------- code/client/pool/send.go | 45 ------ code/client/rule/bench/bench.go | 8 +- code/client/rule/bench/h_http.go | 6 +- code/client/rule/mgr.go | 4 +- code/client/rule/shell/h_new.go | 17 +- code/client/rule/shell/h_resize.go | 4 +- code/client/rule/shell/h_ws.go | 4 +- code/client/rule/shell/link.go | 25 ++- code/client/rule/shell/shell.go | 38 +++-- code/client/rule/vnc/events.go | 18 +-- code/client/rule/vnc/h_clipboard.go | 18 +-- code/client/rule/vnc/h_ctrl.go | 7 +- code/client/rule/vnc/h_new.go | 17 +- code/client/rule/vnc/h_ws.go | 13 +- code/client/rule/vnc/link.go | 32 ++-- code/client/rule/vnc/vnc.go | 40 ++--- code/network/msg.pb.go | 143 ++++++++--------- code/network/msg.proto | 2 - code/server/handler/client.go | 32 ++-- code/server/handler/clients.go | 50 ++---- code/server/handler/handler.go | 143 +++++++---------- conf/common.yaml | 1 - 35 files changed, 600 insertions(+), 898 deletions(-) create mode 100644 code/client/app/callback.go delete mode 100644 code/client/app/connect.go create mode 100644 code/client/conn/conn.go create mode 100644 code/client/conn/send.go rename code/client/{pool => conn}/send_conn.go (83%) rename code/client/{pool => conn}/send_shell.go (70%) rename code/client/{pool => conn}/send_vnc.go (73%) delete mode 100644 code/client/pool/conn.go delete mode 100644 code/client/pool/pool.go delete mode 100644 code/client/pool/send.go diff --git a/code/client/app/app.go b/code/client/app/app.go index 640f291..d8c9eeb 100644 --- a/code/client/app/app.go +++ b/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 {} diff --git a/code/client/app/callback.go b/code/client/app/callback.go new file mode 100644 index 0000000..c560681 --- /dev/null +++ b/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()) +} diff --git a/code/client/app/connect.go b/code/client/app/connect.go deleted file mode 100644 index f2250d9..0000000 --- a/code/client/app/connect.go +++ /dev/null @@ -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()) -} diff --git a/code/client/conn/conn.go b/code/client/conn/conn.go new file mode 100644 index 0000000..2443d74 --- /dev/null +++ b/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 +} diff --git a/code/client/conn/send.go b/code/client/conn/send.go new file mode 100644 index 0000000..25eb198 --- /dev/null +++ b/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): + } +} diff --git a/code/client/pool/send_conn.go b/code/client/conn/send_conn.go similarity index 83% rename from code/client/pool/send_conn.go rename to code/client/conn/send_conn.go index 7a59426..3523fdc 100644 --- a/code/client/pool/send_conn.go +++ b/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): } } diff --git a/code/client/pool/send_shell.go b/code/client/conn/send_shell.go similarity index 70% rename from code/client/pool/send_shell.go rename to code/client/conn/send_shell.go index a7214a5..f5b197a 100644 --- a/code/client/pool/send_shell.go +++ b/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): } } diff --git a/code/client/pool/send_vnc.go b/code/client/conn/send_vnc.go similarity index 73% rename from code/client/pool/send_vnc.go rename to code/client/conn/send_vnc.go index 2eb5393..549d6fe 100644 --- a/code/client/pool/send_vnc.go +++ b/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): } } diff --git a/code/client/dashboard/dashboard.go b/code/client/dashboard/dashboard.go index 1dd7190..18bf6a9 100644 --- a/code/client/dashboard/dashboard.go +++ b/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, } diff --git a/code/client/dashboard/h_info.go b/code/client/dashboard/h_info.go index 7b092f2..2813e16 100644 --- a/code/client/dashboard/h_info.go +++ b/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 diff --git a/code/client/global/conf.go b/code/client/global/conf.go index 5b3144e..a0dbc2b 100644 --- a/code/client/global/conf.go +++ b/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, diff --git a/code/client/pool/conn.go b/code/client/pool/conn.go deleted file mode 100644 index 945fe8e..0000000 --- a/code/client/pool/conn.go +++ /dev/null @@ -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 -} diff --git a/code/client/pool/pool.go b/code/client/pool/pool.go deleted file mode 100644 index cd1c9bb..0000000 --- a/code/client/pool/pool.go +++ /dev/null @@ -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) -} diff --git a/code/client/pool/send.go b/code/client/pool/send.go deleted file mode 100644 index 77d658a..0000000 --- a/code/client/pool/send.go +++ /dev/null @@ -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): - } -} diff --git a/code/client/rule/bench/bench.go b/code/client/rule/bench/bench.go index 8ecf1c9..06ece41 100644 --- a/code/client/rule/bench/bench.go +++ b/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), diff --git a/code/client/rule/bench/h_http.go b/code/client/rule/bench/h_http.go index ebd0d61..78d7abc 100644 --- a/code/client/rule/bench/h_http.go +++ b/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) } diff --git a/code/client/rule/mgr.go b/code/client/rule/mgr.go index 42a1bd6..d76088d 100644 --- a/code/client/rule/mgr.go +++ b/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 diff --git a/code/client/rule/shell/h_new.go b/code/client/rule/shell/h_new.go index e3150b6..dd15df7 100644 --- a/code/client/rule/shell/h_new.go +++ b/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) } diff --git a/code/client/rule/shell/h_resize.go b/code/client/rule/shell/h_resize.go index 8798777..414a63c 100644 --- a/code/client/rule/shell/h_resize.go +++ b/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") diff --git a/code/client/rule/shell/h_ws.go b/code/client/rule/shell/h_ws.go index ac1d933..8a86d03 100644 --- a/code/client/rule/shell/h_ws.go +++ b/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) diff --git a/code/client/rule/shell/link.go b/code/client/rule/shell/link.go index c5d9d0a..b7938c5 100644 --- a/code/client/rule/shell/link.go +++ b/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) } diff --git a/code/client/rule/shell/shell.go b/code/client/rule/shell/shell.go index df20d48..544f63e 100644 --- a/code/client/rule/shell/shell.go +++ b/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() diff --git a/code/client/rule/vnc/events.go b/code/client/rule/vnc/events.go index 7831bd3..c29cf18 100644 --- a/code/client/rule/vnc/events.go +++ b/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) } diff --git a/code/client/rule/vnc/h_clipboard.go b/code/client/rule/vnc/h_clipboard.go index 0a31872..f963cd5 100644 --- a/code/client/rule/vnc/h_clipboard.go +++ b/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") } diff --git a/code/client/rule/vnc/h_ctrl.go b/code/client/rule/vnc/h_ctrl.go index 4cb1569..155f813 100644 --- a/code/client/rule/vnc/h_ctrl.go +++ b/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") } diff --git a/code/client/rule/vnc/h_new.go b/code/client/rule/vnc/h_new.go index 71af36a..8bdc488 100644 --- a/code/client/rule/vnc/h_new.go +++ b/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() diff --git a/code/client/rule/vnc/h_ws.go b/code/client/rule/vnc/h_ws.go index 9a6cff5..67cfad6 100644 --- a/code/client/rule/vnc/h_ws.go +++ b/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 { diff --git a/code/client/rule/vnc/link.go b/code/client/rule/vnc/link.go index b49c813..3cbd843 100644 --- a/code/client/rule/vnc/link.go +++ b/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) } } diff --git a/code/client/rule/vnc/vnc.go b/code/client/rule/vnc/vnc.go index 1d8fa76..377917b 100644 --- a/code/client/rule/vnc/vnc.go +++ b/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() diff --git a/code/network/msg.pb.go b/code/network/msg.pb.go index 66055a7..fdad4c3 100644 --- a/code/network/msg.pb.go +++ b/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 ( diff --git a/code/network/msg.proto b/code/network/msg.proto index 53f3395..62107e9 100644 --- a/code/network/msg.proto +++ b/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; diff --git a/code/server/handler/client.go b/code/server/handler/client.go index f9ded98..4997260 100644 --- a/code/server/handler/client.go +++ b/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) diff --git a/code/server/handler/clients.go b/code/server/handler/clients.go index 8273f71..21b91ff 100644 --- a/code/server/handler/clients.go +++ b/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] } diff --git a/code/server/handler/handler.go b/code/server/handler/handler.go index 971b256..ed25ed6 100644 --- a/code/server/handler/handler.go +++ b/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() } diff --git a/conf/common.yaml b/conf/common.yaml index 064d65f..5349cb7 100644 --- a/conf/common.yaml +++ b/conf/common.yaml @@ -1,6 +1,5 @@ secret: 0123456789 # 预共享密钥，否则握手失败 link: - connections: 3 # 连接数，仅client有效 read_timeout: 1s # 读取数据包超时时间 write_timeout: 1s # 发送数据包超时时间 log: