Implement Unwrap method to responseAdapter to properly handle HTTP event-streaming (like Server Send Events) without buffering (#1090)

Implement Unwrap method to responseAdapter to properly handle HTTP
event-streaming (like Server Send Events).

Add unit test to check if response was flushed.

<!-- Include the issue number below -->
Fixes golang/go#27816

## Proposed Changes

* GoLang developers implemented the streaming events by immediate
flushing the response if detects the proper header. However, the
response writer need to implement Flush or Unwrap methods to make it
work. responseAdapter used in loadBalancer (in reverseProxy) was not
implementing it, so SSE events were pushed to clients **only if buffers
were overloaded** (4K bytes). This change fixes this behavior.

## Release Notes

```release-note
Fixed the issue with not flushing immediately HTTP text/event-streaming in jsessionid-lb.
```

Author: balamut-google

samples/apps/jsessionid-lb/pkg/loadbalancer.go

@@ -37,6 +37,11 @@ type responseAdapter struct {
 	stickyCookie  string
 }
 
+// Unwrap implements http.ResponseWriter to allow the immediate flushing for HTTP Streaming events
+func (a *responseAdapter) Unwrap() http.ResponseWriter {
+	return a.ResponseWriter
+}
+
 func (a *responseAdapter) WriteHeader(statusCode int) {
 	// Append the sticky cookie the response has a session cookie set.
 

samples/apps/jsessionid-lb/pkg/loadbalancer_test.go

@@ -17,10 +17,12 @@ package pkg
 import (
 	"bytes"
 	"errors"
+	"fmt"
 	"log"
 	"net"
 	"net/http"
 	"net/http/httptest"
+	"net/http/httputil"
 	"testing"
 )
 
@@ -240,6 +242,39 @@ func TestLoadBalancer_ServeHTTP(t *testing.T) {
 				AssertCookie(t, "SESSION", "http://1.1.1.1:8080", recorder)
 			},
 		},
+		"SSE Event stream flushed immediately": {
+			loadBalancer: func() *LoadBalancer {
+				backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+					w.Header().Set("Content-Type", "text/event-stream")
+					if flusher, ok := w.(http.Flusher); ok {
+						fmt.Fprintf(w, "data: 1\n\n")
+						flusher.Flush()
+					}
+				}))
+				lb := &LoadBalancer{
+					LookupIP: lookupOneIP,
+					ReverseProxy: &httputil.ReverseProxy{
+						Director: func(req *http.Request) {
+							req.URL.Scheme = "http"
+							req.URL.Host = backend.Listener.Addr().String()
+						},
+					},
+				}
+				defaultValues(nil, lb)
+				return lb
+			}(),
+			request: func() *http.Request {
+				req := defaultRequest()
+				req.Header.Set("Accept", "text/event-stream")
+				return req
+			}(),
+			checkResponse: func(t *testing.T, recorder *httptest.ResponseRecorder) {
+				if !recorder.Flushed {
+					t.Errorf("Expected response to be flushed for SSE")
+				}
+				AssertResponseCode(t, http.StatusOK, recorder)
+			},
+		},
 	} {
 		t.Run(tn, func(t *testing.T) {
 			defaultValues(t, tc.loadBalancer)