1 files changed, 19 insertions, 11 deletions
diff --git a/pkg/server/server.go b/pkg/server/server.go
index 0e1bf24..4f012d0 100644
--- a/pkg/server/server.go
+++ b/pkg/server/server.go
@@ -4,6 +4,7 @@ import (
 	"tunnel/pkg/config"
 	"tunnel/pkg/netstring"
 	"strings"
+	"errors"
 	"bytes"
 	"sync"
 	"time"
@@ -18,10 +19,12 @@ type Server struct {
 	since time.Time
 
 	wg sync.WaitGroup
+	mu sync.Mutex
 
 	once sync.Once
 	done chan struct{}
 
+	streams streams
 	env env
 
 	nextCid uint64
@@ -47,11 +50,13 @@ type request struct {
 	argc int
 	args []string
 
-	failed bool
-
 	out *bytes.Buffer
 }
 
+type requestError string
+
+var errNotImplemented = errors.New("not implemented")
+
 func (c *client) String() string {
 	return fmt.Sprintf("client(%d)", c.id)
 }
@@ -73,15 +78,11 @@ func (r *request) Println(v ...interface{}) {
 }
 
 func (r *request) Fatal(v ...interface{}) {
-	panic(fmt.Sprint(v...))
+	panic(requestError(fmt.Sprint(v...)))
 }
 
 func (r *request) Fatalf(format string, v ...interface{}) {
-	panic(fmt.Sprintf(format, v...))
-}
-
-func (r *request) Fatalln(v ...interface{}) {
-	panic(fmt.Sprintln(v...))
+	panic(requestError(fmt.Sprintf(format, v...)))
 }
 
 func (r *request) expect(c ...int) {
@@ -134,6 +135,7 @@ func New() (*Server, error) {
 		listen: listen,
 		since: time.Now(),
 		done: make(chan struct{}),
+		streams: make(streams),
 	}
 
 	return s, nil
@@ -294,9 +296,12 @@ func (r *request) parse(query string) {
 
 func (r *request) run(query string) {
 	defer func () {
-		if e := recover(); e != nil {
-			r.failed = true
-			r.Print(e)
+		switch err := recover().(type) {
+		case requestError:
+			r.Print(err)
+		default:
+			panic(err)
+		case nil:
 		}
 	}()
 
@@ -304,6 +309,9 @@ func (r *request) run(query string) {
 
 	log.Printf("%s %s run [%s] '%s'", r.c, r, r.cmd.name, strings.Join(r.args, " "))
 
+	r.c.s.mu.Lock()
+	defer r.c.s.mu.Unlock()
+
 	r.cmd.f(r)
 }