add redial socket

1 files changed, 39 insertions, 3 deletions

diff --git a/pkg/server/tunnel.go b/pkg/server/tunnel.go
index e2fce92..afa0f31 100644
--- a/pkg/server/tunnel.go
+++ b/pkg/server/tunnel.go
@@ -36,6 +36,8 @@ type stream struct {
 	wg      sync.WaitGroup
 	in, out socket.Conn
 	pipes   []*hook.Pipe
+	mu      sync.Mutex
+	zombie  bool
 
 	m struct {
 		in  metric
@@ -295,6 +297,10 @@ func (s *stream) channel(c socket.Conn, m *metric, rq, wq queue.Q) {
 		if err != nil {
 			log.Println(s.t, s, c, err)
 		}
+
+		s.mu.Lock()
+		s.zombie = true
+		s.mu.Unlock()
 	}
 
 	counter := func(c *uint64, src, dst queue.Q) {
@@ -499,6 +505,26 @@ func tunnelDel(r *request, id string) {
 	}
 }
 
+func streamKick(r *request, tid string, sid int) {
+	e, ok := r.c.s.tunnels[tid]
+	if !ok {
+		r.Fatal("no such tunnel")
+	}
+
+	t := e.(*tunnel)
+
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	s, ok := t.streams[sid]
+	if !ok {
+		r.Fatal("no such stream")
+	}
+
+	log.Println(s, "kick")
+	s.stop()
+}
+
 func tunnelRename(r *request, old, new string) {
 	if !isOkTunnelName(new) {
 		r.Fatal("bad name")
@@ -551,7 +577,15 @@ func showActive(r *request) {
 		defer t.mu.Unlock()
 
 		foreachStream(t.streams, func(s *stream) {
-			r.Println(t.id, s.id, s.in, s.out, s.info())
+			var opts string
+
+			s.mu.Lock()
+			if s.zombie {
+				opts = "zombie"
+			}
+			s.mu.Unlock()
+
+			r.Println(t.id, s.id, s.in, s.out, s.info(), opts)
 		})
 	})
 }
@@ -577,9 +611,11 @@ func showRecent(r *request) {
 
 func init() {
 	newCmd("add", tunnelAdd, "[name id] [limit N] [single] socket [hook ...] socket")
-	newCmd("del", tunnelDel, "id")
+	newCmd("del", tunnelDel, "tunnel-id")
+
+	newCmd("kick", streamKick, "tunnel-id stream-id")
 
-	newCmd("rename", tunnelRename, "old-id new-id")
+	newCmd("rename", tunnelRename, "tunnel-old-id tunnel-new-id")
 
 	newCmd("show", showTunnels, "")