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

Migrate to stateless app.App (#17542)

Просмотр исходного кода 
* add request context

* move initialialization to server

* use app interface instead of global app functions

* remove app context from webconn

* cleanup

* remove duplicated services

* move context to separate package

* remove finalize init method and move content to NewServer function

* restart workers and schedulers after adding license for tests

* reflect review comments

Co-authored-by: Mattermod <mattermod@users.noreply.github.com>
Этот коммит содержится в:
Ibrahim Serdar Acikgoz
2021-05-11 13:00:44 +03:00
коммит произвёл GitHub
родитель c09369f14a
Коммит 5ea06e51d0
235 изменённых файлов: 4048 добавлений и 3819 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

9
services/remotecluster/recv.go
Просмотреть файл

@@ -6,6 +6,7 @@ package remotecluster
import (
"fmt"
"github.com/mattermost/mattermost-server/v5/app/request"
"github.com/mattermost/mattermost-server/v5/model"
"github.com/mattermost/mattermost-server/v5/shared/mlog"
)
@@ -13,7 +14,7 @@ import (
// ReceiveIncomingMsg is called by the Rest API layer, or websocket layer (future), when a Remote Cluster
// message is received. Here we route the message to any topic listeners.
// `rc` and `msg` cannot be nil.
func (rcs *Service) ReceiveIncomingMsg(rc *model.RemoteCluster, msg model.RemoteClusterMsg) Response {
func (rcs *Service) ReceiveIncomingMsg(c *request.Context, rc *model.RemoteCluster, msg model.RemoteClusterMsg) Response {
rcs.mux.RLock()
defer rcs.mux.RUnlock()
@@ -31,7 +32,7 @@ func (rcs *Service) ReceiveIncomingMsg(rc *model.RemoteCluster, msg model.Remote
listeners := rcs.getTopicListeners(msg.Topic)
for _, l := range listeners {
if err := callback(l, msg, &rcSanitized, &response); err != nil {
if err := callback(l, c, msg, &rcSanitized, &response); err != nil {
rcs.server.GetLogger().Log(mlog.LvlRemoteClusterServiceError, "Error from remote cluster message listener",
mlog.String("msgId", msg.Id), mlog.String("topic", msg.Topic), mlog.String("remote", rc.DisplayName), mlog.Err(err))
@@ -42,12 +43,12 @@ func (rcs *Service) ReceiveIncomingMsg(rc *model.RemoteCluster, msg model.Remote
return response
}
func callback(listener TopicListener, msg model.RemoteClusterMsg, rc *model.RemoteCluster, resp *Response) (err error) {
func callback(listener TopicListener, c *request.Context, msg model.RemoteClusterMsg, rc *model.RemoteCluster, resp *Response) (err error) {
defer func() {
if r := recover(); r != nil {
err = fmt.Errorf("%v", r)
}
}()
err = listener(msg, rc, resp)
err = listener(c, msg, rc, resp)
return
}

5
services/remotecluster/service.go
Просмотреть файл

@@ -10,6 +10,7 @@ import (
"sync"
"time"
"github.com/mattermost/mattermost-server/v5/app/request"
"github.com/mattermost/mattermost-server/v5/einterfaces"
"github.com/mattermost/mattermost-server/v5/model"
"github.com/mattermost/mattermost-server/v5/shared/mlog"
@@ -65,12 +66,12 @@ type RemoteClusterServiceIFace interface {
SendFile(ctx context.Context, us *model.UploadSession, fi *model.FileInfo, rc *model.RemoteCluster, rp ReaderProvider, f SendFileResultFunc) error
SendProfileImage(ctx context.Context, userID string, rc *model.RemoteCluster, provider ProfileImageProvider, f SendProfileImageResultFunc) error
AcceptInvitation(invite *model.RemoteClusterInvite, name string, displayName string, creatorId string, teamId string, siteURL string) (*model.RemoteCluster, error)
ReceiveIncomingMsg(rc *model.RemoteCluster, msg model.RemoteClusterMsg) Response
ReceiveIncomingMsg(c *request.Context, rc *model.RemoteCluster, msg model.RemoteClusterMsg) Response
}
// TopicListener is a callback signature used to listen for incoming messages for
// a specific topic.
type TopicListener func(msg model.RemoteClusterMsg, rc *model.RemoteCluster, resp *Response) error
type TopicListener func(c *request.Context, msg model.RemoteClusterMsg, rc *model.RemoteCluster, resp *Response) error
// ConnectionStateListener is used to listen to remote cluster connection state changes.
type ConnectionStateListener func(rc *model.RemoteCluster, online bool)

23
services/remotecluster/service_test.go
Просмотреть файл

@@ -10,21 +10,22 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/mattermost/mattermost-server/v5/app/request"
"github.com/mattermost/mattermost-server/v5/model"
)
func TestService_AddTopicListener(t *testing.T) {
var count int32
l1 := func(msg model.RemoteClusterMsg, rc *model.RemoteCluster, resp *Response) error {
l1 := func(_ *request.Context, msg model.RemoteClusterMsg, rc *model.RemoteCluster, resp *Response) error {
atomic.AddInt32(&count, 1)
return nil
}
l2 := func(msg model.RemoteClusterMsg, rc *model.RemoteCluster, resp *Response) error {
l2 := func(_ *request.Context, msg model.RemoteClusterMsg, rc *model.RemoteCluster, resp *Response) error {
atomic.AddInt32(&count, 1)
return nil
}
l3 := func(msg model.RemoteClusterMsg, rc *model.RemoteCluster, resp *Response) error {
l3 := func(_ *request.Context, msg model.RemoteClusterMsg, rc *model.RemoteCluster, resp *Response) error {
atomic.AddInt32(&count, 1)
return nil
}
@@ -46,26 +47,28 @@ func TestService_AddTopicListener(t *testing.T) {
msg1 := model.RemoteClusterMsg{Topic: "test"}
msg2 := model.RemoteClusterMsg{Topic: "different"}
service.ReceiveIncomingMsg(rc, msg1)
c := request.EmptyContext()
service.ReceiveIncomingMsg(c, rc, msg1)
assert.Equal(t, int32(2), atomic.LoadInt32(&count))
service.ReceiveIncomingMsg(rc, msg2)
service.ReceiveIncomingMsg(c, rc, msg2)
assert.Equal(t, int32(3), atomic.LoadInt32(&count))
service.RemoveTopicListener(l1id)
service.ReceiveIncomingMsg(rc, msg1)
service.ReceiveIncomingMsg(c, rc, msg1)
assert.Equal(t, int32(4), atomic.LoadInt32(&count))
service.RemoveTopicListener(l2id)
service.ReceiveIncomingMsg(rc, msg1)
service.ReceiveIncomingMsg(c, rc, msg1)
assert.Equal(t, int32(4), atomic.LoadInt32(&count))
service.ReceiveIncomingMsg(rc, msg2)
service.ReceiveIncomingMsg(c, rc, msg2)
assert.Equal(t, int32(5), atomic.LoadInt32(&count))
service.RemoveTopicListener(l3id)
service.ReceiveIncomingMsg(rc, msg1)
service.ReceiveIncomingMsg(rc, msg2)
service.ReceiveIncomingMsg(c, rc, msg1)
service.ReceiveIncomingMsg(c, rc, msg2)
assert.Equal(t, int32(5), atomic.LoadInt32(&count))
listeners = service.getTopicListeners("test")

3
services/sharedchannel/attachment.go
Просмотреть файл

@@ -10,6 +10,7 @@ import (
"fmt"
"sync"
"github.com/mattermost/mattermost-server/v5/app/request"
"github.com/mattermost/mattermost-server/v5/model"
"github.com/mattermost/mattermost-server/v5/services/remotecluster"
"github.com/mattermost/mattermost-server/v5/shared/mlog"
@@ -158,7 +159,7 @@ func (scs *Service) sendAttachmentForRemote(fi *model.FileInfo, post *model.Post
// onReceiveUploadCreate is called when a message requesting to create an upload session is received. An upload session is
// created and the id returned in the response.
func (scs *Service) onReceiveUploadCreate(msg model.RemoteClusterMsg, rc *model.RemoteCluster, response *remotecluster.Response) error {
func (scs *Service) onReceiveUploadCreate(_ *request.Context, msg model.RemoteClusterMsg, rc *model.RemoteCluster, response *remotecluster.Response) error {
var us model.UploadSession
if err := json.Unmarshal(msg.Payload, &us); err != nil {

15
services/sharedchannel/channelinvite.go
Просмотреть файл

@@ -9,6 +9,7 @@ import (
"fmt"
"strings"
"github.com/mattermost/mattermost-server/v5/app/request"
"github.com/mattermost/mattermost-server/v5/model"
"github.com/mattermost/mattermost-server/v5/services/remotecluster"
"github.com/mattermost/mattermost-server/v5/shared/mlog"
@@ -110,7 +111,7 @@ func combineErrors(err error, serror string) string {
return sb.String()
}
func (scs *Service) onReceiveChannelInvite(msg model.RemoteClusterMsg, rc *model.RemoteCluster, _ *remotecluster.Response) error {
func (scs *Service) onReceiveChannelInvite(c *request.Context, msg model.RemoteClusterMsg, rc *model.RemoteCluster, _ *remotecluster.Response) error {
if len(msg.Payload) == 0 {
return nil
}
@@ -131,7 +132,7 @@ func (scs *Service) onReceiveChannelInvite(msg model.RemoteClusterMsg, rc *model
// create channel if it doesn't exist; the channel may already exist, such as if it was shared then unshared at some point.
channel, err := scs.server.GetStore().Channel().Get(invite.ChannelId, true)
if err != nil {
if channel, err = scs.handleChannelCreation(invite, rc); err != nil {
if channel, err = scs.handleChannelCreation(c, invite, rc); err != nil {
return err
}
}
@@ -178,9 +179,9 @@ func (scs *Service) onReceiveChannelInvite(msg model.RemoteClusterMsg, rc *model
return nil
}
func (scs *Service) handleChannelCreation(invite channelInviteMsg, rc *model.RemoteCluster) (*model.Channel, error) {
func (scs *Service) handleChannelCreation(c *request.Context, invite channelInviteMsg, rc *model.RemoteCluster) (*model.Channel, error) {
if invite.Type == model.CHANNEL_DIRECT {
return scs.createDirectChannel(invite)
return scs.createDirectChannel(c, invite)
}
channelNew := &model.Channel{
@@ -196,7 +197,7 @@ func (scs *Service) handleChannelCreation(invite channelInviteMsg, rc *model.Rem
}
// check user perms?
channel, appErr := scs.app.CreateChannelWithUser(channelNew, rc.CreatorId)
channel, appErr := scs.app.CreateChannelWithUser(c, channelNew, rc.CreatorId)
if appErr != nil {
return nil, fmt.Errorf("cannot create channel `%s`: %w", invite.ChannelId, appErr)
}
@@ -204,12 +205,12 @@ func (scs *Service) handleChannelCreation(invite channelInviteMsg, rc *model.Rem
return channel, nil
}
func (scs *Service) createDirectChannel(invite channelInviteMsg) (*model.Channel, error) {
func (scs *Service) createDirectChannel(c *request.Context, invite channelInviteMsg) (*model.Channel, error) {
if len(invite.DirectParticipantIDs) != 2 {
return nil, fmt.Errorf("cannot create direct channel `%s` insufficient participant count `%d`", invite.ChannelId, len(invite.DirectParticipantIDs))
}
channel, err := scs.app.GetOrCreateDirectChannel(invite.DirectParticipantIDs[0], invite.DirectParticipantIDs[1], model.WithID(invite.ChannelId))
channel, err := scs.app.GetOrCreateDirectChannel(c, invite.DirectParticipantIDs[0], invite.DirectParticipantIDs[1], model.WithID(invite.ChannelId))
if err != nil {
return nil, fmt.Errorf("cannot create direct channel `%s`: %w", invite.ChannelId, err)
}

13
services/sharedchannel/channelinvite_test.go
Просмотреть файл

@@ -13,6 +13,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/mattermost/mattermost-server/v5/app/request"
"github.com/mattermost/mattermost-server/v5/model"
"github.com/mattermost/mattermost-server/v5/plugin/plugintest/mock"
"github.com/mattermost/mattermost-server/v5/shared/mlog"
@@ -26,6 +27,8 @@ type mockLogger struct {
func (ml *mockLogger) Log(level mlog.LogLevel, s string, flds ...mlog.Field) {}
func TestOnReceiveChannelInvite(t *testing.T) {
c := request.EmptyContext()
t.Run("when msg payload is empty, it does nothing", func(t *testing.T) {
mockServer := &MockServerIface{}
mockLogger := &mockLogger{}
@@ -43,7 +46,7 @@ func TestOnReceiveChannelInvite(t *testing.T) {
remoteCluster := &model.RemoteCluster{}
msg := model.RemoteClusterMsg{}
err := scs.onReceiveChannelInvite(msg, remoteCluster, nil)
err := scs.onReceiveChannelInvite(c, msg, remoteCluster, nil)
require.NoError(t, err)
mockStore.AssertNotCalled(t, "Channel")
})
@@ -104,7 +107,7 @@ func TestOnReceiveChannelInvite(t *testing.T) {
mockApp.On("PatchChannelModerationsForChannel", channel, readonlyChannelModerations).Return(nil, nil)
defer mockApp.AssertExpectations(t)
err = scs.onReceiveChannelInvite(msg, remoteCluster, nil)
err = scs.onReceiveChannelInvite(c, msg, remoteCluster, nil)
require.NoError(t, err)
})
@@ -145,7 +148,7 @@ func TestOnReceiveChannelInvite(t *testing.T) {
mockApp.On("PatchChannelModerationsForChannel", channel, mock.Anything).Return(nil, appErr)
defer mockApp.AssertExpectations(t)
err = scs.onReceiveChannelInvite(msg, remoteCluster, nil)
err = scs.onReceiveChannelInvite(c, msg, remoteCluster, nil)
require.Error(t, err)
assert.Equal(t, fmt.Sprintf("cannot make channel readonly `%s`: foo: bar, boom", invitation.ChannelId), err.Error())
})
@@ -188,10 +191,10 @@ func TestOnReceiveChannelInvite(t *testing.T) {
mockServer = scs.server.(*MockServerIface)
mockServer.On("GetStore").Return(mockStore)
mockApp.On("GetOrCreateDirectChannel", invitation.DirectParticipantIDs[0], invitation.DirectParticipantIDs[1], mock.AnythingOfType("model.ChannelOption")).Return(channel, nil)
mockApp.On("GetOrCreateDirectChannel", mock.AnythingOfType("*request.Context"), invitation.DirectParticipantIDs[0], invitation.DirectParticipantIDs[1], mock.AnythingOfType("model.ChannelOption")).Return(channel, nil)
defer mockApp.AssertExpectations(t)
err = scs.onReceiveChannelInvite(msg, remoteCluster, nil)
err = scs.onReceiveChannelInvite(c, msg, remoteCluster, nil)
require.NoError(t, err)
})
}

92
services/sharedchannel/mock_AppIface_test.go
Просмотреть файл

@@ -9,6 +9,8 @@ import (
mock "github.com/stretchr/testify/mock"
model "github.com/mattermost/mattermost-server/v5/model"
request "github.com/mattermost/mattermost-server/v5/app/request"
)
// MockAppIface is an autogenerated mock type for the AppIface type
@@ -41,13 +43,13 @@ func (_m *MockAppIface) AddUserToChannel(user *model.User, channel *model.Channe
return r0, r1
}
// AddUserToTeamByTeamId provides a mock function with given fields: teamId, user
func (_m *MockAppIface) AddUserToTeamByTeamId(teamId string, user *model.User) *model.AppError {
ret := _m.Called(teamId, user)
// AddUserToTeamByTeamId provides a mock function with given fields: c, teamId, user
func (_m *MockAppIface) AddUserToTeamByTeamId(c *request.Context, teamId string, user *model.User) *model.AppError {
ret := _m.Called(c, teamId, user)
var r0 *model.AppError
if rf, ok := ret.Get(0).(func(string, *model.User) *model.AppError); ok {
r0 = rf(teamId, user)
if rf, ok := ret.Get(0).(func(*request.Context, string, *model.User) *model.AppError); ok {
r0 = rf(c, teamId, user)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*model.AppError)
@@ -57,13 +59,13 @@ func (_m *MockAppIface) AddUserToTeamByTeamId(teamId string, user *model.User) *
return r0
}
// CreateChannelWithUser provides a mock function with given fields: channel, userId
func (_m *MockAppIface) CreateChannelWithUser(channel *model.Channel, userId string) (*model.Channel, *model.AppError) {
ret := _m.Called(channel, userId)
// CreateChannelWithUser provides a mock function with given fields: c, channel, userId
func (_m *MockAppIface) CreateChannelWithUser(c *request.Context, channel *model.Channel, userId string) (*model.Channel, *model.AppError) {
ret := _m.Called(c, channel, userId)
var r0 *model.Channel
if rf, ok := ret.Get(0).(func(*model.Channel, string) *model.Channel); ok {
r0 = rf(channel, userId)
if rf, ok := ret.Get(0).(func(*request.Context, *model.Channel, string) *model.Channel); ok {
r0 = rf(c, channel, userId)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*model.Channel)
@@ -71,8 +73,8 @@ func (_m *MockAppIface) CreateChannelWithUser(channel *model.Channel, userId str
}
var r1 *model.AppError
if rf, ok := ret.Get(1).(func(*model.Channel, string) *model.AppError); ok {
r1 = rf(channel, userId)
if rf, ok := ret.Get(1).(func(*request.Context, *model.Channel, string) *model.AppError); ok {
r1 = rf(c, channel, userId)
} else {
if ret.Get(1) != nil {
r1 = ret.Get(1).(*model.AppError)
@@ -82,13 +84,13 @@ func (_m *MockAppIface) CreateChannelWithUser(channel *model.Channel, userId str
return r0, r1
}
// CreatePost provides a mock function with given fields: post, channel, triggerWebhooks, setOnline
func (_m *MockAppIface) CreatePost(post *model.Post, channel *model.Channel, triggerWebhooks bool, setOnline bool) (*model.Post, *model.AppError) {
ret := _m.Called(post, channel, triggerWebhooks, setOnline)
// CreatePost provides a mock function with given fields: c, post, channel, triggerWebhooks, setOnline
func (_m *MockAppIface) CreatePost(c *request.Context, post *model.Post, channel *model.Channel, triggerWebhooks bool, setOnline bool) (*model.Post, *model.AppError) {
ret := _m.Called(c, post, channel, triggerWebhooks, setOnline)
var r0 *model.Post
if rf, ok := ret.Get(0).(func(*model.Post, *model.Channel, bool, bool) *model.Post); ok {
r0 = rf(post, channel, triggerWebhooks, setOnline)
if rf, ok := ret.Get(0).(func(*request.Context, *model.Post, *model.Channel, bool, bool) *model.Post); ok {
r0 = rf(c, post, channel, triggerWebhooks, setOnline)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*model.Post)
@@ -96,8 +98,8 @@ func (_m *MockAppIface) CreatePost(post *model.Post, channel *model.Channel, tri
}
var r1 *model.AppError
if rf, ok := ret.Get(1).(func(*model.Post, *model.Channel, bool, bool) *model.AppError); ok {
r1 = rf(post, channel, triggerWebhooks, setOnline)
if rf, ok := ret.Get(1).(func(*request.Context, *model.Post, *model.Channel, bool, bool) *model.AppError); ok {
r1 = rf(c, post, channel, triggerWebhooks, setOnline)
} else {
if ret.Get(1) != nil {
r1 = ret.Get(1).(*model.AppError)
@@ -157,13 +159,13 @@ func (_m *MockAppIface) DeletePost(postID string, deleteByID string) (*model.Pos
return r0, r1
}
// DeleteReactionForPost provides a mock function with given fields: reaction
func (_m *MockAppIface) DeleteReactionForPost(reaction *model.Reaction) *model.AppError {
ret := _m.Called(reaction)
// DeleteReactionForPost provides a mock function with given fields: c, reaction
func (_m *MockAppIface) DeleteReactionForPost(c *request.Context, reaction *model.Reaction) *model.AppError {
ret := _m.Called(c, reaction)
var r0 *model.AppError
if rf, ok := ret.Get(0).(func(*model.Reaction) *model.AppError); ok {
r0 = rf(reaction)
if rf, ok := ret.Get(0).(func(*request.Context, *model.Reaction) *model.AppError); ok {
r0 = rf(c, reaction)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*model.AppError)
@@ -198,20 +200,20 @@ func (_m *MockAppIface) FileReader(path string) (filestore.ReadCloseSeeker, *mod
return r0, r1
}
// GetOrCreateDirectChannel provides a mock function with given fields: userId, otherUserId, channelOptions
func (_m *MockAppIface) GetOrCreateDirectChannel(userId string, otherUserId string, channelOptions ...model.ChannelOption) (*model.Channel, *model.AppError) {
// GetOrCreateDirectChannel provides a mock function with given fields: c, userId, otherUserId, channelOptions
func (_m *MockAppIface) GetOrCreateDirectChannel(c *request.Context, userId string, otherUserId string, channelOptions ...model.ChannelOption) (*model.Channel, *model.AppError) {
_va := make([]interface{}, len(channelOptions))
for _i := range channelOptions {
_va[_i] = channelOptions[_i]
}
var _ca []interface{}
_ca = append(_ca, userId, otherUserId)
_ca = append(_ca, c, userId, otherUserId)
_ca = append(_ca, _va...)
ret := _m.Called(_ca...)
var r0 *model.Channel
if rf, ok := ret.Get(0).(func(string, string, ...model.ChannelOption) *model.Channel); ok {
r0 = rf(userId, otherUserId, channelOptions...)
if rf, ok := ret.Get(0).(func(*request.Context, string, string, ...model.ChannelOption) *model.Channel); ok {
r0 = rf(c, userId, otherUserId, channelOptions...)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*model.Channel)
@@ -219,8 +221,8 @@ func (_m *MockAppIface) GetOrCreateDirectChannel(userId string, otherUserId stri
}
var r1 *model.AppError
if rf, ok := ret.Get(1).(func(string, string, ...model.ChannelOption) *model.AppError); ok {
r1 = rf(userId, otherUserId, channelOptions...)
if rf, ok := ret.Get(1).(func(*request.Context, string, string, ...model.ChannelOption) *model.AppError); ok {
r1 = rf(c, userId, otherUserId, channelOptions...)
} else {
if ret.Get(1) != nil {
r1 = ret.Get(1).(*model.AppError)
@@ -329,13 +331,13 @@ func (_m *MockAppIface) PermanentDeleteChannel(channel *model.Channel) *model.Ap
return r0
}
// SaveReactionForPost provides a mock function with given fields: reaction
func (_m *MockAppIface) SaveReactionForPost(reaction *model.Reaction) (*model.Reaction, *model.AppError) {
ret := _m.Called(reaction)
// SaveReactionForPost provides a mock function with given fields: c, reaction
func (_m *MockAppIface) SaveReactionForPost(c *request.Context, reaction *model.Reaction) (*model.Reaction, *model.AppError) {
ret := _m.Called(c, reaction)
var r0 *model.Reaction
if rf, ok := ret.Get(0).(func(*model.Reaction) *model.Reaction); ok {
r0 = rf(reaction)
if rf, ok := ret.Get(0).(func(*request.Context, *model.Reaction) *model.Reaction); ok {
r0 = rf(c, reaction)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*model.Reaction)
@@ -343,8 +345,8 @@ func (_m *MockAppIface) SaveReactionForPost(reaction *model.Reaction) (*model.Re
}
var r1 *model.AppError
if rf, ok := ret.Get(1).(func(*model.Reaction) *model.AppError); ok {
r1 = rf(reaction)
if rf, ok := ret.Get(1).(func(*request.Context, *model.Reaction) *model.AppError); ok {
r1 = rf(c, reaction)
} else {
if ret.Get(1) != nil {
r1 = ret.Get(1).(*model.AppError)
@@ -370,13 +372,13 @@ func (_m *MockAppIface) SendEphemeralPost(userId string, post *model.Post) *mode
return r0
}
// UpdatePost provides a mock function with given fields: post, safeUpdate
func (_m *MockAppIface) UpdatePost(post *model.Post, safeUpdate bool) (*model.Post, *model.AppError) {
ret := _m.Called(post, safeUpdate)
// UpdatePost provides a mock function with given fields: c, post, safeUpdate
func (_m *MockAppIface) UpdatePost(c *request.Context, post *model.Post, safeUpdate bool) (*model.Post, *model.AppError) {
ret := _m.Called(c, post, safeUpdate)
var r0 *model.Post
if rf, ok := ret.Get(0).(func(*model.Post, bool) *model.Post); ok {
r0 = rf(post, safeUpdate)
if rf, ok := ret.Get(0).(func(*request.Context, *model.Post, bool) *model.Post); ok {
r0 = rf(c, post, safeUpdate)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*model.Post)
@@ -384,8 +386,8 @@ func (_m *MockAppIface) UpdatePost(post *model.Post, safeUpdate bool) (*model.Po
}
var r1 *model.AppError
if rf, ok := ret.Get(1).(func(*model.Post, bool) *model.AppError); ok {
r1 = rf(post, safeUpdate)
if rf, ok := ret.Get(1).(func(*request.Context, *model.Post, bool) *model.AppError); ok {
r1 = rf(c, post, safeUpdate)
} else {
if ret.Get(1) != nil {
r1 = ret.Get(1).(*model.AppError)

15
services/sharedchannel/service.go
Просмотреть файл

@@ -10,6 +10,7 @@ import (
"sync"
"time"
"github.com/mattermost/mattermost-server/v5/app/request"
"github.com/mattermost/mattermost-server/v5/model"
"github.com/mattermost/mattermost-server/v5/services/remotecluster"
"github.com/mattermost/mattermost-server/v5/shared/filestore"
@@ -43,16 +44,16 @@ type ServerIface interface {
type AppIface interface {
SendEphemeralPost(userId string, post *model.Post) *model.Post
CreateChannelWithUser(channel *model.Channel, userId string) (*model.Channel, *model.AppError)
GetOrCreateDirectChannel(userId, otherUserId string, channelOptions ...model.ChannelOption) (*model.Channel, *model.AppError)
CreateChannelWithUser(c *request.Context, channel *model.Channel, userId string) (*model.Channel, *model.AppError)
GetOrCreateDirectChannel(c *request.Context, userId, otherUserId string, channelOptions ...model.ChannelOption) (*model.Channel, *model.AppError)
AddUserToChannel(user *model.User, channel *model.Channel, skipTeamMemberIntegrityCheck bool) (*model.ChannelMember, *model.AppError)
AddUserToTeamByTeamId(teamId string, user *model.User) *model.AppError
AddUserToTeamByTeamId(c *request.Context, teamId string, user *model.User) *model.AppError
PermanentDeleteChannel(channel *model.Channel) *model.AppError
CreatePost(post *model.Post, channel *model.Channel, triggerWebhooks bool, setOnline bool) (savedPost *model.Post, err *model.AppError)
UpdatePost(post *model.Post, safeUpdate bool) (*model.Post, *model.AppError)
CreatePost(c *request.Context, post *model.Post, channel *model.Channel, triggerWebhooks bool, setOnline bool) (savedPost *model.Post, err *model.AppError)
UpdatePost(c *request.Context, post *model.Post, safeUpdate bool) (*model.Post, *model.AppError)
DeletePost(postID, deleteByID string) (*model.Post, *model.AppError)
SaveReactionForPost(reaction *model.Reaction) (*model.Reaction, *model.AppError)
DeleteReactionForPost(reaction *model.Reaction) *model.AppError
SaveReactionForPost(c *request.Context, reaction *model.Reaction) (*model.Reaction, *model.AppError)
DeleteReactionForPost(c *request.Context, reaction *model.Reaction) *model.AppError
PatchChannelModerationsForChannel(channel *model.Channel, channelModerationsPatch []*model.ChannelModerationPatch) ([]*model.ChannelModeration, *model.AppError)
CreateUploadSession(us *model.UploadSession) (*model.UploadSession, *model.AppError)
FileReader(path string) (filestore.ReadCloseSeeker, *model.AppError)

29
services/sharedchannel/sync_recv.go
Просмотреть файл

@@ -11,12 +11,13 @@ import (
"strconv"
"strings"
"github.com/mattermost/mattermost-server/v5/app/request"
"github.com/mattermost/mattermost-server/v5/model"
"github.com/mattermost/mattermost-server/v5/services/remotecluster"
"github.com/mattermost/mattermost-server/v5/shared/mlog"
)
func (scs *Service) onReceiveSyncMessage(msg model.RemoteClusterMsg, rc *model.RemoteCluster, response *remotecluster.Response) error {
func (scs *Service) onReceiveSyncMessage(c *request.Context, msg model.RemoteClusterMsg, rc *model.RemoteCluster, response *remotecluster.Response) error {
if msg.Topic != TopicSync {
return fmt.Errorf("wrong topic, expected `%s`, got `%s`", TopicSync, msg.Topic)
}
@@ -43,10 +44,10 @@ func (scs *Service) onReceiveSyncMessage(msg model.RemoteClusterMsg, rc *model.R
mlog.Int("sync_msg_count", len(syncMessages)),
)
return scs.processSyncMessages(syncMessages, rc, response)
return scs.processSyncMessages(c, syncMessages, rc, response)
}
func (scs *Service) processSyncMessages(syncMessages []syncMsg, rc *model.RemoteCluster, response *remotecluster.Response) error {
func (scs *Service) processSyncMessages(c *request.Context, syncMessages []syncMsg, rc *model.RemoteCluster, response *remotecluster.Response) error {
var channel *model.Channel
var team *model.Team
@@ -73,7 +74,7 @@ func (scs *Service) processSyncMessages(syncMessages []syncMsg, rc *model.Remote
// add/update users before posts
for _, user := range sm.Users {
if userSaved, err := scs.upsertSyncUser(user, channel, rc); err != nil {
if userSaved, err := scs.upsertSyncUser(c, user, channel, rc); err != nil {
scs.server.GetLogger().Log(mlog.LvlSharedChannelServiceError, "Error upserting sync user",
mlog.String("post_id", sm.PostId),
mlog.String("channel_id", sm.ChannelId),
@@ -120,7 +121,7 @@ func (scs *Service) processSyncMessages(syncMessages []syncMsg, rc *model.Remote
}
// add/update post
rpost, err := scs.upsertSyncPost(sm.Post, channel, rc)
rpost, err := scs.upsertSyncPost(c, sm.Post, channel, rc)
if err != nil {
postErrors = append(postErrors, sm.Post.Id)
scs.server.GetLogger().Log(mlog.LvlSharedChannelServiceError, "Error upserting sync post",
@@ -135,7 +136,7 @@ func (scs *Service) processSyncMessages(syncMessages []syncMsg, rc *model.Remote
// add/remove reactions
for _, reaction := range sm.Reactions {
if _, err := scs.upsertSyncReaction(reaction, rc); err != nil {
if _, err := scs.upsertSyncReaction(c, reaction, rc); err != nil {
scs.server.GetLogger().Log(mlog.LvlSharedChannelServiceError, "Error upserting sync reaction",
mlog.String("user_id", reaction.UserId),
mlog.String("post_id", reaction.PostId),
@@ -169,7 +170,7 @@ func (scs *Service) processSyncMessages(syncMessages []syncMsg, rc *model.Remote
return nil
}
func (scs *Service) upsertSyncUser(user *model.User, channel *model.Channel, rc *model.RemoteCluster) (*model.User, error) {
func (scs *Service) upsertSyncUser(c *request.Context, user *model.User, channel *model.Channel, rc *model.RemoteCluster) (*model.User, error) {
var err error
if user.RemoteId == nil || *user.RemoteId == "" {
user.RemoteId = model.NewString(rc.RemoteId)
@@ -212,7 +213,7 @@ func (scs *Service) upsertSyncUser(user *model.User, channel *model.Channel, rc
// Instead of undoing what succeeded on any failure we simply do all steps each
// time. AddUserToChannel & AddUserToTeamByTeamId do not error if user was already
// added and exit quickly.
if err := scs.app.AddUserToTeamByTeamId(channel.TeamId, userSaved); err != nil {
if err := scs.app.AddUserToTeamByTeamId(c, channel.TeamId, userSaved); err != nil {
return nil, fmt.Errorf("error adding sync user to Team: %w", err)
}
@@ -329,7 +330,7 @@ func (scs *Service) updateSyncUser(patch *model.UserPatch, user *model.User, cha
return nil, fmt.Errorf("error updating sync user %s: %w", user.Id, err)
}
func (scs *Service) upsertSyncPost(post *model.Post, channel *model.Channel, rc *model.RemoteCluster) (*model.Post, error) {
func (scs *Service) upsertSyncPost(c *request.Context, post *model.Post, channel *model.Channel, rc *model.RemoteCluster) (*model.Post, error) {
var appErr *model.AppError
post.RemoteId = model.NewString(rc.RemoteId)
@@ -343,7 +344,7 @@ func (scs *Service) upsertSyncPost(post *model.Post, channel *model.Channel, rc
if rpost == nil {
// post doesn't exist; create new one
rpost, appErr = scs.app.CreatePost(post, channel, true, true)
rpost, appErr = scs.app.CreatePost(c, post, channel, true, true)
scs.server.GetLogger().Log(mlog.LvlSharedChannelServiceDebug, "Created sync post",
mlog.String("post_id", post.Id),
mlog.String("channel_id", post.ChannelId),
@@ -357,7 +358,7 @@ func (scs *Service) upsertSyncPost(post *model.Post, channel *model.Channel, rc
)
} else if post.EditAt > rpost.EditAt || post.Message != rpost.Message {
// update post
rpost, appErr = scs.app.UpdatePost(post, false)
rpost, appErr = scs.app.UpdatePost(c, post, false)
scs.server.GetLogger().Log(mlog.LvlSharedChannelServiceDebug, "Updated sync post",
mlog.String("post_id", post.Id),
mlog.String("channel_id", post.ChannelId),
@@ -377,16 +378,16 @@ func (scs *Service) upsertSyncPost(post *model.Post, channel *model.Channel, rc
return rpost, rerr
}
func (scs *Service) upsertSyncReaction(reaction *model.Reaction, rc *model.RemoteCluster) (*model.Reaction, error) {
func (scs *Service) upsertSyncReaction(c *request.Context, reaction *model.Reaction, rc *model.RemoteCluster) (*model.Reaction, error) {
savedReaction := reaction
var appErr *model.AppError
reaction.RemoteId = model.NewString(rc.RemoteId)
if reaction.DeleteAt == 0 {
savedReaction, appErr = scs.app.SaveReactionForPost(reaction)
savedReaction, appErr = scs.app.SaveReactionForPost(c, reaction)
} else {
appErr = scs.app.DeleteReactionForPost(reaction)
appErr = scs.app.DeleteReactionForPost(c, reaction)
}
var err error

6
services/sharedchannel/sync_send.go
Просмотреть файл

@@ -10,6 +10,7 @@ import (
"sync"
"time"
"github.com/mattermost/mattermost-server/v5/app/request"
"github.com/mattermost/mattermost-server/v5/model"
"github.com/mattermost/mattermost-server/v5/services/remotecluster"
"github.com/mattermost/mattermost-server/v5/shared/i18n"
@@ -140,6 +141,7 @@ func stopTimer(timer *time.Timer) {
// doSync checks the task queue for any tasks to be processed and processes all that are ready.
// If any delayed tasks remain in queue then the duration until the next scheduled task is returned.
func (scs *Service) doSync() time.Duration {
c := request.EmptyContext() // TODO: check this
var task syncTask
var ok bool
var shortestWait time.Duration
@@ -149,7 +151,7 @@ func (scs *Service) doSync() time.Duration {
if !ok {
break
}
if err := scs.processTask(task); err != nil {
if err := scs.processTask(c, task); err != nil {
// put task back into map so it will update again
if task.incRetry() {
scs.addTask(task)
@@ -200,7 +202,7 @@ func (scs *Service) removeOldestTask() (syncTask, bool, time.Duration) {
}
// processTask updates one or more remote clusters with any new channel content.
func (scs *Service) processTask(task syncTask) error {
func (scs *Service) processTask(c *request.Context, task syncTask) error {
var err error
var remotes []*model.RemoteCluster