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
```
Этот коммит содержится в:
Agniva De Sarker
2023-12-12 08:49:09 +05:30
коммит произвёл GitHub
родитель f35b0a3781
Коммит e0b5b951f1
9 изменённых файлов: 120 добавлений и 9 удалений

Просмотреть файл

@@ -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) {

Просмотреть файл

@@ -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 {

Просмотреть файл

@@ -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)

Просмотреть файл

@@ -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))

Просмотреть файл

@@ -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 {

Просмотреть файл

@@ -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();

Просмотреть файл

@@ -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<Props, State> {
};
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();

Просмотреть файл

@@ -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<Props, State> {
}
};
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);

Просмотреть файл

@@ -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,