Merge branch 'master' into ftporigin/ftp

4 files changed, 36 insertions, 31 deletions

diff --git a/weed/filer/redis2/redis_cluster_store.go b/weed/filer/redis2/redis_cluster_store.go
index c7742bb19..22d09da25 100644
--- a/weed/filer/redis2/redis_cluster_store.go
+++ b/weed/filer/redis2/redis_cluster_store.go
@@ -3,7 +3,7 @@ package redis2
 import (
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/util"
-	"github.com/go-redis/redis"
+	"github.com/go-redis/redis/v8"
 )
 
 func init() {
@@ -20,8 +20,8 @@ func (store *RedisCluster2Store) GetName() string {
 
 func (store *RedisCluster2Store) Initialize(configuration util.Configuration, prefix string) (err error) {
 
-	configuration.SetDefault(prefix+"useReadOnly", true)
-	configuration.SetDefault(prefix+"routeByLatency", true)
+	configuration.SetDefault(prefix+"useReadOnly", false)
+	configuration.SetDefault(prefix+"routeByLatency", false)
 
 	return store.initialize(
 		configuration.GetStringSlice(prefix+"addresses"),
diff --git a/weed/filer/redis2/redis_store.go b/weed/filer/redis2/redis_store.go
index da404ed4c..8eb97e374 100644
--- a/weed/filer/redis2/redis_store.go
+++ b/weed/filer/redis2/redis_store.go
@@ -3,7 +3,7 @@ package redis2
 import (
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/util"
-	"github.com/go-redis/redis"
+	"github.com/go-redis/redis/v8"
 )
 
 func init() {
diff --git a/weed/filer/redis2/universal_redis_store.go b/weed/filer/redis2/universal_redis_store.go
index 00d02ea14..aab3d1f4a 100644
--- a/weed/filer/redis2/universal_redis_store.go
+++ b/weed/filer/redis2/universal_redis_store.go
@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/go-redis/redis"
+	"github.com/go-redis/redis/v8"
 
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/glog"
@@ -56,7 +56,7 @@ func (store *UniversalRedis2Store) InsertEntry(ctx context.Context, entry *filer
 		value = util.MaybeGzipData(value)
 	}
 
-	if err = store.Client.Set(string(entry.FullPath), value, time.Duration(entry.TtlSec)*time.Second).Err(); err != nil {
+	if err = store.Client.Set(ctx, string(entry.FullPath), value, time.Duration(entry.TtlSec)*time.Second).Err(); err != nil {
 		return fmt.Errorf("persisting %s : %v", entry.FullPath, err)
 	}
 
@@ -66,7 +66,7 @@ func (store *UniversalRedis2Store) InsertEntry(ctx context.Context, entry *filer
 	}
 
 	if name != "" {
-		if err = store.Client.ZAddNX(genDirectoryListKey(dir), redis.Z{Score: 0, Member: name}).Err(); err != nil {
+		if err = store.Client.ZAddNX(ctx, genDirectoryListKey(dir), &redis.Z{Score: 0, Member: name}).Err(); err != nil {
 			return fmt.Errorf("persisting %s in parent dir: %v", entry.FullPath, err)
 		}
 	}
@@ -81,7 +81,7 @@ func (store *UniversalRedis2Store) UpdateEntry(ctx context.Context, entry *filer
 
 func (store *UniversalRedis2Store) FindEntry(ctx context.Context, fullpath util.FullPath) (entry *filer.Entry, err error) {
 
-	data, err := store.Client.Get(string(fullpath)).Result()
+	data, err := store.Client.Get(ctx, string(fullpath)).Result()
 	if err == redis.Nil {
 		return nil, filer_pb.ErrNotFound
 	}
@@ -103,12 +103,12 @@ func (store *UniversalRedis2Store) FindEntry(ctx context.Context, fullpath util.
 
 func (store *UniversalRedis2Store) DeleteEntry(ctx context.Context, fullpath util.FullPath) (err error) {
 
-	_, err = store.Client.Del(genDirectoryListKey(string(fullpath))).Result()
+	_, err = store.Client.Del(ctx, genDirectoryListKey(string(fullpath))).Result()
 	if err != nil {
 		return fmt.Errorf("delete dir list %s : %v", fullpath, err)
 	}
 
-	_, err = store.Client.Del(string(fullpath)).Result()
+	_, err = store.Client.Del(ctx, string(fullpath)).Result()
 	if err != nil {
 		return fmt.Errorf("delete %s : %v", fullpath, err)
 	}
@@ -118,7 +118,7 @@ func (store *UniversalRedis2Store) DeleteEntry(ctx context.Context, fullpath uti
 		return nil
 	}
 	if name != "" {
-		_, err = store.Client.ZRem(genDirectoryListKey(dir), name).Result()
+		_, err = store.Client.ZRem(ctx, genDirectoryListKey(dir), name).Result()
 		if err != nil {
 			return fmt.Errorf("DeleteEntry %s in parent dir: %v", fullpath, err)
 		}
@@ -133,14 +133,14 @@ func (store *UniversalRedis2Store) DeleteFolderChildren(ctx context.Context, ful
 		return nil
 	}
 
-	members, err := store.Client.ZRange(genDirectoryListKey(string(fullpath)), 0, -1).Result()
+	members, err := store.Client.ZRange(ctx, genDirectoryListKey(string(fullpath)), 0, -1).Result()
 	if err != nil {
 		return fmt.Errorf("DeleteFolderChildren %s : %v", fullpath, err)
 	}
 
 	for _, fileName := range members {
 		path := util.NewFullPath(string(fullpath), fileName)
-		_, err = store.Client.Del(string(path)).Result()
+		_, err = store.Client.Del(ctx, string(path)).Result()
 		if err != nil {
 			return fmt.Errorf("DeleteFolderChildren %s in parent dir: %v", fullpath, err)
 		}
@@ -149,45 +149,50 @@ func (store *UniversalRedis2Store) DeleteFolderChildren(ctx context.Context, ful
 	return nil
 }
 
-func (store *UniversalRedis2Store) ListDirectoryPrefixedEntries(ctx context.Context, fullpath util.FullPath, startFileName string, inclusive bool, limit int, prefix string) (entries []*filer.Entry, err error) {
-	return nil, filer.ErrUnsupportedListDirectoryPrefixed
+func (store *UniversalRedis2Store) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
+	return lastFileName, filer.ErrUnsupportedListDirectoryPrefixed
 }
 
-func (store *UniversalRedis2Store) ListDirectoryEntries(ctx context.Context, fullpath util.FullPath, startFileName string, inclusive bool,
-	limit int) (entries []*filer.Entry, err error) {
+func (store *UniversalRedis2Store) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
 
-	dirListKey := genDirectoryListKey(string(fullpath))
+	dirListKey := genDirectoryListKey(string(dirPath))
 	start := int64(0)
 	if startFileName != "" {
-		start, _ = store.Client.ZRank(dirListKey, startFileName).Result()
-		if !inclusive {
+		start, _ = store.Client.ZRank(ctx, dirListKey, startFileName).Result()
+		if !includeStartFile {
 			start++
 		}
 	}
-	members, err := store.Client.ZRange(dirListKey, start, start+int64(limit)-1).Result()
+	members, err := store.Client.ZRange(ctx, dirListKey, start, start+int64(limit)-1).Result()
 	if err != nil {
-		return nil, fmt.Errorf("list %s : %v", fullpath, err)
+		return lastFileName, fmt.Errorf("list %s : %v", dirPath, err)
 	}
 
 	// fetch entry meta
 	for _, fileName := range members {
-		path := util.NewFullPath(string(fullpath), fileName)
+		path := util.NewFullPath(string(dirPath), fileName)
 		entry, err := store.FindEntry(ctx, path)
+		lastFileName = fileName
 		if err != nil {
 			glog.V(0).Infof("list %s : %v", path, err)
+			if err == filer_pb.ErrNotFound {
+				continue
+			}
 		} else {
 			if entry.TtlSec > 0 {
 				if entry.Attr.Crtime.Add(time.Duration(entry.TtlSec) * time.Second).Before(time.Now()) {
-					store.Client.Del(string(path)).Result()
-					store.Client.ZRem(dirListKey, fileName).Result()
+					store.Client.Del(ctx, string(path)).Result()
+					store.Client.ZRem(ctx, dirListKey, fileName).Result()
 					continue
 				}
 			}
-			entries = append(entries, entry)
+			if !eachEntryFunc(entry) {
+				break
+			}
 		}
 	}
 
-	return entries, err
+	return lastFileName, err
 }
 
 func genDirectoryListKey(dir string) (dirList string) {
diff --git a/weed/filer/redis2/universal_redis_store_kv.go b/weed/filer/redis2/universal_redis_store_kv.go
index 658491ddf..bde994dc9 100644
--- a/weed/filer/redis2/universal_redis_store_kv.go
+++ b/weed/filer/redis2/universal_redis_store_kv.go
@@ -5,12 +5,12 @@ import (
 	"fmt"
 
 	"github.com/chrislusf/seaweedfs/weed/filer"
-	"github.com/go-redis/redis"
+	"github.com/go-redis/redis/v8"
 )
 
 func (store *UniversalRedis2Store) KvPut(ctx context.Context, key []byte, value []byte) (err error) {
 
-	_, err = store.Client.Set(string(key), value, 0).Result()
+	_, err = store.Client.Set(ctx, string(key), value, 0).Result()
 
 	if err != nil {
 		return fmt.Errorf("kv put: %v", err)
@@ -21,7 +21,7 @@ func (store *UniversalRedis2Store) KvPut(ctx context.Context, key []byte, value
 
 func (store *UniversalRedis2Store) KvGet(ctx context.Context, key []byte) (value []byte, err error) {
 
-	data, err := store.Client.Get(string(key)).Result()
+	data, err := store.Client.Get(ctx, string(key)).Result()
 
 	if err == redis.Nil {
 		return nil, filer.ErrKvNotFound
@@ -32,7 +32,7 @@ func (store *UniversalRedis2Store) KvGet(ctx context.Context, key []byte) (value
 
 func (store *UniversalRedis2Store) KvDelete(ctx context.Context, key []byte) (err error) {
 
-	_, err = store.Client.Del(string(key)).Result()
+	_, err = store.Client.Del(ctx, string(key)).Result()
 
 	if err != nil {
 		return fmt.Errorf("kv delete: %v", err)