mono, force and simpler channels

1 files changed, 25 insertions, 7 deletions

diff --git a/pkg/server/tunnel.go b/pkg/server/tunnel.go
index 01d7c22..72d4c13 100644
--- a/pkg/server/tunnel.go
+++ b/pkg/server/tunnel.go
@@ -56,6 +56,8 @@ type tunnel struct {
 	quit chan struct{}
 	done chan struct{}
 
+	mono bool
+
 	in, out socket.S
 	hooks   []hook.H
 
@@ -123,6 +125,11 @@ func (t *tunnel) serve() {
 				t.handle(in)
 				wg.Done()
 			}()
+
+			if t.mono {
+				wg.Wait()
+				t.wg.Wait()
+			}
 		}
 	}
 
@@ -240,7 +247,7 @@ func (s *stream) channel(c socket.Channel, m *metric, rq, wq queue.Q) {
 		q := queue.New()
 		go counter(&m.rx, rq, q)
 		watch(q, c.Recv)
-		rq.Dry()
+		q.Dry()
 	}()
 }
 
@@ -311,7 +318,7 @@ func parseHooks(args []string, env env.Env) ([]hook.H, error) {
 	return hooks, nil
 }
 
-func newTunnel(args []string, env env.Env) (*tunnel, error) {
+func newTunnel(mono bool, args []string, env env.Env) (*tunnel, error) {
 	var in, out socket.S
 	var hooks []hook.H
 	var err error
@@ -337,6 +344,7 @@ func newTunnel(args []string, env env.Env) (*tunnel, error) {
 		args:    strings.Join(args, " "),
 		quit:    make(chan struct{}),
 		done:    make(chan struct{}),
+		mono:    mono,
 		hooks:   hooks,
 		in:      in,
 		out:     out,
@@ -356,9 +364,10 @@ func isOkTunnelName(s string) bool {
 func tunnelAdd(r *request) {
 	args := r.args
 	name := ""
+	mono := false
 
-	if len(args) >= 2 {
-		if args[0] == "name" {
+	for len(args) > 1 {
+		if args[0] == "name" && len(args) > 1 {
 			name = args[1]
 			if !isOkTunnelName(name) {
 				r.Fatal("bad name")
@@ -369,14 +378,23 @@ func tunnelAdd(r *request) {
 			}
 
 			args = args[2:]
+			continue
 		}
+
+		if args[0] == "mono" {
+			mono = true
+			args = args[1:]
+			continue
+		}
+
+		break
 	}
 
 	if len(args) < 2 {
 		r.Fatal("not enough args")
 	}
 
-	t, err := newTunnel(args, r.c.s.env)
+	t, err := newTunnel(mono, args, r.c.s.env)
 	if err != nil {
 		r.Fatal(err)
 	}
@@ -459,7 +477,7 @@ func showActive(r *request) {
 		defer t.mu.Unlock()
 
 		foreachStream(t.streams, func(s *stream) {
-			r.Println(t.id, s.id, s.in.Origin(), s.out.Origin(), s.info())
+			r.Println(t.id, s.id, s.in, s.out, s.info())
 		})
 	})
 }
@@ -479,7 +497,7 @@ func showRecent(r *request) {
 
 	for _, s := range streams {
 		when := s.until.Format(config.TimeFormat)
-		r.Println(when, s.t.id, s.id, s.in.Origin(), s.out.Origin(), s.info())
+		r.Println(when, s.t.id, s.id, s.in, s.out, s.info())
 	}
 }