net/http/httputil: rewrite flushing code, disable on Server-Sent Events

* Rewrite the flushing code to not use a persistent goroutine, which
  also simplifies testing.
* Define the meaning of a negative flush interval. Its meaning doesn't
  change, but now it's locked in, and then we can use it to optimize
  the performance of the non-buffered case to avoid use of an AfterFunc.
* Support (internal-only) special casing of FlushInterval values per
  request/response.
* For now, treat Server-Sent Event responses as unbuffered. (or rather,
  immediately flushed from the buffer per-write)

Fixes #27816

Change-Id: Ie0f975c997daa3db539504137c741a96d7022665
Reviewed-on: https://go-review.googlesource.com/c/137335
Run-TryBot: Brad Fitzpatrick <bradfitz@golang.org>
TryBot-Result: Gobot Gobot <gobot@golang.org>
Reviewed-by: Dmitri Shuralyov <dmitshur@golang.org>

1 files changed, 45 insertions, 11 deletions

diff --git a/src/net/http/httputil/reverseproxy_test.go b/src/net/http/httputil/reverseproxy_test.go
index 2f75b4e34e..ddae11b168 100644
--- a/src/net/http/httputil/reverseproxy_test.go
+++ b/src/net/http/httputil/reverseproxy_test.go
@@ -297,10 +297,6 @@ func TestReverseProxyFlushInterval(t *testing.T) {
 	proxyHandler := NewSingleHostReverseProxy(backendURL)
 	proxyHandler.FlushInterval = time.Microsecond
 
-	done := make(chan bool)
-	onExitFlushLoop = func() { done <- true }
-	defer func() { onExitFlushLoop = nil }()
-
 	frontend := httptest.NewServer(proxyHandler)
 	defer frontend.Close()
 
@@ -314,13 +310,6 @@ func TestReverseProxyFlushInterval(t *testing.T) {
 	if bodyBytes, _ := ioutil.ReadAll(res.Body); string(bodyBytes) != expected {
 		t.Errorf("got body %q; expected %q", bodyBytes, expected)
 	}
-
-	select {
-	case <-done:
-		// OK
-	case <-time.After(5 * time.Second):
-		t.Error("maxLatencyWriter flushLoop() never exited")
-	}
 }
 
 func TestReverseProxyCancelation(t *testing.T) {
@@ -946,3 +935,48 @@ func TestReverseProxy_PanicBodyError(t *testing.T) {
 	req, _ := http.NewRequest("GET", "http://foo.tld/", nil)
 	rproxy.ServeHTTP(httptest.NewRecorder(), req)
 }
+
+func TestSelectFlushInterval(t *testing.T) {
+	tests := []struct {
+		name string
+		p    *ReverseProxy
+		req  *http.Request
+		res  *http.Response
+		want time.Duration
+	}{
+		{
+			name: "default",
+			res:  &http.Response{},
+			p:    &ReverseProxy{FlushInterval: 123},
+			want: 123,
+		},
+		{
+			name: "server-sent events overrides non-zero",
+			res: &http.Response{
+				Header: http.Header{
+					"Content-Type": {"text/event-stream"},
+				},
+			},
+			p:    &ReverseProxy{FlushInterval: 123},
+			want: -1,
+		},
+		{
+			name: "server-sent events overrides zero",
+			res: &http.Response{
+				Header: http.Header{
+					"Content-Type": {"text/event-stream"},
+				},
+			},
+			p:    &ReverseProxy{FlushInterval: 0},
+			want: -1,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := tt.p.flushInterval(tt.req, tt.res)
+			if got != tt.want {
+				t.Errorf("flushLatency = %v; want %v", got, tt.want)
+			}
+		})
+	}
+}