Controller API: correctly detect WebSocket closure in Watch RPC (#259)
This commit is contained in:
parent
2dfdb40dea
commit
818f4288c2
|
|
@ -4,6 +4,7 @@ import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
|
"fmt"
|
||||||
"github.com/cirruslabs/orchard/internal/responder"
|
"github.com/cirruslabs/orchard/internal/responder"
|
||||||
v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
|
v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
|
||||||
"github.com/cirruslabs/orchard/rpc"
|
"github.com/cirruslabs/orchard/rpc"
|
||||||
|
|
@ -41,7 +42,7 @@ func (controller *Controller) rpcWatch(ctx *gin.Context) responder.Responder {
|
||||||
// from the connection in the background
|
// from the connection in the background
|
||||||
//
|
//
|
||||||
// Otherwise the wsConn.Ping() will wait forever.
|
// Otherwise the wsConn.Ping() will wait forever.
|
||||||
wsConn.CloseRead(ctx)
|
closeReadCtx := wsConn.CloseRead(ctx)
|
||||||
|
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
|
|
@ -86,10 +87,14 @@ func (controller *Controller) rpcWatch(ctx *gin.Context) responder.Responder {
|
||||||
}
|
}
|
||||||
|
|
||||||
pingCtxCancel()
|
pingCtxCancel()
|
||||||
|
case <-closeReadCtx.Done():
|
||||||
|
// Connection shouldn't be normally closed by the worker
|
||||||
|
return controller.wsErrorNoClose("watch RPC",
|
||||||
|
fmt.Sprintf("worker %s unexpectedly disconnected", workerName), closeReadCtx.Err())
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
// Connection shouldn't be normally closed by the worker
|
// Connection shouldn't be normally closed by the worker
|
||||||
return controller.wsError(wsConn, websocket.StatusAbnormalClosure, "watch RPC",
|
return controller.wsErrorNoClose("watch RPC",
|
||||||
"unexpectedly disconnected worker", err)
|
fmt.Sprintf("worker %s unexpectedly disconnected", workerName), ctx.Err())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -13,14 +13,22 @@ func (controller *Controller) wsError(
|
||||||
reason string,
|
reason string,
|
||||||
err error,
|
err error,
|
||||||
) responder.Responder {
|
) responder.Responder {
|
||||||
message := fmt.Sprintf("%s: %v", reason, err)
|
responder := controller.wsErrorNoClose(component, reason, err)
|
||||||
|
|
||||||
controller.logger.Warn(message)
|
if err := wsConn.Close(code, fmt.Sprintf("%s: %v", reason, err)); err != nil {
|
||||||
|
|
||||||
if err := wsConn.Close(code, message); err != nil {
|
|
||||||
controller.logger.Warnf("%s: failed to close the WebSocket connection that entered error state"+
|
controller.logger.Warnf("%s: failed to close the WebSocket connection that entered error state"+
|
||||||
" due to %s: %v", component, reason, err)
|
" due to %s: %v", component, reason, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
return responder
|
||||||
|
}
|
||||||
|
|
||||||
|
func (controller *Controller) wsErrorNoClose(
|
||||||
|
component string,
|
||||||
|
reason string,
|
||||||
|
err error,
|
||||||
|
) responder.Responder {
|
||||||
|
controller.logger.Warnf("%s: %s: %v", component, reason, err)
|
||||||
|
|
||||||
return responder.Empty()
|
return responder.Empty()
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue