Redis support (#78)

Makefile

@@ -8,7 +8,8 @@ test: fmtcheck
 
 testint: fmtcheck deps
 	go test ./... -cover -race -tags=integration -count=1
-	docker stop badger
+	docker stop harvester-consul
+	docker stop harvester-redis
 
 
 cover: fmtcheck
@@ -29,15 +30,21 @@ deeplint: fmtcheck
 	docker run --env=GOFLAGS=-mod=vendor --rm -v $(CURDIR):/app -w /app golangci/golangci-lint:v1.28.1 golangci-lint run --exclude-use-default=false --enable-all -D dupl --build-tags integration
 
 deps:
-	docker container inspect badger > /dev/null 2>&1 || docker run -d --rm -p 8500:8500 -p 8600:8600/udp --name=badger consul:1.4.3 agent -server -ui -node=server-1 -bootstrap-expect=1 -client=0.0.0.0  -http-port 8500 -log-level=err
+	docker container inspect harvester-consul > /dev/null 2>&1 || docker run -d --rm -p 8500:8500 -p 8600:8600/udp --name=harvester-consul consul:1.4.3 agent -server -ui -node=server-1 -bootstrap-expect=1 -client=0.0.0.0  -http-port 8500 -log-level=err
+	docker container inspect harvester-redis > /dev/null 2>&1 || docker run -d --rm  -p 6379:6379 --name harvester-redis -e ALLOW_EMPTY_PASSWORD=yes bitnami/redis:6.2
+
+deps-stop:
+	docker stop harvester-consul
+	docker stop harvester-redis
 
 ci: fmtcheck lint deps
 	go test ./... -race -cover -tags=integration -coverprofile=coverage.txt -covermode=atomic
-	docker stop badger
+	docker stop harvester-consul
+	docker stop harvester-redis
 
 # disallow any parallelism (-j) for Make. This is necessary since some
 # commands during the build process create temporary files that collide
 # under parallel conditions.
 .NOTPARALLEL:
 
-.PHONY: default test testint cover fmt fmtcheck lint deeplint ci deps
+.PHONY: default test testint cover fmt fmtcheck lint deeplint ci deps deps-stop

README.md

@@ -24,6 +24,7 @@ type Config struct {
     Sandbox        sync.Bool            `seed:"true" env:"ENV_SANDBOX" consul:"/config/sandbox-mode"`
     AccessToken    sync.Secret          `seed:"defaultaccesstoken" env:"ENV_ACCESS_TOKEN" consul:"/config/access-token"`
     WorkDuration   sync.TimeDuration    `seed:"1s" env:"ENV_WORK_DURATION" consul:"/config/work-duration"`
+    OpeningBalance sync.Float64         `seed:"0.0" env:"ENV_OPENING_BALANCE" redis:"opening-balance"`
 }
 ```
 
@@ -32,8 +33,9 @@ The above defines the following fields:
 - IndexName, which will be seeded with the value `customers-v1`
 - CacheRetention, which will be seeded with the value `18`, and if exists, overridden with whatever value the env var `ENV_CACHE_RETENTION_SECONDS` holds
 - LogLevel, which will be seeded with the value `DEBUG`, and if exists, overridden with whatever value the flag `loglevel` holds
-- Sandbox, which will be seeded with the value `true`, and if exists, overridden with whatever value the env var `ENV_SANDBOX` holds and then from consul if the consul seeder and/or watcher are provided.
-- WorkDuration, which will be seeded with the value `1s`, and if exists, overridden with whatever value the env var `ENV_WORK_DURATION` holds and then from consul if the consul seeder and/or watcher are provided.
+- Sandbox, which will be seeded with the value `true`, and if exists, overridden with whatever value the env var `ENV_SANDBOX` holds and then from Consul if the consul seeder and/or watcher are provided.
+- WorkDuration, which will be seeded with the value `1s`, and if exists, overridden with whatever value the env var `ENV_WORK_DURATION` holds and then from Consul if the consul seeder and/or watcher are provided.
+- OpeningBalance, which will be seeded with the value `0.0`, and if exists, overridden with whatever value the env var `ENV_OPENING_BALANCE` holds and then from Redis if the redis seeder and/or watcher are provided.
 
 The fields have to be one of the types that the sync package supports in order to allow concurrent read and write to the fields. The following types are supported:
 
@@ -75,8 +77,8 @@ Seed and env tags are supported by default, the Consul getter has to be setup wh
 
 ## Monitoring phase (Consul only)
 
-- Monitor a key and apply if tag key matches
-- Monitor a key-prefix and apply if tag key matches
+- Monitor a key and apply if tag key matches (Consul and Redis)
+- Monitor a key-prefix and apply if tag key matches (Consul only)
 
 ### Monitor
 
@@ -92,15 +94,19 @@ The `Harvester` builder pattern is used to create a `Harvester` instance. The bu
 
 - Consul seed, for setting up seeding from Consul
 - Consul monitor, for setting up monitoring from Consul
+- Redis seed, for setting up seeding from Redis
+- Redis monitor, for setting up monitoring from Redis
 
 ```go
     h, err := New(&cfg).
                 WithConsulSeed("address", "dc", "token").
                 WithConsulMonitor("address", "dc", "token").
+                WithRedisSeed(redisClient).
+                WithRedisMonitor(redisClient, 10*time.Millisecond).
                 Create()
 ```
 
-The above snippet set's up a `Harvester` instance with consul seed and monitor.
+The above snippet set's up a `Harvester` instance with Consul and Redis seed and monitor.
 
 ## Notification support
 

config/config.go

@@ -19,13 +19,15 @@ const (
 	SourceEnv Source = "env"
 	// SourceConsul defines a value from consul.
 	SourceConsul Source = "consul"
+	// SourceRedis defines a value from redis.
+	SourceRedis Source = "redis"
 	// SourceFlag defines a value from CLI flag.
 	SourceFlag Source = "flag"
 	// SourceFile defines a value from external file.
 	SourceFile Source = "file"
 )
 
-var sourceTags = [...]Source{SourceSeed, SourceEnv, SourceConsul, SourceFlag, SourceFile}
+var sourceTags = [...]Source{SourceSeed, SourceEnv, SourceConsul, SourceRedis, SourceFlag, SourceFile}
 
 // CfgType represents an interface which any config field type must implement.
 type CfgType interface {

config/config_test.go

@@ -58,6 +58,7 @@ func TestNew(t *testing.T) {
 		"cfg duplicate consul key":        {args: args{cfg: &testDuplicateConfig{}}, wantErr: true},
 		"cfg tagged struct not supported": {args: args{cfg: &testInvalidNestedStructWithTags{}}, wantErr: true},
 		"cfg nested duplicate consul key": {args: args{cfg: &testDuplicateNestedConsulConfig{}}, wantErr: true},
+		"cfg nested duplicate redis key":  {args: args{cfg: &testDuplicateNestedRedisConfig{}}, wantErr: true},
 	}
 	for name, tt := range tests {
 		t.Run(name, func(t *testing.T) {
@@ -68,7 +69,7 @@ func TestNew(t *testing.T) {
 			} else {
 				assert.NoError(t, err)
 				assert.NotNil(t, got)
-				assert.Len(t, got.Fields, 6)
+				assert.Len(t, got.Fields, 7)
 				assertField(t, got.Fields[0], "Name", "String",
 					map[Source]string{SourceSeed: "John Doe", SourceEnv: "ENV_NAME"})
 				assertField(t, got.Fields[1], "Age", "Int64",
@@ -81,6 +82,8 @@ func TestNew(t *testing.T) {
 					map[Source]string{SourceSeed: "2000", SourceEnv: "ENV_SALARY"})
 				assertField(t, got.Fields[5], "LevelOneLevelTwoDeepField", "String",
 					map[Source]string{SourceSeed: "foobar"})
+				assertField(t, got.Fields[6], "IsAdult", "Bool",
+					map[Source]string{SourceSeed: "true", SourceEnv: "ENV_IS_ADULT", SourceRedis: "is-adult"})
 			}
 		})
 	}
@@ -122,12 +125,17 @@ func TestConfig_Set(t *testing.T) {
 	assert.NoError(t, err)
 	change = <-chNotify
 	assert.Equal(t, "field [LevelOneLevelTwoDeepField] of type [String] changed from [] to [baz]", change.String())
+	err = cfg.Fields[6].Set("true", 1)
+	assert.NoError(t, err)
+	change = <-chNotify
+	assert.Equal(t, "field [IsAdult] of type [Bool] changed from [false] to [true]", change.String())
 	assert.Equal(t, "John Doe", c.Name.Get())
 	assert.Equal(t, int64(18), c.Age.Get())
 	assert.Equal(t, 99.9, c.Balance.Get())
 	assert.Equal(t, true, c.HasJob.Get())
 	assert.Equal(t, int64(6000), c.Position.Salary.Get())
 	assert.Equal(t, "baz", c.LevelOne.LevelTwo.DeepField.Get())
+	assert.Equal(t, true, c.IsAdult.Get())
 }
 
 type testNestedConfig struct {
@@ -145,6 +153,7 @@ type testConfig struct {
 			DeepField sync.String `seed:"foobar"`
 		}
 	}
+	IsAdult sync.Bool `seed:"true" env:"ENV_IS_ADULT" redis:"is-adult"`
 }
 
 type testDuplicateNestedConsulConfig struct {
@@ -154,6 +163,13 @@ type testDuplicateNestedConsulConfig struct {
 	}
 }
 
+type testDuplicateNestedRedisConfig struct {
+	Age1   sync.Int64 `env:"ENV_AGE" redis:"age"`
+	Nested struct {
+		Age2 sync.Int64 `env:"ENV_AGE" redis:"age"`
+	}
+}
+
 type testInvalidTypeConfig struct {
 	Balance float32 `seed:"99.9" env:"ENV_BALANCE" consul:"/config/balance"`
 }

config/parser.go

@@ -71,7 +71,12 @@ func (p *parser) createField(prefix string, f reflect.StructField, val reflect.V
 			return nil, fmt.Errorf("duplicate value %v for source %s", fld, SourceConsul)
 		}
 	}
-
+	value, ok = fld.Sources()[SourceRedis]
+	if ok {
+		if p.isKeyValueDuplicate(SourceRedis, value) {
+			return nil, fmt.Errorf("duplicate value %v for source %s", fld, SourceRedis)
+		}
+	}
 	return fld, nil
 }
 

examples/07_redis/main.go

@@ -0,0 +1,79 @@
+package main
+
+import (
+	"context"
+	"log"
+	"os"
+	"time"
+
+	"github.com/beatlabs/harvester"
+	"github.com/beatlabs/harvester/sync"
+	"github.com/go-redis/redis/v8"
+)
+
+type config struct {
+	IndexName      sync.String  `seed:"customers-v1"`
+	CacheRetention sync.Int64   `seed:"43200" env:"ENV_CACHE_RETENTION_SECONDS"`
+	LogLevel       sync.String  `seed:"DEBUG" flag:"loglevel"`
+	OpeningBalance sync.Float64 `seed:"0.0" env:"ENV_CONSUL_VAR" redis:"opening-balance"`
+}
+
+var redisClient = redis.NewClient(&redis.Options{})
+
+func main() {
+	ctx, cnl := context.WithCancel(context.Background())
+	defer cnl()
+
+	err := os.Setenv("ENV_CACHE_RETENTION_SECONDS", "86400")
+	if err != nil {
+		log.Fatalf("failed to set env var: %v", err)
+	}
+
+	cfg := config{}
+
+	err = setBalance(ctx, "1000")
+	if err != nil {
+		log.Fatalf("failed to seed balance in redis: %v", err)
+	}
+
+	h, err := harvester.New(&cfg).WithRedisSeed(redisClient).WithRedisMonitor(redisClient, 200*time.Millisecond).Create()
+	if err != nil {
+		log.Fatalf("failed to create harvester: %v", err)
+	}
+
+	err = h.Harvest(ctx)
+	if err != nil {
+		log.Fatalf("failed to harvest configuration: %v", err)
+	}
+
+	log.Printf("Initial Config: IndexName: %s, CacheRetention: %d, LogLevel: %s, OpeningBalance: %f\n",
+		cfg.IndexName.Get(), cfg.CacheRetention.Get(), cfg.LogLevel.Get(), cfg.OpeningBalance.Get())
+
+	err = setBalance(ctx, "2000")
+	if err != nil {
+		log.Fatalf("failed to change balance in redis: %v", err)
+	}
+
+	time.Sleep(1 * time.Second)
+
+	log.Printf("Change balance. Config: IndexName: %s, CacheRetention: %d, LogLevel: %s, OpeningBalance: %f\n",
+		cfg.IndexName.Get(), cfg.CacheRetention.Get(), cfg.LogLevel.Get(), cfg.OpeningBalance.Get())
+
+	err = setBalance(ctx, "1000")
+	if err != nil {
+		log.Fatalf("failed to change balance in redis: %v", err)
+	}
+
+	time.Sleep(1 * time.Second)
+
+	log.Printf("Revert balance. Config: IndexName: %s, CacheRetention: %d, LogLevel: %s, OpeningBalance: %f\n",
+		cfg.IndexName.Get(), cfg.CacheRetention.Get(), cfg.LogLevel.Get(), cfg.OpeningBalance.Get())
+}
+
+func setBalance(ctx context.Context, amount string) error {
+	_, err := redisClient.Set(ctx, "opening-balance", amount, 0).Result()
+	if err != nil {
+		return err
+	}
+	return nil
+}

go.mod

@@ -3,7 +3,10 @@ module github.com/beatlabs/harvester
 go 1.15
 
 require (
+	github.com/go-redis/redis/v8 v8.8.0
 	github.com/hashicorp/consul/api v1.8.1
 	github.com/hashicorp/go-hclog v0.15.0
+	github.com/onsi/ginkgo v1.16.1 // indirect
+	github.com/onsi/gomega v1.11.0 // indirect
 	github.com/stretchr/testify v1.7.0
 )