b7a7d8e7b6
* MM-39612: Make acquiring and removing connections atomic The reconnect phase of a websocket was split into two parts: one where we check if a connection with a given connectionID exists or not. And second, where we remove that connection and insert the new connection again in the index. This would lead to a race where it would be possible for 2 concurrent requests for the same connectionID to go through which would lead to separate goroutines working on the same dead queue. We simplify this by removing the connection from the index in the check connection stage itself. And then just add that during register phase. And to distinguish between a fresh and an old connection, we add a new field called reuseCount. While here, we also cleanup some old comments and add more in some places. https://mattermost.atlassian.net/browse/MM-39612 ```release-note NONE ``` * remove unused method ```release-note NONE ``` * race test ```release-note NONE ```
114 строки
2.9 KiB
Go
114 строки
2.9 KiB
Go
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
|
|
// See LICENSE.txt for license information.
|
|
|
|
package app
|
|
|
|
import (
|
|
"net/http"
|
|
|
|
"github.com/mattermost/mattermost-server/v6/model"
|
|
"github.com/mattermost/mattermost-server/v6/shared/i18n"
|
|
"github.com/mattermost/mattermost-server/v6/shared/mlog"
|
|
)
|
|
|
|
type webSocketHandler interface {
|
|
ServeWebSocket(*WebConn, *model.WebSocketRequest)
|
|
}
|
|
|
|
type WebSocketRouter struct {
|
|
handlers map[string]webSocketHandler
|
|
}
|
|
|
|
func (wr *WebSocketRouter) Handle(action string, handler webSocketHandler) {
|
|
wr.handlers[action] = handler
|
|
}
|
|
|
|
func (wr *WebSocketRouter) ServeWebSocket(conn *WebConn, r *model.WebSocketRequest) {
|
|
if r.Action == "" {
|
|
err := model.NewAppError("ServeWebSocket", "api.web_socket_router.no_action.app_error", nil, "", http.StatusBadRequest)
|
|
returnWebSocketError(conn.App, conn, r, err)
|
|
return
|
|
}
|
|
|
|
if r.Seq <= 0 {
|
|
err := model.NewAppError("ServeWebSocket", "api.web_socket_router.bad_seq.app_error", nil, "", http.StatusBadRequest)
|
|
returnWebSocketError(conn.App, conn, r, err)
|
|
return
|
|
}
|
|
|
|
if r.Action == model.WebsocketAuthenticationChallenge {
|
|
if conn.GetSessionToken() != "" {
|
|
return
|
|
}
|
|
|
|
token, ok := r.Data["token"].(string)
|
|
if !ok {
|
|
conn.WebSocket.Close()
|
|
return
|
|
}
|
|
|
|
session, err := conn.App.GetSession(token)
|
|
if err != nil {
|
|
conn.WebSocket.Close()
|
|
return
|
|
}
|
|
conn.SetSession(session)
|
|
conn.SetSessionToken(session.Token)
|
|
conn.UserId = session.UserId
|
|
|
|
conn.App.HubRegister(conn)
|
|
|
|
conn.App.Srv().Go(func() {
|
|
conn.App.SetStatusOnline(session.UserId, false)
|
|
conn.App.UpdateLastActivityAtIfNeeded(*session)
|
|
})
|
|
|
|
resp := model.NewWebSocketResponse(model.StatusOk, r.Seq, nil)
|
|
hub := conn.App.GetHubForUserId(conn.UserId)
|
|
if hub == nil {
|
|
return
|
|
}
|
|
hub.SendMessage(conn, resp)
|
|
|
|
return
|
|
}
|
|
|
|
if !conn.IsAuthenticated() {
|
|
err := model.NewAppError("ServeWebSocket", "api.web_socket_router.not_authenticated.app_error", nil, "", http.StatusUnauthorized)
|
|
returnWebSocketError(conn.App, conn, r, err)
|
|
return
|
|
}
|
|
|
|
handler, ok := wr.handlers[r.Action]
|
|
if !ok {
|
|
err := model.NewAppError("ServeWebSocket", "api.web_socket_router.bad_action.app_error", nil, "", http.StatusInternalServerError)
|
|
returnWebSocketError(conn.App, conn, r, err)
|
|
return
|
|
}
|
|
|
|
handler.ServeWebSocket(conn, r)
|
|
}
|
|
|
|
func returnWebSocketError(app *App, conn *WebConn, r *model.WebSocketRequest, err *model.AppError) {
|
|
logF := mlog.Error
|
|
if err.StatusCode >= http.StatusBadRequest && err.StatusCode < http.StatusInternalServerError {
|
|
logF = mlog.Debug
|
|
}
|
|
logF(
|
|
"websocket routing error.",
|
|
mlog.Int64("seq", r.Seq),
|
|
mlog.String("user_id", conn.UserId),
|
|
mlog.String("system_message", err.SystemMessage(i18n.T)),
|
|
mlog.Err(err),
|
|
)
|
|
|
|
hub := app.GetHubForUserId(conn.UserId)
|
|
if hub == nil {
|
|
return
|
|
}
|
|
|
|
err.DetailedError = ""
|
|
errorResp := model.NewWebSocketError(r.Seq, err)
|
|
hub.SendMessage(conn, errorResp)
|
|
}
|