Merge branch 'master' into ftporigin/ftp

1 files changed, 12 insertions, 10 deletions

diff --git a/weed/filer/cassandra/cassandra_store.go b/weed/filer/cassandra/cassandra_store.go
index 49f5625d9..fd2ce91a6 100644
--- a/weed/filer/cassandra/cassandra_store.go
+++ b/weed/filer/cassandra/cassandra_store.go
@@ -168,41 +168,43 @@ func (store *CassandraStore) DeleteFolderChildren(ctx context.Context, fullpath
 	return nil
 }
 
-func (store *CassandraStore) 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 *CassandraStore) 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 *CassandraStore) ListDirectoryEntries(ctx context.Context, fullpath util.FullPath, startFileName string, inclusive bool,
-	limit int) (entries []*filer.Entry, err error) {
+func (store *CassandraStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
 
-	if _, ok := store.isSuperLargeDirectory(string(fullpath)); ok {
+	if _, ok := store.isSuperLargeDirectory(string(dirPath)); ok {
 		return // nil, filer.ErrUnsupportedSuperLargeDirectoryListing
 	}
 
 	cqlStr := "SELECT NAME, meta FROM filemeta WHERE directory=? AND name>? ORDER BY NAME ASC LIMIT ?"
-	if inclusive {
+	if includeStartFile {
 		cqlStr = "SELECT NAME, meta FROM filemeta WHERE directory=? AND name>=? ORDER BY NAME ASC LIMIT ?"
 	}
 
 	var data []byte
 	var name string
-	iter := store.session.Query(cqlStr, string(fullpath), startFileName, limit).Iter()
+	iter := store.session.Query(cqlStr, string(dirPath), startFileName, limit+1).Iter()
 	for iter.Scan(&name, &data) {
 		entry := &filer.Entry{
-			FullPath: util.NewFullPath(string(fullpath), name),
+			FullPath: util.NewFullPath(string(dirPath), name),
 		}
+		lastFileName = name
 		if decodeErr := entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data)); decodeErr != nil {
 			err = decodeErr
 			glog.V(0).Infof("list %s : %v", entry.FullPath, err)
 			break
 		}
-		entries = append(entries, entry)
+		if !eachEntryFunc(entry) {
+			break
+		}
 	}
 	if err := iter.Close(); err != nil {
 		glog.V(0).Infof("list iterator close: %v", err)
 	}
 
-	return entries, err
+	return lastFileName, err
 }
 
 func (store *CassandraStore) Shutdown() {