mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-11-27 20:59:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor APIs

Browse Source
This commit is contained in:
Chris Lu 2020-12-12 12:42:53 -08:00
parent 03637d6f57
commit 316d1b4e69
1 changed files with 52 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

98
weed/pb/filer_pb/filer_client.go
View File

@ -85,57 +85,63 @@ func ReadDirAllEntries(filerClient FilerClient, fullDirPath util.FullPath, prefi
}
func List(filerClient FilerClient, parentDirectoryPath, prefix string, fn EachEntryFunciton, startFrom string, inclusive bool, limit uint32) (err error) {
return doList(filerClient, util.FullPath(parentDirectoryPath), prefix, fn, startFrom, inclusive, limit)
return filerClient.WithFilerClient(func(client SeaweedFilerClient) error {
return doSeaweedList(client, util.FullPath(parentDirectoryPath), prefix, fn, startFrom, inclusive, limit)
})
}
func doList(filerClient FilerClient, fullDirPath util.FullPath, prefix string, fn EachEntryFunciton, startFrom string, inclusive bool, limit uint32) (err error) {
err = filerClient.WithFilerClient(func(client SeaweedFilerClient) error {
request := &ListEntriesRequest{
Directory: string(fullDirPath),
Prefix: prefix,
StartFromFileName: startFrom,
Limit: limit,
InclusiveStartFrom: inclusive,
}
glog.V(4).Infof("read directory: %v", request)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
stream, err := client.ListEntries(ctx, request)
if err != nil {
return fmt.Errorf("list %s: %v", fullDirPath, err)
}
var prevEntry *Entry
for {
resp, recvErr := stream.Recv()
if recvErr != nil {
if recvErr == io.EOF {
if prevEntry != nil {
if err := fn(prevEntry, true); err != nil {
return err
}
}
break
} else {
return recvErr
}
}
if prevEntry != nil {
if err := fn(prevEntry, false); err != nil {
return err
}
}
prevEntry = resp.Entry
}
return nil
return filerClient.WithFilerClient(func(client SeaweedFilerClient) error {
return doSeaweedList(client, fullDirPath, prefix, fn, startFrom, inclusive, limit)
})
}
return
func SeaweedList(client SeaweedFilerClient, parentDirectoryPath, prefix string, fn EachEntryFunciton, startFrom string, inclusive bool, limit uint32) (err error) {
return doSeaweedList(client, util.FullPath(parentDirectoryPath), prefix, fn, startFrom, inclusive, limit)
}
func doSeaweedList(client SeaweedFilerClient, fullDirPath util.FullPath, prefix string, fn EachEntryFunciton, startFrom string, inclusive bool, limit uint32) (err error) {
request := &ListEntriesRequest{
Directory: string(fullDirPath),
Prefix: prefix,
StartFromFileName: startFrom,
Limit: limit,
InclusiveStartFrom: inclusive,
}
glog.V(4).Infof("read directory: %v", request)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
stream, err := client.ListEntries(ctx, request)
if err != nil {
return fmt.Errorf("list %s: %v", fullDirPath, err)
}
var prevEntry *Entry
for {
resp, recvErr := stream.Recv()
if recvErr != nil {
if recvErr == io.EOF {
if prevEntry != nil {
if err := fn(prevEntry, true); err != nil {
return err
}
}
break
} else {
return recvErr
}
}
if prevEntry != nil {
if err := fn(prevEntry, false); err != nil {
return err
}
}
prevEntry = resp.Entry
}
return nil
}
func Exists(filerClient FilerClient, parentDirectoryPath string, entryName string, isDirectory bool) (exists bool, err error) {