restic/pipe/pipe.go

247 lines
5.6 KiB
Go
Raw Normal View History

2015-02-15 11:57:09 +00:00
package pipe
import (
2015-03-02 18:44:16 +00:00
"errors"
2015-02-15 11:57:09 +00:00
"fmt"
"os"
"path/filepath"
"sort"
2015-03-02 13:48:47 +00:00
"github.com/restic/restic/debug"
2015-02-15 11:57:09 +00:00
)
2015-03-07 10:53:32 +00:00
type Result interface{}
type Job interface {
Path() string
Fullpath() string
Error() error
Info() os.FileInfo
Result() chan<- Result
}
2015-02-15 11:57:09 +00:00
type Entry struct {
2015-03-07 10:53:32 +00:00
basedir string
path string
info os.FileInfo
error error
result chan<- Result
// points to the old node if available, interface{} is used to prevent
// circular import
Node interface{}
2015-02-15 11:57:09 +00:00
}
2015-03-07 10:53:32 +00:00
func (e Entry) Path() string { return e.path }
func (e Entry) Fullpath() string { return filepath.Join(e.basedir, e.path) }
func (e Entry) Error() error { return e.error }
func (e Entry) Info() os.FileInfo { return e.info }
func (e Entry) Result() chan<- Result { return e.result }
2015-02-15 11:57:09 +00:00
type Dir struct {
2015-03-07 10:53:32 +00:00
basedir string
path string
error error
info os.FileInfo
2015-02-15 11:57:09 +00:00
2015-03-07 10:53:32 +00:00
Entries [](<-chan Result)
result chan<- Result
2015-02-15 11:57:09 +00:00
}
2015-03-07 10:53:32 +00:00
func (e Dir) Path() string { return e.path }
func (e Dir) Fullpath() string { return filepath.Join(e.basedir, e.path) }
func (e Dir) Error() error { return e.error }
func (e Dir) Info() os.FileInfo { return e.info }
func (e Dir) Result() chan<- Result { return e.result }
2015-02-15 11:57:09 +00:00
// readDirNames reads the directory named by dirname and returns
// a sorted list of directory entries.
// taken from filepath/path.go
func readDirNames(dirname string) ([]string, error) {
f, err := os.Open(dirname)
if err != nil {
return nil, err
}
names, err := f.Readdirnames(-1)
f.Close()
if err != nil {
return nil, err
}
sort.Strings(names)
return names, nil
}
func isDir(fi os.FileInfo) bool {
return fi.IsDir()
}
2015-03-02 18:44:16 +00:00
var errCancelled = errors.New("walk cancelled")
// SelectFunc returns true for all items that should be included (files and
// dirs). If false is returned, files are ignored and dirs are not even walked.
type SelectFunc func(item string, fi os.FileInfo) bool
2015-11-06 18:41:57 +00:00
func walk(basedir, dir string, selectFunc SelectFunc, done <-chan struct{}, jobs chan<- Job, res chan<- Result) {
debug.Log("pipe.walk", "start on %v", dir)
relpath, err := filepath.Rel(basedir, dir)
if err != nil {
panic(err)
}
2015-03-15 13:24:58 +00:00
info, err := os.Lstat(dir)
2015-02-15 11:57:09 +00:00
if err != nil {
2015-03-28 15:35:46 +00:00
debug.Log("pipe.walk", "error for %v: %v", dir, err)
2015-11-06 18:41:57 +00:00
select {
case jobs <- Dir{basedir: basedir, path: relpath, info: info, error: err, result: res}:
case <-done:
}
return
2015-02-15 11:57:09 +00:00
}
if !selectFunc(dir, info) {
debug.Log("pipe.walk", "file %v excluded by filter", dir)
2015-11-06 18:41:57 +00:00
return
}
2015-02-15 11:57:09 +00:00
if !info.IsDir() {
2015-03-02 18:44:16 +00:00
select {
2015-03-07 10:53:32 +00:00
case jobs <- Entry{info: info, basedir: basedir, path: relpath, result: res}:
2015-03-02 18:44:16 +00:00
case <-done:
}
2015-11-06 18:41:57 +00:00
return
2015-02-15 11:57:09 +00:00
}
2015-03-15 13:24:58 +00:00
names, err := readDirNames(dir)
2015-02-15 11:57:09 +00:00
if err != nil {
2015-11-06 18:41:57 +00:00
debug.Log("pipe.walk", "Readdirnames(%v) returned error: %v", dir, err)
select {
case <-done:
case jobs <- Dir{basedir: basedir, path: relpath, info: info, error: err, result: res}:
}
return
2015-02-15 11:57:09 +00:00
}
2015-03-15 13:24:58 +00:00
// Insert breakpoint to allow testing behaviour with vanishing files
// between Readdir() and lstat()
debug.RunHook("pipe.walk1", relpath)
2015-03-15 13:24:58 +00:00
2015-03-07 10:53:32 +00:00
entries := make([]<-chan Result, 0, len(names))
2015-02-15 11:57:09 +00:00
for _, name := range names {
2015-03-15 13:24:58 +00:00
subpath := filepath.Join(dir, name)
2015-02-15 13:44:54 +00:00
fi, statErr := os.Lstat(subpath)
if !selectFunc(subpath, fi) {
debug.Log("pipe.walk", "file %v excluded by filter", subpath)
continue
}
2015-03-07 10:53:32 +00:00
ch := make(chan Result, 1)
2015-02-15 13:44:54 +00:00
entries = append(entries, ch)
2015-02-15 11:57:09 +00:00
if statErr != nil {
2015-03-02 18:44:16 +00:00
select {
case jobs <- Entry{info: fi, error: statErr, basedir: basedir, path: filepath.Join(relpath, name), result: ch}:
2015-03-02 18:44:16 +00:00
case <-done:
2015-11-06 18:41:57 +00:00
return
2015-03-02 18:44:16 +00:00
}
continue
2015-02-15 11:57:09 +00:00
}
2015-03-15 13:24:58 +00:00
// Insert breakpoint to allow testing behaviour with vanishing files
// between walk and open
debug.RunHook("pipe.walk2", filepath.Join(relpath, name))
2015-03-15 13:24:58 +00:00
2015-11-06 18:41:57 +00:00
walk(basedir, subpath, selectFunc, done, jobs, ch)
2015-02-15 11:57:09 +00:00
}
2015-03-02 18:44:16 +00:00
select {
2015-03-07 10:53:32 +00:00
case jobs <- Dir{basedir: basedir, path: relpath, info: info, Entries: entries, result: res}:
2015-03-02 18:44:16 +00:00
case <-done:
}
2015-02-15 11:57:09 +00:00
}
2015-03-02 13:48:47 +00:00
// Walk sends a Job for each file and directory it finds below the paths. When
// the channel done is closed, processing stops.
2015-11-06 18:41:57 +00:00
func Walk(paths []string, selectFunc SelectFunc, done chan struct{}, jobs chan<- Job, res chan<- Result) {
debug.Log("pipe.Walk", "start on %v", paths)
2015-03-02 13:48:47 +00:00
defer func() {
debug.Log("pipe.Walk", "output channel closed")
2015-03-07 10:53:32 +00:00
close(jobs)
2015-03-02 13:48:47 +00:00
}()
2015-03-07 10:53:32 +00:00
entries := make([]<-chan Result, 0, len(paths))
2015-03-02 13:48:47 +00:00
for _, path := range paths {
debug.Log("pipe.Walk", "start walker for %v", path)
2015-03-07 10:53:32 +00:00
ch := make(chan Result, 1)
2015-11-06 18:41:57 +00:00
walk(filepath.Dir(path), path, selectFunc, done, jobs, ch)
2015-03-28 15:35:46 +00:00
entries = append(entries, ch)
2015-03-02 13:48:47 +00:00
debug.Log("pipe.Walk", "walker for %v done", path)
}
debug.Log("pipe.Walk", "sending root node")
select {
case <-done:
2015-11-06 18:41:57 +00:00
return
case jobs <- Dir{Entries: entries, result: res}:
}
debug.Log("pipe.Walk", "walker done")
2015-03-02 13:48:47 +00:00
}
// Split feeds all elements read from inChan to dirChan and entChan.
2015-03-07 10:53:32 +00:00
func Split(inChan <-chan Job, dirChan chan<- Dir, entChan chan<- Entry) {
2015-03-02 13:48:47 +00:00
debug.Log("pipe.Split", "start")
defer debug.Log("pipe.Split", "done")
inCh := inChan
dirCh := dirChan
entCh := entChan
var (
dir Dir
ent Entry
)
// deactivate sending until we received at least one job
dirCh = nil
entCh = nil
for {
select {
case job, ok := <-inCh:
if !ok {
// channel is closed
return
}
if job == nil {
panic("nil job received")
}
// disable receiving until the current job has been sent
inCh = nil
switch j := job.(type) {
case Dir:
dir = j
dirCh = dirChan
case Entry:
ent = j
entCh = entChan
default:
panic(fmt.Sprintf("unknown job type %v", j))
}
case dirCh <- dir:
// disable sending, re-enable receiving
dirCh = nil
inCh = inChan
case entCh <- ent:
// disable sending, re-enable receiving
entCh = nil
inCh = inChan
}
}
2015-02-15 11:57:09 +00:00
}