filer: etcd fix pagination

fix https://github.com/chrislusf/seaweedfs/issues/3016

weed/filer/etcd/etcd_store.go

@@ -152,7 +152,7 @@ func (store *EtcdStore) ListDirectoryEntries(ctx context.Context, dirPath weed_u
 	}
 
 	resp, err := store.client.Get(ctx, string(lastFileStart),
-		clientv3.WithPrefix(), clientv3.WithSort(clientv3.SortByKey, clientv3.SortDescend))
+		clientv3.WithFromKey())
 	if err != nil {
 		return lastFileName, fmt.Errorf("list %s : %v", dirPath, err)
 	}

weed/filer/etcd/etcd_store_test.go

@@ -0,0 +1,14 @@
+package etcd
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/filer/store_test"
+	"testing"
+)
+
+func TestStore(t *testing.T) {
+	if false {
+		store := &EtcdStore{}
+		store.initialize("localhost:2379", "3s")
+		store_test.TestFilerStore(t, store)
+	}
+}

weed/filer/store_test/test_suite.go

@@ -0,0 +1,68 @@
+package store_test
+
+import (
+	"context"
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/stretchr/testify/assert"
+	"os"
+	"testing"
+)
+
+func TestFilerStore(t *testing.T, store filer.FilerStore) {
+	ctx := context.Background()
+
+	store.InsertEntry(ctx, makeEntry(util.FullPath("/"), true))
+	store.InsertEntry(ctx, makeEntry(util.FullPath("/a"), true))
+	store.InsertEntry(ctx, makeEntry(util.FullPath("/a/b"), true))
+	store.InsertEntry(ctx, makeEntry(util.FullPath("/a/b/c"), true))
+	store.InsertEntry(ctx, makeEntry(util.FullPath("/a/b/c/f1"), false))
+	store.InsertEntry(ctx, makeEntry(util.FullPath("/a/b/c/f2"), false))
+	store.InsertEntry(ctx, makeEntry(util.FullPath("/a/b/c/f3"), false))
+	store.InsertEntry(ctx, makeEntry(util.FullPath("/a/b/c/f4"), false))
+	store.InsertEntry(ctx, makeEntry(util.FullPath("/a/b/c/f5"), false))
+
+	{
+		var counter int
+		lastFileName, err := store.ListDirectoryEntries(ctx, util.FullPath("/a/b/c"), "", false, 3, func(entry *filer.Entry) bool {
+			counter++
+			return true
+		})
+		if err != nil {
+			t.Errorf("list directory: %v", err)
+		}
+		if counter != 3 {
+			assert.Equal(t, 3, counter, "directory list counter")
+		}
+		if lastFileName != "f3" {
+			assert.Equal(t, "f3", lastFileName, "directory list last file")
+		}
+		lastFileName, err = store.ListDirectoryEntries(ctx, util.FullPath("/a/b/c"), lastFileName, false, 3, func(entry *filer.Entry) bool {
+			counter++
+			return true
+		})
+		if err != nil {
+			t.Errorf("list directory: %v", err)
+		}
+		if counter != 5 {
+			assert.Equal(t, 5, counter, "directory list counter")
+		}
+		if lastFileName != "f5" {
+			assert.Equal(t, "f5", lastFileName, "directory list last file")
+		}
+	}
+
+}
+
+func makeEntry(fullPath util.FullPath, isDirectory bool) *filer.Entry {
+	var mode os.FileMode
+	if isDirectory {
+		mode = os.ModeDir
+	}
+	return &filer.Entry{
+		FullPath: fullPath,
+		Attr: filer.Attr{
+			Mode: mode,
+		},
+	}
+}