WIP: add initial cases

Author: cyberhck

.gitignore

@@ -0,0 +1,3 @@
+vendor
+coverage.out
+.idea/

Gopkg.lock

@@ -0,0 +1,25 @@
+# Gopkg.toml example
+#
+# Refer to https://github.com/golang/dep/blob/master/docs/Gopkg.toml.md
+# for detailed Gopkg.toml documentation.
+#
+# required = ["github.com/user/thing/cmd/thing"]
+# ignored = ["github.com/user/project/pkgX", "bitbucket.org/user/project/pkgA/pkgY"]
+#
+# [[constraint]]
+#   name = "github.com/user/project"
+#   version = "1.0.0"
+#
+# [[constraint]]
+#   name = "github.com/user/project2"
+#   branch = "dev"
+#   source = "github.com/myfork/project2"
+#
+# [[override]]
+#  name = "github.com/x/y"
+#  version = "2.4.0"
+
+
+[[constraint]]
+  name = "github.com/stretchr/testify"
+  version = "1.2.0"

Gopkg.toml

@@ -0,0 +1,111 @@
+package job
+
+import (
+	"time"
+	"errors"
+	"sync"
+)
+
+type Config struct {
+	ResultProcessor            ResultProcessor
+	RuntimeProcessor           RuntimeProcessor
+	RuntimeProcessingFrequency time.Duration
+	LockProvider               LockProvider
+	Worker                     Worker
+}
+
+type ResultProcessor interface{}
+type RuntimeProcessor func (tick time.Time, message string, startTime time.Time) error
+type Worker func(Channel chan string, WaitGroup *sync.WaitGroup)
+
+type LockProvider interface {
+	Acquire() error
+	Release() error
+}
+
+func New() Config {
+	return Config{
+		LockProvider:               nil,
+		RuntimeProcessor:           nil,
+		ResultProcessor:            nil,
+		RuntimeProcessingFrequency: 200 * time.Millisecond,
+	}
+}
+
+func (config *Config) WithLockProvider(lockProvider LockProvider) {
+	config.LockProvider = lockProvider
+}
+
+func (config *Config) WithResultProcessor(processor ResultProcessor) {
+	config.ResultProcessor = processor
+}
+
+func (config *Config) WithRuntimeProcessingFrequency(frequency time.Duration) {
+	config.RuntimeProcessingFrequency = frequency
+}
+
+func (config *Config) WithRuntimeProcessor(processor RuntimeProcessor) {
+	config.RuntimeProcessor = processor
+}
+
+func (config *Config) SetWorker(worker Worker) {
+	config.Worker = worker
+}
+
+func (config *Config) Run() {
+	err := config.ensureLock()
+	if err != nil {
+		panic(err)
+	}
+	err = config.runWorker()
+	if err != nil {
+		panic(err)
+	}
+}
+
+func (config *Config) runWorker() error {
+	if config.Worker == nil {
+		return errors.New("worker not set")
+	}
+	startTime := time.Now()
+	ticker := time.NewTicker(config.RuntimeProcessingFrequency)
+	defer ticker.Stop()
+	commChan := make(chan string)
+	defer close(commChan)
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go config.Worker(commChan, &wg)
+	go config.invokeRuntimeProcessor(ticker, commChan, startTime)
+
+	wg.Wait()
+	return nil
+}
+
+func (config *Config) invokeRuntimeProcessor(ticker *time.Ticker, commChan chan string, startTime time.Time) {
+	for t := range ticker.C {
+		message := getMessage(commChan)
+		if config.RuntimeProcessor == nil {
+			return
+		}
+		err := config.RuntimeProcessor(t, message, startTime)
+		if err != nil {
+			panic(err)
+		}
+	}
+}
+
+func (config *Config) ensureLock() error {
+	if config.LockProvider == nil {
+		return nil
+	}
+	return config.LockProvider.Acquire()
+}
+
+func getMessage(ch chan string) string {
+	select {
+	case msg := <-ch:
+		return msg
+	default:
+		return ""
+	}
+}

job.go

@@ -0,0 +1,157 @@
+package job_test
+
+import (
+	"testing"
+	"github.com/cyberhck/jobs"
+	"time"
+	Assert "github.com/stretchr/testify/assert"
+	"errors"
+	"github.com/stretchr/testify/mock"
+	"sync"
+)
+
+type LockProviderMock struct {
+	mock.Mock
+}
+
+func (m *LockProviderMock) Acquire() error {
+	m.Called()
+	return nil
+}
+
+func (m *LockProviderMock) Release() error {
+	args := m.Called(nil)
+	return args.Error(1)
+}
+
+func TestNew(t *testing.T) {
+	assert := Assert.New(t)
+
+	testJob := job.New()
+
+	assert.Equal(testJob.RuntimeProcessingFrequency, 200*time.Millisecond)
+	assert.Nil(testJob.ResultProcessor)
+	assert.Nil(testJob.RuntimeProcessor)
+	assert.Nil(testJob.LockProvider)
+}
+
+type LockProvider struct {
+	acquire error
+	release error
+}
+
+func (r LockProvider) Acquire() error { return r.acquire }
+func (r LockProvider) Release() error { return r.release }
+
+func getMockedLockProvider(acquire error, release error) job.LockProvider {
+	return LockProvider{
+		acquire: acquire,
+		release: release,
+	}
+}
+func TestWithLockProvider(t *testing.T) {
+	testJob := job.New()
+	testJob.WithLockProvider(getMockedLockProvider(nil, nil))
+	Assert.NotNil(t, testJob.LockProvider)
+}
+
+func TestWithResultProcessor(t *testing.T) {
+	testJob := job.New()
+	testJob.WithResultProcessor(func() {})
+	Assert.NotNil(t, testJob.ResultProcessor)
+}
+
+func TestWithRuntimeProcessingFrequency(t *testing.T) {
+	testJob := job.New()
+	testJob.WithRuntimeProcessingFrequency(1 * time.Second)
+	if testJob.RuntimeProcessingFrequency != 1*time.Second {
+		t.Error("Should be able to change runtime processing frequency")
+	}
+}
+
+func TestWithRuntimeProcessor(t *testing.T) {
+	testJob := job.New()
+	testJob.WithRuntimeProcessor(func(tick time.Time, message string, startTime time.Time) error {
+		return nil
+	})
+	if testJob.RuntimeProcessor == nil {
+		t.Error("Should be able to set run time processor")
+	}
+}
+
+func TestRunPanicsIfLockNotAcquired(t *testing.T) {
+	assert := Assert.New(t)
+	testJob := job.New()
+	testJob.WithLockProvider(getMockedLockProvider(errors.New("couldn't acquire lock"), nil))
+	testJob.SetWorker(func(channel chan string, doneFunc *sync.WaitGroup) {})
+	assert.Panics(testJob.Run)
+}
+
+func TestRunDoesNotPanicIfLockAcquired(t *testing.T) {
+	assert := Assert.New(t)
+	testJob := job.New()
+	testJob.WithLockProvider(getMockedLockProvider(nil, nil))
+	testJob.SetWorker(func(channel chan string, doneFunc *sync.WaitGroup) {
+		doneFunc.Done()
+	})
+	assert.NotPanics(testJob.Run)
+}
+
+func TestRunPanicsIfNoWorkerIsDefined(t *testing.T) {
+	testJob := job.New()
+	mocked := new(LockProviderMock)
+	mocked.On("Acquire").Return(nil)
+	testJob.WithLockProvider(mocked)
+	Assert.Panics(t, testJob.Run)
+}
+
+func TestRunCallsWorker(t *testing.T) {
+	testJob := job.New()
+	mocked := new(LockProviderMock)
+	mocked.On("Acquire").Return(nil)
+	testJob.WithLockProvider(mocked)
+	testJob.SetWorker(func(channel chan string, doneFunc *sync.WaitGroup) {
+		doneFunc.Done()
+	})
+	testJob.Run()
+	mocked.AssertExpectations(t)
+}
+
+func TestRunWorksWithoutLockProvider(t *testing.T) {
+	testJob := job.New()
+	testJob.SetWorker(func(channel chan string, doneFunc *sync.WaitGroup) {
+		doneFunc.Done()
+	})
+	Assert.NotPanics(t, testJob.Run)
+}
+
+func TestDoesNotPanicWhenNoRuntimeProcessorPresent(t *testing.T) {
+	testJob := job.New()
+	testJob.SetWorker(func(channel chan string, doneFunc *sync.WaitGroup) {
+		doneFunc.Done()
+	})
+	Assert.NotPanics(t, testJob.Run)
+}
+func TestRunTimeProcessorGetsCalled(t *testing.T) {
+	testJob := job.New()
+	runtimeProcessorCalled := false
+	testJob.WithRuntimeProcessor(func(tick time.Time, message string, startTime time.Time) error {
+		runtimeProcessorCalled = true
+		return nil
+	})
+	testJob.SetWorker(func(channel chan string, doneFunc *sync.WaitGroup) {
+		time.Sleep(1 * time.Second)
+		doneFunc.Done()
+	})
+	testJob.Run()
+	Assert.True(t, runtimeProcessorCalled)
+}
+func TestLongRunningProcessorWorksWithoutRuntimeProcessor(t *testing.T) {
+	testJob := job.New()
+	testJob.SetWorker(func(channel chan string, doneFunc *sync.WaitGroup) {
+		time.Sleep(10 * time.Millisecond)
+		channel <- "Done..."
+		doneFunc.Done()
+	})
+	Assert.NotPanics(t, testJob.Run)
+}