Handle long messages in event parser

internal/parser/reader_parser.go

@@ -5,9 +5,12 @@ import (
 	"io"
 )
 
-var splitFunc bufio.SplitFunc = func(data []byte, _ bool) (advance int, token []byte, err error) {
+// splitFunc is a split function for a bufio.Scanner that splits a sequence of
+// bytes into SSE events. Each event ends with two consecutive newline sequences,
+// where a newline sequence is defined as either "\n", "\r", or "\r\n".
+var splitFunc bufio.SplitFunc = func(data []byte, atEOF bool) (advance int, token []byte, err error) {
 	if len(data) == 0 {
-		return
+		return 0, nil, nil
 	}
 
 	var start, index, endlineLen int
@@ -22,16 +25,22 @@ var splitFunc bufio.SplitFunc = func(data []byte, _ bool) (advance int, token []
 		}
 	}
 
-	if l := len(data); advance < l {
+	if l := len(data); advance == l && !atEOF {
+		// We have reached the end of the buffer but have not yet seen two consecutive
+		// newline sequences, so we request more data.
+		return 0, nil, nil
+	} else if advance < l {
+		// We have found a newline. Consume the end-of-line sequence.
 		advance++
+		// Consume one more character if end-of-line is \r\n.
 		if advance < l && data[advance-1] == '\r' && data[advance] == '\n' {
 			advance++
 		}
 	}
 
 	token = data[start:advance]
 
-	return
+	return advance, token, nil
 }
 
 // ReaderParser extracts fields from a reader. Reading is buffered using a bufio.Scanner.

internal/parser/reader_parser_test.go

@@ -30,6 +30,8 @@ func TestReaderParser(t *testing.T) {
 		expected []parser.Field
 	}
 
+	longString := strings.Repeat("abcdefghijklmnopqrstuvwxyz", 193)
+
 	tests := []test{
 		{
 			name: "Valid input",
@@ -67,6 +69,14 @@ data: still, here's some data: you deserve it
 				newDataField(t, "still, here's some data: you deserve it"),
 			},
 		},
+		{
+			name:  "Valid input with long string",
+			input: strings.NewReader("\nid:2\ndata:" + longString + "\n"),
+			expected: []parser.Field{
+				newIDField(t, "2"),
+				newDataField(t, longString),
+			},
+		},
 		{
 			name:  "Error",
 			input: errReader{nil},

internal/parser/split_func_test.go

@@ -34,3 +34,33 @@ func TestSplitFunc(t *testing.T) {
 		t.Fatalf("wrong tokens:\nreceived: %#v\nexpected: %#v", tokens, expected)
 	}
 }
+
+func TestSplitFuncWithLongLine(t *testing.T) {
+	t.Parallel()
+
+	longString := strings.Repeat("abcdef\rghijklmn\nopqrstu\r\nvwxyz", 193)
+	text := longString + "\n\n" + longString + "\r\r" + longString + "\r\n\r\n" + longString
+	r := strings.NewReader(text)
+	s := bufio.NewScanner(r)
+	s.Split(splitFunc)
+
+	expected := []string{
+		longString + "\n\n",
+		longString + "\r\r",
+		longString + "\r\n\r\n",
+		longString,
+	}
+	tokens := make([]string, 0, len(expected))
+
+	for s.Scan() {
+		tokens = append(tokens, s.Text())
+	}
+
+	if s.Err() != nil {
+		t.Fatalf("an error occurred: %v", s.Err())
+	}
+
+	if !reflect.DeepEqual(tokens, expected) {
+		t.Fatalf("wrong tokens:\nreceived: %#v\nexpected: %#v", tokens, expected)
+	}
+}