diff --git a/app/app_iface.go b/app/app_iface.go
index a8874fbe18..8cde6f109c 100644
--- a/app/app_iface.go
+++ b/app/app_iface.go
@@ -154,6 +154,8 @@ type AppIface interface {
 	GetEmojiStaticUrl(emojiName string) (string, *model.AppError)
 	// GetEnvironmentConfig returns a map of configuration keys whose values have been overridden by an environment variable.
 	GetEnvironmentConfig() map[string]interface{}
+	// GetHubForUserId returns the hub for a given user id.
+	GetHubForUserId(userId string) *Hub
 	// GetLdapGroup retrieves a single LDAP group by the given LDAP group id.
 	GetLdapGroup(ldapGroupID string) (*model.Group, *model.AppError)
 	// GetMarketplacePlugins returns a list of plugins from the marketplace-server,
@@ -183,6 +185,14 @@ type AppIface interface {
 	GetTeamSchemeChannelRoles(teamId string) (guestRoleName string, userRoleName string, adminRoleName string, err *model.AppError)
 	// GetTotalUsersStats is used for the DM list total
 	GetTotalUsersStats(viewRestrictions *model.ViewUsersRestrictions) (*model.UsersStats, *model.AppError)
+	// HubRegister registers a connection to a hub.
+	HubRegister(webConn *WebConn)
+	// HubStart starts all the hubs.
+	HubStart()
+	// HubStop stops all the hubs.
+	HubStop()
+	// HubUnregister unregisters a connection from a hub.
+	HubUnregister(webConn *WebConn)
 	// InstallMarketplacePlugin installs a plugin listed in the marketplace server. It will get the plugin bundle
 	// from the prepackaged folder, if available, or remotely if EnableRemoteMarketplace is true.
 	InstallMarketplacePlugin(request *model.InstallMarketplacePluginRequest) (*model.Manifest, *model.AppError)
@@ -204,6 +214,10 @@ type AppIface interface {
 	MakeAuditRecord(event string, initialStatus string) *audit.Record
 	// MarkChanelAsUnreadFromPost will take a post and set the channel as unread from that one.
 	MarkChannelAsUnreadFromPost(postID string, userID string) (*model.ChannelUnreadAt, *model.AppError)
+	// NewWebConn returns a new WebConn instance.
+	NewWebConn(ws *websocket.Conn, session model.Session, t goi18n.TranslateFunc, locale string) *WebConn
+	// NewWebHub creates a new Hub.
+	NewWebHub() *Hub
 	// OverrideIconURLIfEmoji changes the post icon override URL prop, if it has an emoji icon,
 	// so that it points to the URL (relative) of the emoji - static if emoji is default, /api if custom.
 	OverrideIconURLIfEmoji(post *model.Post)
@@ -281,6 +295,8 @@ type AppIface interface {
 	UpdateChannel(channel *model.Channel) (*model.Channel, *model.AppError)
 	// UpdateChannelScheme saves the new SchemeId of the channel passed.
 	UpdateChannelScheme(channel *model.Channel) (*model.Channel, *model.AppError)
+	// UpdateWebConnUserActivity sets the LastUserActivityAt of the hub for the given session.
+	UpdateWebConnUserActivity(session model.Session, activityAt int64)
 	// UploadFile uploads a single file in form of a completely constructed byte array for a channel.
 	UploadFile(data []byte, channelId string, filename string) (*model.FileInfo, *model.AppError)
 	// UploadFileX uploads a single file as specified in t. It applies the upload
@@ -523,7 +539,6 @@ type AppIface interface {
 	GetGroupsBySource(groupSource model.GroupSource) ([]*model.Group, *model.AppError)
 	GetGroupsByTeam(teamId string, opts model.GroupSearchOpts) ([]*model.GroupWithSchemeAdmin, int, *model.AppError)
 	GetGroupsByUserId(userId string) ([]*model.Group, *model.AppError)
-	GetHubForUserId(userId string) *Hub
 	GetIncomingWebhook(hookId string) (*model.IncomingWebhook, *model.AppError)
 	GetIncomingWebhooksForTeamPage(teamId string, page, perPage int) ([]*model.IncomingWebhook, *model.AppError)
 	GetIncomingWebhooksForTeamPageByUser(teamId string, userId string, page, perPage int) ([]*model.IncomingWebhook, *model.AppError)
@@ -674,10 +689,6 @@ type AppIface interface {
 	HasPermissionToChannelByPost(askingUserId string, postId string, permission *model.Permission) bool
 	HasPermissionToTeam(askingUserId string, teamId string, permission *model.Permission) bool
 	HasPermissionToUser(askingUserId string, userId string) bool
-	HubRegister(webConn *WebConn)
-	HubStart()
-	HubStop()
-	HubUnregister(webConn *WebConn)
 	ImageProxy() *imageproxy.ImageProxy
 	ImageProxyAdder() func(string) string
 	ImageProxyRemover() (f func(string) string)
@@ -724,8 +735,6 @@ type AppIface interface {
 	MoveFile(oldPath, newPath string) *model.AppError
 	NewClusterDiscoveryService() *ClusterDiscoveryService
 	NewPluginAPI(manifest *model.Manifest) plugin.API
-	NewWebConn(ws *websocket.Conn, session model.Session, t goi18n.TranslateFunc, locale string) *WebConn
-	NewWebHub() *Hub
 	Notification() einterfaces.NotificationInterface
 	NotificationsLog() *mlog.Logger
 	OpenInteractiveDialog(request model.OpenDialogRequest) *model.AppError
@@ -949,7 +958,6 @@ type AppIface interface {
 	UpdateUserAuth(userId string, userAuth *model.UserAuth) (*model.UserAuth, *model.AppError)
 	UpdateUserNotifyProps(userId string, props map[string]string) (*model.User, *model.AppError)
 	UpdateUserRoles(userId string, newRoles string, sendWebSocketEvent bool) (*model.User, *model.AppError)
-	UpdateWebConnUserActivity(session model.Session, activityAt int64)
 	UploadEmojiImage(id string, imageData *multipart.FileHeader) *model.AppError
 	UploadMultipartFiles(teamId string, channelId string, userId string, fileHeaders []*multipart.FileHeader, clientIds []string, now time.Time) (*model.FileUploadResponse, *model.AppError)
 	UpsertGroupMember(groupID string, userID string) (*model.GroupMember, *model.AppError)
diff --git a/app/cluster_handlers.go b/app/cluster_handlers.go
index f14b51903c..43dba74a86 100644
--- a/app/cluster_handlers.go
+++ b/app/cluster_handlers.go
@@ -59,7 +59,7 @@ func (a *App) clusterInvalidateCacheForUserHandler(msg *model.ClusterMessage) {
 }
 
 func (a *App) clusterInvalidateCacheForUserTeamsHandler(msg *model.ClusterMessage) {
-	a.invalidateCacheForUserTeamsSkipClusterSend(msg.Data)
+	a.InvalidateWebConnSessionCacheForUser(msg.Data)
 }
 
 func (a *App) clusterClearSessionCacheForUserHandler(msg *model.ClusterMessage) {
diff --git a/app/web_conn.go b/app/web_conn.go
index 99a93267d2..ba0e63772a 100644
--- a/app/web_conn.go
+++ b/app/web_conn.go
@@ -9,44 +9,49 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/gorilla/websocket"
-	goi18n "github.com/mattermost/go-i18n/i18n"
 	"github.com/mattermost/mattermost-server/v5/mlog"
 	"github.com/mattermost/mattermost-server/v5/model"
+
+	"github.com/gorilla/websocket"
+	goi18n "github.com/mattermost/go-i18n/i18n"
 )
 
 const (
-	SEND_QUEUE_SIZE           = 256
-	SEND_SLOW_WARN            = (SEND_QUEUE_SIZE * 50) / 100
-	SEND_DEADLOCK_WARN        = (SEND_QUEUE_SIZE * 95) / 100
-	WRITE_WAIT                = 30 * time.Second
-	PONG_WAIT                 = 100 * time.Second
-	PING_PERIOD               = (PONG_WAIT * 6) / 10
-	AUTH_TIMEOUT              = 5 * time.Second
-	WEBCONN_MEMBER_CACHE_TIME = 1000 * 60 * 30 // 30 minutes
+	sendQueueSize          = 256
+	sendSlowWarn           = (sendQueueSize * 50) / 100
+	sendFullWarn           = (sendQueueSize * 95) / 100
+	writeWaitTime          = 30 * time.Second
+	pongWaitTime           = 100 * time.Second
+	pingInterval           = (pongWaitTime * 6) / 10
+	authCheckInterval      = 5 * time.Second
+	webConnMemberCacheTime = 1000 * 60 * 30 // 30 minutes
 )
 
+// WebConn represents a single websocket connection to a user.
+// It contains all the necesarry state to manage sending/receiving data to/from
+// a websocket.
 type WebConn struct {
-	sessionExpiresAt          int64 // This should stay at the top for 64-bit alignment of 64-bit words accessed atomically
-	App                       *App
-	WebSocket                 *websocket.Conn
-	Send                      chan model.WebSocketMessage
+	sessionExpiresAt int64 // This should stay at the top for 64-bit alignment of 64-bit words accessed atomically
+	App              *App
+	WebSocket        *websocket.Conn
+	T                goi18n.TranslateFunc
+	Locale           string
+	Sequence         int64
+	UserId           string
+
+	allChannelMembers         map[string]string
+	lastAllChannelMembersTime int64
+	lastUserActivityAt        int64
+	send                      chan model.WebSocketMessage
 	sessionToken              atomic.Value
 	session                   atomic.Value
-	LastUserActivityAt        int64
-	UserId                    string
-	T                         goi18n.TranslateFunc
-	Locale                    string
-	AllChannelMembers         map[string]string
-	LastAllChannelMembersTime int64
-	Sequence                  int64
-	closeOnce                 sync.Once
 	endWritePump              chan struct{}
 	pumpFinished              chan struct{}
 }
 
+// NewWebConn returns a new WebConn instance.
 func (a *App) NewWebConn(ws *websocket.Conn, session model.Session, t goi18n.TranslateFunc, locale string) *WebConn {
-	if len(session.UserId) > 0 {
+	if session.UserId != "" {
 		a.Srv().Go(func() {
 			a.SetStatusOnline(session.UserId, false)
 			a.UpdateLastActivityAtIfNeeded(session)
@@ -55,9 +60,9 @@ func (a *App) NewWebConn(ws *websocket.Conn, session model.Session, t goi18n.Tra
 
 	wc := &WebConn{
 		App:                a,
-		Send:               make(chan model.WebSocketMessage, SEND_QUEUE_SIZE),
+		send:               make(chan model.WebSocketMessage, sendQueueSize),
 		WebSocket:          ws,
-		LastUserActivityAt: model.GetMillis(),
+		lastUserActivityAt: model.GetMillis(),
 		UserId:             session.UserId,
 		T:                  t,
 		Locale:             locale,
@@ -72,34 +77,38 @@ func (a *App) NewWebConn(ws *websocket.Conn, session model.Session, t goi18n.Tra
 	return wc
 }
 
+// Close closes the WebConn.
 func (wc *WebConn) Close() {
 	wc.WebSocket.Close()
-	wc.closeOnce.Do(func() {
-		close(wc.endWritePump)
-	})
 	<-wc.pumpFinished
 }
 
+// GetSessionExpiresAt returns the time at which the session expires.
 func (wc *WebConn) GetSessionExpiresAt() int64 {
 	return atomic.LoadInt64(&wc.sessionExpiresAt)
 }
 
+// SetSessionExpiresAt sets the time at which the session expires.
 func (wc *WebConn) SetSessionExpiresAt(v int64) {
 	atomic.StoreInt64(&wc.sessionExpiresAt, v)
 }
 
+// GetSessionToken returns the session token of the connection.
 func (wc *WebConn) GetSessionToken() string {
 	return wc.sessionToken.Load().(string)
 }
 
+// SetSessionToken sets the session token of the connection.
 func (wc *WebConn) SetSessionToken(v string) {
 	wc.sessionToken.Store(v)
 }
 
+// GetSession returns the session of the connection.
 func (wc *WebConn) GetSession() *model.Session {
 	return wc.session.Load().(*model.Session)
 }
 
+// SetSession sets the session of the connection.
 func (wc *WebConn) SetSession(v *model.Session) {
 	if v != nil {
 		v = v.DeepCopy()
@@ -108,17 +117,18 @@ func (wc *WebConn) SetSession(v *model.Session) {
 	wc.session.Store(v)
 }
 
+// Pump starts the WebConn instance. After this, the websocket
+// is ready to send/receive messages.
 func (wc *WebConn) Pump() {
-	ch := make(chan struct{})
+	var wg sync.WaitGroup
+	wg.Add(1)
 	go func() {
+		defer wg.Done()
 		wc.writePump()
-		close(ch)
 	}()
 	wc.readPump()
-	wc.closeOnce.Do(func() {
-		close(wc.endWritePump)
-	})
-	<-ch
+	close(wc.endWritePump)
+	wg.Wait()
 	wc.App.HubUnregister(wc)
 	close(wc.pumpFinished)
 }
@@ -128,9 +138,9 @@ func (wc *WebConn) readPump() {
 		wc.WebSocket.Close()
 	}()
 	wc.WebSocket.SetReadLimit(model.SOCKET_MAX_MESSAGE_SIZE_KB)
-	wc.WebSocket.SetReadDeadline(time.Now().Add(PONG_WAIT))
+	wc.WebSocket.SetReadDeadline(time.Now().Add(pongWaitTime))
 	wc.WebSocket.SetPongHandler(func(string) error {
-		wc.WebSocket.SetReadDeadline(time.Now().Add(PONG_WAIT))
+		wc.WebSocket.SetReadDeadline(time.Now().Add(pongWaitTime))
 		if wc.IsAuthenticated() {
 			wc.App.Srv().Go(func() {
 				wc.App.SetStatusAwayIfNeeded(wc.UserId, false)
@@ -142,12 +152,7 @@ func (wc *WebConn) readPump() {
 	for {
 		var req model.WebSocketRequest
 		if err := wc.WebSocket.ReadJSON(&req); err != nil {
-			// browsers will appear as CloseNoStatusReceived
-			if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) {
-				mlog.Debug("websocket.read: client side closed socket", mlog.String("user_id", wc.UserId))
-			} else {
-				mlog.Debug("websocket.read: closing websocket", mlog.String("user_id", wc.UserId), mlog.Err(err))
-			}
+			wc.logSocketErr("websocket.read", err)
 			return
 		}
 		wc.App.Srv().WebSocketRouter.ServeWebSocket(wc, &req)
@@ -155,8 +160,8 @@ func (wc *WebConn) readPump() {
 }
 
 func (wc *WebConn) writePump() {
-	ticker := time.NewTicker(PING_PERIOD)
-	authTicker := time.NewTicker(AUTH_TIMEOUT)
+	ticker := time.NewTicker(pingInterval)
+	authTicker := time.NewTicker(authCheckInterval)
 
 	defer func() {
 		ticker.Stop()
@@ -166,9 +171,9 @@ func (wc *WebConn) writePump() {
 
 	for {
 		select {
-		case msg, ok := <-wc.Send:
+		case msg, ok := <-wc.send:
 			if !ok {
-				wc.WebSocket.SetWriteDeadline(time.Now().Add(WRITE_WAIT))
+				wc.WebSocket.SetWriteDeadline(time.Now().Add(writeWaitTime))
 				wc.WebSocket.WriteMessage(websocket.CloseMessage, []byte{})
 				return
 			}
@@ -176,11 +181,12 @@ func (wc *WebConn) writePump() {
 			evt, evtOk := msg.(*model.WebSocketEvent)
 
 			skipSend := false
-			if len(wc.Send) >= SEND_SLOW_WARN {
+			if len(wc.send) >= sendSlowWarn {
 				// When the pump starts to get slow we'll drop non-critical messages
-				if msg.EventType() == model.WEBSOCKET_EVENT_TYPING ||
-					msg.EventType() == model.WEBSOCKET_EVENT_STATUS_CHANGE ||
-					msg.EventType() == model.WEBSOCKET_EVENT_CHANNEL_VIEWED {
+				switch msg.EventType() {
+				case model.WEBSOCKET_EVENT_TYPING,
+					model.WEBSOCKET_EVENT_STATUS_CHANGE,
+					model.WEBSOCKET_EVENT_CHANNEL_VIEWED:
 					mlog.Info(
 						"websocket.slow: dropping message",
 						mlog.String("user_id", wc.UserId),
@@ -201,33 +207,22 @@ func (wc *WebConn) writePump() {
 					msgBytes = []byte(msg.ToJson())
 				}
 
-				if len(wc.Send) >= SEND_DEADLOCK_WARN {
-					if evtOk {
-						mlog.Warn(
-							"websocket.full",
-							mlog.String("user_id", wc.UserId),
-							mlog.String("type", msg.EventType()),
-							mlog.String("channel_id", evt.GetBroadcast().ChannelId),
-							mlog.Int("size", len(msg.ToJson())),
-						)
-					} else {
-						mlog.Warn(
-							"websocket.full",
-							mlog.String("user_id", wc.UserId),
-							mlog.String("type", msg.EventType()),
-							mlog.Int("size", len(msg.ToJson())),
-						)
+				if len(wc.send) >= sendFullWarn {
+					logData := []mlog.Field{
+						mlog.String("user_id", wc.UserId),
+						mlog.String("type", msg.EventType()),
+						mlog.Int("size", len(msgBytes)),
 					}
+					if evtOk {
+						logData = append(logData, mlog.String("channel_id", evt.GetBroadcast().ChannelId))
+					}
+
+					mlog.Warn("websocket.full", logData...)
 				}
 
-				wc.WebSocket.SetWriteDeadline(time.Now().Add(WRITE_WAIT))
+				wc.WebSocket.SetWriteDeadline(time.Now().Add(writeWaitTime))
 				if err := wc.WebSocket.WriteMessage(websocket.TextMessage, msgBytes); err != nil {
-					// browsers will appear as CloseNoStatusReceived
-					if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) {
-						mlog.Debug("websocket.send: client side closed socket", mlog.String("user_id", wc.UserId))
-					} else {
-						mlog.Debug("websocket.send: closing websocket", mlog.String("user_id", wc.UserId), mlog.Err(err))
-					}
+					wc.logSocketErr("websocket.send", err)
 					return
 				}
 
@@ -237,14 +232,9 @@ func (wc *WebConn) writePump() {
 			}
 
 		case <-ticker.C:
-			wc.WebSocket.SetWriteDeadline(time.Now().Add(WRITE_WAIT))
+			wc.WebSocket.SetWriteDeadline(time.Now().Add(writeWaitTime))
 			if err := wc.WebSocket.WriteMessage(websocket.PingMessage, []byte{}); err != nil {
-				// browsers will appear as CloseNoStatusReceived
-				if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) {
-					mlog.Debug("websocket.ticker: client side closed socket", mlog.String("user_id", wc.UserId))
-				} else {
-					mlog.Debug("websocket.ticker: closing websocket", mlog.String("user_id", wc.UserId), mlog.Err(err))
-				}
+				wc.logSocketErr("websocket.ticker", err)
 				return
 			}
 
@@ -261,13 +251,15 @@ func (wc *WebConn) writePump() {
 	}
 }
 
+// InvalidateCache resets all internal data of the WebConn.
 func (wc *WebConn) InvalidateCache() {
-	wc.AllChannelMembers = nil
-	wc.LastAllChannelMembersTime = 0
+	wc.allChannelMembers = nil
+	wc.lastAllChannelMembersTime = 0
 	wc.SetSession(nil)
 	wc.SetSessionExpiresAt(0)
 }
 
+// IsAuthenticated returns whether the given WebConn is authenticated or not.
 func (wc *WebConn) IsAuthenticated() bool {
 	// Check the expiry to see if we need to check for a new session
 	if wc.GetSessionExpiresAt() < model.GetMillis() {
@@ -324,7 +316,8 @@ func (wc *WebConn) shouldSendEventToGuest(msg *model.WebSocketEvent) bool {
 	return canSee
 }
 
-func (wc *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool {
+// shouldSendEvent returns whether the message should be sent or not.
+func (wc *WebConn) shouldSendEvent(msg *model.WebSocketEvent) bool {
 	// IMPORTANT: Do not send event if WebConn does not have a session
 	if !wc.IsAuthenticated() {
 		return false
@@ -353,7 +346,7 @@ func (wc *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool {
 	}
 
 	// If the event is destined to a specific user
-	if len(msg.GetBroadcast().UserId) > 0 {
+	if msg.GetBroadcast().UserId != "" {
 		return wc.UserId == msg.GetBroadcast().UserId
 	}
 
@@ -365,31 +358,31 @@ func (wc *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool {
 	}
 
 	// Only report events to users who are in the channel for the event
-	if len(msg.GetBroadcast().ChannelId) > 0 {
-		if model.GetMillis()-wc.LastAllChannelMembersTime > WEBCONN_MEMBER_CACHE_TIME {
-			wc.AllChannelMembers = nil
-			wc.LastAllChannelMembersTime = 0
+	if msg.GetBroadcast().ChannelId != "" {
+		if model.GetMillis()-wc.lastAllChannelMembersTime > webConnMemberCacheTime {
+			wc.allChannelMembers = nil
+			wc.lastAllChannelMembersTime = 0
 		}
 
-		if wc.AllChannelMembers == nil {
+		if wc.allChannelMembers == nil {
 			result, err := wc.App.Srv().Store.Channel().GetAllChannelMembersForUser(wc.UserId, true, false)
 			if err != nil {
 				mlog.Error("webhub.shouldSendEvent.", mlog.Err(err))
 				return false
 			}
-			wc.AllChannelMembers = result
-			wc.LastAllChannelMembersTime = model.GetMillis()
+			wc.allChannelMembers = result
+			wc.lastAllChannelMembersTime = model.GetMillis()
 		}
 
-		if _, ok := wc.AllChannelMembers[msg.GetBroadcast().ChannelId]; ok {
+		if _, ok := wc.allChannelMembers[msg.GetBroadcast().ChannelId]; ok {
 			return true
 		}
 		return false
 	}
 
 	// Only report events to users who are in the team for the event
-	if len(msg.GetBroadcast().TeamId) > 0 {
-		return wc.IsMemberOfTeam(msg.GetBroadcast().TeamId)
+	if msg.GetBroadcast().TeamId != "" {
+		return wc.isMemberOfTeam(msg.GetBroadcast().TeamId)
 	}
 
 	if wc.GetSession().Props[model.SESSION_PROP_IS_GUEST] == "true" {
@@ -399,10 +392,12 @@ func (wc *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool {
 	return true
 }
 
-func (wc *WebConn) IsMemberOfTeam(teamId string) bool {
+// IsMemberOfTeam returns whether the user of the WebConn
+// is a member of the given teamId or not.
+func (wc *WebConn) isMemberOfTeam(teamId string) bool {
 	currentSession := wc.GetSession()
 
-	if currentSession == nil || len(currentSession.Token) == 0 {
+	if currentSession == nil || currentSession.Token == "" {
 		session, err := wc.App.GetSession(wc.GetSessionToken())
 		if err != nil {
 			mlog.Error("Invalid session.", mlog.Err(err))
@@ -414,3 +409,12 @@ func (wc *WebConn) IsMemberOfTeam(teamId string) bool {
 
 	return currentSession.GetTeamByTeamId(teamId) != nil
 }
+
+func (wc *WebConn) logSocketErr(source string, err error) {
+	// browsers will appear as CloseNoStatusReceived
+	if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) {
+		mlog.Debug(source+": client side closed socket", mlog.String("user_id", wc.UserId))
+	} else {
+		mlog.Debug(source+": closing websocket", mlog.String("user_id", wc.UserId), mlog.Err(err))
+	}
+}
diff --git a/app/web_conn_test.go b/app/web_conn_test.go
index d9a146a8d5..5672bc6a96 100644
--- a/app/web_conn_test.go
+++ b/app/web_conn_test.go
@@ -86,15 +86,15 @@ func TestWebConnShouldSendEvent(t *testing.T) {
 	event := model.NewWebSocketEvent("some_event", "", "", "", nil)
 	for _, c := range cases {
 		event = event.SetBroadcast(c.Broadcast)
-		assert.Equal(t, c.User1Expected, basicUserWc.ShouldSendEvent(event), c.Description)
-		assert.Equal(t, c.User2Expected, basicUser2Wc.ShouldSendEvent(event), c.Description)
-		assert.Equal(t, c.AdminExpected, adminUserWc.ShouldSendEvent(event), c.Description)
+		assert.Equal(t, c.User1Expected, basicUserWc.shouldSendEvent(event), c.Description)
+		assert.Equal(t, c.User2Expected, basicUser2Wc.shouldSendEvent(event), c.Description)
+		assert.Equal(t, c.AdminExpected, adminUserWc.shouldSendEvent(event), c.Description)
 	}
 
 	event2 := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_UPDATE_TEAM, th.BasicTeam.Id, "", "", nil)
-	assert.True(t, basicUserWc.ShouldSendEvent(event2))
-	assert.True(t, basicUser2Wc.ShouldSendEvent(event2))
+	assert.True(t, basicUserWc.shouldSendEvent(event2))
+	assert.True(t, basicUser2Wc.shouldSendEvent(event2))
 
 	event3 := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_UPDATE_TEAM, "wrongId", "", "", nil)
-	assert.False(t, basicUserWc.ShouldSendEvent(event3))
+	assert.False(t, basicUserWc.shouldSendEvent(event3))
 }
diff --git a/app/web_hub.go b/app/web_hub.go
index f7623b2ebe..935ab2da5d 100644
--- a/app/web_hub.go
+++ b/app/web_hub.go
@@ -15,15 +15,23 @@ import (
 )
 
 const (
-	BROADCAST_QUEUE_SIZE = 4096
+	broadcastQueueSize = 4096
 )
 
-type WebConnActivityMessage struct {
-	UserId       string
-	SessionToken string
-	ActivityAt   int64
+type webConnActivityMessage struct {
+	userId       string
+	sessionToken string
+	activityAt   int64
 }
 
+type webConnDirectMessage struct {
+	conn *WebConn
+	msg  model.WebSocketMessage
+}
+
+// Hub is the central place to manage all websocket connections in the server.
+// It handles different websocket events and sending messages to individual
+// user connections.
 type Hub struct {
 	// connectionCount should be kept first.
 	// See https://github.com/mattermost/mattermost-server/pull/7281
@@ -36,21 +44,23 @@ type Hub struct {
 	stop            chan struct{}
 	didStop         chan struct{}
 	invalidateUser  chan string
-	activity        chan *WebConnActivityMessage
-	ExplicitStop    bool
+	activity        chan *webConnActivityMessage
+	directMsg       chan *webConnDirectMessage
+	explicitStop    bool
 }
 
+// NewWebHub creates a new Hub.
 func (a *App) NewWebHub() *Hub {
 	return &Hub{
 		app:            a,
 		register:       make(chan *WebConn, 1),
 		unregister:     make(chan *WebConn, 1),
-		broadcast:      make(chan *model.WebSocketEvent, BROADCAST_QUEUE_SIZE),
+		broadcast:      make(chan *model.WebSocketEvent, broadcastQueueSize),
 		stop:           make(chan struct{}),
 		didStop:        make(chan struct{}),
 		invalidateUser: make(chan string),
-		activity:       make(chan *WebConnActivityMessage),
-		ExplicitStop:   false,
+		activity:       make(chan *webConnActivityMessage),
+		directMsg:      make(chan *webConnDirectMessage),
 	}
 }
 
@@ -58,6 +68,7 @@ func (a *App) TotalWebsocketConnections() int {
 	return a.Srv().TotalWebsocketConnections()
 }
 
+// HubStart starts all the hubs.
 func (a *App) HubStart() {
 	// Total number of hubs is twice the number of CPUs.
 	numberOfHubs := runtime.NumCPU() * 2
@@ -77,6 +88,36 @@ func (a *App) HubStart() {
 	}
 }
 
+func (a *App) PublishSkipClusterSend(message *model.WebSocketEvent) {
+	if message.GetBroadcast().UserId != "" {
+		hub := a.GetHubForUserId(message.GetBroadcast().UserId)
+		if hub != nil {
+			hub.Broadcast(message)
+		}
+		return
+	}
+	for _, hub := range a.Srv().GetHubs() {
+		hub.Broadcast(message)
+	}
+}
+
+func (a *App) invalidateCacheForUserSkipClusterSend(userId string) {
+	a.Srv().Store.Channel().InvalidateAllChannelMembersForUser(userId)
+	a.InvalidateWebConnSessionCacheForUser(userId)
+}
+
+func (a *App) invalidateCacheForWebhook(webhookId string) {
+	a.Srv().Store.Webhook().InvalidateWebhookCache(webhookId)
+}
+
+func (a *App) InvalidateWebConnSessionCacheForUser(userId string) {
+	hub := a.GetHubForUserId(userId)
+	if hub != nil {
+		hub.InvalidateUser(userId)
+	}
+}
+
+// HubStop stops all the hubs.
 func (a *App) HubStop() {
 	mlog.Info("stopping websocket hub connections")
 
@@ -87,6 +128,7 @@ func (a *App) HubStop() {
 	a.Srv().SetHubs([]*Hub{})
 }
 
+// GetHubForUserId returns the hub for a given user id.
 func (a *App) GetHubForUserId(userId string) *Hub {
 	if len(a.Srv().GetHubs()) == 0 {
 		return nil
@@ -103,6 +145,7 @@ func (a *App) GetHubForUserId(userId string) *Hub {
 	return hub
 }
 
+// HubRegister registers a connection to a hub.
 func (a *App) HubRegister(webConn *WebConn) {
 	hub := a.GetHubForUserId(webConn.UserId)
 	if hub != nil {
@@ -113,6 +156,7 @@ func (a *App) HubRegister(webConn *WebConn) {
 	}
 }
 
+// HubUnregister unregisters a connection from a hub.
 func (a *App) HubUnregister(webConn *WebConn) {
 	hub := a.GetHubForUserId(webConn.UserId)
 	if hub != nil {
@@ -149,19 +193,6 @@ func (a *App) Publish(message *model.WebSocketEvent) {
 	}
 }
 
-func (a *App) PublishSkipClusterSend(message *model.WebSocketEvent) {
-	if message.GetBroadcast().UserId != "" {
-		hub := a.GetHubForUserId(message.GetBroadcast().UserId)
-		if hub != nil {
-			hub.Broadcast(message)
-		}
-	} else {
-		for _, hub := range a.Srv().GetHubs() {
-			hub.Broadcast(message)
-		}
-	}
-}
-
 func (a *App) invalidateCacheForChannel(channel *model.Channel) {
 	a.Srv().Store.Channel().InvalidateChannel(channel.Id)
 	a.invalidateCacheForChannelByNameSkipClusterSend(channel.TeamId, channel.Name)
@@ -237,7 +268,7 @@ func (a *App) InvalidateCacheForUser(userId string) {
 }
 
 func (a *App) invalidateCacheForUserTeams(userId string) {
-	a.invalidateCacheForUserTeamsSkipClusterSend(userId)
+	a.InvalidateWebConnSessionCacheForUser(userId)
 	a.Srv().Store.Team().InvalidateAllTeamIdsForUser(userId)
 
 	if a.Cluster() != nil {
@@ -250,33 +281,7 @@ func (a *App) invalidateCacheForUserTeams(userId string) {
 	}
 }
 
-func (a *App) invalidateCacheForUserSkipClusterSend(userId string) {
-	a.Srv().Store.Channel().InvalidateAllChannelMembersForUser(userId)
-
-	hub := a.GetHubForUserId(userId)
-	if hub != nil {
-		hub.InvalidateUser(userId)
-	}
-}
-
-func (a *App) invalidateCacheForUserTeamsSkipClusterSend(userId string) {
-	hub := a.GetHubForUserId(userId)
-	if hub != nil {
-		hub.InvalidateUser(userId)
-	}
-}
-
-func (a *App) invalidateCacheForWebhook(webhookId string) {
-	a.Srv().Store.Webhook().InvalidateWebhookCache(webhookId)
-}
-
-func (a *App) InvalidateWebConnSessionCacheForUser(userId string) {
-	hub := a.GetHubForUserId(userId)
-	if hub != nil {
-		hub.InvalidateUser(userId)
-	}
-}
-
+// UpdateWebConnUserActivity sets the LastUserActivityAt of the hub for the given session.
 func (a *App) UpdateWebConnUserActivity(session model.Session, activityAt int64) {
 	hub := a.GetHubForUserId(session.UserId)
 	if hub != nil {
@@ -284,13 +289,15 @@ func (a *App) UpdateWebConnUserActivity(session model.Session, activityAt int64)
 	}
 }
 
+// Register registers a connection to the hub.
 func (h *Hub) Register(webConn *WebConn) {
 	select {
 	case h.register <- webConn:
-	case <-h.didStop:
+	case <-h.stop:
 	}
 }
 
+// Unregister unregisters a connection from the hub.
 func (h *Hub) Unregister(webConn *WebConn) {
 	select {
 	case h.unregister <- webConn:
@@ -298,37 +305,64 @@ func (h *Hub) Unregister(webConn *WebConn) {
 	}
 }
 
+// Broadcast broadcasts the message to all connections in the hub.
 func (h *Hub) Broadcast(message *model.WebSocketEvent) {
-	if h != nil && h.broadcast != nil && message != nil {
+	// XXX: The hub nil check is because of the way we setup our tests. We call `app.NewServer()`
+	// which returns a server, but only after that, we call `wsapi.Init()` through our FakeApp adapter
+	// to initialize the hub. But in the `NewServer` call itself, we call `RunOldAppInitialization`
+	// which directly proceeds to broadcast some messages happily.
+	// This needs to be fixed once the FakeApp adapter goes away. And possibly, we can look into
+	// doing hub initialization inside NewServer itself.
+	if h != nil && message != nil {
 		if metrics := h.app.Metrics(); metrics != nil {
 			metrics.IncrementWebSocketBroadcastBufferSize(strconv.Itoa(h.connectionIndex), 1)
 		}
 		select {
 		case h.broadcast <- message:
-		case <-h.didStop:
+		case <-h.stop:
 		}
 	}
 }
 
+// InvalidateUser invalidates the cache for the given user.
 func (h *Hub) InvalidateUser(userId string) {
 	select {
 	case h.invalidateUser <- userId:
-	case <-h.didStop:
+	case <-h.stop:
 	}
 }
 
+// UpdateActivity sets the LastUserActivityAt field for the connection
+// of the user.
 func (h *Hub) UpdateActivity(userId, sessionToken string, activityAt int64) {
 	select {
-	case h.activity <- &WebConnActivityMessage{UserId: userId, SessionToken: sessionToken, ActivityAt: activityAt}:
-	case <-h.didStop:
+	case h.activity <- &webConnActivityMessage{
+		userId:       userId,
+		sessionToken: sessionToken,
+		activityAt:   activityAt,
+	}:
+	case <-h.stop:
 	}
 }
 
+// SendMessage sends the given message to the given connection.
+func (h *Hub) SendMessage(conn *WebConn, msg model.WebSocketMessage) {
+	select {
+	case h.directMsg <- &webConnDirectMessage{
+		conn: conn,
+		msg:  msg,
+	}:
+	case <-h.stop:
+	}
+}
+
+// Stop stops the hub.
 func (h *Hub) Stop() {
 	close(h.stop)
 	<-h.didStop
 }
 
+// Start starts the hub.
 func (h *Hub) Start() {
 	var doStart func()
 	var doRecoverableStart func()
@@ -337,85 +371,94 @@ func (h *Hub) Start() {
 	doStart = func() {
 		mlog.Debug("Hub is starting", mlog.Int("index", h.connectionIndex))
 
-		connections := newHubConnectionIndex()
+		connIndex := newHubConnectionIndex()
 
 		for {
 			select {
-			case webCon := <-h.register:
-				connections.Add(webCon)
-				atomic.StoreInt64(&h.connectionCount, int64(len(connections.All())))
-				if webCon.IsAuthenticated() {
-					webCon.Send <- webCon.createHelloMessage()
+			case webConn := <-h.register:
+				connIndex.Add(webConn)
+				atomic.StoreInt64(&h.connectionCount, int64(len(connIndex.All())))
+				if webConn.IsAuthenticated() {
+					webConn.send <- webConn.createHelloMessage()
 				}
-			case webCon := <-h.unregister:
-				connections.Remove(webCon)
-				atomic.StoreInt64(&h.connectionCount, int64(len(connections.All())))
+			case webConn := <-h.unregister:
+				connIndex.Remove(webConn)
+				atomic.StoreInt64(&h.connectionCount, int64(len(connIndex.All())))
 
-				if len(webCon.UserId) == 0 {
+				if len(webConn.UserId) == 0 {
 					continue
 				}
 
-				conns := connections.ForUser(webCon.UserId)
+				conns := connIndex.ForUser(webConn.UserId)
 				if len(conns) == 0 {
 					h.app.Srv().Go(func() {
-						h.app.SetStatusOffline(webCon.UserId, false)
+						h.app.SetStatusOffline(webConn.UserId, false)
 					})
-				} else {
-					var latestActivity int64 = 0
-					for _, conn := range conns {
-						if conn.LastUserActivityAt > latestActivity {
-							latestActivity = conn.LastUserActivityAt
-						}
-					}
-					if h.app.IsUserAway(latestActivity) {
-						h.app.Srv().Go(func() {
-							h.app.SetStatusLastActivityAt(webCon.UserId, latestActivity)
-						})
+					continue
+				}
+				var latestActivity int64 = 0
+				for _, conn := range conns {
+					if conn.lastUserActivityAt > latestActivity {
+						latestActivity = conn.lastUserActivityAt
 					}
 				}
+
+				if h.app.IsUserAway(latestActivity) {
+					h.app.Srv().Go(func() {
+						h.app.SetStatusLastActivityAt(webConn.UserId, latestActivity)
+					})
+				}
 			case userId := <-h.invalidateUser:
-				for _, webCon := range connections.ForUser(userId) {
-					webCon.InvalidateCache()
+				for _, webConn := range connIndex.ForUser(userId) {
+					webConn.InvalidateCache()
 				}
 			case activity := <-h.activity:
-				for _, webCon := range connections.ForUser(activity.UserId) {
-					if webCon.GetSessionToken() == activity.SessionToken {
-						webCon.LastUserActivityAt = activity.ActivityAt
+				for _, webConn := range connIndex.ForUser(activity.userId) {
+					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:
+					mlog.Error("webhub.broadcast: cannot send, closing websocket for user", mlog.String("user_id", directMsg.conn.UserId))
+					close(directMsg.conn.send)
+					connIndex.Remove(directMsg.conn)
+				}
 			case msg := <-h.broadcast:
 				if metrics := h.app.Metrics(); metrics != nil {
 					metrics.DecrementWebSocketBroadcastBufferSize(strconv.Itoa(h.connectionIndex), 1)
 				}
-				candidates := connections.All()
+				candidates := connIndex.All()
 				if msg.GetBroadcast().UserId != "" {
-					candidates = connections.ForUser(msg.GetBroadcast().UserId)
+					candidates = connIndex.ForUser(msg.GetBroadcast().UserId)
 				}
 				msg = msg.PrecomputeJSON()
-				for _, webCon := range candidates {
-					if webCon.ShouldSendEvent(msg) {
+				for _, webConn := range candidates {
+					if !connIndex.Has(webConn) {
+						continue
+					}
+					if webConn.shouldSendEvent(msg) {
 						select {
-						case webCon.Send <- msg:
+						case webConn.send <- msg:
 						default:
-							mlog.Error("webhub.broadcast: cannot send, closing websocket for user", mlog.String("user_id", webCon.UserId))
-							close(webCon.Send)
-							connections.Remove(webCon)
+							mlog.Error("webhub.broadcast: cannot send, closing websocket for user", mlog.String("user_id", webConn.UserId))
+							close(webConn.send)
+							connIndex.Remove(webConn)
 						}
 					}
 				}
 			case <-h.stop:
-				userIds := make(map[string]bool)
-
-				for _, webCon := range connections.All() {
-					userIds[webCon.UserId] = true
-					webCon.Close()
+				for _, webConn := range connIndex.All() {
+					webConn.Close()
+					h.app.SetStatusOffline(webConn.UserId, false)
 				}
 
-				for userId := range userIds {
-					h.app.SetStatusOffline(userId, false)
-				}
-
-				h.ExplicitStop = true
+				h.explicitStop = true
 				close(h.didStop)
 
 				return
@@ -429,7 +472,7 @@ func (h *Hub) Start() {
 	}
 
 	doRecover = func() {
-		if !h.ExplicitStop {
+		if !h.explicitStop {
 			if r := recover(); r != nil {
 				mlog.Error("Recovering from Hub panic.", mlog.Any("panic", r))
 			} else {
@@ -494,6 +537,11 @@ func (i *hubConnectionIndex) Remove(wc *WebConn) {
 	delete(i.connectionIndexes, wc)
 }
 
+func (i *hubConnectionIndex) Has(wc *WebConn) bool {
+	_, ok := i.connectionIndexes[wc]
+	return ok
+}
+
 func (i *hubConnectionIndex) ForUser(id string) []*WebConn {
 	return i.connectionsByUserId[id]
 }
diff --git a/app/web_hub_test.go b/app/web_hub_test.go
index 9f36c3442c..d4a27f8d9d 100644
--- a/app/web_hub_test.go
+++ b/app/web_hub_test.go
@@ -90,7 +90,7 @@ func TestHubStopRaceCondition(t *testing.T) {
 
 		hub.UpdateActivity("userId", "sessionToken", 0)
 
-		for i := 0; i <= BROADCAST_QUEUE_SIZE; i++ {
+		for i := 0; i <= broadcastQueueSize; i++ {
 			hub.Broadcast(model.NewWebSocketEvent("", "", "", "", nil))
 		}
 
diff --git a/app/websocket_router.go b/app/websocket_router.go
index 236fca56de..80cc0e7980 100644
--- a/app/websocket_router.go
+++ b/app/websocket_router.go
@@ -27,13 +27,13 @@ func (wr *WebSocketRouter) Handle(action string, handler webSocketHandler) {
 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, r, err)
+		returnWebSocketError(wr.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, r, err)
+		returnWebSocketError(wr.app, conn, r, err)
 		return
 	}
 
@@ -66,28 +66,32 @@ func (wr *WebSocketRouter) ServeWebSocket(conn *WebConn, r *model.WebSocketReque
 		wr.app.HubRegister(conn)
 
 		resp := model.NewWebSocketResponse(model.STATUS_OK, r.Seq, nil)
-		conn.Send <- resp
+		hub := wr.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, r, err)
+		returnWebSocketError(wr.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, r, err)
+		returnWebSocketError(wr.app, conn, r, err)
 		return
 	}
 
 	handler.ServeWebSocket(conn, r)
 }
 
-func ReturnWebSocketError(conn *WebConn, r *model.WebSocketRequest, err *model.AppError) {
+func returnWebSocketError(app *App, conn *WebConn, r *model.WebSocketRequest, err *model.AppError) {
 	mlog.Error(
 		"websocket routing error.",
 		mlog.Int64("seq", r.Seq),
@@ -96,8 +100,12 @@ func ReturnWebSocketError(conn *WebConn, r *model.WebSocketRequest, err *model.A
 		mlog.Err(err),
 	)
 
+	hub := app.GetHubForUserId(conn.UserId)
+	if hub == nil {
+		return
+	}
+
 	err.DetailedError = ""
 	errorResp := model.NewWebSocketError(r.Seq, err)
-
-	conn.Send <- errorResp
+	hub.SendMessage(conn, errorResp)
 }
diff --git a/wsapi/websocket_handler.go b/wsapi/websocket_handler.go
index b0da94844a..46ed07f67b 100644
--- a/wsapi/websocket_handler.go
+++ b/wsapi/websocket_handler.go
@@ -24,6 +24,10 @@ type webSocketHandler struct {
 func (wh webSocketHandler) ServeWebSocket(conn *app.WebConn, r *model.WebSocketRequest) {
 	mlog.Debug("Websocket request", mlog.String("action", r.Action))
 
+	hub := wh.app.GetHubForUserId(conn.UserId)
+	if hub == nil {
+		return
+	}
 	session, sessionErr := wh.app.GetSession(conn.GetSessionToken())
 	if sessionErr != nil {
 		mlog.Error(
@@ -36,8 +40,7 @@ func (wh webSocketHandler) ServeWebSocket(conn *app.WebConn, r *model.WebSocketR
 		)
 		sessionErr.DetailedError = ""
 		errResp := model.NewWebSocketError(r.Seq, sessionErr)
-
-		conn.Send <- errResp
+		hub.SendMessage(conn, errResp)
 		return
 	}
 
@@ -59,14 +62,12 @@ func (wh webSocketHandler) ServeWebSocket(conn *app.WebConn, r *model.WebSocketR
 		)
 		err.DetailedError = ""
 		errResp := model.NewWebSocketError(r.Seq, err)
-
-		conn.Send <- errResp
+		hub.SendMessage(conn, errResp)
 		return
 	}
 
 	resp := model.NewWebSocketResponse(model.STATUS_OK, r.Seq, data)
-
-	conn.Send <- resp
+	hub.SendMessage(conn, resp)
 }
 
 func NewInvalidWebSocketParamError(action string, name string) *model.AppError {