Add header to request and response

emitter.go

@@ -56,7 +56,9 @@ func CopyMulty(src io.Reader, writers ...io.Writer) (err error) {
 		if nr > 0 && len(buf) > nr {
 			payload := buf[0:nr]
 
-			Debug("[EMITTER] input:", string(payload))
+			if Settings.debug {
+				Debug("[EMITTER] input:", string(payload[0:500]))
+			}
 
 			if modifier != nil {
 				payload = modifier.Rewrite(payload)
@@ -65,10 +67,10 @@ func CopyMulty(src io.Reader, writers ...io.Writer) (err error) {
 				if len(payload) == 0 {
 					continue
 				}
-			}
 
-			if Settings.debug {
-				Debug("[EMITTER] Sending payload, size:", len(payload), "First 500 bytes:", string(payload[0:500]))
+				if Settings.debug {
+					Debug("[EMITTER] Rewrittern input:", len(payload), "First 500 bytes:", string(payload[0:500]))
+				}
 			}
 
 			if Settings.splitOutput {

examples/echo_modifier.sh

@@ -1,8 +1,27 @@
 #!/usr/bin/env bash
 while read line; do
     decoded=$(echo "$line" | xxd -r -p)
-    encoded=$(echo "$decoded" | xxd -p | tr -d "\\n")
-    echo "$encoded"
+
+    header=$(echo "$decoded" | head -n +1)
+    payload=$(echo "$decoded" | tail -n +2)
+
+    encoded=$(echo -e "$header\n$payload" | xxd -p | tr -d "\\n")
+
+    >&2 echo ""
+    >&2 echo "[DEBUG][MIDDLEWARE] ==================================="
+
+    case ${header:0:1} in
+    "2")
+        >&2 echo "[DEBUG][MIDDLEWARE] Request type: Replayed Response"
+        ;;
+    "1")
+        >&2 echo "[DEBUG][MIDDLEWARE] Request type: Request"
+        echo "$encoded"
+        ;;
+    *)
+        >&2 echo "[DEBUG][MIDDLEWARE] Unknown request type $header"
+    esac
+    >&2 echo "[DEBUG][MIDDLEWARE] ==================================="
 
     >&2 echo "[DEBUG][MIDDLEWARE] Original data: $line"
     >&2 echo "[DEBUG][MIDDLEWARE] Decoded request: $decoded"

input_raw.go

@@ -29,7 +29,14 @@ func NewRAWInput(address string, expire time.Duration) (i *RAWInput) {
 
 func (i *RAWInput) Read(data []byte) (int, error) {
 	buf := <-i.data
-	copy(data, buf)
+
+	if len(Settings.middleware) > 0 {
+		header := []byte("1\n")
+		copy(data[0:len(header)], header)
+		copy(data[len(header):], buf)
+	} else {
+		copy(data, buf)
+	}
 
 	return len(buf), nil
 }

middleware.go

@@ -74,7 +74,7 @@ func (m *Middleware) read(from io.Reader) {
 		bytes := scanner.Bytes()
 		hex.Decode(buf, bytes)
 
-		Debug("Received:", buf[0:len(bytes)/2])
+		Debug("[MIDDLEWARE-MASTER] Received:", string(buf[0:len(bytes)/2]))
 
 		m.data <- buf[0 : len(bytes)/2]
 	}

middleware_test.go

@@ -99,6 +99,7 @@ func TestEchoMiddleware(t *testing.T) {
 		wg.Done()
 	}))
 	to := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		panic("Asdasd")
 		wg.Done()
 	}))
 
@@ -108,7 +109,7 @@ func TestEchoMiddleware(t *testing.T) {
 	input := NewRAWInput(from.Listener.Addr().String(), testRawExpire)
 
 	// And redirect to another
-	output := NewHTTPOutput(to.URL, &HTTPOutputConfig{})
+	output := NewHTTPOutput(to.URL, &HTTPOutputConfig{Debug: true})
 
 	Plugins.Inputs = []io.Reader{input}
 	Plugins.Outputs = []io.Writer{output}
@@ -122,7 +123,7 @@ func TestEchoMiddleware(t *testing.T) {
 	// Should receive 2 requests from original + 2 from replayed
 	wg.Add(4)
 
-	client := NewHTTPClient(from.URL, &HTTPClientConfig{Debug: true})
+	client := NewHTTPClient(from.URL, &HTTPClientConfig{Debug: false})
 
 	// Request should be echoed
 	client.Get("/")
@@ -131,6 +132,8 @@ func TestEchoMiddleware(t *testing.T) {
 	wg.Wait()
 	close(quit)
 	Settings.middleware = ""
+
+	time.Sleep(10*time.Millisecond)
 }
 
 func TestTokenMiddleware(t *testing.T) {

output_http.go

@@ -60,6 +60,7 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) io.Writer {
 	}
 
 	o.queue = make(chan []byte, 100)
+	o.responses = make(chan []byte, 100)
 	o.needWorker = make(chan int, 1)
 
 	// Initial workers count
@@ -152,15 +153,34 @@ func (o *HTTPOutput) Write(data []byte) (n int, err error) {
 	return len(data), nil
 }
 
+func (o *HTTPOutput) Read(data []byte) (int, error) {
+	buf := <- o.responses
+	header := []byte("2\n")
+	copy(data[0:2], header)
+	copy(data[2:], buf)
+
+	return len(buf) + len(header), nil
+}
+
 func (o *HTTPOutput) sendRequest(client *HTTPClient, request []byte) {
+	if len(Settings.middleware) > 0 {
+		request = request[2:]
+	}
+
 	start := time.Now()
 	resp, err := client.Send(request)
 	stop := time.Now()
 
+	panic(string(resp))
+
 	if err != nil {
 		log.Println("Request error:", err)
 	}
 
+	if len(Settings.middleware) > 0 {
+		o.responses <- resp
+	}
+
 	if o.elasticSearch != nil {
 		o.elasticSearch.ResponseAnalyze(request, resp, start, stop)
 	}