[server] add client, request

7 files changed, 156 insertions, 49 deletions

diff --git a/pkg/client/client.go b/pkg/client/client.go
index d94b5a3..2f0a16e 100644
--- a/pkg/client/client.go
+++ b/pkg/client/client.go
@@ -25,7 +25,7 @@ func New() (*Client, error) {
 }
 
 func (c *Client) Send(args []string) (string, error) {
-	c.conn.SetDeadline(time.Now().Add(time.Second));
+	c.conn.SetDeadline(time.Now().Add(config.IoTimeout));
 
 	defer func () {
 		var t time.Time
diff --git a/pkg/config/config.go b/pkg/config/config.go
index c620e4a..1253f04 100644
--- a/pkg/config/config.go
+++ b/pkg/config/config.go
@@ -1,8 +1,12 @@
 package config
 
+import "time"
+
 const SockType = "unixpacket"
 const SockPath = "/tmp/tunnel.sock"
 
 const TimeFormat = "2006-01-02 15:04:05"
 
 const BufSize = 1024
+
+const IoTimeout = 5 * time.Second
diff --git a/pkg/server/echo.go b/pkg/server/echo.go
index 74c0450..c8dce31 100644
--- a/pkg/server/echo.go
+++ b/pkg/server/echo.go
@@ -3,13 +3,12 @@ package server
 import (
 	"strings"
 	"fmt"
-	"io"
 )
 
 func init() {
-	setHandler("echo", echo)
+	setHandler(echo, "echo")
 }
 
-func echo(s *Server, args []string, out io.Writer) {
-	fmt.Fprint(out, strings.Join(args[1:], " "))
+func echo(r *request) {
+	fmt.Fprint(r.out, strings.Join(r.args, " "))
 }
diff --git a/pkg/server/exit.go b/pkg/server/exit.go
index 504c7d1..159dd8d 100644
--- a/pkg/server/exit.go
+++ b/pkg/server/exit.go
@@ -1,13 +1,9 @@
 package server
 
-import (
-	"io"
-)
-
 func init() {
-	setHandler("exit", exit)
+	setHandler(exit, "exit")
 }
 
-func exit(s *Server, args []string, out io.Writer) {
-	s.Stop()
+func exit(r *request) {
+	r.c.s.Stop()
 }
diff --git a/pkg/server/server.go b/pkg/server/server.go
index ed0106b..3dedfb8 100644
--- a/pkg/server/server.go
+++ b/pkg/server/server.go
@@ -12,7 +12,7 @@ import (
 	"io"
 )
 
-type handler func (*Server, []string, io.Writer)
+type handler func (r *request)
 
 var handlers = map[string]handler{}
 
@@ -24,14 +24,54 @@ type Server struct {
 	m sync.Mutex
 
 	done bool
+
+	nextCid uint64
 }
 
-func setHandler(cmd string, h handler) {
-	if _, ok := handlers[cmd]; ok {
-		panic(fmt.Sprintf("handler %s already registered", cmd))
-	}
+type client struct {
+	s *Server
+
+	id uint64
+
+	conn net.Conn
+
+	nextRid uint64
+}
+
+type request struct {
+	c *client
+
+	id uint64
+
+	name string
+	args []string
+
+	out *bytes.Buffer
+}
 
-	handlers[cmd] = h
+func (c *client) String() string {
+	return fmt.Sprintf("client(%d)", c.id)
+}
+
+func (r *request) String() string {
+	return fmt.Sprintf("request(%d)", r.id)
+}
+
+func setHandler(h handler, names ...string) {
+	var path []string
+
+	for _, s := range names {
+		if _, ok := handlers[s]; ok {
+			err := fmt.Sprintf("handler '%s' already registered at '%s'",
+				s, strings.Join(path, " "))
+			panic(err)
+		}
+
+		path = append(path, s)
+
+		handlers[s] = h
+		break
+	}
 }
 
 func (s *Server) isDone() bool {
@@ -66,11 +106,13 @@ func (s *Server) Run() {
 			continue
 		}
 
-		log.Print("new client")
+		c := s.newClient(conn)
+
+		log.Println(c, "accept")
 
 		s.wg.Add(1)
 
-		go s.handle(conn)
+		go c.handle()
 	}
 
 	s.wg.Wait()
@@ -84,38 +126,73 @@ func (s *Server) Stop() {
 	s.listen.Close()
 }
 
-func (s *Server) handle(c net.Conn) {
-	defer c.Close()
-	defer s.wg.Done()
+func (s *Server) newClient(conn net.Conn) *client {
+	c := &client{
+		s:	s,
+		conn:	conn,
+		id:	s.nextCid,
+	}
+
+	s.nextCid++
+
+	return c
+}
+
+func (c *client) newRequest(msg string) *request {
+	args := strings.Split(msg, " ")
+
+	r := &request{
+		c:	c,
+		id:	c.nextRid,
+		name:	args[0],
+		args:	args[1:],
+		out:	bytes.NewBuffer(nil),
+	}
+
+	c.nextRid++
+
+	return r
+}
+
+func (c *client) handle() {
+	defer c.close()
 
 	buf := make([]byte, config.BufSize)
 
-	nr, er := c.Read(buf)
-	if er != nil {
-		if er != io.EOF {
-			log.Print("handle: ", er)
+	for {
+		nr, er := c.conn.Read(buf)
+		if er != nil {
+			if er != io.EOF {
+				log.Println(c, "handle:", er)
+			}
+			break
 		}
-		return
-	}
 
-	data := string(buf[:nr])
-	args := strings.Split(data, " ")
-	out := bytes.NewBuffer(nil)
+		msg := string(buf[:nr])
+		r := c.newRequest(msg)
 
-	if h, ok := handlers[args[0]]; ok {
-		log.Printf("handle: run: %s", data)
-		h(s, args, out)
-	} else {
-		fmt.Fprint(out, "unknown command")
-	}
+		if h, ok := handlers[r.name]; ok {
+			log.Println(c, r, "run:", msg)
+			h(r)
+		} else {
+			fmt.Fprint(r.out, "unknown command")
+		}
 
-	if out.Len() == 0 {
-		out.Write([]byte("\n"))
-	}
+		if r.out.Len() == 0 {
+			r.out.Write([]byte("\n"))
+		}
 
-	_, ew := c.Write(out.Bytes())
-	if ew != nil {
-		log.Print("handle: ", ew)
-		return
+		_, ew := c.conn.Write(r.out.Bytes())
+		if ew != nil {
+			log.Println(c, "handle:", ew)
+			break
+		}
 	}
 }
+
+func (c *client) close() {
+	log.Println(c, "close")
+
+	c.conn.Close()
+	c.s.wg.Done()
+}
diff --git a/pkg/server/sleep.go b/pkg/server/sleep.go
new file mode 100644
index 0000000..53c85ea
--- /dev/null
+++ b/pkg/server/sleep.go
@@ -0,0 +1,32 @@
+package server
+
+import (
+	"strconv"
+	"time"
+	"fmt"
+)
+
+const maxSleep = 10
+
+func init() {
+	setHandler(sleep, "sleep")
+}
+
+func sleep(r *request) {
+	if len(r.args) == 0 {
+		return
+	}
+
+	n, err := strconv.Atoi(r.args[0])
+	if err != nil || n < 0 {
+		fmt.Fprintf(r.out, "invalid time interval '%s'", r.args[0])
+		return
+	}
+
+	if n > maxSleep {
+		fmt.Fprintf(r.out, "no more than %d", maxSleep)
+		return
+	}
+
+	time.Sleep(time.Duration(n) * time.Second)
+}
diff --git a/pkg/server/status.go b/pkg/server/status.go
index 6fb2a2a..462ac76 100644
--- a/pkg/server/status.go
+++ b/pkg/server/status.go
@@ -3,13 +3,12 @@ package server
 import (
 	"tunnel/pkg/config"
 	"fmt"
-	"io"
 )
 
 func init() {
-	setHandler("status", status)
+	setHandler(status, "status")
 }
 
-func status(s *Server, args []string, out io.Writer) {
-	fmt.Fprintf(out, "since %s", s.since.Format(config.TimeFormat))
+func status(r *request) {
+	fmt.Fprintf(r.out, "since %s", r.c.s.since.Format(config.TimeFormat))
 }