mirrors/mostlymatter
1
0
Форкнуть 0
Код Задачи Запросы на слияние Действия Пакеты Проекты Релизы Вики Активность

MM-30048 added thread related socket messages (#16234)

Просмотр исходного кода 
Co-authored-by: Mattermod <mattermod@users.noreply.github.com>
Этот коммит содержится в:
Eli Yukelzon
2020-11-20 11:00:52 +02:00
коммит произвёл GitHub
родитель 7b515d27f2
Коммит cf4df5fcd2
11 изменённых файлов: 190 добавлений и 26 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

92
api4/user_test.go
Просмотреть файл

@@ -5401,6 +5401,98 @@ func TestGetThreadsForUser(t *testing.T) {
})
}
func TestThreadSocketEvents(t *testing.T) {
th := Setup(t).InitBasic()
defer th.TearDown()
th.App.UpdateConfig(func(cfg *model.Config) {
*cfg.ServiceSettings.ThreadAutoFollow = true
*cfg.ServiceSettings.CollapsedThreads = model.COLLAPSED_THREADS_DEFAULT_ON
})
userWSClient, err := th.CreateWebSocketClient()
require.Nil(t, err)
defer userWSClient.Close()
userWSClient.Listen()
Client := th.Client
rpost, resp := Client.CreatePost(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testMsg"})
CheckNoError(t, resp)
CheckCreatedStatus(t, resp)
_, err = th.App.CreatePostAsUser(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testReply", UserId: th.BasicUser2.Id, RootId: rpost.Id}, th.App.Session().Id, false)
require.Nil(t, err)
defer th.App.Srv().Store.Post().PermanentDeleteByUser(th.BasicUser.Id)
defer th.App.Srv().Store.Post().PermanentDeleteByUser(th.BasicUser2.Id)
t.Run("Listed for update event", func(t *testing.T) {
var caught bool
func() {
for {
select {
case ev := <-userWSClient.EventChannel:
if ev.EventType() == model.WEBSOCKET_EVENT_THREAD_UPDATED {
caught = true
thread, err := model.ThreadFromJson(ev.GetData()["thread"].(string))
require.Nil(t, err)
require.Contains(t, thread.Participants, th.BasicUser.Id)
require.Contains(t, thread.Participants, th.BasicUser2.Id)
}
case <-time.After(1 * time.Second):
return
}
}
}()
require.Truef(t, caught, "User should have received %s event", model.WEBSOCKET_EVENT_THREAD_UPDATED)
})
resp = th.Client.UpdateThreadFollowForUser(th.BasicUser.Id, rpost.Id, false)
CheckNoError(t, resp)
CheckOKStatus(t, resp)
t.Run("Listed for follow event", func(t *testing.T) {
var caught bool
func() {
for {
select {
case ev := <-userWSClient.EventChannel:
if ev.EventType() == model.WEBSOCKET_EVENT_THREAD_FOLLOW_CHANGED {
caught = true
require.Equal(t, ev.GetData()["state"], false)
}
case <-time.After(1 * time.Second):
return
}
}
}()
require.Truef(t, caught, "User should have received %s event", model.WEBSOCKET_EVENT_THREAD_FOLLOW_CHANGED)
})
resp = th.Client.UpdateThreadReadForUser(th.BasicUser.Id, rpost.Id, 123)
CheckNoError(t, resp)
CheckOKStatus(t, resp)
t.Run("Listed for read event", func(t *testing.T) {
var caught bool
func() {
for {
select {
case ev := <-userWSClient.EventChannel:
if ev.EventType() == model.WEBSOCKET_EVENT_THREAD_READ_CHANGED {
caught = true
require.EqualValues(t, ev.GetData()["timestamp"], 123)
}
case <-time.After(1 * time.Second):
return
}
}
}()
require.Truef(t, caught, "User should have received %s event", model.WEBSOCKET_EVENT_THREAD_READ_CHANGED)
})
}
func TestFollowThreads(t *testing.T) {
th := Setup(t).InitBasic()
defer th.TearDown()

52
app/notification.go
Просмотреть файл

@@ -6,6 +6,7 @@ package app
import (
"net/http"
"sort"
"strconv"
"strings"
"unicode"
"unicode/utf8"
@@ -160,24 +161,31 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
mentionedUsersList := make([]string, 0, len(mentions.Mentions))
updateMentionChans := []chan *model.AppError{}
mentionAutofollowChans := []chan *model.AppError{}
threadParticipants := []string{post.UserId}
if *a.Config().ServiceSettings.ThreadAutoFollow && post.RootId != "" {
if parentPostList != nil {
threadParticipants = append(threadParticipants, parentPostList.Posts[parentPostList.Order[0]].UserId)
}
for id := range mentions.Mentions {
threadParticipants = append(threadParticipants, id)
}
// for each mention, make sure to update thread autofollow
for _, id := range threadParticipants {
mac := make(chan *model.AppError, 1)
go func(userId string) {
defer close(mac)
// for each mention, make sure to update thread autofollow
for id := range mentions.Mentions {
mac := make(chan *model.AppError, 1)
go func(userId string) {
defer close(mac)
if *a.Config().ServiceSettings.ThreadAutoFollow && post.RootId != "" {
nErr := a.Srv().Store.Thread().CreateMembershipIfNeeded(userId, post.RootId, true)
if nErr != nil {
mac <- model.NewAppError("SendNotifications", "app.channel.autofollow.app_error", nil, nErr.Error(), http.StatusInternalServerError)
return
}
}
mac <- nil
}(id)
mentionAutofollowChans = append(mentionAutofollowChans, mac)
}
mac <- nil
}(id)
mentionAutofollowChans = append(mentionAutofollowChans, mac)
}
}
for id := range mentions.Mentions {
mentionedUsersList = append(mentionedUsersList, id)
@@ -403,6 +411,28 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
}
a.Publish(message)
// If this is a reply in a thread, notify participants
if *a.Config().ServiceSettings.CollapsedThreads != model.COLLAPSED_THREADS_DISABLED && post.RootId != "" {
thread, err := a.Srv().Store.Thread().Get(post.RootId)
if err != nil {
mlog.Error("Cannot get thread", mlog.String("id", post.RootId))
return nil, err
}
payload := thread.ToJson()
for _, uid := range thread.Participants {
sendEvent := *a.Config().ServiceSettings.CollapsedThreads == model.COLLAPSED_THREADS_DEFAULT_ON
// check if a participant has overridden collapsed threads settings
if preference, err := a.Srv().Store.Preference().Get(uid, model.PREFERENCE_CATEGORY_COLLAPSED_THREADS_SETTINGS, model.PREFERENCE_NAME_COLLAPSED_THREADS_ENABLED); err == nil {
sendEvent, _ = strconv.ParseBool(preference.Value)
}
if sendEvent {
message := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_THREAD_UPDATED, "", "", uid, nil)
message.Add("thread", payload)
a.Publish(message)
}
}
}
return mentionedUsersList, nil
}

11
app/user.go
Просмотреть файл

@@ -2388,6 +2388,9 @@ func (a *App) UpdateThreadsReadForUser(userId string, timestamp int64) *model.Ap
if err != nil {
return model.NewAppError("UpdateThreadsReadForUser", "app.user.update_threads_read_for_user.app_error", nil, err.Error(), http.StatusInternalServerError)
}
message := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_THREAD_READ_CHANGED, "", "", userId, nil)
message.Add("timestamp", timestamp)
a.Publish(message)
return nil
}
@@ -2396,6 +2399,10 @@ func (a *App) UpdateThreadFollowForUser(userId, threadId string, state bool) *mo
if err != nil {
return model.NewAppError("UpdateThreadFollowForUser", "app.user.update_thread_follow_for_user.app_error", nil, err.Error(), http.StatusInternalServerError)
}
message := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_THREAD_FOLLOW_CHANGED, "", "", userId, nil)
message.Add("thread_id", threadId)
message.Add("state", state)
a.Publish(message)
return nil
}
@@ -2404,5 +2411,9 @@ func (a *App) UpdateThreadReadForUser(userId, threadId string, timestamp int64)
if err != nil {
return model.NewAppError("UpdateThreadReadForUser", "app.user.update_thread_read_for_user.app_error", nil, err.Error(), http.StatusInternalServerError)
}
message := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_THREAD_READ_CHANGED, "", "", userId, nil)
message.Add("thread_id", threadId)
message.Add("timestamp", timestamp)
a.Publish(message)
return nil
}

4
i18n/en.json
Просмотреть файл

@@ -7022,6 +7022,10 @@
"id": "model.config.is_valid.cluster_email_batching.app_error",
"translation": "Unable to enable email batching when clustering is enabled."
},
{
"id": "model.config.is_valid.collapsed_threads.app_error",
"translation": "CollapsedThreads setting must be either disabled,default_on or default_off"
},
{
"id": "model.config.is_valid.data_retention.deletion_job_start_time.app_error",
"translation": "Data retention job start time must be a 24-hour time stamp in the form HH:MM."

15
model/config.go
Просмотреть файл

@@ -83,6 +83,10 @@ const (
GROUP_UNREAD_CHANNELS_DEFAULT_ON = "default_on"
GROUP_UNREAD_CHANNELS_DEFAULT_OFF = "default_off"
COLLAPSED_THREADS_DISABLED = "disabled"
COLLAPSED_THREADS_DEFAULT_ON = "default_on"
COLLAPSED_THREADS_DEFAULT_OFF = "default_off"
EMAIL_BATCHING_BUFFER_SIZE = 256
EMAIL_BATCHING_INTERVAL = 30
@@ -351,6 +355,7 @@ type ServiceSettings struct {
FeatureFlagSyncIntervalSeconds *int `access:"environment,write_restrictable"`
DebugSplit *bool `access:"environment,write_restrictable"`
ThreadAutoFollow *bool `access:"experimental"`
CollapsedThreads *string `access:"experimental"`
ManagedResourcePaths *string `access:"environment,write_restrictable,cloud_restrictable"`
}
@@ -786,6 +791,10 @@ func (s *ServiceSettings) SetDefaults(isUpdate bool) {
s.ThreadAutoFollow = NewBool(true)
}
if s.CollapsedThreads == nil {
s.CollapsedThreads = NewString(COLLAPSED_THREADS_DISABLED)
}
if s.ManagedResourcePaths == nil {
s.ManagedResourcePaths = NewString("")
}
@@ -3437,6 +3446,12 @@ func (s *ServiceSettings) isValid() *AppError {
return NewAppError("Config.IsValid", "model.config.is_valid.group_unread_channels.app_error", nil, "", http.StatusBadRequest)
}
if *s.CollapsedThreads != COLLAPSED_THREADS_DISABLED &&
*s.CollapsedThreads != COLLAPSED_THREADS_DEFAULT_ON &&
*s.CollapsedThreads != COLLAPSED_THREADS_DEFAULT_OFF {
return NewAppError("Config.IsValid", "model.config.is_valid.collapsed_threads.app_error", nil, "", http.StatusBadRequest)
}
return nil
}

28
model/preference.go
Просмотреть файл

@@ -13,20 +13,22 @@ import (
)
const (
PREFERENCE_CATEGORY_DIRECT_CHANNEL_SHOW = "direct_channel_show"
PREFERENCE_CATEGORY_GROUP_CHANNEL_SHOW = "group_channel_show"
PREFERENCE_CATEGORY_TUTORIAL_STEPS = "tutorial_step"
PREFERENCE_CATEGORY_ADVANCED_SETTINGS = "advanced_settings"
PREFERENCE_CATEGORY_FLAGGED_POST = "flagged_post"
PREFERENCE_CATEGORY_FAVORITE_CHANNEL = "favorite_channel"
PREFERENCE_CATEGORY_SIDEBAR_SETTINGS = "sidebar_settings"
PREFERENCE_CATEGORY_DIRECT_CHANNEL_SHOW = "direct_channel_show"
PREFERENCE_CATEGORY_GROUP_CHANNEL_SHOW = "group_channel_show"
PREFERENCE_CATEGORY_TUTORIAL_STEPS = "tutorial_step"
PREFERENCE_CATEGORY_ADVANCED_SETTINGS = "advanced_settings"
PREFERENCE_CATEGORY_FLAGGED_POST = "flagged_post"
PREFERENCE_CATEGORY_FAVORITE_CHANNEL = "favorite_channel"
PREFERENCE_CATEGORY_SIDEBAR_SETTINGS = "sidebar_settings"
PREFERENCE_CATEGORY_COLLAPSED_THREADS_SETTINGS = "collapsed_threads_settings"
PREFERENCE_CATEGORY_DISPLAY_SETTINGS = "display_settings"
PREFERENCE_NAME_CHANNEL_DISPLAY_MODE = "channel_display_mode"
PREFERENCE_NAME_COLLAPSE_SETTING = "collapse_previews"
PREFERENCE_NAME_MESSAGE_DISPLAY = "message_display"
PREFERENCE_NAME_NAME_FORMAT = "name_format"
PREFERENCE_NAME_USE_MILITARY_TIME = "use_military_time"
PREFERENCE_CATEGORY_DISPLAY_SETTINGS = "display_settings"
PREFERENCE_NAME_COLLAPSED_THREADS_ENABLED = "collapsed_threads_enabled"
PREFERENCE_NAME_CHANNEL_DISPLAY_MODE = "channel_display_mode"
PREFERENCE_NAME_COLLAPSE_SETTING = "collapse_previews"
PREFERENCE_NAME_MESSAGE_DISPLAY = "message_display"
PREFERENCE_NAME_NAME_FORMAT = "name_format"
PREFERENCE_NAME_USE_MILITARY_TIME = "use_military_time"
PREFERENCE_CATEGORY_THEME = "theme"
// the name for theme props is the team id

6
model/thread.go
Просмотреть файл

@@ -56,6 +56,12 @@ func (o *Thread) ToJson() string {
return string(b)
}
func ThreadFromJson(s string) (*Thread, error) {
var t Thread
err := json.Unmarshal([]byte(s), &t)
return &t, err
}
func (o *Thread) Etag() string {
return Etag(o.PostId, o.LastReplyAt)
}

3
model/websocket_message.go
Просмотреть файл

@@ -70,6 +70,9 @@ const (
WEBSOCKET_WARN_METRIC_STATUS_RECEIVED = "warn_metric_status_received"
WEBSOCKET_WARN_METRIC_STATUS_REMOVED = "warn_metric_status_removed"
WEBSOCKET_EVENT_CLOUD_PAYMENT_STATUS_UPDATED = "cloud_payment_status_updated"
WEBSOCKET_EVENT_THREAD_UPDATED = "thread_updated"
WEBSOCKET_EVENT_THREAD_FOLLOW_CHANGED = "thread_follow_changed"
WEBSOCKET_EVENT_THREAD_READ_CHANGED = "thread_read_changed"
)
type WebSocketMessage interface {

1
services/telemetry/telemetry.go
Просмотреть файл

@@ -405,6 +405,7 @@ func (ts *TelemetryService) trackConfig() {
"enable_tutorial": *cfg.ServiceSettings.EnableTutorial,
"experimental_enable_default_channel_leave_join_messages": *cfg.ServiceSettings.ExperimentalEnableDefaultChannelLeaveJoinMessages,
"experimental_group_unread_channels": *cfg.ServiceSettings.ExperimentalGroupUnreadChannels,
"collapsed_threads": *cfg.ServiceSettings.CollapsedThreads,
"websocket_url": isDefault(*cfg.ServiceSettings.WebsocketURL, ""),
"allow_cookies_for_subdomains": *cfg.ServiceSettings.AllowCookiesForSubdomains,
"enable_api_team_deletion": *cfg.ServiceSettings.EnableAPITeamDeletion,

2
store/sqlstore/post_store.go
Просмотреть файл

@@ -1976,7 +1976,7 @@ func (s *SqlPostStore) updateThreadsFromPosts(transaction *gorp.Transaction, pos
if thread, found := threadByRoot[rootId]; !found {
// calculate participants
var participants model.StringArray
if _, err := transaction.Select(&participants, "SELECT DISTINCT UserId FROM Posts WHERE RootId=:RootId", map[string]interface{}{"RootId": rootId}); err != nil {
if _, err := transaction.Select(&participants, "SELECT DISTINCT UserId FROM Posts WHERE RootId=:RootId OR Id=:RootId", map[string]interface{}{"RootId": rootId}); err != nil {
return err
}
// calculate reply count

2
store/storetest/thread_store.go
Просмотреть файл

@@ -221,7 +221,7 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) {
thread1, err := ss.Thread().Get(newPosts1[0].Id)
require.Nil(t, err)
require.EqualValues(t, thread1.ReplyCount, 1)
require.Len(t, thread1.Participants, 1)
require.Len(t, thread1.Participants, 2)
err = ss.Post().PermanentDeleteByUser(rootPost.UserId)
require.Nil(t, err)