From 97558f6a6ec4c53fa69035fb430ead209d9c222d Mon Sep 17 00:00:00 2001 From: Joram Wilander Date: Fri, 13 Jan 2017 13:53:37 -0500 Subject: PLT-4938 Add app package and move logic over from api package (#4931) * Add app package and move logic over from api package * Change app package functions to return errors * Move non-api tests into app package * Fix merge --- app/web_conn.go | 254 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 254 insertions(+) create mode 100644 app/web_conn.go (limited to 'app/web_conn.go') diff --git a/app/web_conn.go b/app/web_conn.go new file mode 100644 index 000000000..02c3b2642 --- /dev/null +++ b/app/web_conn.go @@ -0,0 +1,254 @@ +// Copyright (c) 2015 Mattermost, Inc. All Rights Reserved. +// See License.txt for license information. + +package app + +import ( + "fmt" + "time" + + "github.com/mattermost/platform/einterfaces" + "github.com/mattermost/platform/model" + "github.com/mattermost/platform/utils" + + l4g "github.com/alecthomas/log4go" + "github.com/gorilla/websocket" + goi18n "github.com/nicksnyder/go-i18n/i18n" +) + +const ( + WRITE_WAIT = 30 * time.Second + PONG_WAIT = 100 * time.Second + PING_PERIOD = (PONG_WAIT * 6) / 10 + AUTH_TIMEOUT = 5 * time.Second +) + +type WebConn struct { + WebSocket *websocket.Conn + Send chan model.WebSocketMessage + SessionToken string + SessionExpiresAt int64 + UserId string + T goi18n.TranslateFunc + Locale string + AllChannelMembers map[string]string + LastAllChannelMembersTime int64 +} + +func NewWebConn(ws *websocket.Conn, session model.Session, t goi18n.TranslateFunc, locale string) *WebConn { + if len(session.UserId) > 0 { + go SetStatusOnline(session.UserId, session.Id, false) + } + + return &WebConn{ + Send: make(chan model.WebSocketMessage, 256), + WebSocket: ws, + UserId: session.UserId, + SessionToken: session.Token, + SessionExpiresAt: session.ExpiresAt, + T: t, + Locale: locale, + } +} + +func (c *WebConn) ReadPump() { + defer func() { + HubUnregister(c) + c.WebSocket.Close() + }() + c.WebSocket.SetReadLimit(model.SOCKET_MAX_MESSAGE_SIZE_KB) + c.WebSocket.SetReadDeadline(time.Now().Add(PONG_WAIT)) + c.WebSocket.SetPongHandler(func(string) error { + c.WebSocket.SetReadDeadline(time.Now().Add(PONG_WAIT)) + if c.IsAuthenticated() { + go SetStatusAwayIfNeeded(c.UserId, false) + } + return nil + }) + + for { + var req model.WebSocketRequest + if err := c.WebSocket.ReadJSON(&req); err != nil { + // browsers will appear as CloseNoStatusReceived + if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) { + l4g.Debug(fmt.Sprintf("websocket.read: client side closed socket userId=%v", c.UserId)) + } else { + l4g.Debug(fmt.Sprintf("websocket.read: closing websocket for userId=%v error=%v", c.UserId, err.Error())) + } + + return + } else { + Srv.WebSocketRouter.ServeWebSocket(c, &req) + } + } +} + +func (c *WebConn) WritePump() { + ticker := time.NewTicker(PING_PERIOD) + authTicker := time.NewTicker(AUTH_TIMEOUT) + + defer func() { + ticker.Stop() + authTicker.Stop() + c.WebSocket.Close() + }() + + for { + select { + case msg, ok := <-c.Send: + if !ok { + c.WebSocket.SetWriteDeadline(time.Now().Add(WRITE_WAIT)) + c.WebSocket.WriteMessage(websocket.CloseMessage, []byte{}) + return + } + + c.WebSocket.SetWriteDeadline(time.Now().Add(WRITE_WAIT)) + if err := c.WebSocket.WriteMessage(websocket.TextMessage, msg.GetPreComputeJson()); err != nil { + // browsers will appear as CloseNoStatusReceived + if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) { + l4g.Debug(fmt.Sprintf("websocket.send: client side closed socket userId=%v", c.UserId)) + } else { + l4g.Debug(fmt.Sprintf("websocket.send: closing websocket for userId=%v, error=%v", c.UserId, err.Error())) + } + + return + } + + if msg.EventType() == model.WEBSOCKET_EVENT_POSTED { + if einterfaces.GetMetricsInterface() != nil { + einterfaces.GetMetricsInterface().IncrementPostBroadcast() + } + } + + case <-ticker.C: + c.WebSocket.SetWriteDeadline(time.Now().Add(WRITE_WAIT)) + if err := c.WebSocket.WriteMessage(websocket.PingMessage, []byte{}); err != nil { + // browsers will appear as CloseNoStatusReceived + if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) { + l4g.Debug(fmt.Sprintf("websocket.ticker: client side closed socket userId=%v", c.UserId)) + } else { + l4g.Debug(fmt.Sprintf("websocket.ticker: closing websocket for userId=%v error=%v", c.UserId, err.Error())) + } + + return + } + + case <-authTicker.C: + if c.SessionToken == "" { + l4g.Debug(fmt.Sprintf("websocket.authTicker: did not authenticate ip=%v", c.WebSocket.RemoteAddr())) + return + } + authTicker.Stop() + } + } +} + +func (webCon *WebConn) InvalidateCache() { + webCon.AllChannelMembers = nil + webCon.LastAllChannelMembersTime = 0 + webCon.SessionExpiresAt = 0 +} + +func (webCon *WebConn) IsAuthenticated() bool { + // Check the expiry to see if we need to check for a new session + if webCon.SessionExpiresAt < model.GetMillis() { + if webCon.SessionToken == "" { + return false + } + + session, err := GetSession(webCon.SessionToken) + if err != nil { + l4g.Error(utils.T("api.websocket.invalid_session.error"), err.Error()) + webCon.SessionToken = "" + webCon.SessionExpiresAt = 0 + return false + } + + webCon.SessionToken = session.Token + webCon.SessionExpiresAt = session.ExpiresAt + } + + return true +} + +func (webCon *WebConn) SendHello() { + msg := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_HELLO, "", "", webCon.UserId, nil) + msg.Add("server_version", fmt.Sprintf("%v.%v.%v", model.CurrentVersion, model.BuildNumber, utils.CfgHash)) + msg.DoPreComputeJson() + webCon.Send <- msg +} + +func (webCon *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool { + // IMPORTANT: Do not send event if WebConn does not have a session + if !webCon.IsAuthenticated() { + return false + } + + // If the event is destined to a specific user + if len(msg.Broadcast.UserId) > 0 && webCon.UserId != msg.Broadcast.UserId { + return false + } + + // if the user is omitted don't send the message + if len(msg.Broadcast.OmitUsers) > 0 { + if _, ok := msg.Broadcast.OmitUsers[webCon.UserId]; ok { + return false + } + } + + // Only report events to users who are in the channel for the event + if len(msg.Broadcast.ChannelId) > 0 { + + // Only broadcast typing messages if less than 1K people in channel + if msg.Event == model.WEBSOCKET_EVENT_TYPING { + if Srv.Store.Channel().GetMemberCountFromCache(msg.Broadcast.ChannelId) > *utils.Cfg.TeamSettings.MaxNotificationsPerChannel { + return false + } + } + + if model.GetMillis()-webCon.LastAllChannelMembersTime > 1000*60*15 { // 15 minutes + webCon.AllChannelMembers = nil + webCon.LastAllChannelMembersTime = 0 + } + + if webCon.AllChannelMembers == nil { + if result := <-Srv.Store.Channel().GetAllChannelMembersForUser(webCon.UserId, true); result.Err != nil { + l4g.Error("webhub.shouldSendEvent: " + result.Err.Error()) + return false + } else { + webCon.AllChannelMembers = result.Data.(map[string]string) + webCon.LastAllChannelMembersTime = model.GetMillis() + } + } + + if _, ok := webCon.AllChannelMembers[msg.Broadcast.ChannelId]; ok { + return true + } else { + return false + } + } + + // Only report events to users who are in the team for the event + if len(msg.Broadcast.TeamId) > 0 { + return webCon.IsMemberOfTeam(msg.Broadcast.TeamId) + + } + + return true +} + +func (webCon *WebConn) IsMemberOfTeam(teamId string) bool { + session, err := GetSession(webCon.SessionToken) + if err != nil { + l4g.Error(utils.T("api.websocket.invalid_session.error"), err.Error()) + return false + } else { + member := session.GetTeamByTeamId(teamId) + + if member != nil { + return true + } else { + return false + } + } +} -- cgit v1.2.3-1-g7c22