Allow pipelining

parser.go

@@ -8,8 +8,7 @@ import (
 	"strings"
 )
 
-func parseRequest(conn io.ReadCloser) (*Request, error) {
-	r := bufio.NewReader(conn)
+func parseRequest(r *bufio.Reader) (*Request, error) {
 	// first line of redis request should be:
 	// *<number of arguments>CRLF
 	line, err := r.ReadString('\n')
@@ -43,7 +42,6 @@ func parseRequest(conn io.ReadCloser) (*Request, error) {
 		return &Request{
 			Name: strings.ToLower(string(firstArg)),
 			Args: args,
-			Body: conn,
 		}, nil
 	}
 
@@ -59,7 +57,6 @@ func parseRequest(conn io.ReadCloser) (*Request, error) {
 	return &Request{
 		Name: strings.ToLower(string(fields[0])),
 		Args: args,
-		Body: conn,
 	}, nil
 
 }

parser_test.go

@@ -3,6 +3,7 @@ package redis
 import (
 	"bufio"
 	"bytes"
+	"io"
 	"io/ioutil"
 	"strings"
 	"testing"
@@ -36,7 +37,7 @@ func TestParseBadRequests(t *testing.T) {
 		"*2\r\n$3\r\ngEt\r\n$100\r\nx\r\n",
 	}
 	for _, v := range requests {
-		_, err := parseRequest(ioutil.NopCloser(strings.NewReader(v)))
+		_, err := parseRequest(bufio.NewReader(strings.NewReader(v)))
 		if err == nil {
 			t.Fatalf("Expected error for request [%s]", v)
 		}
@@ -55,7 +56,7 @@ func TestSucess(t *testing.T) {
 	}
 
 	for _, p := range expected {
-		request, err := parseRequest(ioutil.NopCloser(strings.NewReader(p.s)))
+		request, err := parseRequest(bufio.NewReader(strings.NewReader(p.s)))
 		if err != nil {
 			t.Fatalf("Un xxpected eror %s when parsting", err, p.s)
 		}
@@ -73,6 +74,22 @@ func TestSucess(t *testing.T) {
 	}
 }
 
+func TestPipielines(t *testing.T) {
+	chain := strings.Repeat("*2\r\n$3\r\ngEt\r\n$1\r\nx\r\n", 10)
+	reader := bufio.NewReader(strings.NewReader(chain))
+	for i := 0; i < 10; i++ {
+		_, err := parseRequest(reader)
+		if err != nil {
+			t.Fatalf("Unexpected error %s when parsing", err)
+		}
+	}
+	_, err := parseRequest(reader)
+	if err != io.EOF {
+		t.Fatalf("Expected EOF but received %+v", err)
+	}
+
+}
+
 func b(args ...string) [][]byte {
 	arr := make([][]byte, len(args))
 	for i := 0; i < len(args); i += 1 {

request.go

@@ -1,7 +1,6 @@
 package redis
 
 import (
-	"io"
 	"strconv"
 )
 
@@ -10,7 +9,6 @@ type Request struct {
 	Args       [][]byte
 	Host       string
 	ClientChan chan struct{}
-	Body       io.ReadCloser
 }
 
 func (r *Request) HasArgument(index int) bool {

server.go

@@ -5,6 +5,7 @@
 package redis
 
 import (
+	"bufio"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -88,8 +89,9 @@ func (srv *Server) ServeClient(conn net.Conn) (err error) {
 		clientAddr = co.RemoteAddr().String()
 	}
 
+	reader := bufio.NewReader(conn)
 	for {
-		request, err := parseRequest(conn)
+		request, err := parseRequest(reader)
 		if err != nil {
 			return err
 		}