chore: keep stores together (numaproj#167)

Signed-off-by: Vigith Maurice <[email protected]>
dseapy · Sep 9, 2022 · 9a85860 · 9a85860
1 parent 560a300
commit 9a85860
Show file tree

Hide file tree

Showing 32 changed files with 27 additions and 28 deletions.
diff --git 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
@@ -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 → pkg/isb/message.go b/pkg/isb/format.go → pkg/isb/message.go
diff --git a/pkg/isb/jetstream/metrics.go → pkg/isb/stores/jetstream/metrics.go b/pkg/isb/jetstream/metrics.go → pkg/isb/stores/jetstream/metrics.go
diff --git a/pkg/isb/jetstream/options.go → pkg/isb/stores/jetstream/options.go b/pkg/isb/jetstream/options.go → pkg/isb/stores/jetstream/options.go
diff --git a/pkg/isb/jetstream/reader.go → pkg/isb/stores/jetstream/reader.go b/pkg/isb/jetstream/reader.go → pkg/isb/stores/jetstream/reader.go
diff --git a/pkg/isb/jetstream/reader_test.go → pkg/isb/stores/jetstream/reader_test.go b/pkg/isb/jetstream/reader_test.go → pkg/isb/stores/jetstream/reader_test.go
diff --git a/pkg/isb/jetstream/types.go → pkg/isb/stores/jetstream/types.go b/pkg/isb/jetstream/types.go → pkg/isb/stores/jetstream/types.go
diff --git a/pkg/isb/jetstream/writer.go → pkg/isb/stores/jetstream/writer.go b/pkg/isb/jetstream/writer.go → pkg/isb/stores/jetstream/writer.go
diff --git a/pkg/isb/jetstream/writer_test.go → pkg/isb/stores/jetstream/writer_test.go b/pkg/isb/jetstream/writer_test.go → pkg/isb/stores/jetstream/writer_test.go
diff --git a/pkg/isb/redis/exactlyOnceInsert.lua → pkg/isb/stores/redis/exactlyOnceInsert.lua b/pkg/isb/redis/exactlyOnceInsert.lua → pkg/isb/stores/redis/exactlyOnceInsert.lua
diff --git a/pkg/isb/redis/metrics.go → pkg/isb/stores/redis/metrics.go b/pkg/isb/redis/metrics.go → pkg/isb/stores/redis/metrics.go
diff --git a/pkg/isb/redis/options.go → pkg/isb/stores/redis/options.go b/pkg/isb/redis/options.go → pkg/isb/stores/redis/options.go
diff --git a/pkg/isb/redis/read.go → pkg/isb/stores/redis/read.go b/pkg/isb/redis/read.go → pkg/isb/stores/redis/read.go
diff --git a/pkg/isb/redis/read_test.go → pkg/isb/stores/redis/read_test.go b/pkg/isb/redis/read_test.go → pkg/isb/stores/redis/read_test.go
diff --git a/pkg/isb/redis/write.go → pkg/isb/stores/redis/write.go b/pkg/isb/redis/write.go → pkg/isb/stores/redis/write.go
diff --git a/pkg/isb/redis/write_info.go → pkg/isb/stores/redis/write_info.go b/pkg/isb/redis/write_info.go → pkg/isb/stores/redis/write_info.go
diff --git a/pkg/isb/redis/write_test.go → pkg/isb/stores/redis/write_test.go b/pkg/isb/redis/write_test.go → 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/isb/simplebuffer/buffer.go → pkg/isb/stores/simplebuffer/buffer.go b/pkg/isb/simplebuffer/buffer.go → pkg/isb/stores/simplebuffer/buffer.go
diff --git a/pkg/isb/simplebuffer/buffer_test.go → pkg/isb/stores/simplebuffer/buffer_test.go b/pkg/isb/simplebuffer/buffer_test.go → pkg/isb/stores/simplebuffer/buffer_test.go
diff --git 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/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
@@ -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
@@ -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/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/generator/tickgen_test.go b/pkg/sources/generator/tickgen_test.go
@@ -5,14 +5,14 @@ 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"
 	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
@@ -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
@@ -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
@@ -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
@@ -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"
 )

diff --git 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
@@ -6,21 +6,21 @@ 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"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	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"
 )