@@ -9,9 +9,12 @@ import (
99 "io"
1010 "log"
1111 "net/http"
12+ "net/http/httputil"
1213 "os"
14+ "strings"
1315 "time"
1416
17+ fhttputil "github.com/openfaas/faas-provider/httputil"
1518 "github.com/openfaas/faas/gateway/pkg/middleware"
1619 "github.com/openfaas/faas/gateway/types"
1720)
@@ -28,7 +31,10 @@ func MakeForwardingProxyHandler(proxy *types.HTTPClientReverseProxy,
2831 writeRequestURI = exists
2932 }
3033
34+ reverseProxy := makeRewriteProxy (baseURLResolver , urlPathTransformer )
35+
3136 return func (w http.ResponseWriter , r * http.Request ) {
37+
3238 baseURL := baseURLResolver .Resolve (r )
3339 originalURL := r .URL .String ()
3440 requestURL := urlPathTransformer .Transform (r )
@@ -39,13 +45,13 @@ func MakeForwardingProxyHandler(proxy *types.HTTPClientReverseProxy,
3945
4046 start := time .Now ()
4147
42- statusCode , err := forwardRequest (w , r , proxy .Client , baseURL , requestURL , proxy .Timeout , writeRequestURI , serviceAuthInjector )
43-
44- seconds := time .Since (start )
48+ statusCode , err := forwardRequest (w , r , proxy .Client , baseURL , requestURL , proxy .Timeout , writeRequestURI , serviceAuthInjector , reverseProxy )
4549 if err != nil {
4650 log .Printf ("error with upstream request to: %s, %s\n " , requestURL , err .Error ())
4751 }
4852
53+ seconds := time .Since (start )
54+
4955 for _ , notifier := range notifiers {
5056 notifier .Notify (r .Method , requestURL , originalURL , statusCode , "completed" , seconds )
5157 }
@@ -86,7 +92,12 @@ func forwardRequest(w http.ResponseWriter,
8692 requestURL string ,
8793 timeout time.Duration ,
8894 writeRequestURI bool ,
89- serviceAuthInjector middleware.AuthInjector ) (int , error ) {
95+ serviceAuthInjector middleware.AuthInjector ,
96+ reverseProxy * httputil.ReverseProxy ) (int , error ) {
97+
98+ if r .Body != nil {
99+ defer r .Body .Close ()
100+ }
90101
91102 upstreamReq := buildUpstreamRequest (r , baseURL , requestURL )
92103 if upstreamReq .Body != nil {
@@ -101,14 +112,20 @@ func forwardRequest(w http.ResponseWriter,
101112 log .Printf ("forwardRequest: %s %s\n " , upstreamReq .Host , upstreamReq .URL .String ())
102113 }
103114
115+ if strings .HasPrefix (r .Header .Get ("Accept" ), "text/event-stream" ) {
116+ ww := fhttputil .NewHttpWriteInterceptor (w )
117+ reverseProxy .ServeHTTP (ww , upstreamReq )
118+ return ww .Status (), nil
119+ }
120+
104121 ctx , cancel := context .WithTimeout (r .Context (), timeout )
105122 defer cancel ()
106123
107- res , resErr := proxyClient .Do (upstreamReq .WithContext (ctx ))
108- if resErr != nil {
124+ res , err := proxyClient .Do (upstreamReq .WithContext (ctx ))
125+ if err != nil {
109126 badStatus := http .StatusBadGateway
110127 w .WriteHeader (badStatus )
111- return badStatus , resErr
128+ return badStatus , err
112129 }
113130
114131 if res .Body != nil {
@@ -117,12 +134,10 @@ func forwardRequest(w http.ResponseWriter,
117134
118135 copyHeaders (w .Header (), & res .Header )
119136
120- // Write status code
121137 w .WriteHeader (res .StatusCode )
122138
123139 if res .Body != nil {
124- // Copy the body over
125- io .CopyBuffer (w , res .Body , nil )
140+ io .Copy (w , res .Body )
126141 }
127142
128143 return res .StatusCode , nil
@@ -159,3 +174,14 @@ var hopHeaders = []string{
159174 "Transfer-Encoding" ,
160175 "Upgrade" ,
161176}
177+
178+ func makeRewriteProxy (baseURLResolver middleware.BaseURLResolver , urlPathTransformer middleware.URLPathTransformer ) * httputil.ReverseProxy {
179+ return & httputil.ReverseProxy {
180+ ErrorLog : log .New (io .Discard , "proxy:" , 0 ),
181+ Transport : http .DefaultClient .Transport ,
182+ ErrorHandler : func (w http.ResponseWriter , r * http.Request , err error ) {
183+ },
184+ Director : func (r * http.Request ) {
185+ },
186+ }
187+ }
0 commit comments