storj/cmd/uplinkng/cmd_mv.go
Michał Niewrzał 24cf7e8ea6 cmd/uplinkng: add mv command
Add ability to move files and objects.

Change-Id: I4929da730984c06aa578678b1d8c8e9b4aceade8
2021-11-22 09:07:24 +00:00

197 lines
4.8 KiB
Go

// Copyright (C) 2021 Storj Labs, Inc.
// See LICENSE for copying information.
package main
import (
"fmt"
"io"
"strconv"
"sync"
"github.com/zeebo/clingy"
"github.com/zeebo/errs"
"storj.io/common/sync2"
"storj.io/storj/cmd/uplinkng/ulext"
"storj.io/storj/cmd/uplinkng/ulfs"
"storj.io/storj/cmd/uplinkng/ulloc"
)
type cmdMv struct {
ex ulext.External
access string
recursive bool
parallelism int
dryrun bool
progress bool
source ulloc.Location
dest ulloc.Location
}
func newCmdMv(ex ulext.External) *cmdMv {
return &cmdMv{ex: ex}
}
func (c *cmdMv) Setup(params clingy.Parameters) {
c.access = params.Flag("access", "Access name or value to use", "").(string)
c.recursive = params.Flag("recursive", "Move all objects or files under the specified prefix or directory", false,
clingy.Short('r'),
clingy.Transform(strconv.ParseBool),
).(bool)
c.parallelism = params.Flag("parallelism", "Controls how many objects will be moved in parallel", 1,
clingy.Short('p'),
clingy.Transform(strconv.Atoi),
clingy.Transform(func(n int) (int, error) {
if n <= 0 {
return 0, errs.New("parallelism must be at least 1")
}
return n, nil
}),
).(int)
c.dryrun = params.Flag("dryrun", "Print what operations would happen but don't execute them", false,
clingy.Transform(strconv.ParseBool),
).(bool)
c.progress = params.Flag("progress", "Show a progress bar when possible", true,
clingy.Transform(strconv.ParseBool),
).(bool)
c.source = params.Arg("source", "Source to move", clingy.Transform(ulloc.Parse)).(ulloc.Location)
c.dest = params.Arg("dest", "Destination to move", clingy.Transform(ulloc.Parse)).(ulloc.Location)
}
func (c *cmdMv) Execute(ctx clingy.Context) error {
fs, err := c.ex.OpenFilesystem(ctx, c.access)
if err != nil {
return err
}
defer func() { _ = fs.Close() }()
switch {
case c.source.Std() || c.dest.Std():
return errs.New("cannot move to stdin/stdout")
case c.source.String() == "" || c.dest.String() == "": // TODO maybe add Empty() method
return errs.New("both source and dest cannot be empty")
case (c.source.Local() && c.dest.Remote()) || (c.source.Remote() && c.dest.Local()):
return errs.New("source and dest must be both local or both remote")
case c.source.String() == c.dest.String():
return errs.New("source and dest cannot be equal")
case c.recursive && (!c.source.Directoryish() || !c.dest.Directoryish()):
return errs.New("with --recursive flag source and destination must end with '/'")
}
// we ensure the source and destination are lexically directoryish
// if they map to directories. the destination is always converted to be
// directoryish if the copy is recursive.
if fs.IsLocalDir(ctx, c.source) {
c.source = c.source.AsDirectoryish()
}
if c.recursive || fs.IsLocalDir(ctx, c.dest) {
c.dest = c.dest.AsDirectoryish()
}
if c.recursive {
return c.moveRecursive(ctx, fs)
}
// if the destination is directoryish, we add the basename of the source
// to the end of the destination to pick a filename.
var base string
if c.dest.Directoryish() && !c.source.Std() {
// we undirectoryish the source so that we ignore any trailing slashes
// when finding the base name.
var ok bool
base, ok = c.source.Undirectoryish().Base()
if !ok {
return errs.New("destination is a directory and cannot find base name for source %q", c.source)
}
}
c.dest = joinDestWith(c.dest, base)
return c.moveFile(ctx, fs, c.source, c.dest)
}
func (c *cmdMv) moveRecursive(ctx clingy.Context, fs ulfs.Filesystem) error {
iter, err := fs.List(ctx, c.source, &ulfs.ListOptions{
Recursive: true,
})
if err != nil {
return errs.Wrap(err)
}
var (
limiter = sync2.NewLimiter(c.parallelism)
es errs.Group
mu sync.Mutex
)
fprintln := func(w io.Writer, args ...interface{}) {
mu.Lock()
defer mu.Unlock()
fmt.Fprintln(w, args...)
}
addError := func(err error) {
mu.Lock()
defer mu.Unlock()
es.Add(err)
}
items := make([]ulfs.ObjectInfo, 0, 10)
for iter.Next() {
item := iter.Item()
if item.IsPrefix {
continue
}
items = append(items, item)
}
if err := iter.Err(); err != nil {
return errs.Wrap(err)
}
for _, item := range items {
source := item.Loc
rel, err := c.source.RelativeTo(source)
if err != nil {
return err
}
dest := joinDestWith(c.dest, rel)
ok := limiter.Go(ctx, func() {
if c.progress {
fprintln(ctx.Stdout(), "Move", source, "to", dest)
}
if err := c.moveFile(ctx, fs, source, dest); err != nil {
fprintln(ctx.Stderr(), "Move", "failed:", err.Error())
addError(err)
}
})
if !ok {
break
}
}
limiter.Wait()
if len(es) > 0 {
return errs.Wrap(es.Err())
}
return nil
}
func (c *cmdMv) moveFile(ctx clingy.Context, fs ulfs.Filesystem, source, dest ulloc.Location) error {
if c.dryrun {
return nil
}
return errs.Wrap(fs.Move(ctx, source, dest))
}