Merge pull request #157 from buger/encoding-chunked

Handle Transfer-Encoding: chunked

output_http.go

@@ -7,6 +7,7 @@ import (
 	"io/ioutil"
 	"log"
 	"net/http"
+	"net/http/httputil"
 	"net/url"
 	"strings"
 	"sync/atomic"
@@ -29,18 +30,31 @@ func (o *HTTPOutput) customCheckRedirect(req *http.Request, via []*http.Request)
 
 // ParseRequest in []byte returns a http request or an error
 func ParseRequest(data []byte) (request *http.Request, err error) {
+	var body []byte
+
+	// Test if request have Transfer-Encoding: chunked
+	isChunked := bytes.Contains(data, []byte(": chunked\r\n"))
+
 	buf := bytes.NewBuffer(data)
 	reader := bufio.NewReader(buf)
 
+	// ReadRequest does not read POST bodies, we have to do it by ourseves
 	request, err = http.ReadRequest(reader)
 
 	if err != nil {
 		return
 	}
 
 	if request.Method == "POST" {
-		body, _ := ioutil.ReadAll(reader)
+		// This works, because ReadRequest method modify buffer and strips all headers, leaving only body
+		if isChunked {
+			body, _ = ioutil.ReadAll(httputil.NewChunkedReader(reader))
+		} else {
+			body, _ = ioutil.ReadAll(reader)
+		}
+
 		bodyBuf := bytes.NewBuffer(body)
+
 		request.Body = ioutil.NopCloser(bodyBuf)
 		request.ContentLength = int64(bodyBuf.Len())
 	}
@@ -62,7 +76,7 @@ type HTTPOutput struct {
 
 	redirectLimit int
 
-	needWorker    chan int
+	needWorker chan int
 
 	urlRegexp            HTTPUrlRegexp
 	headerFilters        HTTPHeaderFilters

output_http_test.go

@@ -67,7 +67,7 @@ func TestHTTPOutput(t *testing.T) {
 			defer req.Body.Close()
 			body, _ := ioutil.ReadAll(req.Body)
 
-			if string(body) != "a=1&b=2\r\n\r\n" {
+			if string(body) != "a=1&b=2" {
 				buf, _ := httputil.DumpRequest(req, true)
 				t.Error("Wrong POST body:", string(buf))
 			}
@@ -95,6 +95,42 @@ func TestHTTPOutput(t *testing.T) {
 	close(quit)
 }
 
+func TestHTTPOutputChunkedEncoding(t *testing.T) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	input := NewTestInput()
+
+	headers := HTTPHeaders{HTTPHeader{"User-Agent", "Gor"}}
+	methods := HTTPMethods{"GET", "PUT", "POST"}
+
+	listener := startHTTP(func(req *http.Request) {
+		defer req.Body.Close()
+		body, _ := ioutil.ReadAll(req.Body)
+
+		if string(body) != "Wikipedia in\r\n\r\nchunks." {
+			buf, _ := httputil.DumpRequest(req, true)
+			t.Error("Wrong POST body:", buf, body, []byte("Wikipedia in\r\n\r\nchunks."))
+		}
+
+		wg.Done()
+	})
+
+	output := NewHTTPOutput(listener.Addr().String(), headers, methods, HTTPUrlRegexp{}, HTTPHeaderFilters{}, HTTPHeaderHashFilters{}, "", UrlRewriteMap{}, 0)
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	go Start(quit)
+
+	wg.Add(1)
+	input.EmitChunkedPOST()
+
+	wg.Wait()
+
+	close(quit)
+}
+
 func BenchmarkHTTPOutput(b *testing.B) {
 	wg := new(sync.WaitGroup)
 	quit := make(chan int)

settings_header_hash_filters.go

@@ -39,7 +39,7 @@ func (h *HTTPHeaderHashFilters) Set(value string) error {
 		panic("need positive numerators and denominators, with the former less than the latter.")
 	}
 
-	if den & (den - 1) != 0 {
+	if den&(den-1) != 0 {
 		return errors.New("must have a denominator which is a power of two.")
 	}
 

test_input.go

@@ -28,7 +28,11 @@ func (i *TestInput) EmitGET() {
 }
 
 func (i *TestInput) EmitPOST() {
-	i.data <- []byte("POST /pub/WWW/ HTTP/1.1\nHost: www.w3.org\r\n\r\na=1&b=2\r\n\r\n")
+	i.data <- []byte("POST /pub/WWW/ HTTP/1.1\nHost: www.w3.org\r\n\r\na=1&b=2")
+}
+
+func (i *TestInput) EmitChunkedPOST() {
+	i.data <- []byte("POST /pub/WWW/ HTTP/1.1\nHost: www.w3.org\nTransfer-Encoding: chunked\r\n\r\n4\r\nWiki\r\n5\r\npedia\r\ne\r\n in\r\n\r\nchunks.\r\n0\r\n\r\n")
 }
 
 func (i *TestInput) EmitFile() {