From e0b5b951f1dff2c224b7e296b8637c3384f70fc8 Mon Sep 17 00:00:00 2001 From: Agniva De Sarker Date: Tue, 12 Dec 2023 08:49:09 +0530 Subject: [PATCH] MM-56060: Create base scaffolding for websocket pub-sub (#25654) We create a new websocket action called "presence" which can contain the active_channel and the active_team for a given client connection. On the client side, for every channel or team switch, we send out this message. https://mattermost.atlassian.net/browse/MM-56060 ```release-note NONE ``` --- server/channels/api4/websocket_test.go | 23 +++++++++++- server/channels/app/platform/web_conn.go | 36 +++++++++++++++---- .../channels/app/platform/websocket_router.go | 19 ++++++++++ server/public/model/websocket_client.go | 16 +++++++++ server/public/model/websocket_message.go | 1 + .../src/actions/websocket_actions.jsx | 4 +++ .../components/channel_view/channel_view.tsx | 6 ++++ .../components/team_sidebar/team_sidebar.tsx | 8 +++++ webapp/platform/client/src/websocket.ts | 16 ++++++++- 9 files changed, 120 insertions(+), 9 deletions(-) diff --git a/server/channels/api4/websocket_test.go b/server/channels/api4/websocket_test.go index 6d0ec68875..f6687e8296 100644 --- a/server/channels/api4/websocket_test.go +++ b/server/channels/api4/websocket_test.go @@ -417,8 +417,29 @@ func TestWebSocketStatuses(t *testing.T) { require.True(t, awayHit, "didn't get away event") time.Sleep(500 * time.Millisecond) +} - WebSocketClient.Close() +func TestWebSocketPresence(t *testing.T) { + th := Setup(t).InitBasic() + defer th.TearDown() + + wsClient, err := th.CreateWebSocketClient() + require.NoError(t, err) + defer wsClient.Close() + wsClient.Listen() + + resp := <-wsClient.ResponseChannel + require.Equal(t, resp.Status, model.StatusOk, "should have responded OK to authentication challenge") + + wsClient.UpdateActiveChannel("chID") + resp = <-wsClient.ResponseChannel + require.Nil(t, resp.Error) + require.Equal(t, resp.SeqReply, wsClient.Sequence-1, "bad sequence number") + + wsClient.UpdateActiveTeam("teamID") + resp = <-wsClient.ResponseChannel + require.Nil(t, resp.Error) + require.Equal(t, resp.SeqReply, wsClient.Sequence-1, "bad sequence number") } func TestWebSocketUpgrade(t *testing.T) { diff --git a/server/channels/app/platform/web_conn.go b/server/channels/app/platform/web_conn.go index cce589989f..d643f0f827 100644 --- a/server/channels/app/platform/web_conn.go +++ b/server/channels/app/platform/web_conn.go @@ -105,13 +105,15 @@ type WebConn struct { // a reused connection. // It's theoretically possible for this number to wrap around. But we // leave that as an edge-case. - reuseCount int - sessionToken atomic.Value - session atomic.Pointer[model.Session] - connectionID atomic.Value - endWritePump chan struct{} - pumpFinished chan struct{} - pluginPosted chan pluginWSPostedHook + reuseCount int + sessionToken atomic.Value + session atomic.Pointer[model.Session] + connectionID atomic.Value + activeChannelID atomic.Value + activeTeamID atomic.Value + endWritePump chan struct{} + pumpFinished chan struct{} + pluginPosted chan pluginWSPostedHook // These counters are to suppress spammy websocket.slow // and websocket.full logs which happen continuously, if they @@ -290,6 +292,26 @@ func (wc *WebConn) GetConnectionID() string { return wc.connectionID.Load().(string) } +// SetActiveChannelID sets the active channel id of the connection. +func (wc *WebConn) SetActiveChannelID(id string) { + wc.activeChannelID.Store(id) +} + +// GetActiveChannelID returns the active channel id of the connection. +func (wc *WebConn) GetActiveChannelID() string { + return wc.activeChannelID.Load().(string) +} + +// SetActiveTeamID sets the active team id of the connection. +func (wc *WebConn) SetActiveTeamID(id string) { + wc.activeTeamID.Store(id) +} + +// GetActiveTeamID returns the active team id of the connection. +func (wc *WebConn) GetActiveTeamID() string { + return wc.activeTeamID.Load().(string) +} + // areAllInactive returns whether all of the connections // are inactive or not. func areAllInactive(conns []*WebConn) bool { diff --git a/server/channels/app/platform/websocket_router.go b/server/channels/app/platform/websocket_router.go index 9f15b5355d..b6a1f9cb46 100644 --- a/server/channels/app/platform/websocket_router.go +++ b/server/channels/app/platform/websocket_router.go @@ -73,6 +73,25 @@ func (wr *WebSocketRouter) ServeWebSocket(conn *WebConn, r *model.WebSocketReque return } + if r.Action == string(model.WebsocketPresenceIndicator) { + if chID, ok := r.Data["channel_id"].(string); ok { + // Set active channel + conn.SetActiveChannelID(chID) + } + if teamID, ok := r.Data["team_id"].(string); ok { + // Set active team + conn.SetActiveTeamID(teamID) + } + + resp := model.NewWebSocketResponse(model.StatusOk, r.Seq, nil) + hub := conn.Platform.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.Platform, conn, r, err) diff --git a/server/public/model/websocket_client.go b/server/public/model/websocket_client.go index 7ba82c17d1..55f7560633 100644 --- a/server/public/model/websocket_client.go +++ b/server/public/model/websocket_client.go @@ -326,6 +326,22 @@ func (wsc *WebSocketClient) GetStatusesByIds(userIds []string) { wsc.SendMessage("get_statuses_by_ids", data) } +// UpdateActiveChannel sets the current channel that the user is viewing. +func (wsc *WebSocketClient) UpdateActiveChannel(channelID string) { + data := map[string]any{ + "channel_id": channelID, + } + wsc.SendMessage(string(WebsocketPresenceIndicator), data) +} + +// UpdateActiveTeam sets the current team that the user is in. +func (wsc *WebSocketClient) UpdateActiveTeam(teamID string) { + data := map[string]any{ + "team_id": teamID, + } + wsc.SendMessage(string(WebsocketPresenceIndicator), data) +} + func (wsc *WebSocketClient) configurePingHandling() { wsc.Conn.SetPingHandler(wsc.pingHandler) wsc.pingTimeoutTimer = time.NewTimer(time.Second * (60 + PingTimeoutBufferSeconds)) diff --git a/server/public/model/websocket_message.go b/server/public/model/websocket_message.go index c15ed4fb2a..d3e3893202 100644 --- a/server/public/model/websocket_message.go +++ b/server/public/model/websocket_message.go @@ -86,6 +86,7 @@ const ( WebsocketEventAcknowledgementRemoved WebsocketEventType = "post_acknowledgement_removed" WebsocketEventPersistentNotificationTriggered WebsocketEventType = "persistent_notification_triggered" WebsocketEventHostedCustomerSignupProgressUpdated WebsocketEventType = "hosted_customer_signup_progress_updated" + WebsocketPresenceIndicator WebsocketEventType = "presence" ) type WebSocketMessage interface { diff --git a/webapp/channels/src/actions/websocket_actions.jsx b/webapp/channels/src/actions/websocket_actions.jsx index 5d4cc67c59..2452adfd4e 100644 --- a/webapp/channels/src/actions/websocket_actions.jsx +++ b/webapp/channels/src/actions/websocket_actions.jsx @@ -256,6 +256,10 @@ export function reconnect() { syncThreads(team.id, currentUserId); } } + + // Re-syncing the current channel and team ids. + WebSocketClient.updateActiveChannel(currentChannelId); + WebSocketClient.updateActiveTeam(currentTeamId); } loadPluginsIfNecessary(); diff --git a/webapp/channels/src/components/channel_view/channel_view.tsx b/webapp/channels/src/components/channel_view/channel_view.tsx index e0030bb75f..d66caa1c42 100644 --- a/webapp/channels/src/components/channel_view/channel_view.tsx +++ b/webapp/channels/src/components/channel_view/channel_view.tsx @@ -12,6 +12,8 @@ import FileUploadOverlay from 'components/file_upload_overlay'; import FormattedMarkdownMessage from 'components/formatted_markdown_message'; import PostView from 'components/post_view'; +import WebSocketClient from 'client/web_websocket_client'; + import type {PropsFromRedux} from './index'; export type Props = PropsFromRedux & RouteComponentProps<{ @@ -86,6 +88,10 @@ export default class ChannelView extends React.PureComponent { }; componentDidUpdate(prevProps: Props) { + // TODO: debounce + if (prevProps.channelId !== this.props.channelId) { + WebSocketClient.updateActiveChannel(this.props.channelId); + } if (prevProps.channelId !== this.props.channelId || prevProps.channelIsArchived !== this.props.channelIsArchived) { if (this.props.channelIsArchived && !this.props.viewArchivedChannels) { this.props.goToLastViewedChannel(); diff --git a/webapp/channels/src/components/team_sidebar/team_sidebar.tsx b/webapp/channels/src/components/team_sidebar/team_sidebar.tsx index 8587b07619..78b97f643d 100644 --- a/webapp/channels/src/components/team_sidebar/team_sidebar.tsx +++ b/webapp/channels/src/components/team_sidebar/team_sidebar.tsx @@ -16,6 +16,7 @@ import Permissions from 'mattermost-redux/constants/permissions'; import SystemPermissionGate from 'components/permissions_gates/system_permission_gate'; import TeamButton from 'components/team_sidebar/components/team_button'; +import WebSocketClient from 'client/web_websocket_client'; import Pluggable from 'plugins/pluggable'; import {Constants} from 'utils/constants'; import * as Keyboard from 'utils/keyboard'; @@ -146,6 +147,13 @@ export default class TeamSidebar extends React.PureComponent { } }; + componentDidUpdate(prevProps: Props) { + // TODO: debounce + if (prevProps.currentTeamId !== this.props.currentTeamId) { + WebSocketClient.updateActiveTeam(this.props.currentTeamId); + } + } + componentDidMount() { this.props.actions.getTeams(0, 200); document.addEventListener('keydown', this.handleKeyDown); diff --git a/webapp/platform/client/src/websocket.ts b/webapp/platform/client/src/websocket.ts index b026891933..a932c86b2c 100644 --- a/webapp/platform/client/src/websocket.ts +++ b/webapp/platform/client/src/websocket.ts @@ -355,7 +355,7 @@ export default class WebSocketClient { } } - sendMessage(action: string, data: any, responseCallback?: () => void) { + sendMessage(action: string, data: any, responseCallback?: (msg: any) => void) { const msg = { action, seq: this.responseSequence++, @@ -382,6 +382,20 @@ export default class WebSocketClient { this.sendMessage('user_typing', data, callback); } + updateActiveChannel(channelId: string, callback?: (msg: any) => void) { + const data = { + channel_id: channelId, + }; + this.sendMessage('presence', data, callback); + } + + updateActiveTeam(teamId: string, callback?: (msg: any) => void) { + const data = { + team_id: teamId, + }; + this.sendMessage('presence', data, callback); + } + userUpdateActiveStatus(userIsActive: boolean, manual: boolean, callback?: () => void) { const data = { user_is_active: userIsActive,