Support forwarding of ndjson streams

Additional fixes:
- Support Accept headers with multiple Accept values
- Accept header values should be matched case insensitive.
  As stated in RFC 7321: https://datatracker.ietf.org/doc/html/rfc7231#section-3.1.1.1
  Media Types are case-insensitive

Signed-off-by: Han Verstraete (OpenFaaS Ltd) <[email protected]>

Author: welteki

Makefile

@@ -56,7 +56,7 @@ print-image:
 # Example:
 # SERVER=docker.io OWNER=alexellis2 TAG=ready make publish
 .PHONY: publish
-publish:
+publish: dist
 	@echo  $(SERVER)/$(OWNER)/$(IMG_NAME):$(TAG) && \
 	docker buildx create --use --name=multiarch --node=multiarch && \
 	docker buildx build \

executor/http_runner.go

@@ -132,8 +132,7 @@ func (f *HTTPFunctionRunner) Run(req FunctionRequest, contentLength int64, r *ht
 	}
 	defer cancel()
 
-	if strings.HasPrefix(r.Header.Get("Accept"), "text/event-stream") ||
-		r.Header.Get("Upgrade") == "websocket" {
+	if requiresStdlibProxy(r) {
 		ww := fhttputil.NewHttpWriteInterceptor(w)
 
 		f.ReverseProxy.ServeHTTP(w, r)
@@ -251,3 +250,13 @@ func makeProxyClient(dialTimeout time.Duration) *http.Client {
 
 	return &proxyClient
 }
+
+// requiresStdlibProxy checks if the request should be proxied using the standard library reverse proxy.
+// Support SSE, NDSJON and WebSockets through the stdlib reverse proxy
+func requiresStdlibProxy(req *http.Request) bool {
+	acceptHeader := strings.ToLower(req.Header.Get("Accept"))
+
+	return strings.Contains(acceptHeader, "text/event-stream") ||
+		strings.Contains(acceptHeader, "application/x-ndjson") ||
+		req.Header.Get("Upgrade") == "websocket"
+}

executor/http_runner_test.go

@@ -1,6 +1,7 @@
 package executor
 
 import (
+	"net/http"
 	"net/http/httptest"
 	"testing"
 	"time"
@@ -47,3 +48,83 @@ func TestGetTimeout_NoDefaultMeansNoOverride(t *testing.T) {
 		t.Errorf("getTimeout() got: %v, want %v", got, want)
 	}
 }
+
+func Test_requiresStdlibProxy(t *testing.T) {
+	testCases := []struct {
+		name    string
+		headers map[string]string
+		want    bool
+	}{
+		{
+			name:    "SSE request",
+			headers: map[string]string{"Accept": "text/event-stream"},
+			want:    true,
+		},
+		{
+			name:    "SSE request with multiple accept values",
+			headers: map[string]string{"Accept": "application/json, text/event-stream;q=0.9, text/plain"},
+			want:    true,
+		},
+		{
+			name:    "NDJSON request",
+			headers: map[string]string{"Accept": "application/x-ndjson"},
+			want:    true,
+		},
+		{
+			name:    "NDJSON request with multiple accept values",
+			headers: map[string]string{"Accept": "text/plain, application/x-ndjson;q=0.9, application/json;q=0.8"},
+			want:    true,
+		},
+		{
+			name:    "WebSocket request",
+			headers: map[string]string{"Upgrade": "websocket"},
+			want:    true,
+		},
+		{
+			name:    "Regular JSON request",
+			headers: map[string]string{"Accept": "application/json"},
+			want:    false,
+		},
+		{
+			name:    "Regular request with multiple values",
+			headers: map[string]string{"Accept": "text/plain, application/json;q=0.9"},
+			want:    false,
+		},
+		{
+			name:    "Request without headers",
+			headers: map[string]string{},
+			want:    false,
+		},
+		{
+			name:    "Request with non-websocket Upgrade header",
+			headers: map[string]string{"Accept": "application/json", "Upgrade": "h2c"},
+			want:    false,
+		},
+
+		{
+			name:    "Case insensitive headers",
+			headers: map[string]string{"Accept": "APPLICATION/X-NDJSON"},
+			want:    true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			req, err := http.NewRequest("GET", "/test", nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			// Set headers from test case
+			for key, value := range tc.headers {
+				req.Header.Set(key, value)
+			}
+
+			got := requiresStdlibProxy(req)
+
+			if got != tc.want {
+				t.Errorf("Want %t, got %t", tc.want, got)
+			}
+		})
+	}
+}