mirror of
https://github.com/caddyserver/caddy.git
synced 2024-12-27 14:13:48 +03:00
reverseproxy: Wait for both ends of websocket to close (#6175)
This commit is contained in:
parent
81413caea2
commit
b40cacf5ce
2 changed files with 9 additions and 3 deletions
|
@ -922,7 +922,9 @@ func (h *Handler) finalizeResponse(
|
||||||
) error {
|
) error {
|
||||||
// deal with 101 Switching Protocols responses: (WebSocket, h2c, etc)
|
// deal with 101 Switching Protocols responses: (WebSocket, h2c, etc)
|
||||||
if res.StatusCode == http.StatusSwitchingProtocols {
|
if res.StatusCode == http.StatusSwitchingProtocols {
|
||||||
h.handleUpgradeResponse(logger, rw, req, res)
|
var wg sync.WaitGroup
|
||||||
|
h.handleUpgradeResponse(logger, &wg, rw, req, res)
|
||||||
|
wg.Wait()
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -34,7 +34,7 @@ import (
|
||||||
"golang.org/x/net/http/httpguts"
|
"golang.org/x/net/http/httpguts"
|
||||||
)
|
)
|
||||||
|
|
||||||
func (h *Handler) handleUpgradeResponse(logger *zap.Logger, rw http.ResponseWriter, req *http.Request, res *http.Response) {
|
func (h *Handler) handleUpgradeResponse(logger *zap.Logger, wg *sync.WaitGroup, rw http.ResponseWriter, req *http.Request, res *http.Response) {
|
||||||
reqUpType := upgradeType(req.Header)
|
reqUpType := upgradeType(req.Header)
|
||||||
resUpType := upgradeType(res.Header)
|
resUpType := upgradeType(res.Header)
|
||||||
|
|
||||||
|
@ -121,7 +121,7 @@ func (h *Handler) handleUpgradeResponse(logger *zap.Logger, rw http.ResponseWrit
|
||||||
defer deleteFrontConn()
|
defer deleteFrontConn()
|
||||||
defer deleteBackConn()
|
defer deleteBackConn()
|
||||||
|
|
||||||
spc := switchProtocolCopier{user: conn, backend: backConn}
|
spc := switchProtocolCopier{user: conn, backend: backConn, wg: wg}
|
||||||
|
|
||||||
// setup the timeout if requested
|
// setup the timeout if requested
|
||||||
var timeoutc <-chan time.Time
|
var timeoutc <-chan time.Time
|
||||||
|
@ -132,6 +132,7 @@ func (h *Handler) handleUpgradeResponse(logger *zap.Logger, rw http.ResponseWrit
|
||||||
}
|
}
|
||||||
|
|
||||||
errc := make(chan error, 1)
|
errc := make(chan error, 1)
|
||||||
|
wg.Add(2)
|
||||||
go spc.copyToBackend(errc)
|
go spc.copyToBackend(errc)
|
||||||
go spc.copyFromBackend(errc)
|
go spc.copyFromBackend(errc)
|
||||||
select {
|
select {
|
||||||
|
@ -529,16 +530,19 @@ func (m *maxLatencyWriter) stop() {
|
||||||
// forth have nice names in stacks.
|
// forth have nice names in stacks.
|
||||||
type switchProtocolCopier struct {
|
type switchProtocolCopier struct {
|
||||||
user, backend io.ReadWriteCloser
|
user, backend io.ReadWriteCloser
|
||||||
|
wg *sync.WaitGroup
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c switchProtocolCopier) copyFromBackend(errc chan<- error) {
|
func (c switchProtocolCopier) copyFromBackend(errc chan<- error) {
|
||||||
_, err := io.Copy(c.user, c.backend)
|
_, err := io.Copy(c.user, c.backend)
|
||||||
errc <- err
|
errc <- err
|
||||||
|
c.wg.Done()
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c switchProtocolCopier) copyToBackend(errc chan<- error) {
|
func (c switchProtocolCopier) copyToBackend(errc chan<- error) {
|
||||||
_, err := io.Copy(c.backend, c.user)
|
_, err := io.Copy(c.backend, c.user)
|
||||||
errc <- err
|
errc <- err
|
||||||
|
c.wg.Done()
|
||||||
}
|
}
|
||||||
|
|
||||||
var streamingBufPool = sync.Pool{
|
var streamingBufPool = sync.Pool{
|
||||||
|
|
Loading…
Reference in a new issue