Repository: numaproj/numaflow
Pull request #167 (2022-09-09 15:19:00)

chore: keep stores together
Signed-off-by: Vigith Maurice <vigith@gmail.com>

This is to clean up the `isb` module before we can start refactoring `isb.Message` to have an interface. 


Patch:
diff --git a/pkg/isb/simplebuffer/buffer.go b/pkg/isb/stores/simplebuffer/buffer.go
similarity index 100%
rename from pkg/isb/simplebuffer/buffer.go
rename to pkg/isb/stores/simplebuffer/buffer.go
diff --git a/pkg/isb/simplebuffer/buffer_test.go b/pkg/isb/stores/simplebuffer/buffer_test.go
similarity index 100%
rename from pkg/isb/simplebuffer/buffer_test.go
rename to pkg/isb/stores/simplebuffer/buffer_test.go
diff --git a/pkg/isbsvc/redis_service.go b/pkg/isbsvc/redis_service.go
index 61dcd586..7391bc8b 100644
--- a/pkg/isbsvc/redis_service.go
+++ b/pkg/isbsvc/redis_service.go
@@ -4,10 +4,10 @@ import (
 	"context"
 	"fmt"
 
+	redis2 "github.com/numaproj/numaflow/pkg/isb/stores/redis"
 	"go.uber.org/zap"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
-	"github.com/numaproj/numaflow/pkg/isb/redis"
 	redisclient "github.com/numaproj/numaflow/pkg/shared/clients/redis"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 )
@@ -111,8 +111,8 @@ func (r *isbsRedisSvc) GetBufferInfo(ctx context.Context, buffer dfv1.Buffer) (*
 		return nil, fmt.Errorf("buffer infomation inquiry is not supported for type %q", buffer.Type)
 	}
 	group := fmt.Sprintf("%s-group", buffer.Name)
-	rqw := redis.NewBufferWrite(ctx, redisclient.NewInClusterRedisClient(), buffer.Name, group, redis.WithRefreshBufferWriteInfo(false))
-	var bufferWrite = rqw.(*redis.BufferWrite)
+	rqw := redis2.NewBufferWrite(ctx, redisclient.NewInClusterRedisClient(), buffer.Name, group, redis2.WithRefreshBufferWriteInfo(false))
+	var bufferWrite = rqw.(*redis2.BufferWrite)
 
 	bufferInfo := &BufferInfo{
 		Name:            buffer.Name,
diff --git a/pkg/sinks/sink.go b/pkg/sinks/sink.go
index b2ad65fc..a786fdf1 100644
--- a/pkg/sinks/sink.go
+++ b/pkg/sinks/sink.go
@@ -5,15 +5,12 @@ import (
 	"fmt"
 	"sync"
 
-	"github.com/numaproj/numaflow/pkg/watermark/generic"
-	"github.com/numaproj/numaflow/pkg/watermark/generic/jetstream"
-
 	"go.uber.org/zap"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
-	jetstreamisb "github.com/numaproj/numaflow/pkg/isb/jetstream"
-	redisisb "github.com/numaproj/numaflow/pkg/isb/redis"
+	jetstreamisb "github.com/numaproj/numaflow/pkg/isb/stores/jetstream"
+	redisisb "github.com/numaproj/numaflow/pkg/isb/stores/redis"
 	"github.com/numaproj/numaflow/pkg/metrics"
 	jsclient "github.com/numaproj/numaflow/pkg/shared/clients/jetstream"
 	redisclient "github.com/numaproj/numaflow/pkg/shared/clients/redis"
@@ -22,6 +19,8 @@ import (
 	logsink "github.com/numaproj/numaflow/pkg/sinks/logger"
 	udsink "github.com/numaproj/numaflow/pkg/sinks/udsink"
 	"github.com/numaproj/numaflow/pkg/watermark/fetch"
+	"github.com/numaproj/numaflow/pkg/watermark/generic"
+	"github.com/numaproj/numaflow/pkg/watermark/generic/jetstream"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
 )
 
diff --git a/pkg/sources/source.go b/pkg/sources/source.go
index e275a6a9..19504124 100644
--- a/pkg/sources/source.go
+++ b/pkg/sources/source.go
@@ -5,14 +5,12 @@ import (
 	"fmt"
 	"sync"
 
-	"github.com/numaproj/numaflow/pkg/watermark/generic/jetstream"
-
 	"go.uber.org/zap"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
-	jetstreamisb "github.com/numaproj/numaflow/pkg/isb/jetstream"
-	redisisb "github.com/numaproj/numaflow/pkg/isb/redis"
+	jetstreamisb "github.com/numaproj/numaflow/pkg/isb/stores/jetstream"
+	redisisb "github.com/numaproj/numaflow/pkg/isb/stores/redis"
 	"github.com/numaproj/numaflow/pkg/metrics"
 	jsclient "github.com/numaproj/numaflow/pkg/shared/clients/jetstream"
 	redisclient "github.com/numaproj/numaflow/pkg/shared/clients/redis"
@@ -22,6 +20,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/sources/kafka"
 	"github.com/numaproj/numaflow/pkg/watermark/fetch"
 	"github.com/numaproj/numaflow/pkg/watermark/generic"
+	"github.com/numaproj/numaflow/pkg/watermark/generic/jetstream"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
 	"github.com/numaproj/numaflow/pkg/watermark/store/noop"
 )
diff --git a/pkg/udf/udf.go b/pkg/udf/udf.go
index e298bfe1..674aa1f7 100644
--- a/pkg/udf/udf.go
+++ b/pkg/udf/udf.go
@@ -6,14 +6,13 @@ import (
 	"sync"
 	"time"
 
-	"github.com/numaproj/numaflow/pkg/watermark/generic/jetstream"
 	"go.uber.org/zap"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/forward"
-	jetstreamisb "github.com/numaproj/numaflow/pkg/isb/jetstream"
-	redisisb "github.com/numaproj/numaflow/pkg/isb/redis"
+	jetstreamisb "github.com/numaproj/numaflow/pkg/isb/stores/jetstream"
+	redisisb "github.com/numaproj/numaflow/pkg/isb/stores/redis"
 	"github.com/numaproj/numaflow/pkg/metrics"
 	jsclient "github.com/numaproj/numaflow/pkg/shared/clients/jetstream"
 	redisclient "github.com/numaproj/numaflow/pkg/shared/clients/redis"
@@ -21,6 +20,7 @@ import (
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
 	"github.com/numaproj/numaflow/pkg/udf/applier"
 	"github.com/numaproj/numaflow/pkg/watermark/fetch"
+	"github.com/numaproj/numaflow/pkg/watermark/generic/jetstream"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
 )
 


Test patch:
diff --git a/pkg/isb/forward/forward_test.go b/pkg/isb/forward/forward_test.go
index c75cc1f7..28a8f3ba 100644
--- a/pkg/isb/forward/forward_test.go
+++ b/pkg/isb/forward/forward_test.go
@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/numaproj/numaflow/pkg/isb/stores/simplebuffer"
 	"github.com/numaproj/numaflow/pkg/watermark/generic"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
@@ -15,7 +16,6 @@ import (
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 
 	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/isb/simplebuffer"
 	"github.com/numaproj/numaflow/pkg/isb/testutils"
 	udfapplier "github.com/numaproj/numaflow/pkg/udf/applier"
 	"github.com/stretchr/testify/assert"
diff --git a/pkg/isb/forward/shutdown_test.go b/pkg/isb/forward/shutdown_test.go
index c69f042f..21024c68 100644
--- a/pkg/isb/forward/shutdown_test.go
+++ b/pkg/isb/forward/shutdown_test.go
@@ -5,12 +5,12 @@ import (
 	"testing"
 	"time"
 
+	"github.com/numaproj/numaflow/pkg/isb/stores/simplebuffer"
 	"github.com/numaproj/numaflow/pkg/watermark/generic"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 
 	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/isb/simplebuffer"
 	"github.com/numaproj/numaflow/pkg/isb/testutils"
 	"github.com/stretchr/testify/assert"
 )
diff --git a/pkg/isb/format.go b/pkg/isb/message.go
similarity index 100%
rename from pkg/isb/format.go
rename to pkg/isb/message.go
diff --git a/pkg/isb/jetstream/metrics.go b/pkg/isb/stores/jetstream/metrics.go
similarity index 100%
rename from pkg/isb/jetstream/metrics.go
rename to pkg/isb/stores/jetstream/metrics.go
diff --git a/pkg/isb/jetstream/options.go b/pkg/isb/stores/jetstream/options.go
similarity index 100%
rename from pkg/isb/jetstream/options.go
rename to pkg/isb/stores/jetstream/options.go
diff --git a/pkg/isb/jetstream/reader.go b/pkg/isb/stores/jetstream/reader.go
similarity index 100%
rename from pkg/isb/jetstream/reader.go
rename to pkg/isb/stores/jetstream/reader.go
diff --git a/pkg/isb/jetstream/reader_test.go b/pkg/isb/stores/jetstream/reader_test.go
similarity index 100%
rename from pkg/isb/jetstream/reader_test.go
rename to pkg/isb/stores/jetstream/reader_test.go
diff --git a/pkg/isb/jetstream/types.go b/pkg/isb/stores/jetstream/types.go
similarity index 100%
rename from pkg/isb/jetstream/types.go
rename to pkg/isb/stores/jetstream/types.go
diff --git a/pkg/isb/jetstream/writer.go b/pkg/isb/stores/jetstream/writer.go
similarity index 100%
rename from pkg/isb/jetstream/writer.go
rename to pkg/isb/stores/jetstream/writer.go
diff --git a/pkg/isb/jetstream/writer_test.go b/pkg/isb/stores/jetstream/writer_test.go
similarity index 100%
rename from pkg/isb/jetstream/writer_test.go
rename to pkg/isb/stores/jetstream/writer_test.go
diff --git a/pkg/isb/redis/exactlyOnceInsert.lua b/pkg/isb/stores/redis/exactlyOnceInsert.lua
similarity index 100%
rename from pkg/isb/redis/exactlyOnceInsert.lua
rename to pkg/isb/stores/redis/exactlyOnceInsert.lua
diff --git a/pkg/isb/redis/metrics.go b/pkg/isb/stores/redis/metrics.go
similarity index 100%
rename from pkg/isb/redis/metrics.go
rename to pkg/isb/stores/redis/metrics.go
diff --git a/pkg/isb/redis/options.go b/pkg/isb/stores/redis/options.go
similarity index 100%
rename from pkg/isb/redis/options.go
rename to pkg/isb/stores/redis/options.go
diff --git a/pkg/isb/redis/read.go b/pkg/isb/stores/redis/read.go
similarity index 100%
rename from pkg/isb/redis/read.go
rename to pkg/isb/stores/redis/read.go
diff --git a/pkg/isb/redis/read_test.go b/pkg/isb/stores/redis/read_test.go
similarity index 100%
rename from pkg/isb/redis/read_test.go
rename to pkg/isb/stores/redis/read_test.go
diff --git a/pkg/isb/redis/write.go b/pkg/isb/stores/redis/write.go
similarity index 100%
rename from pkg/isb/redis/write.go
rename to pkg/isb/stores/redis/write.go
diff --git a/pkg/isb/redis/write_info.go b/pkg/isb/stores/redis/write_info.go
similarity index 100%
rename from pkg/isb/redis/write_info.go
rename to pkg/isb/stores/redis/write_info.go
diff --git a/pkg/isb/redis/write_test.go b/pkg/isb/stores/redis/write_test.go
similarity index 99%
rename from pkg/isb/redis/write_test.go
rename to pkg/isb/stores/redis/write_test.go
index f552a12f..279b0514 100644
--- a/pkg/isb/redis/write_test.go
+++ b/pkg/isb/stores/redis/write_test.go
@@ -191,14 +191,14 @@ func TestLua(t *testing.T) {
 
 	// first insert
 	res, err := script.Run(ctx, client, []string{hashName, streamName}, "10", message.Header, message.Payload, "0-0").Result()
-	assert.NoErrorf(t, err, "lua script execution failed, %s")
+	assert.NoErrorf(t, err, "lua script execution failed, %s", err)
 	id, err := splitId(res.(string))
 	assert.NoError(t, err)
 	assert.Positive(t, id, res)
 
 	// duplicate insert
 	res, err = script.Run(ctx, client, []string{hashName, streamName}, "10", message.Header, message.Payload, "0-0").Result()
-	assert.NoErrorf(t, err, "lua script execution failed, %s")
+	assert.NoErrorf(t, err, "lua script execution failed, %s", err)
 	id, err = splitId(res.(string))
 	assert.NoError(t, err)
 	assert.Positive(t, id, res)
diff --git a/pkg/isbsvc/redis_service_test.go b/pkg/isbsvc/redis_service_test.go
index f60b3166..e40b1270 100644
--- a/pkg/isbsvc/redis_service_test.go
+++ b/pkg/isbsvc/redis_service_test.go
@@ -8,10 +8,10 @@ import (
 	"time"
 
 	goredis "github.com/go-redis/redis/v8"
+	"github.com/numaproj/numaflow/pkg/isb/stores/redis"
 	"github.com/stretchr/testify/assert"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
-	"github.com/numaproj/numaflow/pkg/isb/redis"
 	"github.com/numaproj/numaflow/pkg/isb/testutils"
 	redisclient "github.com/numaproj/numaflow/pkg/shared/clients/redis"
 )
diff --git a/pkg/sinks/kafka/kafka_test.go b/pkg/sinks/kafka/kafka_test.go
index 60fd741a..58e62abd 100644
--- a/pkg/sinks/kafka/kafka_test.go
+++ b/pkg/sinks/kafka/kafka_test.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/numaproj/numaflow/pkg/isb/stores/simplebuffer"
 	"github.com/numaproj/numaflow/pkg/watermark/generic"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
@@ -14,7 +15,6 @@ import (
 
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/forward"
-	"github.com/numaproj/numaflow/pkg/isb/simplebuffer"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	"github.com/numaproj/numaflow/pkg/udf/applier"
 )
diff --git a/pkg/sinks/logger/log_test.go b/pkg/sinks/logger/log_test.go
index a4420560..73218e2d 100644
--- a/pkg/sinks/logger/log_test.go
+++ b/pkg/sinks/logger/log_test.go
@@ -2,14 +2,15 @@ package logger
 
 import (
 	"context"
-	"github.com/numaproj/numaflow/pkg/watermark/generic"
 	"testing"
 	"time"
 
+	"github.com/numaproj/numaflow/pkg/isb/stores/simplebuffer"
+	"github.com/numaproj/numaflow/pkg/watermark/generic"
+
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/forward"
-	"github.com/numaproj/numaflow/pkg/isb/simplebuffer"
 	"github.com/numaproj/numaflow/pkg/isb/testutils"
 	"github.com/numaproj/numaflow/pkg/udf/applier"
 	"github.com/stretchr/testify/assert"
diff --git a/pkg/sources/generator/tickgen_test.go b/pkg/sources/generator/tickgen_test.go
index 44633682..85cfadb5 100644
--- a/pkg/sources/generator/tickgen_test.go
+++ b/pkg/sources/generator/tickgen_test.go
@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/numaproj/numaflow/pkg/isb/stores/simplebuffer"
 	"github.com/numaproj/numaflow/pkg/watermark/generic"
 	"github.com/numaproj/numaflow/pkg/watermark/store/noop"
 
@@ -12,7 +13,6 @@ import (
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/isb/simplebuffer"
 	"github.com/stretchr/testify/assert"
 )
 
diff --git a/pkg/sources/generator/watermark_test.go b/pkg/sources/generator/watermark_test.go
index 7f57fe5c..6a5e412a 100644
--- a/pkg/sources/generator/watermark_test.go
+++ b/pkg/sources/generator/watermark_test.go
@@ -6,12 +6,12 @@ import (
 	"testing"
 	"time"
 
+	"github.com/numaproj/numaflow/pkg/isb/stores/simplebuffer"
 	"github.com/numaproj/numaflow/pkg/watermark/generic"
 	"github.com/numaproj/numaflow/pkg/watermark/store/noop"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/isb/simplebuffer"
 	"github.com/stretchr/testify/assert"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
diff --git a/pkg/sources/http/http_test.go b/pkg/sources/http/http_test.go
index 0a01418c..33d8058d 100644
--- a/pkg/sources/http/http_test.go
+++ b/pkg/sources/http/http_test.go
@@ -6,7 +6,7 @@ import (
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/isb/simplebuffer"
+	"github.com/numaproj/numaflow/pkg/isb/stores/simplebuffer"
 	"github.com/stretchr/testify/assert"
 )
 
diff --git a/pkg/sources/kafka/handler_test.go b/pkg/sources/kafka/handler_test.go
index bddde457..7604a039 100644
--- a/pkg/sources/kafka/handler_test.go
+++ b/pkg/sources/kafka/handler_test.go
@@ -9,7 +9,7 @@ import (
 	"github.com/Shopify/sarama"
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/isb/simplebuffer"
+	"github.com/numaproj/numaflow/pkg/isb/stores/simplebuffer"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	"github.com/stretchr/testify/assert"
 )
diff --git a/pkg/sources/kafka/reader_test.go b/pkg/sources/kafka/reader_test.go
index 04618e4f..291a7415 100644
--- a/pkg/sources/kafka/reader_test.go
+++ b/pkg/sources/kafka/reader_test.go
@@ -7,7 +7,7 @@ import (
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/isb/simplebuffer"
+	"github.com/numaproj/numaflow/pkg/isb/stores/simplebuffer"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	"github.com/stretchr/testify/assert"
 )