Merge pull request #81 from cankush625/ankushchavan/multi-line-commands

feat(resp): support multi-line commands split across TCP packets

Author: cankush625

resp/handler.go

@@ -1,22 +1,26 @@
 package resp
 
 import (
+	"bufio"
+	"io"
 	"net"
 )
 
-// HandleConn does a read and write to the connection
+// HandleConn reads RESP commands from the connection and writes replies.
+// It uses a bufio.Reader to read one complete RESP message at a time,
+// so commands split across multiple TCP packets are handled correctly.
 func HandleConn(conn net.Conn) {
-	buffer := make([]byte, 1024)
+	reader := bufio.NewReader(conn)
 	for {
-		n, err := conn.Read(buffer)
+		msg, err := readMessage(reader)
 		if err != nil {
+			// io.EOF means the client closed the connection cleanly.
+			if err == io.EOF {
+				return
+			}
 			conn.Write([]byte("-Error processing\r\n"))
 			return
 		}
-		if n == 0 {
-			conn.Write([]byte("-Error processing\r\n"))
-			return
-		}
-		PerformRequest(buffer[:n], conn)
+		PerformRequest(msg, conn)
 	}
 }

resp/reader.go

@@ -0,0 +1,65 @@
+package resp
+
+import (
+	"bufio"
+	"bytes"
+	"fmt"
+	"io"
+	"strconv"
+	"strings"
+)
+
+// readMessage reads exactly one complete RESP message from the reader and
+// returns the raw bytes. It follows the RESP framing so it works correctly
+// regardless of how many TCP packets the message arrives in.
+//
+// Only Array messages are expected for incoming commands. Any other first byte
+// is read as a single line and returned as-is so ParseCommand can reject it.
+func readMessage(reader *bufio.Reader) ([]byte, error) {
+	// Read the first line, e.g. "*3\r\n"
+	line, err := reader.ReadString('\n')
+	if err != nil {
+		return nil, err
+	}
+
+	var buf bytes.Buffer
+	buf.WriteString(line)
+
+	if len(line) == 0 || line[0] != '*' {
+		// Not an array — return the single line for ParseCommand to reject.
+		return buf.Bytes(), nil
+	}
+
+	// Parse element count from "*N\r\n"
+	count, err := strconv.Atoi(strings.TrimRight(line[1:], "\r\n"))
+	if err != nil {
+		return nil, fmt.Errorf("invalid array length: %w", err)
+	}
+
+	for i := 0; i < count; i++ {
+		// Read the bulk string length line, e.g. "$5\r\n"
+		lenLine, err := reader.ReadString('\n')
+		if err != nil {
+			return nil, err
+		}
+		buf.WriteString(lenLine)
+
+		if len(lenLine) == 0 || lenLine[0] != '$' {
+			return nil, fmt.Errorf("expected bulk string, got %q", lenLine)
+		}
+
+		n, err := strconv.Atoi(strings.TrimRight(lenLine[1:], "\r\n"))
+		if err != nil {
+			return nil, fmt.Errorf("invalid bulk string length: %w", err)
+		}
+
+		// Read exactly n bytes + \r\n
+		data := make([]byte, n+2)
+		if _, err := io.ReadFull(reader, data); err != nil {
+			return nil, err
+		}
+		buf.Write(data)
+	}
+
+	return buf.Bytes(), nil
+}

resp/reader_test.go

@@ -0,0 +1,100 @@
+package resp
+
+import (
+	"bufio"
+	"bytes"
+	"io"
+	"strings"
+	"testing"
+)
+
+// TestReadMessage tests the readMessage function for complete and partial reads.
+func TestReadMessage(t *testing.T) {
+	tests := []struct {
+		input   string
+		want    string
+		wantErr bool
+	}{
+		// Single-element array (PING)
+		{"*1\r\n$4\r\nPING\r\n", "*1\r\n$4\r\nPING\r\n", false},
+		// Two-element array (ECHO hello)
+		{"*2\r\n$4\r\nECHO\r\n$5\r\nhello\r\n", "*2\r\n$4\r\nECHO\r\n$5\r\nhello\r\n", false},
+		// Three-element array (SET key value)
+		{"*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n", "*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n", false},
+		// Non-array first byte — returned as-is for ParseCommand to reject
+		{"+OK\r\n", "+OK\r\n", false},
+	}
+	for _, test := range tests {
+		reader := bufio.NewReader(strings.NewReader(test.input))
+		got, err := readMessage(reader)
+		if test.wantErr && err == nil {
+			t.Errorf("readMessage(%q) expected error, got nil", test.input)
+			continue
+		}
+		if !test.wantErr && err != nil {
+			t.Errorf("readMessage(%q) unexpected error: %v", test.input, err)
+			continue
+		}
+		if string(got) != test.want {
+			t.Errorf("readMessage(%q) = %q; want %q", test.input, got, test.want)
+		}
+	}
+}
+
+// TestReadMessage_MultipleCommands tests that readMessage reads one command at
+// a time from a stream containing multiple pipelined commands.
+func TestReadMessage_MultipleCommands(t *testing.T) {
+	// Two commands sent back-to-back in the same stream (pipelining)
+	input := "*1\r\n$4\r\nPING\r\n" +
+		"*2\r\n$4\r\nECHO\r\n$5\r\nhello\r\n"
+
+	reader := bufio.NewReader(strings.NewReader(input))
+
+	first, err := readMessage(reader)
+	if err != nil {
+		t.Fatalf("first readMessage error: %v", err)
+	}
+	if string(first) != "*1\r\n$4\r\nPING\r\n" {
+		t.Errorf("first message = %q; want %q", first, "*1\r\n$4\r\nPING\r\n")
+	}
+
+	second, err := readMessage(reader)
+	if err != nil {
+		t.Fatalf("second readMessage error: %v", err)
+	}
+	if string(second) != "*2\r\n$4\r\nECHO\r\n$5\r\nhello\r\n" {
+		t.Errorf("second message = %q; want %q", second, "*2\r\n$4\r\nECHO\r\n$5\r\nhello\r\n")
+	}
+
+	// Third read should return EOF
+	_, err = readMessage(reader)
+	if err != io.EOF {
+		t.Errorf("expected io.EOF after last command, got %v", err)
+	}
+}
+
+// TestReadMessage_SplitAcrossReads simulates a command arriving in two separate
+// TCP chunks by writing to a pipe with a delay between writes.
+func TestReadMessage_SplitAcrossReads(t *testing.T) {
+	// Simulate split delivery: write two parts with a pipe
+	pr, pw := io.Pipe()
+	reader := bufio.NewReader(pr)
+
+	done := make(chan []byte, 1)
+	go func() {
+		msg, _ := readMessage(reader)
+		done <- msg
+	}()
+
+	// Write the first half
+	pw.Write([]byte("*3\r\n$3\r\nSET\r\n"))
+	// Write the second half
+	pw.Write([]byte("$3\r\nkey\r\n$5\r\nvalue\r\n"))
+	pw.Close()
+
+	got := <-done
+	want := "*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n"
+	if !bytes.Equal(got, []byte(want)) {
+		t.Errorf("split read = %q; want %q", got, want)
+	}
+}