cluster: support LockWithContext

cluster/job.go

@@ -18,7 +18,7 @@ const (
 // JobPluginAPI is the plugin API interface required to schedule jobs.
 type JobPluginAPI interface {
 	MutexPluginAPI
-	KVGet(key string) ([]byte, *model.AppError)
+	KVSet(key string, value []byte) *model.AppError
 }
 
 // JobConfig defines the configuration of a scheduled job.
@@ -102,8 +102,8 @@ func (j *Job) saveMetadata(metadata jobMetadata) error {
 		return errors.Wrap(err, "failed to marshal data")
 	}
 
-	ok, appErr := j.pluginAPI.KVSetWithOptions(j.key, data, model.PluginKVSetOptions{})
-	if appErr != nil || !ok {
+	appErr := j.pluginAPI.KVSet(j.key, data)
+	if appErr != nil {
 		return errors.Wrap(appErr, "failed to set data")
 	}
 

cluster/job_example_test.go

@@ -1,10 +1,14 @@
 package cluster
 
-import "time"
+import (
+	"time"
+
+	"github.com/mattermost/mattermost-server/v5/plugin"
+)
 
 func ExampleSchedule() {
 	// Use p.API from your plugin instead.
-	pluginAPI := NewMockMutexPluginAPI(nil)
+	pluginAPI := plugin.API(nil)
 
 	callback := func() {
 		// periodic work to do

cluster/job_test.go

@@ -6,38 +6,39 @@ import (
 	"testing"
 	"time"
 
+	"github.com/mattermost/mattermost-server/v5/model"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
-type MockJobPluginAPI struct {
-	*MockMutexPluginAPI
-}
+func TestSchedule(t *testing.T) {
+	t.Parallel()
 
-func NewMockJobPluginAPI(t *testing.T) *MockJobPluginAPI {
-	return &MockJobPluginAPI{
-		MockMutexPluginAPI: NewMockMutexPluginAPI(t),
+	makeKey := func() string {
+		return model.NewId()
 	}
-}
 
-func TestSchedule(t *testing.T) {
 	t.Run("invalid interval", func(t *testing.T) {
-		mockPluginAPI := NewMockJobPluginAPI(t)
+		t.Parallel()
+
+		mockPluginAPI := newMockPluginAPI(t)
 
-		job, err := Schedule(mockPluginAPI, "key", JobConfig{}, func() {})
+		job, err := Schedule(mockPluginAPI, makeKey(), JobConfig{}, func() {})
 		require.Error(t, err, "must specify non-zero job config interval")
 		require.Nil(t, job)
 	})
 
 	t.Run("single-threaded", func(t *testing.T) {
-		mockPluginAPI := NewMockJobPluginAPI(t)
+		t.Parallel()
+
+		mockPluginAPI := newMockPluginAPI(t)
 
 		count := new(int32)
 		callback := func() {
 			atomic.AddInt32(count, 1)
 		}
 
-		job, err := Schedule(mockPluginAPI, "key", JobConfig{Interval: 100 * time.Millisecond}, callback)
+		job, err := Schedule(mockPluginAPI, makeKey(), JobConfig{Interval: 100 * time.Millisecond}, callback)
 		require.NoError(t, err)
 		require.NotNil(t, job)
 
@@ -56,7 +57,9 @@ func TestSchedule(t *testing.T) {
 	})
 
 	t.Run("multi-threaded, single job", func(t *testing.T) {
-		mockPluginAPI := NewMockJobPluginAPI(t)
+		t.Parallel()
+
+		mockPluginAPI := newMockPluginAPI(t)
 
 		count := new(int32)
 		callback := func() {
@@ -65,8 +68,10 @@ func TestSchedule(t *testing.T) {
 
 		var jobs []*Job
 
+		key := makeKey()
+
 		for i := 0; i < 3; i++ {
-			job, err := Schedule(mockPluginAPI, "key", JobConfig{Interval: 100 * time.Millisecond}, callback)
+			job, err := Schedule(mockPluginAPI, key, JobConfig{Interval: 100 * time.Millisecond}, callback)
 			require.NoError(t, err)
 			require.NotNil(t, job)
 
@@ -97,7 +102,9 @@ func TestSchedule(t *testing.T) {
 	})
 
 	t.Run("multi-threaded, multiple jobs", func(t *testing.T) {
-		mockPluginAPI := NewMockJobPluginAPI(t)
+		t.Parallel()
+
+		mockPluginAPI := newMockPluginAPI(t)
 
 		countA := new(int32)
 		callbackA := func() {
@@ -109,15 +116,18 @@ func TestSchedule(t *testing.T) {
 			atomic.AddInt32(countB, 1)
 		}
 
+		keyA := makeKey()
+		keyB := makeKey()
+
 		var jobs []*Job
 		for i := 0; i < 3; i++ {
 			var key string
 			var callback func()
 			if i <= 1 {
-				key = "keyA"
+				key = keyA
 				callback = callbackA
 			} else {
-				key = "keyB"
+				key = keyB
 				callback = callbackB
 			}
 

cluster/mock_plugin_api_test.go

@@ -0,0 +1,115 @@
+package cluster
+
+import (
+	"bytes"
+	"sync"
+	"testing"
+
+	"github.com/mattermost/mattermost-server/v5/model"
+)
+
+type mockPluginAPI struct {
+	t *testing.T
+
+	lock      sync.Mutex
+	keyValues map[string][]byte
+	failing   bool
+}
+
+func newMockPluginAPI(t *testing.T) *mockPluginAPI {
+	return &mockPluginAPI{
+		t:         t,
+		keyValues: make(map[string][]byte),
+	}
+}
+
+func (pluginAPI *mockPluginAPI) setFailing(failing bool) {
+	pluginAPI.lock.Lock()
+	defer pluginAPI.lock.Unlock()
+
+	pluginAPI.failing = failing
+}
+
+func (pluginAPI *mockPluginAPI) KVGet(key string) ([]byte, *model.AppError) {
+	pluginAPI.lock.Lock()
+	defer pluginAPI.lock.Unlock()
+
+	if pluginAPI.failing {
+		return nil, &model.AppError{Message: "fake error"}
+	}
+
+	return pluginAPI.keyValues[key], nil
+}
+
+func (pluginAPI *mockPluginAPI) KVSet(key string, value []byte) *model.AppError {
+	pluginAPI.lock.Lock()
+	defer pluginAPI.lock.Unlock()
+
+	if pluginAPI.failing {
+		return &model.AppError{Message: "fake error"}
+	}
+
+	pluginAPI.keyValues[key] = value
+
+	return nil
+}
+
+func (pluginAPI *mockPluginAPI) KVDelete(key string) *model.AppError {
+	pluginAPI.lock.Lock()
+	defer pluginAPI.lock.Unlock()
+
+	if pluginAPI.failing {
+		return &model.AppError{Message: "fake error"}
+	}
+
+	delete(pluginAPI.keyValues, key)
+
+	return nil
+}
+
+func (pluginAPI *mockPluginAPI) KVCompareAndSet(key string, oldValue []byte, value []byte) (bool, *model.AppError) {
+	pluginAPI.lock.Lock()
+	defer pluginAPI.lock.Unlock()
+
+	if pluginAPI.failing {
+		return false, &model.AppError{Message: "fake error"}
+	}
+
+	if actualValue := pluginAPI.keyValues[key]; !bytes.Equal(actualValue, oldValue) {
+		return false, nil
+	}
+
+	pluginAPI.keyValues[key] = value
+
+	return true, nil
+}
+
+func (pluginAPI *mockPluginAPI) KVCompareAndDelete(key string, oldValue []byte) (bool, *model.AppError) {
+	pluginAPI.lock.Lock()
+	defer pluginAPI.lock.Unlock()
+
+	if pluginAPI.failing {
+		return false, &model.AppError{Message: "fake error"}
+	}
+
+	if actualValue := pluginAPI.keyValues[key]; !bytes.Equal(actualValue, oldValue) {
+		return false, nil
+	}
+
+	delete(pluginAPI.keyValues, key)
+
+	return true, nil
+}
+
+func (pluginAPI *mockPluginAPI) LogError(msg string, keyValuePairs ...interface{}) {
+	if pluginAPI.t == nil {
+		return
+	}
+
+	pluginAPI.t.Helper()
+
+	params := []interface{}{msg}
+	params = append(params, keyValuePairs...)
+
+	pluginAPI.t.Log(params...)
+}