Fix --input-tcp for payloads > 64kb

Makefile

@@ -32,7 +32,7 @@ race:
 	$(RUN) go test ./... $(ARGS) -v -race -timeout 15s
 
 test:
-	$(RUN) go test ./. -race -timeout 30s $(LDFLAGS) $(ARGS)  -v
+	$(RUN) go test ./. -race -timeout 60s $(LDFLAGS) $(ARGS)  -v
 
 test_all:
 	$(RUN) go test ./... -timeout 60s $(LDFLAGS) $(ARGS) -v

input_file.go

@@ -71,7 +71,7 @@ func (i *FileInput) emit() {
 		}
 
 		if bytes.Equal(payloadSeparatorAsBytes[1:], line) {
-			asBytes := buffer.Bytes()[:buffer.Len()-1]
+			asBytes := buffer.Bytes()
 			buffer.Reset()
 
 			meta := payloadMeta(asBytes)
@@ -93,7 +93,7 @@ func (i *FileInput) emit() {
 			}
 
 			// Bytes() returns only pointer, so to remove data-race copy the data to an array
-			newBuf := make([]byte, len(asBytes))
+			newBuf := make([]byte, len(asBytes)-1)
 			copy(newBuf, asBytes)
 
 			i.data <- newBuf

input_tcp.go

@@ -2,7 +2,9 @@ package main
 
 import (
 	"bufio"
+	"bytes"
 	"fmt"
+	"io"
 	"log"
 	"net"
 	"os"
@@ -18,7 +20,7 @@ type TCPInput struct {
 // NewTCPInput constructor for TCPInput, accepts address with port
 func NewTCPInput(address string) (i *TCPInput) {
 	i = new(TCPInput)
-	i.data = make(chan []byte)
+	i.data = make(chan []byte, 1000)
 	i.address = address
 
 	i.listen(address)
@@ -58,16 +60,32 @@ func (i *TCPInput) listen(address string) {
 func (i *TCPInput) handleConnection(conn net.Conn) {
 	defer conn.Close()
 
+	payloadSeparatorAsBytes := []byte(payloadSeparator)
 	reader := bufio.NewReader(conn)
-	scanner := bufio.NewScanner(reader)
-	scanner.Split(payloadScanner)
+	var buffer bytes.Buffer
 
-	for scanner.Scan() {
-		i.data <- scanner.Bytes()
-	}
+	for {
+		line, err := reader.ReadBytes('\n')
 
-	if err := scanner.Err(); err != nil {
-		fmt.Fprintln(os.Stderr, "Unexpected error in input tcp connection:", err)
+		if err != nil {
+			if err != io.EOF {
+				fmt.Fprintln(os.Stderr, "Unexpected error in input tcp connection:", err)
+			}
+			break
+
+		}
+
+		if bytes.Equal(payloadSeparatorAsBytes[1:], line) {
+			asBytes := buffer.Bytes()
+			buffer.Reset()
+
+			newBuf := make([]byte, len(asBytes)-1)
+			copy(newBuf, asBytes)
+
+			i.data <- newBuf
+		} else {
+			buffer.Write(line)
+		}
 	}
 }
 

input_tcp_test.go

@@ -46,65 +46,3 @@ func TestTCPInput(t *testing.T) {
 
 	close(quit)
 }
-
-func BenchmarkTCPInput(b *testing.B) {
-	wg := new(sync.WaitGroup)
-	quit := make(chan int)
-
-	input := NewTCPInput("127.0.0.1:0")
-	output := NewTestOutput(func(data []byte) {
-		wg.Done()
-	})
-
-	Plugins.Inputs = []io.Reader{input}
-	Plugins.Outputs = []io.Writer{output}
-
-	go Start(quit)
-
-	tcpAddr, err := net.ResolveTCPAddr("tcp", input.listener.Addr().String())
-
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	var connections []net.Conn
-
-	// Creating simple pool of workers, same as output_tcp have
-	dataChan := make(chan []byte, 1000)
-
-	for i := 0; i < 10; i++ {
-		conn, _ := net.DialTCP("tcp", nil, tcpAddr)
-		connections = append(connections, conn)
-
-		go func(conn net.Conn) {
-			for {
-				data := <-dataChan
-
-				buf := make([]byte, len(data)+2)
-				data = append(data, []byte("¶")...)
-				copy(buf, data)
-				conn.Write(buf)
-			}
-		}(conn)
-	}
-
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	msg := []byte("GET / HTTP/1.1\r\n\r\n")
-
-	b.ResetTimer()
-	for i := 0; i < b.N; i++ {
-		wg.Add(1)
-		dataChan <- msg
-	}
-
-	wg.Wait()
-
-	for _, conn := range connections {
-		conn.Close()
-	}
-
-	close(quit)
-}