mirror of
https://github.com/octoleo/restic.git
synced 2024-11-22 21:05:10 +00:00
commit
4c329110c5
30
pipe/pipe.go
30
pipe/pipe.go
@ -83,7 +83,7 @@ var errCancelled = errors.New("walk cancelled")
|
||||
type SelectFunc func(item string, fi os.FileInfo) bool
|
||||
|
||||
func walk(basedir, dir string, selectFunc SelectFunc, done <-chan struct{}, jobs chan<- Job, res chan<- Result) {
|
||||
debug.Log("pipe.walk", "start on %v", dir)
|
||||
debug.Log("pipe.walk", "start on %q, basedir %q", dir, basedir)
|
||||
|
||||
relpath, err := filepath.Rel(basedir, dir)
|
||||
if err != nil {
|
||||
@ -158,15 +158,41 @@ func walk(basedir, dir string, selectFunc SelectFunc, done <-chan struct{}, jobs
|
||||
walk(basedir, subpath, selectFunc, done, jobs, ch)
|
||||
}
|
||||
|
||||
debug.Log("pipe.walk", "sending dirjob for %q, basedir %q", dir, basedir)
|
||||
select {
|
||||
case jobs <- Dir{basedir: basedir, path: relpath, info: info, Entries: entries, result: res}:
|
||||
case <-done:
|
||||
}
|
||||
}
|
||||
|
||||
// cleanupPath is used to clean a path. For a normal path, a slice with just
|
||||
// the path is returned. For special cases such as "." and "/" the list of
|
||||
// names within those paths is returned.
|
||||
func cleanupPath(path string) ([]string, error) {
|
||||
path = filepath.Clean(path)
|
||||
if filepath.Dir(path) != path {
|
||||
return []string{path}, nil
|
||||
}
|
||||
|
||||
return readDirNames(path)
|
||||
}
|
||||
|
||||
// Walk sends a Job for each file and directory it finds below the paths. When
|
||||
// the channel done is closed, processing stops.
|
||||
func Walk(paths []string, selectFunc SelectFunc, done chan struct{}, jobs chan<- Job, res chan<- Result) {
|
||||
func Walk(walkPaths []string, selectFunc SelectFunc, done chan struct{}, jobs chan<- Job, res chan<- Result) {
|
||||
var paths []string
|
||||
|
||||
for _, p := range walkPaths {
|
||||
ps, err := cleanupPath(p)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "Readdirnames(%v): %v, skipping\n", p, err)
|
||||
debug.Log("pipe.Walk", "Readdirnames(%v) returned error: %v, skipping", p, err)
|
||||
continue
|
||||
}
|
||||
|
||||
paths = append(paths, ps...)
|
||||
}
|
||||
|
||||
debug.Log("pipe.Walk", "start on %v", paths)
|
||||
defer func() {
|
||||
debug.Log("pipe.Walk", "output channel closed")
|
||||
|
@ -4,6 +4,7 @@ import (
|
||||
"io/ioutil"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
@ -53,6 +54,12 @@ func TestPipelineWalkerWithSplit(t *testing.T) {
|
||||
t.Skipf("walkerpath not set, skipping TestPipelineWalker")
|
||||
}
|
||||
|
||||
var err error
|
||||
if !filepath.IsAbs(TestWalkerPath) {
|
||||
TestWalkerPath, err = filepath.Abs(TestWalkerPath)
|
||||
OK(t, err)
|
||||
}
|
||||
|
||||
before, err := statPath(TestWalkerPath)
|
||||
OK(t, err)
|
||||
|
||||
@ -143,6 +150,12 @@ func TestPipelineWalker(t *testing.T) {
|
||||
t.Skipf("walkerpath not set, skipping TestPipelineWalker")
|
||||
}
|
||||
|
||||
var err error
|
||||
if !filepath.IsAbs(TestWalkerPath) {
|
||||
TestWalkerPath, err = filepath.Abs(TestWalkerPath)
|
||||
OK(t, err)
|
||||
}
|
||||
|
||||
before, err := statPath(TestWalkerPath)
|
||||
OK(t, err)
|
||||
|
||||
@ -421,6 +434,7 @@ func TestPipelineWalkerMultiple(t *testing.T) {
|
||||
}
|
||||
|
||||
paths, err := filepath.Glob(filepath.Join(TestWalkerPath, "*"))
|
||||
OK(t, err)
|
||||
|
||||
before, err := statPath(TestWalkerPath)
|
||||
OK(t, err)
|
||||
@ -491,3 +505,95 @@ func TestPipelineWalkerMultiple(t *testing.T) {
|
||||
|
||||
Assert(t, before == after, "stats do not match, expected %v, got %v", before, after)
|
||||
}
|
||||
|
||||
func dirsInPath(path string) int {
|
||||
if path == "/" || path == "." || path == "" {
|
||||
return 0
|
||||
}
|
||||
|
||||
n := 0
|
||||
for dir := path; dir != "/" && dir != "."; dir = filepath.Dir(dir) {
|
||||
n++
|
||||
}
|
||||
|
||||
return n
|
||||
}
|
||||
|
||||
func TestPipeWalkerRoot(t *testing.T) {
|
||||
if runtime.GOOS == "windows" {
|
||||
t.Skipf("not running TestPipeWalkerRoot on %s", runtime.GOOS)
|
||||
return
|
||||
}
|
||||
|
||||
cwd, err := os.Getwd()
|
||||
OK(t, err)
|
||||
|
||||
testPaths := []string{
|
||||
string(filepath.Separator),
|
||||
".",
|
||||
cwd,
|
||||
}
|
||||
|
||||
for _, path := range testPaths {
|
||||
testPipeWalkerRootWithPath(path, t)
|
||||
}
|
||||
}
|
||||
|
||||
func testPipeWalkerRootWithPath(path string, t *testing.T) {
|
||||
pattern := filepath.Join(path, "*")
|
||||
rootPaths, err := filepath.Glob(pattern)
|
||||
OK(t, err)
|
||||
|
||||
for i, p := range rootPaths {
|
||||
rootPaths[i], err = filepath.Rel(path, p)
|
||||
OK(t, err)
|
||||
}
|
||||
|
||||
t.Logf("paths in %v (pattern %q) expanded to %v items", path, pattern, len(rootPaths))
|
||||
|
||||
done := make(chan struct{})
|
||||
defer close(done)
|
||||
|
||||
jobCh := make(chan pipe.Job)
|
||||
var jobs []pipe.Job
|
||||
|
||||
worker := func(wg *sync.WaitGroup) {
|
||||
defer wg.Done()
|
||||
for job := range jobCh {
|
||||
jobs = append(jobs, job)
|
||||
}
|
||||
}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(1)
|
||||
go worker(&wg)
|
||||
|
||||
filter := func(p string, fi os.FileInfo) bool {
|
||||
p, err := filepath.Rel(path, p)
|
||||
OK(t, err)
|
||||
return dirsInPath(p) <= 1
|
||||
}
|
||||
|
||||
resCh := make(chan pipe.Result, 1)
|
||||
pipe.Walk([]string{path}, filter, done, jobCh, resCh)
|
||||
|
||||
wg.Wait()
|
||||
|
||||
t.Logf("received %d jobs", len(jobs))
|
||||
|
||||
for i, job := range jobs[:len(jobs)-1] {
|
||||
path := job.Path()
|
||||
if path == "." || path == ".." || path == string(filepath.Separator) {
|
||||
t.Errorf("job %v has invalid path %q", i, path)
|
||||
}
|
||||
}
|
||||
|
||||
lastPath := jobs[len(jobs)-1].Path()
|
||||
if lastPath != "" {
|
||||
t.Errorf("last job has non-empty path %q", lastPath)
|
||||
}
|
||||
|
||||
if len(jobs) < len(rootPaths) {
|
||||
t.Errorf("want at least %v jobs, got %v for path %v\n", len(rootPaths), len(jobs), path)
|
||||
}
|
||||
}
|
||||
|
Loading…
Reference in New Issue
Block a user