Streaming fixes (#970)v1.21.0

- Allow DisablePreParseMultipartForm in combination with
StreamRequestBody.
- Support streaming into MultipartForm instead of reading the whole body
  first.
- Support calling ctx.PostBody() when streaming is enabled.

1 files changed, 90 insertions, 0 deletions

diff --git a/streaming_test.go b/streaming_test.go
index e99033c..a943cb8 100644
--- a/streaming_test.go
+++ b/streaming_test.go
@@ -6,10 +6,100 @@ import (
 	"io/ioutil"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/valyala/fasthttp/fasthttputil"
 )
 
+func TestStreamingPipeline(t *testing.T) {
+	t.Parallel()
+
+	reqS := `POST /one HTTP/1.1
+Host: example.com
+Content-Length: 10
+
+aaaaaaaaaa
+POST /two HTTP/1.1
+Host: example.com
+Content-Length: 10
+
+aaaaaaaaaa`
+
+	ln := fasthttputil.NewInmemoryListener()
+
+	s := &Server{
+		StreamRequestBody: true,
+		Handler: func(ctx *RequestCtx) {
+			body := ""
+			expected := "aaaaaaaaaa"
+			if string(ctx.Path()) == "/one" {
+				body = string(ctx.PostBody())
+			} else {
+				all, err := ioutil.ReadAll(ctx.RequestBodyStream())
+				if err != nil {
+					t.Error(err)
+				}
+				body = string(all)
+			}
+			if body != expected {
+				t.Errorf("expected %q got %q", expected, body)
+			}
+		},
+	}
+
+	ch := make(chan struct{})
+	go func() {
+		if err := s.Serve(ln); err != nil {
+			t.Errorf("unexpected error: %s", err)
+		}
+		close(ch)
+	}()
+
+	conn, err := ln.Dial()
+	if err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+	if _, err = conn.Write([]byte(reqS)); err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+
+	var resp Response
+	br := bufio.NewReader(conn)
+	respCh := make(chan struct{})
+	go func() {
+		if err := resp.Read(br); err != nil {
+			t.Errorf("error when reading response: %s", err)
+		}
+		if resp.StatusCode() != StatusOK {
+			t.Errorf("unexpected status code %d. Expecting %d", resp.StatusCode(), StatusOK)
+		}
+
+		if err := resp.Read(br); err != nil {
+			t.Errorf("error when reading response: %s", err)
+		}
+		if resp.StatusCode() != StatusOK {
+			t.Errorf("unexpected status code %d. Expecting %d", resp.StatusCode(), StatusOK)
+		}
+		close(respCh)
+	}()
+
+	select {
+	case <-respCh:
+	case <-time.After(time.Second):
+		t.Fatal("timeout")
+	}
+
+	if err := ln.Close(); err != nil {
+		t.Fatalf("error when closing listener: %s", err)
+	}
+
+	select {
+	case <-ch:
+	case <-time.After(time.Second):
+		t.Fatal("timeout when waiting for the server to stop")
+	}
+}
+
 func TestRequestStream(t *testing.T) {
 	body := createFixedBody(3)
 	chunkedBody := createChunkedBody(body)