Add compression and better error handling
This commit is contained in:
parent
8239e04a19
commit
b5ee5c34f2
3 changed files with 35 additions and 11 deletions
2
Gopkg.lock
generated
2
Gopkg.lock
generated
|
@ -1026,7 +1026,7 @@
|
||||||
"roundrobin",
|
"roundrobin",
|
||||||
"utils"
|
"utils"
|
||||||
]
|
]
|
||||||
revision = "812cebb8c764f2a78cb806267648b8728b4599ad"
|
revision = "fd0f370c961f6aa304379f4106e76ffe5ed7e97a"
|
||||||
source = "https://github.com/containous/oxy.git"
|
source = "https://github.com/containous/oxy.git"
|
||||||
|
|
||||||
[[projects]]
|
[[projects]]
|
||||||
|
|
43
vendor/github.com/vulcand/oxy/forward/fwd.go
generated
vendored
43
vendor/github.com/vulcand/oxy/forward/fwd.go
generated
vendored
|
@ -5,6 +5,7 @@ package forward
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"crypto/tls"
|
"crypto/tls"
|
||||||
|
"fmt"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
"net/http/httputil"
|
"net/http/httputil"
|
||||||
|
@ -281,6 +282,9 @@ func (f *httpForwarder) serveWebSocket(w http.ResponseWriter, req *http.Request,
|
||||||
outReq := f.copyWebSocketRequest(req)
|
outReq := f.copyWebSocketRequest(req)
|
||||||
|
|
||||||
dialer := websocket.DefaultDialer
|
dialer := websocket.DefaultDialer
|
||||||
|
|
||||||
|
dialer.EnableCompression = strings.Contains(req.Header.Get("Sec-Websocket-Extensions"), "permessage-deflate")
|
||||||
|
|
||||||
if outReq.URL.Scheme == "wss" && f.tlsClientConfig != nil {
|
if outReq.URL.Scheme == "wss" && f.tlsClientConfig != nil {
|
||||||
dialer.TLSClientConfig = f.tlsClientConfig.Clone()
|
dialer.TLSClientConfig = f.tlsClientConfig.Clone()
|
||||||
// WebSocket is only in http/1.1
|
// WebSocket is only in http/1.1
|
||||||
|
@ -322,7 +326,10 @@ func (f *httpForwarder) serveWebSocket(w http.ResponseWriter, req *http.Request,
|
||||||
return true
|
return true
|
||||||
}}
|
}}
|
||||||
|
|
||||||
|
upgrader.EnableCompression = strings.Contains(resp.Header.Get("Sec-Websocket-Extensions"), "permessage-deflate")
|
||||||
|
|
||||||
utils.RemoveHeaders(resp.Header, WebsocketUpgradeHeaders...)
|
utils.RemoveHeaders(resp.Header, WebsocketUpgradeHeaders...)
|
||||||
|
|
||||||
underlyingConn, err := upgrader.Upgrade(w, req, resp.Header)
|
underlyingConn, err := upgrader.Upgrade(w, req, resp.Header)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Errorf("vulcand/oxy/forward/websocket: Error while upgrading connection : %v", err)
|
log.Errorf("vulcand/oxy/forward/websocket: Error while upgrading connection : %v", err)
|
||||||
|
@ -331,29 +338,45 @@ func (f *httpForwarder) serveWebSocket(w http.ResponseWriter, req *http.Request,
|
||||||
defer underlyingConn.Close()
|
defer underlyingConn.Close()
|
||||||
defer targetConn.Close()
|
defer targetConn.Close()
|
||||||
|
|
||||||
errc := make(chan error, 2)
|
errClient := make(chan error)
|
||||||
|
errBackend := make(chan error)
|
||||||
replicateWebsocketConn := func(dst, src *websocket.Conn, dstName, srcName string) {
|
replicateWebsocketConn := func(dst, src *websocket.Conn, errc chan error) {
|
||||||
var err error
|
|
||||||
for {
|
for {
|
||||||
msgType, msg, err := src.ReadMessage()
|
msgType, msg, err := src.ReadMessage()
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
f.log.Errorf("vulcand/oxy/forward/websocket: Error when copying from %s to %s using ReadMessage: %v", srcName, dstName, err)
|
m := websocket.FormatCloseMessage(websocket.CloseNormalClosure, fmt.Sprintf("%v", err))
|
||||||
|
if e, ok := err.(*websocket.CloseError); ok {
|
||||||
|
if e.Code != websocket.CloseNoStatusReceived {
|
||||||
|
m = websocket.FormatCloseMessage(e.Code, e.Text)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
errc <- err
|
||||||
|
dst.WriteMessage(websocket.CloseMessage, m)
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
err = dst.WriteMessage(msgType, msg)
|
err = dst.WriteMessage(msgType, msg)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
f.log.Errorf("vulcand/oxy/forward/websocket: Error when copying from %s to %s using WriteMessage: %v", srcName, dstName, err)
|
errc <- err
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
errc <- err
|
|
||||||
}
|
}
|
||||||
|
|
||||||
go replicateWebsocketConn(underlyingConn, targetConn, "client", "backend")
|
go replicateWebsocketConn(underlyingConn, targetConn, errClient)
|
||||||
go replicateWebsocketConn(targetConn, underlyingConn, "backend", "client")
|
go replicateWebsocketConn(targetConn, underlyingConn, errBackend)
|
||||||
|
|
||||||
<-errc
|
var message string
|
||||||
|
select {
|
||||||
|
case err = <-errClient:
|
||||||
|
message = "vulcand/oxy/forward/websocket: Error when copying from backend to client: %v"
|
||||||
|
case err = <-errBackend:
|
||||||
|
message = "vulcand/oxy/forward/websocket: Error when copying from client to backend: %v"
|
||||||
|
|
||||||
|
}
|
||||||
|
if e, ok := err.(*websocket.CloseError); !ok || e.Code == websocket.CloseAbnormalClosure {
|
||||||
|
f.log.Errorf(message, err)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// copyWebsocketRequest makes a copy of the specified request.
|
// copyWebsocketRequest makes a copy of the specified request.
|
||||||
|
|
1
vendor/github.com/vulcand/oxy/forward/headers.go
generated
vendored
1
vendor/github.com/vulcand/oxy/forward/headers.go
generated
vendored
|
@ -49,6 +49,7 @@ var WebsocketUpgradeHeaders = []string{
|
||||||
Upgrade,
|
Upgrade,
|
||||||
Connection,
|
Connection,
|
||||||
SecWebsocketAccept,
|
SecWebsocketAccept,
|
||||||
|
SecWebsocketExtensions,
|
||||||
}
|
}
|
||||||
|
|
||||||
var XHeaders = []string{
|
var XHeaders = []string{
|
||||||
|
|
Loading…
Reference in a new issue