MM-27493 Shared channels (MVP) (#17301)

Remote Cluster Service
- provides ability for multiple Mattermost cluster instances to create a trusted connection with each other and exchange messages
- trusted connections are managed via slash commands (for now)
- facilitates features requiring inter-cluster communication, such as Shared Channels
Shared Channels Service
- provides ability to shared channels between one or more Mattermost cluster instances (using trusted connection)
- sharing/unsharing of channels is managed via slash commands (for now)

services/remotecluster/error.go

@@ -0,0 +1,24 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import "fmt"
+
+type BufferFullError struct {
+	capacity int
+}
+
+func NewBufferFullError(capacity int) BufferFullError {
+	return BufferFullError{
+		capacity: capacity,
+	}
+}
+
+func (e BufferFullError) Capacity() int {
+	return e.capacity
+}
+
+func (e BufferFullError) Error() string {
+	return fmt.Sprintf("buffer capacity (%d) exceeded", e.capacity)
+}

services/remotecluster/invitation.go

@@ -0,0 +1,82 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/mattermost/mattermost-server/v5/model"
+)
+
+// AcceptInvitation is called when accepting an invitation to connect with a remote cluster.
+func (rcs *Service) AcceptInvitation(invite *model.RemoteClusterInvite, name string, creatorId string, teamId string, siteURL string) (*model.RemoteCluster, error) {
+	rc := &model.RemoteCluster{
+		RemoteId:     invite.RemoteId,
+		RemoteTeamId: invite.RemoteTeamId,
+		DisplayName:  name,
+		Token:        model.NewId(),
+		RemoteToken:  invite.Token,
+		SiteURL:      invite.SiteURL,
+		CreatorId:    creatorId,
+	}
+
+	rcSaved, err := rcs.server.GetStore().RemoteCluster().Save(rc)
+	if err != nil {
+		return nil, err
+	}
+
+	// confirm the invitation with the originating site
+	frame, err := makeConfirmFrame(rcSaved, teamId, siteURL)
+	if err != nil {
+		return nil, err
+	}
+
+	url := fmt.Sprintf("%s/%s", rcSaved.SiteURL, ConfirmInviteURL)
+
+	resp, err := rcs.sendFrameToRemote(PingTimeout, rc, frame, url)
+	if err != nil {
+		rcs.server.GetStore().RemoteCluster().Delete(rcSaved.RemoteId)
+		return nil, err
+	}
+
+	var response Response
+	err = json.Unmarshal(resp, &response)
+	if err != nil {
+		rcs.server.GetStore().RemoteCluster().Delete(rcSaved.RemoteId)
+		return nil, fmt.Errorf("invalid response from remote server: %w", err)
+	}
+
+	if !response.IsSuccess() {
+		rcs.server.GetStore().RemoteCluster().Delete(rcSaved.RemoteId)
+		return nil, errors.New(response.Err)
+	}
+
+	// issue the first ping right away. The goroutine will exit when ping completes or PingTimeout exceeded.
+	go rcs.pingRemote(rcSaved)
+
+	return rcSaved, nil
+}
+
+func makeConfirmFrame(rc *model.RemoteCluster, teamId string, siteURL string) (*model.RemoteClusterFrame, error) {
+	confirm := model.RemoteClusterInvite{
+		RemoteId:     rc.RemoteId,
+		RemoteTeamId: teamId,
+		SiteURL:      siteURL,
+		Token:        rc.Token,
+	}
+	confirmRaw, err := json.Marshal(confirm)
+	if err != nil {
+		return nil, err
+	}
+
+	msg := model.NewRemoteClusterMsg(InvitationTopic, confirmRaw)
+
+	frame := &model.RemoteClusterFrame{
+		RemoteId: rc.RemoteId,
+		Msg:      msg,
+	}
+	return frame, nil
+}

services/remotecluster/mocks_test.go

@@ -0,0 +1,104 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+
+	"go.uber.org/zap/zapcore"
+
+	"github.com/mattermost/mattermost-server/v5/einterfaces"
+	"github.com/mattermost/mattermost-server/v5/model"
+	"github.com/mattermost/mattermost-server/v5/plugin/plugintest/mock"
+	"github.com/mattermost/mattermost-server/v5/shared/mlog"
+	"github.com/mattermost/mattermost-server/v5/store"
+	"github.com/mattermost/mattermost-server/v5/store/storetest/mocks"
+)
+
+type mockServer struct {
+	remotes []*model.RemoteCluster
+	logger  *mockLogger
+}
+
+func newMockServer(t *testing.T, remotes []*model.RemoteCluster) *mockServer {
+	return &mockServer{
+		remotes: remotes,
+		logger:  &mockLogger{t: t},
+	}
+}
+
+func (ms *mockServer) Config() *model.Config                                  { return nil }
+func (ms *mockServer) GetMetrics() einterfaces.MetricsInterface               { return nil }
+func (ms *mockServer) IsLeader() bool                                         { return true }
+func (ms *mockServer) AddClusterLeaderChangedListener(listener func()) string { return model.NewId() }
+func (ms *mockServer) RemoveClusterLeaderChangedListener(id string)           {}
+func (ms *mockServer) GetLogger() mlog.LoggerIFace {
+	return ms.logger
+}
+func (ms *mockServer) GetStore() store.Store {
+	anyFilter := mock.MatchedBy(func(filter model.RemoteClusterQueryFilter) bool {
+		return true
+	})
+
+	remoteClusterStoreMock := &mocks.RemoteClusterStore{}
+	remoteClusterStoreMock.On("GetByTopic", "share").Return(ms.remotes, nil)
+	remoteClusterStoreMock.On("GetAll", anyFilter).Return(ms.remotes, nil)
+
+	storeMock := &mocks.Store{}
+	storeMock.On("RemoteCluster").Return(remoteClusterStoreMock)
+	return storeMock
+}
+
+type mockLogger struct {
+	t *testing.T
+}
+
+func (ml *mockLogger) IsLevelEnabled(level mlog.LogLevel) bool {
+	return true
+}
+func (ml *mockLogger) Debug(s string, flds ...mlog.Field) {
+	ml.t.Log("debug", s, fieldsToStrings(flds))
+}
+func (ml *mockLogger) Info(s string, flds ...mlog.Field) {
+	ml.t.Log("info", s, fieldsToStrings(flds))
+}
+func (ml *mockLogger) Warn(s string, flds ...mlog.Field) {
+	ml.t.Log("warn", s, fieldsToStrings(flds))
+}
+func (ml *mockLogger) Error(s string, flds ...mlog.Field) {
+	ml.t.Log("error", s, fieldsToStrings(flds))
+}
+func (ml *mockLogger) Critical(s string, flds ...mlog.Field) {
+	ml.t.Log("crit", s, fieldsToStrings(flds))
+}
+func (ml *mockLogger) Log(level mlog.LogLevel, s string, flds ...mlog.Field) {
+	ml.t.Log(level.Name, s, fieldsToStrings(flds))
+}
+func (ml *mockLogger) LogM(levels []mlog.LogLevel, s string, flds ...mlog.Field) {
+	ml.t.Log(levelsToString(levels), s, fieldsToStrings(flds))
+}
+
+func levelsToString(levels []mlog.LogLevel) string {
+	sb := strings.Builder{}
+	for _, l := range levels {
+		sb.WriteString(l.Name)
+		sb.WriteString(",")
+	}
+	return sb.String()
+}
+
+func fieldsToStrings(fields []mlog.Field) []string {
+	encoder := zapcore.NewMapObjectEncoder()
+	for _, zapField := range fields {
+		zapField.AddTo(encoder)
+	}
+
+	var result []string
+	for k, v := range encoder.Fields {
+		result = append(result, fmt.Sprintf("%s:%v", k, v))
+	}
+	return result
+}

services/remotecluster/ping.go

@@ -0,0 +1,174 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/mattermost/mattermost-server/v5/model"
+	"github.com/mattermost/mattermost-server/v5/shared/mlog"
+)
+
+// pingLoop periodically sends a ping to all remote clusters.
+func (rcs *Service) pingLoop(done <-chan struct{}) {
+	pingChan := make(chan *model.RemoteCluster, MaxConcurrentSends*2)
+
+	// create a thread pool to send pings concurrently to remotes.
+	for i := 0; i < MaxConcurrentSends; i++ {
+		go rcs.pingEmitter(pingChan, done)
+	}
+
+	go rcs.pingGenerator(pingChan, done)
+}
+
+func (rcs *Service) pingGenerator(pingChan chan *model.RemoteCluster, done <-chan struct{}) {
+	defer close(pingChan)
+
+	for {
+		start := time.Now()
+
+		// get all remotes, including any previously offline.
+		remotes, err := rcs.server.GetStore().RemoteCluster().GetAll(model.RemoteClusterQueryFilter{})
+		if err != nil {
+			rcs.server.GetLogger().Log(mlog.LvlRemoteClusterServiceError, "Ping remote cluster failed (could not get list of remotes)", mlog.Err(err))
+			select {
+			case <-time.After(PingFreq):
+				continue
+			case <-done:
+				return
+			}
+		}
+
+		for _, rc := range remotes {
+			if rc.SiteURL != "" { // filter out unconfirmed invites
+				pingChan <- rc
+			}
+		}
+
+		// try to maintain frequency
+		elapsed := time.Since(start)
+		if elapsed < PingFreq {
+			sleep := time.Until(start.Add(PingFreq))
+			select {
+			case <-time.After(sleep):
+			case <-done:
+				return
+			}
+		}
+	}
+}
+
+// pingEmitter pulls Remotes from the ping queue (pingChan) and pings them.
+// Pinging a remote cannot take longer than PingTimeoutMillis.
+func (rcs *Service) pingEmitter(pingChan <-chan *model.RemoteCluster, done <-chan struct{}) {
+	for {
+		select {
+		case rc := <-pingChan:
+			if rc == nil {
+				return
+			}
+
+			online := rc.IsOnline()
+
+			if err := rcs.pingRemote(rc); err != nil {
+				rcs.server.GetLogger().Log(mlog.LvlRemoteClusterServiceWarn, "Remote cluster ping failed",
+					mlog.String("remote", rc.DisplayName),
+					mlog.String("remoteId", rc.RemoteId),
+					mlog.Err(err),
+				)
+			}
+
+			if online != rc.IsOnline() {
+				if metrics := rcs.server.GetMetrics(); metrics != nil {
+					metrics.IncrementRemoteClusterConnStateChangeCounter(rc.RemoteId, rc.IsOnline())
+				}
+				rcs.fireConnectionStateChgEvent(rc)
+			}
+		case <-done:
+			return
+		}
+	}
+}
+
+// pingRemote make a synchronous ping to a remote cluster. Return is error if ping is
+// unsuccessful and nil on success.
+func (rcs *Service) pingRemote(rc *model.RemoteCluster) error {
+	frame, err := makePingFrame(rc)
+	if err != nil {
+		return err
+	}
+	url := fmt.Sprintf("%s/%s", rc.SiteURL, PingURL)
+
+	resp, err := rcs.sendFrameToRemote(PingTimeout, rc, frame, url)
+	if err != nil {
+		return err
+	}
+
+	ping := model.RemoteClusterPing{}
+	err = json.Unmarshal(resp, &ping)
+	if err != nil {
+		return err
+	}
+
+	if err := rcs.server.GetStore().RemoteCluster().SetLastPingAt(rc.RemoteId); err != nil {
+		rcs.server.GetLogger().Log(mlog.LvlRemoteClusterServiceError, "Failed to update LastPingAt for remote cluster",
+			mlog.String("remote", rc.DisplayName),
+			mlog.String("remoteId", rc.RemoteId),
+			mlog.Err(err),
+		)
+	}
+	rc.LastPingAt = model.GetMillis()
+
+	if metrics := rcs.server.GetMetrics(); metrics != nil {
+		sentAt := time.Unix(0, ping.SentAt*int64(time.Millisecond))
+		elapsed := time.Since(sentAt).Seconds()
+		metrics.ObserveRemoteClusterPingDuration(rc.RemoteId, elapsed)
+
+		// we approximate clock skew between remotes.
+		skew := elapsed/2 - float64(ping.RecvAt-ping.SentAt)/1000
+		metrics.ObserveRemoteClusterClockSkew(rc.RemoteId, skew)
+	}
+
+	rcs.server.GetLogger().Log(mlog.LvlRemoteClusterServiceDebug, "Remote cluster ping",
+		mlog.String("remote", rc.DisplayName),
+		mlog.String("remoteId", rc.RemoteId),
+		mlog.Int64("SentAt", ping.SentAt),
+		mlog.Int64("RecvAt", ping.RecvAt),
+		mlog.Int64("Diff", ping.RecvAt-ping.SentAt),
+	)
+	return nil
+}
+
+func makePingFrame(rc *model.RemoteCluster) (*model.RemoteClusterFrame, error) {
+	ping := model.RemoteClusterPing{
+		SentAt: model.GetMillis(),
+	}
+	pingRaw, err := json.Marshal(ping)
+	if err != nil {
+		return nil, err
+	}
+
+	msg := model.NewRemoteClusterMsg(PingTopic, pingRaw)
+
+	frame := &model.RemoteClusterFrame{
+		RemoteId: rc.RemoteId,
+		Msg:      msg,
+	}
+	return frame, nil
+}
+
+func (rcs *Service) fireConnectionStateChgEvent(rc *model.RemoteCluster) {
+	rcs.mux.RLock()
+	listeners := make([]ConnectionStateListener, 0, len(rcs.connectionStateListeners))
+	for _, l := range rcs.connectionStateListeners {
+		listeners = append(listeners, l)
+	}
+	rcs.mux.RUnlock()
+
+	for _, l := range listeners {
+		l(rc, rc.IsOnline())
+	}
+}

services/remotecluster/ping_test.go

@@ -0,0 +1,133 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/wiggin77/merror"
+
+	"github.com/mattermost/mattermost-server/v5/model"
+)
+
+const (
+	Recent = 60000
+)
+
+func TestPing(t *testing.T) {
+	disablePing = false
+
+	t.Run("No error", func(t *testing.T) {
+		var countWebReq int32
+		merr := merror.New()
+
+		wg := &sync.WaitGroup{}
+		wg.Add(NumRemotes)
+
+		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			defer wg.Done()
+			defer w.WriteHeader(200)
+			atomic.AddInt32(&countWebReq, 1)
+
+			frame, err := model.RemoteClusterFrameFromJSON(r.Body)
+			if err != nil {
+				merr.Append(err)
+				return
+			}
+			if len(frame.Msg.Payload) == 0 {
+				merr.Append(fmt.Errorf("Payload should not be empty; remote_id=%s", frame.RemoteId))
+				return
+			}
+
+			ping, err := model.RemoteClusterPingFromRawJSON(frame.Msg.Payload)
+			if err != nil {
+				merr.Append(err)
+				return
+			}
+			if !checkRecent(ping.SentAt, Recent) {
+				merr.Append(fmt.Errorf("timestamp out of range, got %d", ping.SentAt))
+				return
+			}
+			if ping.RecvAt != 0 {
+				merr.Append(fmt.Errorf("timestamp should be 0, got %d", ping.RecvAt))
+				return
+			}
+		}))
+		defer ts.Close()
+
+		mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL))
+		service, err := NewRemoteClusterService(mockServer)
+		require.NoError(t, err)
+
+		err = service.Start()
+		require.NoError(t, err)
+		defer service.Shutdown()
+
+		wg.Wait()
+
+		assert.NoError(t, merr.ErrorOrNil())
+
+		assert.Equal(t, int32(NumRemotes), atomic.LoadInt32(&countWebReq))
+		t.Log(fmt.Sprintf("%d web requests counted;  %d expected",
+			atomic.LoadInt32(&countWebReq), NumRemotes))
+	})
+
+	t.Run("HTTP errors", func(t *testing.T) {
+		var countWebReq int32
+		merr := merror.New()
+
+		wg := &sync.WaitGroup{}
+		wg.Add(NumRemotes)
+
+		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			defer wg.Done()
+			atomic.AddInt32(&countWebReq, 1)
+
+			frame, err := model.RemoteClusterFrameFromJSON(r.Body)
+			if err != nil {
+				merr.Append(err)
+			}
+			ping, err := model.RemoteClusterPingFromRawJSON(frame.Msg.Payload)
+			if err != nil {
+				merr.Append(err)
+			}
+			if !checkRecent(ping.SentAt, Recent) {
+				merr.Append(fmt.Errorf("timestamp out of range, got %d", ping.SentAt))
+			}
+			if ping.RecvAt != 0 {
+				merr.Append(fmt.Errorf("timestamp should be 0, got %d", ping.RecvAt))
+			}
+			w.WriteHeader(500)
+		}))
+		defer ts.Close()
+
+		mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL))
+		service, err := NewRemoteClusterService(mockServer)
+		require.NoError(t, err)
+
+		err = service.Start()
+		require.NoError(t, err)
+		defer service.Shutdown()
+
+		wg.Wait()
+
+		assert.Nil(t, merr.ErrorOrNil())
+
+		assert.Equal(t, int32(NumRemotes), atomic.LoadInt32(&countWebReq))
+		t.Log(fmt.Sprintf("%d web requests counted;  %d expected",
+			atomic.LoadInt32(&countWebReq), NumRemotes))
+	})
+}
+
+func checkRecent(millis int64, within int64) bool {
+	now := model.GetMillis()
+	return millis > now-within && millis < now+within
+}

services/remotecluster/recv.go

@@ -0,0 +1,53 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"fmt"
+
+	"github.com/mattermost/mattermost-server/v5/model"
+	"github.com/mattermost/mattermost-server/v5/shared/mlog"
+)
+
+// 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 {
+	rcs.mux.RLock()
+	defer rcs.mux.RUnlock()
+
+	if metrics := rcs.server.GetMetrics(); metrics != nil {
+		metrics.IncrementRemoteClusterMsgReceivedCounter(rc.RemoteId)
+	}
+
+	rcSanitized := *rc
+	rcSanitized.Token = ""
+	rcSanitized.RemoteToken = ""
+
+	var response Response
+	response.Status = ResponseStatusOK
+
+	listeners := rcs.getTopicListeners(msg.Topic)
+
+	for _, l := range listeners {
+		if err := callback(l, 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))
+
+			response.Status = ResponseStatusFail
+			response.Err = err.Error()
+		}
+	}
+	return response
+}
+
+func callback(listener TopicListener, 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)
+	return
+}

services/remotecluster/response.go

@@ -0,0 +1,30 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"encoding/json"
+)
+
+// Response represents the bytes replied from a remote server when a message is sent.
+type Response struct {
+	Status  string          `json:"status"`
+	Err     string          `json:"err"`
+	Payload json.RawMessage `json:"payload"`
+}
+
+// IsSuccess returns true if the response status indicates success.
+func (r *Response) IsSuccess() bool {
+	return r.Status == ResponseStatusOK
+}
+
+// SetPayload serializes an arbitrary struct as a RawMessage.
+func (r *Response) SetPayload(v interface{}) error {
+	raw, err := json.Marshal(v)
+	if err != nil {
+		return err
+	}
+	r.Payload = raw
+	return nil
+}

services/remotecluster/send.go

@@ -0,0 +1,56 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"context"
+	"hash/fnv"
+)
+
+// enqueueTask adds a task to one of the send channels based on remoteId.
+//
+// There are a number  of send channels (`MaxConcurrentSends`) to allow for sending to multiple
+// remotes concurrently, while preserving message order for each remote.
+func (rcs *Service) enqueueTask(ctx context.Context, remoteId string, task interface{}) error {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+
+	h := hash(remoteId)
+	idx := h % uint32(len(rcs.send))
+
+	select {
+	case rcs.send[idx] <- task:
+		return nil
+	case <-ctx.Done():
+		return NewBufferFullError(cap(rcs.send))
+	}
+}
+
+func hash(s string) uint32 {
+	h := fnv.New32a()
+	h.Write([]byte(s))
+	return h.Sum32()
+}
+
+// sendLoop is called by each goroutine created for the send pool and waits for sendTask's until the
+// done channel is signalled.
+//
+// Each goroutine in the pool is assigned a specific channel, and tasks are placed in the
+// channel corresponding to the remoteId.
+func (rcs *Service) sendLoop(idx int, done chan struct{}) {
+	for {
+		select {
+		case t := <-rcs.send[idx]:
+			switch task := t.(type) {
+			case sendMsgTask:
+				rcs.sendMsg(task)
+			case sendFileTask:
+				rcs.sendFile(task)
+			}
+		case <-done:
+			return
+		}
+	}
+}

services/remotecluster/send_test.go

@@ -0,0 +1,200 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/wiggin77/merror"
+
+	"github.com/mattermost/mattermost-server/v5/model"
+)
+
+const (
+	TestTopics  = " share incident "
+	TestTopic   = "share"
+	NumRemotes  = 50
+	NoteContent = "Woot!!"
+)
+
+type testPayload struct {
+	Note string `json:"note"`
+}
+
+func TestBroadcastMsg(t *testing.T) {
+	msgId := model.NewId()
+	disablePing = true
+
+	t.Run("No error", func(t *testing.T) {
+		var countCallbacks int32
+		var countWebReq int32
+		merr := merror.New()
+
+		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			defer func() {
+				w.WriteHeader(200)
+				var resp Response
+				b, errMarshall := json.Marshal(&resp)
+				if errMarshall != nil {
+					merr.Append(errMarshall)
+					return
+				}
+				w.Write(b)
+			}()
+
+			atomic.AddInt32(&countWebReq, 1)
+
+			frame, appErr := model.RemoteClusterFrameFromJSON(r.Body)
+			if appErr != nil {
+				merr.Append(appErr)
+				return
+			}
+			if len(frame.Msg.Payload) == 0 {
+				merr.Append(fmt.Errorf("webrequest missing Msg.Payload"))
+			}
+			if msgId != frame.Msg.Id {
+				merr.Append(fmt.Errorf("webrequest msgId expected %s, got %s", msgId, frame.Msg.Id))
+				return
+			}
+
+			note := testPayload{}
+			err := json.Unmarshal(frame.Msg.Payload, &note)
+			if err != nil {
+				merr.Append(err)
+				return
+			}
+			if note.Note != NoteContent {
+				merr.Append(fmt.Errorf("webrequest payload expected %s, got %s", NoteContent, note.Note))
+				return
+			}
+		}))
+		defer ts.Close()
+
+		mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL))
+		service, err := NewRemoteClusterService(mockServer)
+		require.NoError(t, err)
+
+		err = service.Start()
+		require.NoError(t, err)
+		defer service.Shutdown()
+
+		wg := &sync.WaitGroup{}
+		wg.Add(NumRemotes)
+
+		msg := makeRemoteClusterMsg(msgId, NoteContent)
+
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second*15)
+		defer cancel()
+
+		err = service.BroadcastMsg(ctx, msg, func(msg model.RemoteClusterMsg, remote *model.RemoteCluster, resp *Response, err error) {
+			defer wg.Done()
+			atomic.AddInt32(&countCallbacks, 1)
+
+			if err != nil {
+				merr.Append(err)
+			}
+			if msgId != msg.Id {
+				merr.Append(fmt.Errorf("result callback msgId expected %s, got %s", msgId, msg.Id))
+			}
+
+			var note testPayload
+			err2 := json.Unmarshal(msg.Payload, &note)
+			if err2 != nil {
+				merr.Append(fmt.Errorf("unmarshal payload error: %w", err2))
+				return
+			}
+			if note.Note != NoteContent {
+				merr.Append(fmt.Errorf("compare payload failed: expected '%s', got '%s'", NoteContent, note))
+			}
+		})
+		assert.NoError(t, err)
+
+		wg.Wait()
+
+		assert.NoError(t, merr.ErrorOrNil())
+
+		assert.Equal(t, int32(NumRemotes), atomic.LoadInt32(&countCallbacks))
+		assert.Equal(t, int32(NumRemotes), atomic.LoadInt32(&countWebReq))
+		t.Log(fmt.Sprintf("%d callbacks counted;  %d web requests counted;  %d expected",
+			atomic.LoadInt32(&countCallbacks), atomic.LoadInt32(&countWebReq), NumRemotes))
+	})
+
+	t.Run("HTTP error", func(t *testing.T) {
+		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.WriteHeader(500)
+		}))
+		defer ts.Close()
+
+		mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL))
+		service, err := NewRemoteClusterService(mockServer)
+		require.NoError(t, err)
+
+		err = service.Start()
+		require.NoError(t, err)
+		defer service.Shutdown()
+
+		msg := makeRemoteClusterMsg(msgId, NoteContent)
+		var countCallbacks int32
+		var countErrors int32
+		wg := &sync.WaitGroup{}
+		wg.Add(NumRemotes)
+
+		err = service.BroadcastMsg(context.Background(), msg, func(msg model.RemoteClusterMsg, remote *model.RemoteCluster, resp *Response, err error) {
+			defer wg.Done()
+			atomic.AddInt32(&countCallbacks, 1)
+			if err != nil {
+				atomic.AddInt32(&countErrors, 1)
+			}
+		})
+		assert.NoError(t, err)
+
+		wg.Wait()
+
+		assert.Equal(t, int32(NumRemotes), atomic.LoadInt32(&countCallbacks))
+		assert.Equal(t, int32(NumRemotes), atomic.LoadInt32(&countErrors))
+	})
+}
+
+func makeRemoteClusters(num int, siteURL string) []*model.RemoteCluster {
+	var remotes []*model.RemoteCluster
+	for i := 0; i < num; i++ {
+		rc := makeRemoteCluster(fmt.Sprintf("test cluster %d", i+1), siteURL, TestTopics)
+		remotes = append(remotes, rc)
+	}
+	return remotes
+}
+
+func makeRemoteCluster(name string, siteURL string, topics string) *model.RemoteCluster {
+	return &model.RemoteCluster{
+		RemoteId:    model.NewId(),
+		DisplayName: name,
+		SiteURL:     siteURL,
+		Token:       model.NewId(),
+		Topics:      topics,
+		CreateAt:    model.GetMillis(),
+		LastPingAt:  model.GetMillis(),
+		CreatorId:   model.NewId(),
+	}
+}
+
+func makeRemoteClusterMsg(id string, note string) model.RemoteClusterMsg {
+	payload := testPayload{Note: note}
+	raw, _ := json.Marshal(payload)
+
+	return model.RemoteClusterMsg{
+		Id:       id,
+		Topic:    TestTopic,
+		CreateAt: model.GetMillis(),
+		Payload:  raw}
+}

services/remotecluster/sendfile.go

@@ -0,0 +1,147 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"path"
+	"time"
+
+	"github.com/mattermost/mattermost-server/v5/model"
+	"github.com/mattermost/mattermost-server/v5/shared/filestore"
+	"github.com/mattermost/mattermost-server/v5/shared/mlog"
+)
+
+type SendFileResultFunc func(us *model.UploadSession, rc *model.RemoteCluster, resp *Response, err error)
+
+type sendFileTask struct {
+	rc *model.RemoteCluster
+	us *model.UploadSession
+	fi *model.FileInfo
+	rp ReaderProvider
+	f  SendFileResultFunc
+}
+
+type ReaderProvider interface {
+	FileReader(path string) (filestore.ReadCloseSeeker, *model.AppError)
+}
+
+// SendFile asynchronously sends a file to a remote cluster.
+//
+// `ctx` determines behaviour when the outbound queue is full. A timeout or deadline context will return a
+// BufferFullError if the file cannot be enqueued before the timeout. A background context will block indefinitely.
+//
+// Nil or error return indicates success or failure of file enqueue only.
+//
+// An optional callback can be provided that receives the response from the remote cluster. The `err` provided to the
+// callback is regarding file delivery only. The `resp` contains the decoded bytes returned from the remote.
+// If a callback is provided it should return quickly.
+func (rcs *Service) SendFile(ctx context.Context, us *model.UploadSession, fi *model.FileInfo, rc *model.RemoteCluster, rp ReaderProvider, f SendFileResultFunc) error {
+	task := sendFileTask{
+		rc: rc,
+		us: us,
+		fi: fi,
+		rp: rp,
+		f:  f,
+	}
+	return rcs.enqueueTask(ctx, rc.RemoteId, task)
+}
+
+// sendFile is called when a sendFileTask is popped from the send channel.
+func (rcs *Service) sendFile(task sendFileTask) {
+	// Ensure a panic from the callback does not exit the goroutine.
+	defer func() {
+		if r := recover(); r != nil {
+			rcs.server.GetLogger().Log(mlog.LvlRemoteClusterServiceError, "Remote Cluster sendFile panic",
+				mlog.String("remote", task.rc.DisplayName),
+				mlog.String("uploadId", task.us.Id),
+				mlog.Any("panic", r),
+			)
+		}
+	}()
+
+	fi, err := rcs.sendFileToRemote(SendTimeout, task)
+	var response Response
+
+	if err != nil {
+		rcs.server.GetLogger().Log(mlog.LvlRemoteClusterServiceError, "Remote Cluster send file failed",
+			mlog.String("remote", task.rc.DisplayName),
+			mlog.String("uploadId", task.us.Id),
+			mlog.Err(err),
+		)
+		response.Status = ResponseStatusFail
+		response.Err = err.Error()
+	} else {
+		rcs.server.GetLogger().Log(mlog.LvlRemoteClusterServiceDebug, "Remote Cluster file sent successfully",
+			mlog.String("remote", task.rc.DisplayName),
+			mlog.String("uploadId", task.us.Id),
+		)
+		response.Status = ResponseStatusOK
+		response.SetPayload(fi)
+	}
+
+	// If callback provided then call it with the results.
+	if task.f != nil {
+		task.f(task.us, task.rc, &response, err)
+	}
+}
+
+func (rcs *Service) sendFileToRemote(timeout time.Duration, task sendFileTask) (*model.FileInfo, error) {
+	rcs.server.GetLogger().Log(mlog.LvlRemoteClusterServiceDebug, "sending file to remote...",
+		mlog.String("remote", task.rc.DisplayName),
+		mlog.String("uploadId", task.us.Id),
+		mlog.String("file_path", task.us.Path),
+	)
+
+	r, appErr := task.rp.FileReader(task.fi.Path) // get Reader for the file
+	if appErr != nil {
+		return nil, fmt.Errorf("error opening file while sending file to remote %s: %w", task.rc.RemoteId, appErr)
+	}
+	defer r.Close()
+
+	u, err := url.Parse(task.rc.SiteURL)
+	if err != nil {
+		return nil, fmt.Errorf("invalid siteURL while sending file to remote %s: %w", task.rc.RemoteId, err)
+	}
+	u.Path = path.Join(u.Path, model.API_URL_SUFFIX, "remotecluster", "upload", task.us.Id)
+
+	req, err := http.NewRequest("POST", u.String(), r)
+	if err != nil {
+		return nil, err
+	}
+
+	req.Header.Set(model.HEADER_REMOTECLUSTER_ID, task.rc.RemoteId)
+	req.Header.Set(model.HEADER_REMOTECLUSTER_TOKEN, task.rc.RemoteToken)
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	resp, err := rcs.httpClient.Do(req.WithContext(ctx))
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("unexpected response: %d - %s", resp.StatusCode, resp.Status)
+	}
+
+	// body should be a FileInfo
+	var fi model.FileInfo
+	if err := json.Unmarshal(body, &fi); err != nil {
+		return nil, fmt.Errorf("unexpected response body: %w", err)
+	}
+
+	return &fi, nil
+}

services/remotecluster/sendmsg.go

@@ -0,0 +1,180 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"os"
+	"path"
+	"time"
+
+	"github.com/wiggin77/merror"
+
+	"github.com/mattermost/mattermost-server/v5/model"
+	"github.com/mattermost/mattermost-server/v5/shared/mlog"
+)
+
+type SendMsgResultFunc func(msg model.RemoteClusterMsg, rc *model.RemoteCluster, resp *Response, err error)
+
+type sendMsgTask struct {
+	rc  *model.RemoteCluster
+	msg model.RemoteClusterMsg
+	f   SendMsgResultFunc
+}
+
+// BroadcastMsg asynchronously sends a message to all remote clusters interested in the message's topic.
+//
+// `ctx` determines behaviour when the outbound queue is full. A timeout or deadline context will return a
+// BufferFullError if the message cannot be enqueued before the timeout. A background context will block indefinitely.
+//
+// An optional callback can be provided that receives the success or fail result of sending to each remote cluster.
+// Success or fail is regarding message delivery only.  If a callback is provided it should return quickly.
+func (rcs *Service) BroadcastMsg(ctx context.Context, msg model.RemoteClusterMsg, f SendMsgResultFunc) error {
+	// get list of interested remotes.
+	filter := model.RemoteClusterQueryFilter{
+		Topic: msg.Topic,
+	}
+	list, err := rcs.server.GetStore().RemoteCluster().GetAll(filter)
+	if err != nil {
+		return err
+	}
+
+	errs := merror.New()
+
+	for _, rc := range list {
+		if err := rcs.SendMsg(ctx, msg, rc, f); err != nil {
+			errs.Append(err)
+		}
+	}
+	return errs.ErrorOrNil()
+}
+
+// SendMsg asynchronously sends a message to a remote cluster.
+//
+// `ctx` determines behaviour when the outbound queue is full. A timeout or deadline context will return a
+// BufferFullError if the message cannot be enqueued before the timeout. A background context will block indefinitely.
+//
+// Nil or error return indicates success or failure of message enqueue only.
+//
+// An optional callback can be provided that receives the response from the remote cluster. The `err` provided to the
+// callback is regarding response decoding only. The `resp` contains the decoded bytes returned from the remote.
+// If a callback is provided it should return quickly.
+func (rcs *Service) SendMsg(ctx context.Context, msg model.RemoteClusterMsg, rc *model.RemoteCluster, f SendMsgResultFunc) error {
+	task := sendMsgTask{
+		rc:  rc,
+		msg: msg,
+		f:   f,
+	}
+	return rcs.enqueueTask(ctx, rc.RemoteId, task)
+}
+
+// sendMsg is called when a sendMsgTask is popped from the send channel.
+func (rcs *Service) sendMsg(task sendMsgTask) {
+	var errResp error
+	var response Response
+
+	// Ensure a panic from the callback does not exit the pool goroutine.
+	defer func() {
+		if r := recover(); r != nil {
+			rcs.server.GetLogger().Log(mlog.LvlRemoteClusterServiceError, "Remote Cluster sendMsg panic",
+				mlog.String("remote", task.rc.DisplayName), mlog.String("msgId", task.msg.Id), mlog.Any("panic", r))
+		}
+
+		if errResp != nil {
+			response.Err = errResp.Error()
+		}
+
+		// If callback provided then call it with the results.
+		if task.f != nil {
+			task.f(task.msg, task.rc, &response, errResp)
+		}
+	}()
+
+	frame := &model.RemoteClusterFrame{
+		RemoteId: task.rc.RemoteId,
+		Msg:      task.msg,
+	}
+
+	u, err := url.Parse(task.rc.SiteURL)
+	if err != nil {
+		rcs.server.GetLogger().Log(mlog.LvlRemoteClusterServiceError, "Invalid siteURL while sending message to remote",
+			mlog.String("remote", task.rc.DisplayName),
+			mlog.String("msgId", task.msg.Id),
+			mlog.Err(err),
+		)
+		errResp = err
+		return
+	}
+	u.Path = path.Join(u.Path, SendMsgURL)
+
+	respJSON, err := rcs.sendFrameToRemote(SendTimeout, task.rc, frame, u.String())
+
+	if err != nil {
+		rcs.server.GetLogger().Log(mlog.LvlRemoteClusterServiceError, "Remote Cluster send message failed",
+			mlog.String("remote", task.rc.DisplayName),
+			mlog.String("msgId", task.msg.Id),
+			mlog.Err(err),
+		)
+		errResp = err
+	} else {
+		rcs.server.GetLogger().Log(mlog.LvlRemoteClusterServiceDebug, "Remote Cluster message sent successfully",
+			mlog.String("remote", task.rc.DisplayName),
+			mlog.String("msgId", task.msg.Id),
+		)
+
+		if err = json.Unmarshal(respJSON, &response); err != nil {
+			rcs.server.GetLogger().Error("Invalid response sending message to remote cluster",
+				mlog.String("remote", task.rc.DisplayName),
+				mlog.Err(err),
+			)
+			errResp = err
+		}
+	}
+}
+
+func (rcs *Service) sendFrameToRemote(timeout time.Duration, rc *model.RemoteCluster, frame *model.RemoteClusterFrame, url string) ([]byte, error) {
+	body, err := json.Marshal(frame)
+	if err != nil {
+		return nil, err
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	req, err := http.NewRequest("POST", url, bytes.NewReader(body))
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set(model.HEADER_REMOTECLUSTER_ID, rc.RemoteId)
+	req.Header.Set(model.HEADER_REMOTECLUSTER_TOKEN, rc.RemoteToken)
+
+	resp, err := rcs.httpClient.Do(req.WithContext(ctx))
+	if metrics := rcs.server.GetMetrics(); metrics != nil {
+		if err != nil || resp.StatusCode != http.StatusOK {
+			metrics.IncrementRemoteClusterMsgErrorsCounter(frame.RemoteId, os.IsTimeout(err))
+		} else {
+			metrics.IncrementRemoteClusterMsgSentCounter(frame.RemoteId)
+		}
+	}
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	body, err = ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		return body, fmt.Errorf("unexpected response: %d - %s", resp.StatusCode, resp.Status)
+	}
+	return body, nil
+}

services/remotecluster/service.go

@@ -0,0 +1,261 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"context"
+	"net"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/mattermost/mattermost-server/v5/einterfaces"
+	"github.com/mattermost/mattermost-server/v5/model"
+	"github.com/mattermost/mattermost-server/v5/shared/mlog"
+	"github.com/mattermost/mattermost-server/v5/store"
+)
+
+const (
+	SendChanBuffer                = 50
+	RecvChanBuffer                = 50
+	ResultsChanBuffer             = 50
+	ResultQueueDrainTimeoutMillis = 10000
+	MaxConcurrentSends            = 10
+	SendMsgURL                    = "api/v4/remotecluster/msg"
+	SendTimeout                   = time.Minute
+	SendFileTimeout               = time.Minute * 5
+	PingURL                       = "api/v4/remotecluster/ping"
+	PingFreq                      = time.Minute
+	PingTimeout                   = time.Second * 15
+	ConfirmInviteURL              = "api/v4/remotecluster/confirm_invite"
+	InvitationTopic               = "invitation"
+	PingTopic                     = "ping"
+	ResponseStatusOK              = model.STATUS_OK
+	ResponseStatusFail            = model.STATUS_FAIL
+	InviteExpiresAfter            = time.Hour * 48
+)
+
+var (
+	disablePing bool // override for testing
+)
+
+type ServerIface interface {
+	Config() *model.Config
+	IsLeader() bool
+	AddClusterLeaderChangedListener(listener func()) string
+	RemoveClusterLeaderChangedListener(id string)
+	GetStore() store.Store
+	GetLogger() mlog.LoggerIFace
+	GetMetrics() einterfaces.MetricsInterface
+}
+
+// RemoteClusterServiceIFace is used to allow mocking where a remote cluster service is used (for testing).
+// Unfortunately it lives here because the shared channel service, app layer, and server interface all need it.
+// Putting it in app layer means shared channel service must import app package.
+type RemoteClusterServiceIFace interface {
+	Shutdown() error
+	Start() error
+	Active() bool
+	AddTopicListener(topic string, listener TopicListener) string
+	RemoveTopicListener(listenerId string)
+	AddConnectionStateListener(listener ConnectionStateListener) string
+	RemoveConnectionStateListener(listenerId string)
+	SendMsg(ctx context.Context, msg model.RemoteClusterMsg, rc *model.RemoteCluster, f SendMsgResultFunc) error
+	SendFile(ctx context.Context, us *model.UploadSession, fi *model.FileInfo, rc *model.RemoteCluster, rp ReaderProvider, f SendFileResultFunc) error
+	AcceptInvitation(invite *model.RemoteClusterInvite, name string, creatorId string, teamId string, siteURL string) (*model.RemoteCluster, error)
+	ReceiveIncomingMsg(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
+
+// ConnectionStateListener is used to listen to remote cluster connection state changes.
+type ConnectionStateListener func(rc *model.RemoteCluster, online bool)
+
+// Service provides inter-cluster communication via topic based messages.
+type Service struct {
+	server     ServerIface
+	httpClient *http.Client
+	send       []chan interface{}
+
+	// everything below guarded by `mux`
+	mux                      sync.RWMutex
+	active                   bool
+	leaderListenerId         string
+	topicListeners           map[string]map[string]TopicListener // maps topic id to a map of listenerid->listener
+	connectionStateListeners map[string]ConnectionStateListener  // maps listener id to listener
+	done                     chan struct{}
+}
+
+// NewRemoteClusterService creates a RemoteClusterService instance.
+func NewRemoteClusterService(server ServerIface) (*Service, error) {
+	transport := &http.Transport{
+		Proxy: http.ProxyFromEnvironment,
+		DialContext: (&net.Dialer{
+			Timeout:   30 * time.Second,
+			KeepAlive: 30 * time.Second,
+			DualStack: true,
+		}).DialContext,
+		ForceAttemptHTTP2:     true,
+		MaxIdleConns:          200,
+		MaxIdleConnsPerHost:   2,
+		IdleConnTimeout:       90 * time.Second,
+		TLSHandshakeTimeout:   10 * time.Second,
+		ExpectContinueTimeout: 1 * time.Second,
+		DisableCompression:    false,
+	}
+
+	client := &http.Client{
+		Transport: transport,
+		Timeout:   SendTimeout,
+	}
+
+	service := &Service{
+		server:                   server,
+		httpClient:               client,
+		topicListeners:           make(map[string]map[string]TopicListener),
+		connectionStateListeners: make(map[string]ConnectionStateListener),
+	}
+
+	service.send = make([]chan interface{}, MaxConcurrentSends)
+	for i := range service.send {
+		service.send[i] = make(chan interface{}, SendChanBuffer)
+	}
+
+	return service, nil
+}
+
+// Start is called by the server on server start-up.
+func (rcs *Service) Start() error {
+	rcs.mux.Lock()
+	rcs.leaderListenerId = rcs.server.AddClusterLeaderChangedListener(rcs.onClusterLeaderChange)
+	rcs.mux.Unlock()
+
+	rcs.onClusterLeaderChange()
+
+	return nil
+}
+
+// Shutdown is called by the server on server shutdown.
+func (rcs *Service) Shutdown() error {
+	rcs.server.RemoveClusterLeaderChangedListener(rcs.leaderListenerId)
+	rcs.pause()
+	return nil
+}
+
+// Active returns true if this instance of the remote cluster service is active.
+// The active instance is responsible for pinging and sending messages to remotes.
+func (rcs *Service) Active() bool {
+	rcs.mux.Lock()
+	defer rcs.mux.Unlock()
+	return rcs.active
+}
+
+// AddTopicListener registers a callback
+func (rcs *Service) AddTopicListener(topic string, listener TopicListener) string {
+	rcs.mux.Lock()
+	defer rcs.mux.Unlock()
+
+	id := model.NewId()
+
+	listeners, ok := rcs.topicListeners[topic]
+	if !ok || listeners == nil {
+		rcs.topicListeners[topic] = make(map[string]TopicListener)
+	}
+	rcs.topicListeners[topic][id] = listener
+	return id
+}
+
+func (rcs *Service) RemoveTopicListener(listenerId string) {
+	rcs.mux.Lock()
+	defer rcs.mux.Unlock()
+
+	for topic, listeners := range rcs.topicListeners {
+		if _, ok := listeners[listenerId]; ok {
+			delete(listeners, listenerId)
+			if len(listeners) == 0 {
+				delete(rcs.topicListeners, topic)
+			}
+			break
+		}
+	}
+}
+
+func (rcs *Service) getTopicListeners(topic string) []TopicListener {
+	rcs.mux.RLock()
+	defer rcs.mux.RUnlock()
+
+	listeners, ok := rcs.topicListeners[topic]
+	if !ok {
+		return nil
+	}
+
+	listenersCopy := make([]TopicListener, 0, len(listeners))
+	for _, l := range listeners {
+		listenersCopy = append(listenersCopy, l)
+	}
+	return listenersCopy
+}
+
+func (rcs *Service) AddConnectionStateListener(listener ConnectionStateListener) string {
+	id := model.NewId()
+
+	rcs.mux.Lock()
+	defer rcs.mux.Unlock()
+
+	rcs.connectionStateListeners[id] = listener
+	return id
+}
+
+func (rcs *Service) RemoveConnectionStateListener(listenerId string) {
+	rcs.mux.Lock()
+	defer rcs.mux.Unlock()
+	delete(rcs.connectionStateListeners, listenerId)
+}
+
+// onClusterLeaderChange is called whenever the cluster leader may have changed.
+func (rcs *Service) onClusterLeaderChange() {
+	if rcs.server.IsLeader() {
+		rcs.resume()
+	} else {
+		rcs.pause()
+	}
+}
+
+func (rcs *Service) resume() {
+	rcs.mux.Lock()
+	defer rcs.mux.Unlock()
+
+	if rcs.active {
+		return // already active
+	}
+	rcs.active = true
+	rcs.done = make(chan struct{})
+
+	if !disablePing {
+		rcs.pingLoop(rcs.done)
+	}
+
+	// create thread pool for concurrent message sending.
+	for i := range rcs.send {
+		go rcs.sendLoop(i, rcs.done)
+	}
+
+	rcs.server.GetLogger().Debug("Remote Cluster Service active")
+}
+
+func (rcs *Service) pause() {
+	rcs.mux.Lock()
+	defer rcs.mux.Unlock()
+
+	if !rcs.active {
+		return // already inactive
+	}
+	rcs.active = false
+	close(rcs.done)
+	rcs.done = nil
+
+	rcs.server.GetLogger().Debug("Remote Cluster Service inactive")
+}

services/remotecluster/service_test.go

@@ -0,0 +1,71 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"sync/atomic"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"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 {
+		atomic.AddInt32(&count, 1)
+		return nil
+	}
+	l2 := func(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 {
+		atomic.AddInt32(&count, 1)
+		return nil
+	}
+
+	mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ""))
+	service, err := NewRemoteClusterService(mockServer)
+	require.NoError(t, err)
+
+	l1id := service.AddTopicListener("test", l1)
+	l2id := service.AddTopicListener("test", l2)
+	l3id := service.AddTopicListener("different", l3)
+
+	listeners := service.getTopicListeners("test")
+	assert.Len(t, listeners, 2)
+
+	rc := &model.RemoteCluster{}
+	msg1 := model.RemoteClusterMsg{Topic: "test"}
+	msg2 := model.RemoteClusterMsg{Topic: "different"}
+
+	service.ReceiveIncomingMsg(rc, msg1)
+	assert.Equal(t, int32(2), atomic.LoadInt32(&count))
+
+	service.ReceiveIncomingMsg(rc, msg2)
+	assert.Equal(t, int32(3), atomic.LoadInt32(&count))
+
+	service.RemoveTopicListener(l1id)
+	service.ReceiveIncomingMsg(rc, msg1)
+	assert.Equal(t, int32(4), atomic.LoadInt32(&count))
+
+	service.RemoveTopicListener(l2id)
+	service.ReceiveIncomingMsg(rc, msg1)
+	assert.Equal(t, int32(4), atomic.LoadInt32(&count))
+
+	service.ReceiveIncomingMsg(rc, msg2)
+	assert.Equal(t, int32(5), atomic.LoadInt32(&count))
+
+	service.RemoveTopicListener(l3id)
+	service.ReceiveIncomingMsg(rc, msg1)
+	service.ReceiveIncomingMsg(rc, msg2)
+	assert.Equal(t, int32(5), atomic.LoadInt32(&count))
+
+	listeners = service.getTopicListeners("test")
+	assert.Empty(t, listeners)
+}