"msg_typing.isTyping": "{user}が入力しています..."

return (
    <FormattedMessage
        id='msg_typing.isTyping'
        defaultMessage='{user} is typing...'
        values={{
            user: users[0],
        }}
    />
);

export function useWebSocket({handler}: UseWebSocketOptions) {
    const wsClient = useWebSocketClient();

    useEffect(() => {
        wsClient.addMessageListener(handler);

        return () => {
            wsClient.removeMessageListener(handler);
        };
    }, [wsClient, handler]);
}

this.conn = new WebSocket(`${connectionUrl}?connection_id=${this.connectionId}&sequence_number=${this.serverSequence}`);

WebSocketClient.initialize(connUrl); // 182行目

let connUrl = '';
    if (config.WebsocketURL) {
        connUrl = config.WebsocketURL;
    } else {
        connUrl = new URL(getSiteURL());

        // replace the protocol with a websocket one
        if (connUrl.protocol === 'https:') {
            connUrl.protocol = 'wss:';
        } else {
            connUrl.protocol = 'ws:';
        }

        // append a port number if one isn't already specified
        if (!(/:\d+$/).test(connUrl.host)) {
            if (connUrl.protocol === 'wss:') {
                connUrl.host += ':' + config.WebsocketSecurePort;
            } else {
                connUrl.host += ':' + config.WebsocketPort;
            }
        }

        connUrl = connUrl.toString();
    }

    // Strip any trailing slash before appending the pathname below.
    if (connUrl.length > 0 && connUrl[connUrl.length - 1] === '/') {
        connUrl = connUrl.substring(0, connUrl.length - 1);
    }

    connUrl += Client4.getUrlVersion() + '/websocket';

const getTypingText = () => {
    let users: string[] = [];
    let numUsers = 0;
    if (props.typingUsers) {
        users = [...props.typingUsers];
        numUsers = users.length;
    }

    if (numUsers === 0) {
        return '';
    }
    // 中略
    return (
        <FormattedMessage
            id='msg_typing.areTyping'
            defaultMessage='{users} and {last} are typing...'
            values={{
                users: (users.join(', ')),
                last,
            }}
        />
    );
};

<MsgTyping
    channelId={channelId}
    postId={postId}
/>

function makeMapStateToProps() {
    const getUsersTypingByChannelAndPost = makeGetUsersTypingByChannelAndPost();

    return function mapStateToProps(state: GlobalState, ownProps: OwnProps) {
        const typingUsers = getUsersTypingByChannelAndPost(state, {channelId: ownProps.channelId, postId: ownProps.postId});

        return {
            typingUsers,
        };
    };
}

const mapDispatchToProps = {
    userStartedTyping,
    userStoppedTyping,
};

export default connect(makeMapStateToProps, mapDispatchToProps)(MsgTyping);

export function userStartedTyping(userId: string, channelId: string, rootId: string, now: number) {
    return (dispatch: DispatchFunc, getState: GetStateFunc) => {
        const state = getState();

        if (
            isPerformanceDebuggingEnabled(state) &&
            getBool(state, Preferences.CATEGORY_PERFORMANCE_DEBUGGING, Preferences.NAME_DISABLE_TYPING_MESSAGES)
        ) {
            return;
        }

        dispatch({
            type: WebsocketEvents.TYPING,
            data: {
                id: channelId + rootId,
                userId,
                now,
            },
        });

        // Ideally this followup loading would be done by someone else
        dispatch(fillInMissingInfo(userId));

        setTimeout(() => {
            dispatch(userStoppedTyping(userId, channelId, rootId, now));
        }, getTimeBetweenTypingEvents(state));
    };
}

useWebSocket({
    handler: useCallback((msg: WebSocketMessage) => {
        if (msg.event === SocketEvents.TYPING) {
            const channelId = msg.broadcast.channel_id;
            const rootId = msg.data.parent_id;
            const userId = msg.data.user_id;

            if (props.channelId === channelId && props.postId === rootId) {
                userStartedTyping(userId, channelId, rootId, Date.now());
            }
        } else if (msg.event === SocketEvents.POSTED) {
            const post = JSON.parse(msg.data.post);

            const channelId = post.channel_id;
            const rootId = post.root_id;
            const userId = post.user_id;

            if (props.channelId === channelId && props.postId === rootId) {
                userStoppedTyping(userId, channelId, rootId, Date.now());
            }
        }
    }, [props.channelId, props.postId, userStartedTyping, userStoppedTyping]),
});

export function useWebSocket({handler}: UseWebSocketOptions) {
    const wsClient = useWebSocketClient();

    useEffect(() => {
        wsClient.addMessageListener(handler);

        return () => {
            wsClient.removeMessageListener(handler);
        };
    }, [wsClient, handler]);
}

export type MessageListener = (msg: WebSocketMessage) => void;
private messageListeners = new Set<MessageListener>();
export type WebSocketMessage<T = any> = {
    event: string;
    data: T;
    broadcast: WebSocketBroadcast;
    seq: number;
};

// 中略

export default class WebSocketClient {
    // 中略
    addMessageListener(listener: MessageListener) {
        this.messageListeners.add(listener);

        if (this.messageListeners.size > 5) {
            // eslint-disable-next-line no-console
            console.warn(
                `WebSocketClient has ${this.messageListeners.size} message listeners registered`
            );
        }
    }
}

export default class WebSocketClient {
    initialize(connectionUrl = this.connectionUrl, token?: string) {
    // 中略
    this.conn = new WebSocket(`${connectionUrl}?connection_id=${this.connectionId}&sequence_number=${this.serverSequence}`);
    // 中略
    this.conn.onmessage = (evt) => {
            const msg = JSON.parse(evt.data);
            // たくさん中略
            this.messageListeners.forEach((listener) => listener(msg));
        };
    }
}

func (api *API) InitWebSocket() {
	// Optionally supports a trailing slash
	api.BaseRoutes.APIRoot.Handle("/{websocket:websocket(?:\\/)?}", api.APIHandlerTrustRequester(connectWebSocket)).Methods("GET")
}

func connectWebSocket(c *Context, w http.ResponseWriter, r *http.Request) {
	upgrader := websocket.Upgrader{
		ReadBufferSize:  model.SocketMaxMessageSizeKb,
		WriteBufferSize: model.SocketMaxMessageSizeKb,
		CheckOrigin:     c.App.OriginChecker(),
	}

	ws, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		c.Err = model.NewAppError("connect", "api.web_socket.connect.upgrade.app_error", nil, err.Error(), http.StatusBadRequest)
		return
	}

	// We initialize webconn with all the necessary data.
	// If the queues are empty, they are initialized in the constructor.
	cfg := &platform.WebConnConfig{
		WebSocket: ws,
		Session:   *c.AppContext.Session(),
		TFunc:     c.AppContext.T,
		Locale:    "",
		Active:    true,
	}

	cfg.ConnectionID = r.URL.Query().Get(connectionIDParam)
	if cfg.ConnectionID == "" || c.AppContext.Session().UserId == "" {
		// If not present, we assume client is not capable yet, or it's a fresh connection.
		// We just create a new ID.
		cfg.ConnectionID = model.NewId()
		// In case of fresh connection id, sequence number is already zero.
	} else {
		cfg, err = c.App.Srv().Platform().PopulateWebConnConfig(c.AppContext.Session(), cfg, r.URL.Query().Get(sequenceNumberParam))
		if err != nil {
			c.Logger.Warn("Error while populating webconn config", mlog.String("id", r.URL.Query().Get(connectionIDParam)), mlog.Err(err))
			ws.Close()
			return
		}
	}

	wc := c.App.Srv().Platform().NewWebConn(cfg, c.App, c.App.Srv().Channels())
	if c.AppContext.Session().UserId != "" {
		c.App.Srv().Platform().HubRegister(wc)
	}

	wc.Pump()
}

// Pump starts the WebConn instance. After this, the websocket
// is ready to send/receive messages.
func (wc *WebConn) Pump() {
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		wc.writePump()
	}()

	wg.Add(1)
	go wc.pluginPostedConsumer(&wg)

	wc.readPump()
	close(wc.endWritePump)
	close(wc.pluginPosted)
	wg.Wait()
	wc.Platform.HubUnregister(wc)
	close(wc.pumpFinished)

	userID := wc.UserId
	wc.Platform.Go(func() {
		wc.HookRunner.RunMultiHook(func(hooks plugin.Hooks) bool {
			hooks.OnWebSocketDisconnect(wc.GetConnectionID(), userID)
			return true
		}, plugin.OnWebSocketDisconnectID)
	})
}

go func() {
	defer wg.Done()
	wc.writePump()
}()

for {
	msgType, rd, err := wc.WebSocket.NextReader()
	if err != nil {
		wc.logSocketErr("websocket.NextReader", err)
		return
	}

	var decoder interface {
		Decode(v any) error
	}
	if msgType == websocket.TextMessage {
		decoder = json.NewDecoder(rd)
	} else {
		decoder = msgpack.NewDecoder(rd)
	}
	var req model.WebSocketRequest
	if err = decoder.Decode(&req); err != nil {
		wc.logSocketErr("websocket.Decode", err)
		return
	}

	// Messages which actions are prefixed with the plugin prefix
	// should only be dispatched to the plugins
	if !strings.HasPrefix(req.Action, websocketMessagePluginPrefix) {
		wc.Platform.WebSocketRouter.ServeWebSocket(wc, &req)
	}

	clonedReq, err := req.Clone()
	if err != nil {
		wc.logSocketErr("websocket.cloneRequest", err)
		continue
	}

	wc.pluginPosted <- pluginWSPostedHook{wc.GetConnectionID(), wc.UserId, clonedReq}
}

func (ps *PlatformService) Start(broadcastHooks map[string]BroadcastHook) error {
    ps.hubStart(broadcastHooks)

    ps.configListenerId = ps.AddConfigListener(func(_, _ *model.Config) {
        ps.regenerateClientConfig()

        message := model.NewWebSocketEvent(model.WebsocketEventConfigChanged, "", "", "", nil, "")

        message.Add("config", ps.ClientConfigWithComputed())
        ps.Go(func() {
            ps.Publish(message)
        })

        if err := ps.ReconfigureLogger(); err != nil {
            mlog.Error("Error re-configuring logging after config change", mlog.Err(err))
            return
        }
    })

    ps.licenseListenerId = ps.AddLicenseListener(func(oldLicense, newLicense *model.License) {
        ps.regenerateClientConfig()

        message := model.NewWebSocketEvent(model.WebsocketEventLicenseChanged, "", "", "", nil, "")
        message.Add("license", ps.GetSanitizedClientLicense())
        ps.Go(func() {
            ps.Publish(message)
        })
    })
    return nil
}

func (ps *PlatformService) hubStart(broadcastHooks map[string]BroadcastHook) {
    // Total number of hubs is twice the number of CPUs.
    numberOfHubs := runtime.NumCPU() * 2
    ps.logger.Info("Starting websocket hubs", mlog.Int("number_of_hubs", numberOfHubs))

    hubs := make([]*Hub, numberOfHubs)

    for i := 0; i < numberOfHubs; i++ {
        hubs[i] = newWebHub(ps)
        hubs[i].connectionIndex = i
        hubs[i].broadcastHooks = broadcastHooks
        hubs[i].Start()
    }
    // Assigning to the hubs slice without any mutex is fine because it is only assigned once
    // during the start of the program and always read from after that.
    ps.hubs = hubs
}

func connectWebSocket(c *Context, w http.ResponseWriter, r *http.Request) {  
    upgrader := websocket.Upgrader{  
       ReadBufferSize:  model.SocketMaxMessageSizeKb,  
       WriteBufferSize: model.SocketMaxMessageSizeKb,  
       CheckOrigin:     c.App.OriginChecker(),  
    }  
  
    ws, err := upgrader.Upgrade(w, r, nil)  
    if err != nil {  
       c.Err = model.NewAppError("connect", "api.web_socket.connect.upgrade.app_error", nil, err.Error(), http.StatusBadRequest)  
       return  
    }  
  
    // We initialize webconn with all the necessary data.  
    // If the queues are empty, they are initialized in the constructor.    cfg := &platform.WebConnConfig{  
       WebSocket: ws,  
       Session:   *c.AppContext.Session(),  
       TFunc:     c.AppContext.T,  
       Locale:    "",  
       Active:    true,  
    }  
  
    cfg.ConnectionID = r.URL.Query().Get(connectionIDParam)  
    if cfg.ConnectionID == "" || c.AppContext.Session().UserId == "" {  
       // If not present, we assume client is not capable yet, or it's a fresh connection.  
       // We just create a new ID.       cfg.ConnectionID = model.NewId()  
       // In case of fresh connection id, sequence number is already zero.  
    } else {  
       cfg, err = c.App.Srv().Platform().PopulateWebConnConfig(c.AppContext.Session(), cfg, r.URL.Query().Get(sequenceNumberParam))  
       if err != nil {  
          c.Logger.Warn("Error while populating webconn config", mlog.String("id", r.URL.Query().Get(connectionIDParam)), mlog.Err(err))  
          ws.Close()  
          return  
       }  
    }  
  
    wc := c.App.Srv().Platform().NewWebConn(cfg, c.App, c.App.Srv().Channels())  
    if c.AppContext.Session().UserId != "" {  
       c.App.Srv().Platform().HubRegister(wc)  
    }  
 
    wc.Pump()  
}

// Start starts the hub.
func (h *Hub) Start() {
	var doStart func()
	var doRecoverableStart func()
	var doRecover func()

	doStart = func() {
		mlog.Debug("Hub is starting", mlog.Int("index", h.connectionIndex))

		ticker := time.NewTicker(inactiveConnReaperInterval)
		defer ticker.Stop()

		connIndex := newHubConnectionIndex(inactiveConnReaperInterval)

		for {
			select {
			case webSessionMessage := <-h.checkRegistered:
				conns := connIndex.ForUser(webSessionMessage.userID)
				var isRegistered bool
				for _, conn := range conns {
					if !conn.active.Load() {
						continue
					}
					if conn.GetSessionToken() == webSessionMessage.sessionToken {
						isRegistered = true
					}
				}
				webSessionMessage.isRegistered <- isRegistered
			case req := <-h.checkConn:
				var res *CheckConnResult
				conn := connIndex.RemoveInactiveByConnectionID(req.userID, req.connectionID)
				if conn != nil {
					res = &CheckConnResult{
						ConnectionID:     req.connectionID,
						UserID:           req.userID,
						ActiveQueue:      conn.send,
						DeadQueue:        conn.deadQueue,
						DeadQueuePointer: conn.deadQueuePointer,
						ReuseCount:       conn.reuseCount + 1,
					}
				}
				req.result <- res
			case <-ticker.C:
				connIndex.RemoveInactiveConnections()
			case webConn := <-h.register:
				// Mark the current one as active.
				// There is no need to check if it was inactive or not,
				// we will anyways need to make it active.
				webConn.active.Store(true)

				connIndex.Add(webConn)
				atomic.StoreInt64(&h.connectionCount, int64(connIndex.AllActive()))

				if webConn.IsAuthenticated() && webConn.reuseCount == 0 {
					// The hello message should only be sent when the reuseCount is 0.
					// i.e in server restart, or long timeout, or fresh connection case.
					// In case of seq number not found in dead queue, it is handled by
					// the webconn write pump.
					webConn.send <- webConn.createHelloMessage()
				}
			case webConn := <-h.unregister:
				// If already removed (via queue full), then removing again becomes a noop.
				// But if not removed, mark inactive.
				webConn.active.Store(false)

				atomic.StoreInt64(&h.connectionCount, int64(connIndex.AllActive()))

				if webConn.UserId == "" {
					continue
				}

				conns := connIndex.ForUser(webConn.UserId)
				if len(conns) == 0 || areAllInactive(conns) {
					userID := webConn.UserId
					h.platform.Go(func() {
						h.platform.SetStatusOffline(userID, false)
					})
					continue
				}
				var latestActivity int64
				for _, conn := range conns {
					if !conn.active.Load() {
						continue
					}
					if conn.lastUserActivityAt > latestActivity {
						latestActivity = conn.lastUserActivityAt
					}
				}

				if h.platform.isUserAway(latestActivity) {
					userID := webConn.UserId
					h.platform.Go(func() {
						h.platform.SetStatusLastActivityAt(userID, latestActivity)
					})
				}
			case userID := <-h.invalidateUser:
				for _, webConn := range connIndex.ForUser(userID) {
					webConn.InvalidateCache()
				}
			case activity := <-h.activity:
				for _, webConn := range connIndex.ForUser(activity.userID) {
					if !webConn.active.Load() {
						continue
					}
					if webConn.GetSessionToken() == activity.sessionToken {
						webConn.lastUserActivityAt = activity.activityAt
					}
				}
			case directMsg := <-h.directMsg:
				if !connIndex.Has(directMsg.conn) {
					continue
				}
				select {
				case directMsg.conn.send <- directMsg.msg:
				default:
					// Don't log the warning if it's an inactive connection.
					if directMsg.conn.active.Load() {
						mlog.Error("webhub.broadcast: cannot send, closing websocket for user",
							mlog.String("user_id", directMsg.conn.UserId),
							mlog.String("conn_id", directMsg.conn.GetConnectionID()))
					}
					close(directMsg.conn.send)
					connIndex.Remove(directMsg.conn)
				}
			case msg := <-h.broadcast:
				if metrics := h.platform.metricsIFace; metrics != nil {
					metrics.DecrementWebSocketBroadcastBufferSize(strconv.Itoa(h.connectionIndex), 1)
				}

				// Remove the broadcast hook information before precomputing the JSON so that those aren't included in it
				msg, broadcastHooks, broadcastHookArgs := msg.WithoutBroadcastHooks()

				msg = msg.PrecomputeJSON()

				broadcast := func(webConn *WebConn) {
					if !connIndex.Has(webConn) {
						return
					}
					if webConn.ShouldSendEvent(msg) {
						select {
						case webConn.send <- h.runBroadcastHooks(msg, webConn, broadcastHooks, broadcastHookArgs):
						default:
							// Don't log the warning if it's an inactive connection.
							if webConn.active.Load() {
								mlog.Error("webhub.broadcast: cannot send, closing websocket for user",
									mlog.String("user_id", webConn.UserId),
									mlog.String("conn_id", webConn.GetConnectionID()))
							}
							close(webConn.send)
							connIndex.Remove(webConn)
						}
					}
				}

				if connID := msg.GetBroadcast().ConnectionId; connID != "" {
					if webConn := connIndex.byConnectionId[connID]; webConn != nil {
						broadcast(webConn)
						continue
					}
				} else if msg.GetBroadcast().UserId != "" {
					candidates := connIndex.ForUser(msg.GetBroadcast().UserId)
					for _, webConn := range candidates {
						broadcast(webConn)
					}
					continue
				}

				candidates := connIndex.All()
				for webConn := range candidates {
					broadcast(webConn)
				}
			case <-h.stop:
				for webConn := range connIndex.All() {
					webConn.Close()
					h.platform.SetStatusOffline(webConn.UserId, false)
				}

				h.explicitStop = true
				close(h.didStop)

				return
			}
		}
	}

	doRecoverableStart = func() {
		defer doRecover()
		doStart()
	}

	doRecover = func() {
		if !h.explicitStop {
			if r := recover(); r != nil {
				mlog.Error("Recovering from Hub panic.", mlog.Any("panic", r))
			} else {
				mlog.Error("Webhub stopped unexpectedly. Recovering.")
			}

			mlog.Error(string(debug.Stack()))

			go doRecoverableStart()
		}
	}

	go doRecoverableStart()
}

type WebsocketEventType string

const (
	WebsocketEventTyping                              WebsocketEventType = "typing"
)

func (a *App) PublishUserTyping(userID, channelID, parentId string) *model.AppError {
	omitUsers := make(map[string]bool, 1)
	omitUsers[userID] = true

	event := model.NewWebSocketEvent(model.WebsocketEventTyping, "", channelID, "", omitUsers, "")
	event.Add("parent_id", parentId)
	event.Add("user_id", userID)
	a.Publish(event)

	return nil
}

func (api *API) InitUser() {
	api.Router.Handle("user_typing", api.APIWebSocketHandler(api.userTyping))
	api.Router.Handle("user_update_active_status", api.APIWebSocketHandler(api.userUpdateActiveStatus))
}

func (api *API) userTyping(req *model.WebSocketRequest) (map[string]any, *model.AppError) {
	api.App.ExtendSessionExpiryIfNeeded(request.EmptyContext(api.App.Log()), &req.Session)

	if api.App.Srv().Platform().Busy.IsBusy() {
		// this is considered a non-critical service and will be disabled when server busy.
		return nil, NewServerBusyWebSocketError(req.Action)
	}

	var ok bool
	var channelId string
	if channelId, ok = req.Data["channel_id"].(string); !ok || !model.IsValidId(channelId) {
		return nil, NewInvalidWebSocketParamError(req.Action, "channel_id")
	}

	if !api.App.SessionHasPermissionToChannel(request.EmptyContext(api.App.Log()), req.Session, channelId, model.PermissionCreatePost) {
		return nil, NewInvalidWebSocketParamError(req.Action, "channel_id")
	}

	var parentId string
	if parentId, ok = req.Data["parent_id"].(string); !ok {
		parentId = ""
	}

	appErr := api.App.PublishUserTyping(req.Session.UserId, channelId, parentId)

	return nil, appErr
}

userTyping(channelId: string, parentId: string, callback?: () => void) {
    const data = {
        channel_id: channelId,
        parent_id: parentId,
    };
    this.sendMessage("user_typing", data, callback);
}

export function emitLocalUserTypingEvent(channelId: string, parentPostId: string) {
    const userTyping = async (actionDispatch: DispatchFunc, actionGetState: GetStateFunc) => {
        const state = actionGetState();
        const config = getConfig(state);

        if (
            isPerformanceDebuggingEnabled(state) &&
            getBool(state, Preferences.CATEGORY_PERFORMANCE_DEBUGGING, Preferences.NAME_DISABLE_TYPING_MESSAGES)
        ) {
            return {data: false};
        }

        const t = Date.now();
        const stats = getCurrentChannelStats(state);
        const membersInChannel = stats ? stats.member_count : 0;

        const timeBetweenUserTypingUpdatesMilliseconds = Utils.stringToNumber(config.TimeBetweenUserTypingUpdatesMilliseconds);
        const maxNotificationsPerChannel = Utils.stringToNumber(config.MaxNotificationsPerChannel);

        if (((t - lastTimeTypingSent) > timeBetweenUserTypingUpdatesMilliseconds) &&
            (membersInChannel < maxNotificationsPerChannel) && (config.EnableUserTypingMessages === 'true')) {
            WebSocketClient.userTyping(channelId, parentPostId);
            lastTimeTypingSent = t;
        }

        return {data: true};
    };

    return dispatch(userTyping);
}

postMsgKeyPress = (e: React.KeyboardEvent<TextboxElement>) => {
    const {ctrlSend, codeBlockOnCtrlEnter} = this.props;

    const {allowSending, withClosedCodeBlock, ignoreKeyPress, message} = postMessageOnKeyPress(
        e,
        this.state.message,
        Boolean(ctrlSend),
        Boolean(codeBlockOnCtrlEnter),
        Date.now(),
        this.lastChannelSwitchAt,
        this.state.caretPosition,
    ) as {
        allowSending: boolean;
        withClosedCodeBlock?: boolean;
        ignoreKeyPress?: boolean;
        message?: string;
    };

    if (ignoreKeyPress) {
        e.preventDefault();
        e.stopPropagation();
        return;
    }

    if (allowSending && this.isValidPersistentNotifications()) {
        if (e.persist) {
            e.persist();
        }
        if (this.textboxRef.current) {
            this.isDraftSubmitting = true;
            this.textboxRef.current.blur();
        }

        if (withClosedCodeBlock && message) {
            this.setState({message}, () => this.handleSubmit(e));
        } else {
            this.handleSubmit(e);
        }

        this.setShowPreview(false);
    }

    this.emitTypingEvent();
};

emitTypingEvent = () => {
    const channelId = this.props.currentChannel.id;
    GlobalActions.emitLocalUserTypingEvent(channelId, '');
};