diff --git a/go/filer/redis_store/directory_manager.go b/go/filer/redis_store/directory_manager.go index 138cd918e..bf66a5a30 100644 --- a/go/filer/redis_store/directory_manager.go +++ b/go/filer/redis_store/directory_manager.go @@ -3,6 +3,7 @@ package redis_store import ( "fmt" "path/filepath" + "sort" "strconv" "strings" @@ -86,26 +87,38 @@ func (dm *DirectoryManager) MakeDirectory(dirPath string) (filer.DirectoryId, er } //delete directory dirPath and its sub-directories recursively; -//it's not this function's responsibility to check whether dirPath is empty. +//it's not this function's responsibility to check whether dirPath is en empty directory. func (dm *DirectoryManager) DeleteDirectory(dirPath string) error { dirPath = embedded_filer.CleanFilePath(dirPath) if dirPath == "/" { return fmt.Errorf("Can not delete %s", dirPath) } + dirPathParent := filepath.Dir(dirPath) + dirPathName := filepath.Base(dirPath) + /* + lua comments: + 1. delete dirPath's sub-directories recursively + 2. delete dirPath itself from its parent dir + + */ script := redis.NewScript(` local delSubs delSubs = function(dir) local subs=redis.call('hgetall', dir); for i, v in ipairs(subs) do if i%2 ~= 0 then + redis.call('hdel', dir, v) local subd=dir..'/'..v; delSubs(subd); end end end delSubs(KEYS[1]) + redis.call('hdel', KEYS[2], KEYS[3]) + return 0 `) - err := script.Run(dm.Client, []string{dm.dirKeyPrefix + dirPath}, nil).Err() + //we do not use lua to get dirPath's parent dir and its basename, so do it in golang + err := script.Run(dm.Client, []string{dm.dirKeyPrefix + dirPath, dm.dirKeyPrefix + dirPathParent, dirPathName}, nil).Err() return err } @@ -208,6 +221,7 @@ func (dm *DirectoryManager) MoveUnderDirectory(oldDirPath string, newParentDirPa return err } +//return a dir's sub-directories, directories' name are sorted lexicographically func (dm *DirectoryManager) ListDirectories(dirPath string) (dirNames []filer.DirectoryEntry, err error) { dirPath = embedded_filer.CleanFilePath(dirPath) result, le := dm.Client.HGetAllMap(dm.dirKeyPrefix + dirPath).Result() @@ -215,7 +229,16 @@ func (dm *DirectoryManager) ListDirectories(dirPath string) (dirNames []filer.Di glog.Errorf("get sub-directories of %s error:%v", dirPath, err) return dirNames, le } - for k, v := range result { + //sort entries by directories' name + keys := make(sort.StringSlice, len(result)) + i := 0 + for k, _ := range result { + keys[i] = k + i++ + } + keys.Sort() + for _, k := range keys { + v := result[k] did, _ := strconv.Atoi(v) entry := filer.DirectoryEntry{Name: k, Id: filer.DirectoryId(did)} dirNames = append(dirNames, entry) diff --git a/go/filer/redis_store/directory_manager_test.go b/go/filer/redis_store/directory_manager_test.go index 30325f3cf..71fecd734 100644 --- a/go/filer/redis_store/directory_manager_test.go +++ b/go/filer/redis_store/directory_manager_test.go @@ -145,6 +145,41 @@ func TestDirectory(t *testing.T) { t.Errorf("new dir %s has a wrong dir id:%d, 6 is expected", dir, did) } + /* + * list a dir's sub-directories + * + */ + dir = "/a/b" + entries, err := dm.ListDirectories(dir) + if !(entries[0].Name == "d" && entries[1].Name == "f") { + t.Errorf("get entries:%v, expect 'd', 'f'", entries) + } + /* + * delete a dir: /a/b/f/e + * + */ + dir = "/a/b/f" + err = dm.DeleteDirectory(dir) + if err != nil { + t.Errorf("delete dir:%s error:%v", dir, err) + } + // now the deleted dir should not exist + did, err = dm.FindDirectory(dir) + if err != nil { + t.Errorf("get deleted dir %s error:%v", dir, err) + } + if did != 0 { + t.Errorf("the dir %s is not deleted!", dir) + } + // now the deleted dir's sub-directory also should not exist + dir = "/a/b/f/e" + did, err = dm.FindDirectory(dir) + if err != nil { + t.Errorf("get deleted dir %s error:%v", dir, err) + } + if did != 0 { + t.Errorf("the dir %s is not deleted!", dir) + } } func clearRedisKeys(client *redis.Client, dirKeyPrefix string, dirMaxIdKey string) error { diff --git a/go/filer/redis_store/redis_store.go b/go/filer/redis_store/redis_store.go index a31f98232..59317219a 100644 --- a/go/filer/redis_store/redis_store.go +++ b/go/filer/redis_store/redis_store.go @@ -8,6 +8,7 @@ import ( type RedisStore struct { Client *redis.Client + dm *DirectoryManager } func NewRedisStore(hostPort string, database int) *RedisStore { @@ -16,7 +17,8 @@ func NewRedisStore(hostPort string, database int) *RedisStore { Password: "", // no password set DB: int64(database), }) - return &RedisStore{Client: client} + dm := InitDirectoryManger(client) + return &RedisStore{Client: client, dm: dm} } /* @@ -75,10 +77,10 @@ func (s *RedisStore) Close() { } func (c *RedisStore) FindDirectory(dirPath string) (dirId filer.DirectoryId, err error) { - return 0, flat_namespace.ErrNotImplemented + return c.dm.FindDirectory(dirPath) } func (c *RedisStore) ListDirectories(dirPath string) (dirs []filer.DirectoryEntry, err error) { - return nil, flat_namespace.ErrNotImplemented + return c.dm.ListDirectories(dirPath) } func (c *RedisStore) ListFiles(dirPath string, lastFileName string, limit int) (files []filer.FileEntry, err error) { return nil, flat_namespace.ErrNotImplemented