Refactor + some fixes and testing (#5)

* Refactor and some fixes to collector.

Signed-off-by: Artur Souza <asouza.pro@gmail.com>

* Refactor rhythm too.

Signed-off-by: Artur Souza <asouza.pro@gmail.com>

* Update to Go 1.21

Signed-off-by: Artur Souza <asouza.pro@gmail.com>

* Update License for Diagrid.

Signed-off-by: Artur Souza <asouza.pro@gmail.com>

* Test and fix for collector.

Signed-off-by: Artur Souza <asouza.pro@gmail.com>

* Use Collector as interface.

Signed-off-by: Artur Souza <asouza.pro@gmail.com>

---------

Signed-off-by: Artur Souza <asouza.pro@gmail.com>

.github/workflows/build.yml

@@ -42,4 +42,4 @@ jobs:
           git diff --exit-code ./go.mod # check no changes
           git diff --exit-code ./go.sum # check no changes
       - name: Run test
-        run: go test -v -timeout 300s --race
+        run: make test

LICENSE

@@ -1,4 +1,4 @@
-Copyright (C) 2012 Rob Figueiredo
+Copyright (C) 2024 Diagrid Inc.
 All Rights Reserved.
 
 MIT LICENSE

LICENSE-robfig

@@ -0,0 +1,21 @@
+Copyright (C) 2012 Rob Figueiredo
+All Rights Reserved.
+
+MIT LICENSE
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of
+this software and associated documentation files (the "Software"), to deal in
+the Software without restriction, including without limitation the rights to
+use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+the Software, and to permit persons to whom the Software is furnished to do so,
+subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

Makefile

@@ -34,12 +34,15 @@ check-proto-version: ## Checking the version of proto related tools
 define genProtoc
 .PHONY: gen-proto-$(1)
 gen-proto-$(1):
-	$(PROTOC) --go_out=. --go_opt=module=$(PROTO_PREFIX) --go-grpc_out=. --go-grpc_opt=require_unimplemented_servers=false,module=$(PROTO_PREFIX) ./proto/$(1)
+	$(PROTOC) --go_out=. --go_opt=module=$(PROTO_PREFIX) --go-grpc_out=. --go-grpc_opt=require_unimplemented_servers=false,module=$(PROTO_PREFIX) ./proto/$(1)/*
 endef
 
 $(foreach ITEM,$(PROTOS),$(eval $(call genProtoc,$(ITEM))))
 
 GEN_PROTOS:=$(foreach ITEM,$(PROTOS),gen-proto-$(ITEM))
 
 .PHONY: gen-proto
-gen-proto: check-proto-version $(GEN_PROTOS) modtidy
+gen-proto: check-proto-version $(GEN_PROTOS) modtidy
+
+test:
+	go test -timeout 300s --race ./...

README.md

@@ -24,7 +24,10 @@ This package aims at implementing a distributed and fault tolerant cron in order
 By default the library creates an etcd client on `127.0.0.1:2379`
 
 ```go
-c, _ := etcdcron.NewEtcdMutexBuilder(clientv3.Config{
+import etcdclientv3 "go.etcd.io/etcd/client/v3"
+import etcdcron "github.com/diagridio/go-etcd-cron"
+
+c, _ := etcdclientv3.New(etcdclientv3.Config{
   Endpoints: []string{"etcd-host1:2379", "etcd-host2:2379"},
 })
 cron, _ := etcdcron.New(
@@ -79,11 +82,15 @@ Pre-requisites to run the tests locally:
   - Docker: [Running a single node etcd](https://etcd.io/docs/v3.5/op-guide/container/#running-a-single-node-etcd-1)
 
 ```bash
-go test -v --race
+make test
+```
+OR
+```bash
+go test -timeout 300s --race ./...
 ```
 
 ## History
 
 This is a fork of [https://github.com/Scalingo/go-etcd-cron](https://github.com/Scalingo/go-etcd-cron), which had been based on [https://github.com/robfig/cron](https://github.com/robfig/cron).
 
-This fork has similar but still different goals from Scalingo's go-etcd-cron library.
+This fork has different goals from Scalingo's go-etcd-cron library.

collector.go → collector/collector.go

@@ -3,18 +3,24 @@ Copyright (c) 2024 Diagrid Inc.
 Licensed under the MIT License.
 */
 
-package etcdcron
+package collector
 
 import (
 	"context"
 	"sync"
 	"time"
 )
 
-// collector garbage collects items after a globally configured TTL.
-type Collector struct {
-	ttl        int64 // time to wait to perform collection
-	bufferTime int64 // arbitrary delay to allow buffering of operations
+// Collector garbage collects items after a globally configured TTL.
+type Collector interface {
+	Start(ctx context.Context)
+	Add(func(ctx context.Context))
+	Wait()
+}
+
+type collector struct {
+	ttl        time.Duration // time to wait to perform collection
+	bufferTime time.Duration // arbitrary delay to allow buffering of operations
 
 	running         bool
 	mutex           sync.RWMutex
@@ -24,12 +30,12 @@ type Collector struct {
 }
 
 type collectorEntry struct {
-	expiration int64
+	expiration time.Time
 	op         func(ctx context.Context)
 }
 
-func NewCollector(ttl int64, bufferTime int64) *Collector {
-	return &Collector{
+func New(ttl time.Duration, bufferTime time.Duration) Collector {
+	return &collector{
 		ttl:        ttl,
 		bufferTime: bufferTime,
 		running:    false,
@@ -38,7 +44,7 @@ func NewCollector(ttl int64, bufferTime int64) *Collector {
 	}
 }
 
-func (c *Collector) Start(ctx context.Context) {
+func (c *collector) Start(ctx context.Context) {
 	if c.running {
 		return
 	}
@@ -49,10 +55,10 @@ func (c *Collector) Start(ctx context.Context) {
 		c.mutex.Lock()
 		defer c.mutex.Unlock()
 
-		now := time.Now().Unix()
+		now := time.Now()
 		nextStartIndex := -1
 		for i, o := range c.operations {
-			if o.expiration <= now {
+			if o.expiration.Before(now) {
 				o.op(ctx)
 			} else {
 				nextStartIndex = i
@@ -71,14 +77,14 @@ func (c *Collector) Start(ctx context.Context) {
 		c.mutex.RLock()
 		defer c.mutex.RUnlock()
 
-		now := time.Now().Unix()
+		now := time.Now()
 		if len(c.operations) > 0 {
-			diff := c.operations[0].expiration - now
-			if diff <= 0 {
+			op := c.operations[0]
+			if op.expiration.Before(now) {
 				return 0
 			}
 
-			return time.Duration(diff)
+			return now.Sub(op.expiration)
 		}
 
 		// Some arbitrarily large number that gives us certainty that some record will be added.
@@ -102,14 +108,15 @@ func (c *Collector) Start(ctx context.Context) {
 	}(ctx)
 }
 
-func (c *Collector) Add(op func(ctx context.Context)) {
+func (c *collector) Add(op func(ctx context.Context)) {
 	c.mutex.Lock()
 	defer c.mutex.Unlock()
 	c.operations = append(c.operations, &collectorEntry{
-		expiration: time.Now().Unix() + c.ttl,
+		expiration: time.Now().Add(c.ttl),
+		op:         op,
 	})
 }
 
-func (c *Collector) Wait() {
+func (c *collector) Wait() {
 	c.runWaitingGroup.Wait()
 }

collector/collector_test.go

@@ -0,0 +1,133 @@
+/*
+Copyright (c) 2024 Diagrid Inc.
+Licensed under the MIT License.
+*/
+
+package collector
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNoEntries(t *testing.T) {
+	collector := New(time.Second, time.Second)
+	ctx, cancel := context.WithCancel(context.Background())
+	collector.Start(ctx)
+
+	select {
+	case <-time.After(3 * time.Second):
+		t.FailNow()
+	case <-stop(collector, cancel):
+	}
+}
+
+func TestAddEntryAfterStart(t *testing.T) {
+	collector := New(time.Second, time.Second)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	collector.Start(ctx)
+
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+
+	collector.Add(func(ctx context.Context) {
+		wg.Done()
+	})
+
+	select {
+	case <-time.After(2 * time.Second):
+		t.FailNow()
+	case <-wait(wg):
+	}
+}
+
+func TestAddEntryBeforeStart(t *testing.T) {
+	collector := New(time.Second, time.Second)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+
+	collector.Add(func(ctx context.Context) {
+		wg.Done()
+	})
+
+	collector.Start(ctx)
+
+	select {
+	case <-time.After(2 * time.Second):
+		t.FailNow()
+	case <-wait(wg):
+	}
+}
+
+func TestExpireOnlyOneOutOfTwo(t *testing.T) {
+	collector := New(time.Second, time.Second)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	wg1 := &sync.WaitGroup{}
+	wg1.Add(1)
+
+	wg2 := &sync.WaitGroup{}
+	wg2.Add(1)
+
+	called1 := atomic.Bool{}
+	called2 := atomic.Bool{}
+
+	collector.Add(func(ctx context.Context) {
+		called1.Store(true)
+		wg1.Done()
+	})
+
+	time.Sleep(3 * time.Second)
+
+	collector.Add(func(ctx context.Context) {
+		called2.Store(true)
+		wg2.Done()
+	})
+
+	collector.Start(ctx)
+
+	select {
+	case <-time.After(2 * time.Second):
+		t.FailNow()
+	case <-wait(wg1):
+		assert.True(t, called1.Load())
+		assert.False(t, called2.Load())
+	}
+
+	select {
+	case <-time.After(2 * time.Second):
+		t.FailNow()
+	case <-wait(wg2):
+		assert.True(t, called1.Load())
+		assert.True(t, called2.Load())
+	}
+}
+
+func wait(wg *sync.WaitGroup) chan bool {
+	ch := make(chan bool)
+	go func() {
+		wg.Wait()
+		ch <- true
+	}()
+	return ch
+}
+
+func stop(collector Collector, cancel context.CancelFunc) chan bool {
+	ch := make(chan bool)
+	go func() {
+		cancel()
+		collector.Wait()
+		ch <- true
+	}()
+	return ch
+}