Repository: peak/s5cmd (go)

Issue:
Add option to stdin to cp
Basically the other end of https://github.com/peak/s5cmd/issues/20, which resulted in `s5cmd cat`. It doesn't look like using s5cmd run or cp has the magic "-" option to take in a piped input from stdin.

Use cases: Pipe a tar archive (or any streamed command output) into s5cmd without needing a intermediary temp file. This also solves cases where we want to preserve timestamps, permissions, etc.

Example:
`tar -cf - foo | s5cmd cp - s3://bar/foo.tar`

or if using `s5cmd run`

`tar -cf - foo | echo "cp - s3://bar/foo.tar" | s5cmd run`

Pull request:
command: add pipe command (#182)
Resolves #182 and #522
Added `pipe` command to upload `stdin` output to a bucket.

Example usage:
`echo "content" | s5cmd pipe s3://bucket/file.txt`


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 20e8e05..91a66c2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,7 @@
 #### Features
 - Added `--content-disposition` flag to `cp` command. ([#569](https://github.com/peak/s5cmd/issues/569))
 - Added `--show-fullpath` flag to `ls`. ([#596](https://github.com/peak/s5cmd/issues/596))
+- Added `pipe` command. ([#182](https://github.com/peak/s5cmd/issues/182))
 
 #### Improvements
 - Implemented concurrent multipart download support for `cat`. ([#245](https://github.com/peak/s5cmd/issues/245)) 
diff --git a/command/app.go b/command/app.go
index 2f0e6b1..f776541 100644
--- a/command/app.go
+++ b/command/app.go
@@ -204,6 +204,7 @@ func Commands() []*cli.Command {
 		NewSelectCommand(),
 		NewSizeCommand(),
 		NewCatCommand(),
+		NewPipeCommand(),
 		NewRunCommand(),
 		NewSyncCommand(),
 		NewVersionCommand(),
diff --git a/command/error.go b/command/error.go
index 66b91a3..37294ce 100644
--- a/command/error.go
+++ b/command/error.go
@@ -14,7 +14,9 @@ import (
 func printDebug(op string, err error, urls ...*url.URL) {
 	command := op
 	for _, url := range urls {
-		command += fmt.Sprintf(" %s", url)
+		if url != nil {
+			command += fmt.Sprintf(" %s", url)
+		}
 	}
 
 	msg := log.DebugMessage{
diff --git a/command/pipe.go b/command/pipe.go
new file mode 100644
index 0000000..a091eed
--- /dev/null
+++ b/command/pipe.go
@@ -0,0 +1,322 @@
+package command
+
+import (
+	"context"
+	"fmt"
+	"mime"
+	"os"
+	"path/filepath"
+
+	"github.com/urfave/cli/v2"
+
+	errorpkg "github.com/peak/s5cmd/v2/error"
+	"github.com/peak/s5cmd/v2/log"
+	"github.com/peak/s5cmd/v2/log/stat"
+	"github.com/peak/s5cmd/v2/storage"
+	"github.com/peak/s5cmd/v2/storage/url"
+)
+
+var pipeHelpTemplate = `Name:
+	{{.HelpName}} - {{.Usage}}
+
+Usage:
+	{{.HelpName}} [options] destination
+
+Options:
+	{{range .VisibleFlags}}{{.}}
+	{{end}}
+Examples:
+	01. Stream stdin to an object
+		 > echo "content" | gzip | s5cmd {{.HelpName}} s3://bucket/prefix/object.gz
+`
+
+func NewPipeCommandFlags() []cli.Flag {
+	pipeFlags := []cli.Flag{
+		&cli.StringFlag{
+			Name:  "storage-class",
+			Usage: "set storage class for target ('STANDARD','REDUCED_REDUNDANCY','GLACIER','STANDARD_IA','ONEZONE_IA','INTELLIGENT_TIERING','DEEP_ARCHIVE')",
+		},
+		&cli.IntFlag{
+			Name:    "concurrency",
+			Aliases: []string{"c"},
+			Value:   defaultCopyConcurrency,
+			Usage:   "number of concurrent parts transferred between host and remote server",
+		},
+		&cli.IntFlag{
+			Name:    "part-size",
+			Aliases: []string{"p"},
+			Value:   defaultPartSize,
+			Usage:   "size of each part transferred between host and remote server, in MiB",
+		},
+		&cli.StringFlag{
+			Name:  "sse",
+			Usage: "perform server side encryption of the data at its destination, e.g. aws:kms",
+		},
+		&cli.StringFlag{
+			Name:  "sse-kms-key-id",
+			Usage: "customer master key (CMK) id for SSE-KMS encryption; leave it out if server-side generated key is desired",
+		},
+		&cli.StringFlag{
+			Name:  "acl",
+			Usage: "set acl for target: defines granted accesses and their types on different accounts/groups, e.g. pipe --acl 'public-read'",
+		},
+		&cli.StringFlag{
+			Name:  "cache-control",
+			Usage: "set cache control for target: defines cache control header for object, e.g. pipe --cache-control 'public, max-age=345600'",
+		},
+		&cli.StringFlag{
+			Name:  "expires",
+			Usage: "set expires for target (uses RFC3339 format): defines expires header for object, e.g. pipe  --expires '2024-10-01T20:30:00Z'",
+		},
+		&cli.BoolFlag{
+			Name:  "raw",
+			Usage: "disable the wildcard operations, useful with filenames that contains glob characters",
+		},
+		&cli.StringFlag{
+			Name:  "content-type",
+			Usage: "set content type for target: defines content type header for object, e.g. --content-type text/plain",
+		},
+		&cli.StringFlag{
+			Name:  "content-encoding",
+			Usage: "set content encoding for target: defines content encoding header for object, e.g. --content-encoding gzip",
+		},
+		&cli.StringFlag{
+			Name:  "content-disposition",
+			Usage: "set content disposition for target: defines content disposition header for object, e.g. --content-disposition 'attachment; filename=\"filename.jpg\"'",
+		},
+		&cli.BoolFlag{
+			Name:    "no-clobber",
+			Aliases: []string{"n"},
+			Usage:   "do not overwrite destination if already exists",
+		},
+	}
+	return pipeFlags
+}
+
+func NewPipeCommand() *cli.Command {
+	cmd := &cli.Command{
+		Name:               "pipe",
+		HelpName:           "pipe",
+		Usage:              "stream to remote from stdin",
+		Flags:              NewPipeCommandFlags(),
+		CustomHelpTemplate: pipeHelpTemplate,
+		Before: func(c *cli.Context) error {
+			err := validatePipeCommand(c)
+			if err != nil {
+				printError(commandFromContext(c), c.Command.Name, err)
+			}
+			return err
+		},
+		Action: func(c *cli.Context) (err error) {
+			defer stat.Collect(c.Command.FullName(), &err)()
+
+			pipe, err := NewPipe(c, false)
+			if err != nil {
+				return err
+			}
+			return pipe.Run(c.Context)
+		},
+	}
+
+	cmd.BashComplete = getBashCompleteFn(cmd, false, false)
+	return cmd
+}
+
+// Pipe holds pipe operation flags and states.
+type Pipe struct {
+	dst         *url.URL
+	op          string
+	fullCommand string
+
+	deleteSource bool
+
+	// flags
+	noClobber          bool
+	storageClass       storage.StorageClass
+	encryptionMethod   string
+	encryptionKeyID    string
+	acl                string
+	cacheControl       string
+	expires            string
+	contentType        string
+	contentEncoding    string
+	contentDisposition string
+
+	// s3 options
+	concurrency int
+	partSize    int64
+	storageOpts storage.Options
+}
+
+// NewPipe creates Pipe from cli.Context.
+func NewPipe(c *cli.Context, deleteSource bool) (*Pipe, error) {
+	fullCommand := commandFromContext(c)
+
+	dst, err := url.New(c.Args().Get(0), url.WithRaw(c.Bool("raw")))
+	if err != nil {
+		printError(fullCommand, c.Command.Name, err)
+		return nil, err
+	}
+
+	return &Pipe{
+		dst:          dst,
+		op:           c.Command.Name,
+		fullCommand:  fullCommand,
+		deleteSource: deleteSource,
+		// flags
+		noClobber:          c.Bool("no-clobber"),
+		storageClass:       storage.StorageClass(c.String("storage-class")),
+		concurrency:        c.Int("concurrency"),
+		partSize:           c.Int64("part-size") * megabytes,
+		encryptionMethod:   c.String("sse"),
+		encryptionKeyID:    c.String("sse-kms-key-id"),
+		acl:                c.String("acl"),
+		cacheControl:       c.String("cache-control"),
+		expires:            c.String("expires"),
+		contentType:        c.String("content-type"),
+		contentEncoding:    c.String("content-encoding"),
+		contentDisposition: c.String("content-disposition"),
+
+		// s3 options
+		storageOpts: NewStorageOpts(c),
+	}, nil
+}
+
+// Run starts copying stdin output to destination.
+func (c Pipe) Run(ctx context.Context) error {
+	if c.dst.IsBucket() || c.dst.IsPrefix() {
+		return fmt.Errorf("target %q must be an object", c.dst)
+	}
+
+	err := c.shouldOverride(ctx, c.dst)
+	if err != nil {
+		if errorpkg.IsWarning(err) {
+			printDebug(c.op, err, nil, c.dst)
+			return nil
+		}
+		return err
+	}
+
+	client, err := storage.NewRemoteClient(ctx, c.dst, c.storageOpts)
+	if err != nil {
+		return err
+	}
+
+	metadata := storage.NewMetadata().
+		SetStorageClass(string(c.storageClass)).
+		SetSSE(c.encryptionMethod).
+		SetSSEKeyID(c.encryptionKeyID).
+		SetACL(c.acl).
+		SetCacheControl(c.cacheControl).
+		SetExpires(c.expires)
+
+	if c.contentType != "" {
+		metadata.SetContentType(c.contentType)
+	} else {
+		metadata.SetContentType(guessContentTypeByExtension(c.dst))
+	}
+
+	if c.contentEncoding != "" {
+		metadata.SetContentEncoding(c.contentEncoding)
+	}
+
+	if c.contentDisposition != "" {
+		metadata.SetContentDisposition(c.contentDisposition)
+	}
+
+	err = client.Put(ctx, &stdin{file: os.Stdin}, c.dst, metadata, c.concurrency, c.partSize)
+	if err != nil {
+		return err
+	}
+
+	msg := log.InfoMessage{
+		Operation:   c.op,
+		Source:      nil,
+		Destination: c.dst,
+		Object: &storage.Object{
+			StorageClass: c.storageClass,
+		},
+	}
+	log.Info(msg)
+
+	return nil
+}
+
+// shouldOverride function checks if the destination should be overridden if
+// the destination object and given pipe flags conform to the
+// override criteria.
+func (c Pipe) shouldOverride(ctx context.Context, dsturl *url.URL) error {
+	// if not asked to override, ignore.
+	if !c.noClobber {
+		return nil
+	}
+
+	client, err := storage.NewClient(ctx, dsturl, c.storageOpts)
+	if err != nil {
+		return err
+	}
+
+	obj, err := getObject(ctx, dsturl, client)
+	if err != nil {
+		return err
+	}
+
+	// if destination not exists, no conditions apply.
+	if obj == nil {
+		return nil
+	}
+
+	if c.noClobber {
+		return errorpkg.ErrObjectExists
+	}
+
+	return nil
+}
+
+func validatePipeCommand(c *cli.Context) error {
+	if c.Args().Len() != 1 {
+		return fmt.Errorf("expected destination argument")
+	}
+
+	dst := c.Args().Get(0)
+
+	dsturl, err := url.New(dst, url.WithRaw(c.Bool("raw")))
+	if err != nil {
+		return err
+	}
+
+	if !dsturl.IsRemote() {
+		return fmt.Errorf("destination must be a bucket")
+	}
+
+	if dsturl.IsBucket() || dsturl.IsPrefix() {
+		return fmt.Errorf("target %q must be an object", dsturl)
+	}
+
+	// wildcard destination can not be used with pipe
+	if dsturl.IsWildcard() {
+		return fmt.Errorf("target %q can not contain glob characters", dst)
+	}
+
+	return nil
+}
+
+func guessContentTypeByExtension(dsturl *url.URL) string {
+	contentType := mime.TypeByExtension(filepath.Ext(dsturl.Absolute()))
+	if contentType == "" {
+		return "application/octet-stream"
+	}
+	return contentType
+}
+
+// stdin is an io.Reader adapter for os.File, enabling it to function solely as
+// an io.Reader. The AWS SDK, which accepts an io.Reader for multipart uploads,
+// will attempt to use io.Seek if the reader supports it. However, os.Stdin is
+// a specific type of file that can not seekable.
+type stdin struct {
+	file *os.File
+}
+
+func (s *stdin) Read(p []byte) (n int, err error) {
+	return s.file.Read(p)
+}
diff --git a/go.sum b/go.sum
index e21e16c..664d545 100644
--- a/go.sum
+++ b/go.sum
@@ -1,7 +1,4 @@
 github.com/BurntSushi/toml v1.1.0/go.mod h1:CxXYINrC8qIiEnFrOxCa7Jy5BFHlXnUU2pbicEuybxQ=
-github.com/aws/aws-sdk-go v1.17.4/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=
-github.com/aws/aws-sdk-go v1.40.25 h1:Depnx7O86HWgOCLD5nMto6F9Ju85Q1QuFDnbpZYQWno=
-github.com/aws/aws-sdk-go v1.40.25/go.mod h1:585smgzpB/KqRA+K3y/NL/oYRqQvpNJYvLm+LY1U59Q=
 github.com/aws/aws-sdk-go v1.44.256 h1:O8VH+bJqgLDguqkH/xQBFz5o/YheeZqgcOYIgsTVWY4=
 github.com/aws/aws-sdk-go v1.44.256/go.mod h1:aVsgQcEevwlmQ7qHE9I3h+dtQgpqhFB+i8Phjh7fkwI=
 github.com/cpuguy83/go-md2man/v2 v2.0.2 h1:p1EgwI/C7NhT0JmVkwCD2ZBK8j4aeHQX2pMHHBfMQ6w=
@@ -21,11 +18,8 @@ github.com/hashicorp/go-multierror v1.0.0 h1:iVjPR7a6H0tWELX5NxNe7bYopibicUzc7uP
 github.com/hashicorp/go-multierror v1.0.0/go.mod h1:dHtQlpGsu+cZNNAkkCN/P3hoUDHhCYQXV3UM06sGGrk=
 github.com/iancoleman/strcase v0.0.0-20191112232945-16388991a334 h1:VHgatEHNcBFEB7inlalqfNqw65aNkM1lGX2yt3NmbS8=
 github.com/iancoleman/strcase v0.0.0-20191112232945-16388991a334/go.mod h1:SK73tn/9oHe+/Y0h39VT4UCxmurVJkR5NA7kMEAOgSE=
-github.com/igungor/gofakes3 v0.0.13 h1:rMi9KRo3jrqMyNnR9rc+d+cWTFR3mIQXnO1y+5gL98A=
-github.com/igungor/gofakes3 v0.0.13/go.mod h1:id+2y7yvTVoN1HW6VOtnfSqFt4EAfH5piVL4p9JsmQc=
 github.com/igungor/gofakes3 v0.0.14 h1:jpKujPDFe6iF3rJ34AyafvK7+LatbSbjRTnmBVPnsuY=
 github.com/igungor/gofakes3 v0.0.14/go.mod h1:+rwAKRO9RTGCIeE8SRvRPLSj7PVhaMBLlm1zPXzu7Cs=
-github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
 github.com/jmespath/go-jmespath v0.4.0 h1:BEgLn5cpjn8UN1mAw4NjwDrS35OdebyEtFe+9YPoQUg=
 github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
 github.com/jmespath/go-jmespath/internal/testify v1.5.1 h1:shLQSRRSCCPj3f2gpwzGwWFoC7ycTf1rcQZHOlsJ6N8=
@@ -54,14 +48,11 @@ github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/ryszard/goskiplist v0.0.0-20150312221310-2dfbae5fcf46 h1:GHRpF1pTW19a8tTFrMLUcfWwyC0pnifVo2ClaLq+hP8=
 github.com/ryszard/goskiplist v0.0.0-20150312221310-2dfbae5fcf46/go.mod h1:uAQ5PCi+MFsC7HjREoAz1BU+Mq60+05gifQSsHSDG/8=
-github.com/shabbyrobe/gocovmerge v0.0.0-20180507124511-f6ea450bfb63 h1:J6qvD6rbmOil46orKqJaRPG+zTpoGlBTUdyv8ki63L0=
-github.com/shabbyrobe/gocovmerge v0.0.0-20180507124511-f6ea450bfb63/go.mod h1:n+VKSARF5y/tS9XFSP7vWDfS+GUC5vs/YT7M5XDTUEM=
 github.com/shabbyrobe/gocovmerge v0.0.0-20190829150210-3e036491d500 h1:WnNuhiq+FOY3jNj6JXFT+eLN3CQ/oPIsDPRanvwsmbI=
 github.com/shabbyrobe/gocovmerge v0.0.0-20190829150210-3e036491d500/go.mod h1:+njLrG5wSeoG4Ds61rFgEzKvenR2UHbjMoDHsczxly0=
 github.com/spf13/afero v1.2.1/go.mod h1:9ZxEEn6pIJ8Rxe320qSDBk6AsU0r9pR7Q4OcevTdifk=
 github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/termie/go-shutil v0.0.0-20140729215957-bcacb06fecae h1:vgGSvdW5Lqg+I1aZOlG32uyE6xHpLdKhZzcTEktz5wM=
 github.com/termie/go-shutil v0.0.0-20140729215957-bcacb06fecae/go.mod h1:quDq6Se6jlGwiIKia/itDZxqC5rj6/8OdFyMMAwTxCs=
 github.com/urfave/cli/v2 v2.11.2 h1:FVfNg4m3vbjbBpLYxW//WjxUoHvJ9TlppXcqY9Q9ZfA=
@@ -79,21 +70,17 @@ golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
 golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
 golang.org/x/mod v0.10.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
-golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
-golang.org/x/net v0.0.0-20190310074541-c10a0554eabf/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
-golang.org/x/net v0.0.0-20210614182718-04defd469f4e h1:XpT3nA5TvE525Ne3hInMh6+GETgn27Zfm9dxsThnX2Q=
-golang.org/x/net v0.0.0-20210614182718-04defd469f4e/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.1.0/go.mod h1:Cx3nUiGt4eDBEyega/BKRp+/AlGL8hYe7U9odMt2Cco=
 golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
+golang.org/x/net v0.9.0 h1:aWJ/m6xSmxWBx+V0XRHTlrYrPG56jKsLdTFmsSsCzOM=
 golang.org/x/net v0.9.0/go.mod h1:d48xBJpPfHeWQsugry2m+kC02ZBRGRgulfHnEXEuWns=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
@@ -103,11 +90,8 @@ golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20200923182605-d9f96fdee20d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220405210540-1e041c57c461 h1:kHVeDEnfKn3T238CvrUcz6KeEsFHVaKh4kMTt6Wsysg=
-golang.org/x/sys v0.0.0-20220405210540-1e041c57c461/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -121,18 +105,15 @@ golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
 golang.org/x/term v0.7.0/go.mod h1:P32HKFT3hSsZrRxla30E9HqToFYAQPCMs/zFMBUFqPY=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.3.7 h1:olpwvP2KacW1ZWvsR7uQhoyTYvKAupfQrRGBFM352Gk=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE=
 golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
-golang.org/x/tools v0.0.0-20190308174544-00c44ba9c14f/go.mod h1:25r3+/G6/xytQM8iWZKq3Hn0kr0rgFKPUNVEL/dr3z4=
 golang.org/x/tools v0.0.0-20190624222133-a101b041ded4/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20190829051458-42f498d34c4d/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
-golang.org/x/tools v0.1.1 h1:wGiQel/hW0NnEkJUk8lbzkX2gFJU6PFxf1v5OlCfuOs=
 golang.org/x/tools v0.1.1/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
 golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
@@ -149,7 +130,6 @@ gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/mgo.v2 v2.0.0-20180705113604-9856a29383ce h1:xcEWjVhvbDy+nHP67nPDDpbYrY+ILlfndk4bRioVHaU=
 gopkg.in/mgo.v2 v2.0.0-20180705113604-9856a29383ce/go.mod h1:yeKp02qBN3iKW1OzL3MGk2IdtZzaj7SFntXj72NppTA=
-gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.8 h1:obN1ZagJSUGI0Ek/LBmuj4SNLPfIny3KsKFopxRdj10=
 gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/log/message.go b/log/message.go
index a4b92ff..df485a9 100644
--- a/log/message.go
+++ b/log/message.go
@@ -17,7 +17,7 @@ type Message interface {
 type InfoMessage struct {
 	Operation   string   `json:"operation"`
 	Success     bool     `json:"success"`
-	Source      *url.URL `json:"source"`
+	Source      *url.URL `json:"source,omitempty"`
 	Destination *url.URL `json:"destination,omitempty"`
 	Object      Message  `json:"object,omitempty"`
 
@@ -28,12 +28,15 @@ type InfoMessage struct {
 
 // String is the string representation of InfoMessage.
 func (i InfoMessage) String() string {
-	if i.Destination != nil {
+	if i.Source != nil && i.Destination != nil {
 		return fmt.Sprintf("%v %v %v", i.Operation, i.Source, i.Destination)
 	}
 	if i.Source != nil && i.Source.VersionID != "" {
 		return fmt.Sprintf("%v %-50v %v", i.Operation, i.Source, i.Source.VersionID)
 	}
+	if i.Destination != nil {
+		return fmt.Sprintf("%v %v", i.Operation, i.Destination)
+	}
 	return fmt.Sprintf("%v %v", i.Operation, i.Source)
 }
 


Test patch:
diff --git a/e2e/pipe_test.go b/e2e/pipe_test.go
new file mode 100644
index 0000000..04dd159
--- /dev/null
+++ b/e2e/pipe_test.go
@@ -0,0 +1,450 @@
+package e2e
+
+import (
+	"bytes"
+	"fmt"
+	"runtime"
+	"testing"
+
+	"gotest.tools/v3/assert"
+	"gotest.tools/v3/icmd"
+)
+
+// pipe s3://bucket/object.zip
+func TestUploadStdinToS3(t *testing.T) {
+	t.Parallel()
+
+	const (
+		filename = "verylargefile.zip"
+		content  = "Lorem ipsum dolor sit amet"
+	)
+
+	expectedContentType := "application/zip"
+
+	if runtime.GOOS == "windows" {
+		expectedContentType = "application/x-zip-compressed"
+	}
+
+	bucket := s3BucketFromTestName(t)
+
+	s3client, s5cmd := setup(t)
+
+	createBucket(t, s3client, bucket)
+
+	reader := bytes.NewBufferString(content)
+
+	dstpath := fmt.Sprintf("s3://%v/%v", bucket, filename)
+
+	cmd := s5cmd("pipe", dstpath)
+
+	result := icmd.RunCmd(cmd, icmd.WithStdin(reader))
+
+	result.Assert(t, icmd.Success)
+
+	assertLines(t, result.Stdout(), map[int]compareFunc{
+		0: equals(`pipe %v`, dstpath),
+	})
+
+	// assert that all was read from the fake stdin
+	assert.Equal(t, 0, reader.Len())
+
+	// assert S3
+	assert.Assert(t, ensureS3Object(s3client, bucket, filename, content, ensureContentType(expectedContentType)))
+}
+
+// pipe s3://bucket/object
+func TestUploadStdinToS3WithoutFileExtension(t *testing.T) {
+	t.Parallel()
+
+	s3client, s5cmd := setup(t)
+
+	bucket := s3BucketFromTestName(t)
+	createBucket(t, s3client, bucket)
+
+	const (
+		// make sure that Put reads the file header and guess Content-Type correctly.
+		filename = "index"
+		content  = `
+<html lang="en">
+	<head>
+	<meta charset="utf-8">
+	<body>
+		<div id="foo">
+			<div class="bar"></div>
+		</div>
+		<div id="baz">
+			<style data-hey="naber"></style>
+		</div>
+	</body>
+</html>
+`
+		expectedContentType = "application/octet-stream"
+	)
+
+	reader := bytes.NewBufferString(content)
+
+	dstpath := fmt.Sprintf("s3://%v/%v", bucket, filename)
+
+	cmd := s5cmd("pipe", dstpath)
+	result := icmd.RunCmd(cmd, icmd.WithStdin(reader))
+
+	result.Assert(t, icmd.Success)
+
+	assertLines(t, result.Stdout(), map[int]compareFunc{
+		0: suffix(`pipe %v`, dstpath),
+	})
+
+	// assert S3
+	assert.Assert(t, ensureS3Object(s3client, bucket, filename, content, ensureContentType(expectedContentType)))
+}
+
+// pipe --raw s3://bucket/object
+func TestUploadStdinToS3WithRawMode(t *testing.T) {
+	t.Parallel()
+
+	s3client, s5cmd := setup(t)
+
+	bucket := s3BucketFromTestName(t)
+	createBucket(t, s3client, bucket)
+
+	const (
+		filename = "ex*mple.txt"
+		content  = "Some *** text"
+	)
+
+	reader := bytes.NewBufferString(content)
+	dstpath := fmt.Sprintf("s3://%v/%v", bucket, filename)
+
+	cmd := s5cmd("pipe", "--raw", dstpath)
+	result := icmd.RunCmd(cmd, icmd.WithStdin(reader))
+
+	result.Assert(t, icmd.Success)
+
+	assertLines(t, result.Stdout(), map[int]compareFunc{
+		0: suffix(`pipe %v`, dstpath),
+	})
+
+	// assert S3
+	assert.Assert(t, ensureS3Object(s3client, bucket, filename, content))
+}
+
+// pipe s3://bucket/prefix/target
+func TestUploadStdinToS3WithPrefix(t *testing.T) {
+	t.Parallel()
+
+	bucket := s3BucketFromTestName(t)
+
+	s3client, s5cmd := setup(t)
+
+	const (
+		filename = "testfile1.txt"
+		content  = "this is the content"
+	)
+
+	createBucket(t, s3client, bucket)
+
+	reader := bytes.NewBufferString(content)
+
+	dstpath := fmt.Sprintf("s3://%v/s5cmdtest/%v", bucket, filename)
+
+	cmd := s5cmd("pipe", dstpath)
+	result := icmd.RunCmd(cmd, icmd.WithStdin(reader))
+
+	result.Assert(t, icmd.Success)
+
+	assertLines(t, result.Stdout(), map[int]compareFunc{
+		0: equals(`pipe %v`, dstpath),
+	})
+
+	// assert s3 object
+	assert.Assert(t, ensureS3Object(s3client, bucket, fmt.Sprintf("s5cmdtest/%v", filename), content))
+}
+
+// pipe -n s3://bucket/object
+func TestUploadStdinToS3WithNoClobber(t *testing.T) {
+	t.Parallel()
+
+	bucket := s3BucketFromTestName(t)
+
+	s3client, s5cmd := setup(t)
+
+	const (
+		filename   = "test.txt"
+		content    = "this text should be preserved"
+		newContent = "new content should not be written"
+	)
+
+	createBucket(t, s3client, bucket)
+	putFile(t, s3client, bucket, filename, content)
+
+	dst := "s3://" + bucket + "/" + filename
+	cmd := s5cmd("pipe", "-n", dst)
+	reader := bytes.NewBufferString(content)
+	result := icmd.RunCmd(cmd, icmd.WithStdin(reader))
+
+	result.Assert(t, icmd.Success)
+
+	assertLines(t, result.Stdout(), map[int]compareFunc{})
+
+	assertLines(t, result.Stderr(), map[int]compareFunc{})
+
+	// expect s3 object is not overridden
+	assert.Assert(t, ensureS3Object(s3client, bucket, filename, content))
+}
+
+// -log=debug pipe -n s3://bucket/object (bucket/file exists)
+func TestUploadStdinToS3WithSameFilenameWithNoClobber(t *testing.T) {
+	t.Parallel()
+
+	bucket := s3BucketFromTestName(t)
+
+	s3client, s5cmd := setup(t)
+
+	const (
+		filename   = "testfile1.txt"
+		content    = "this is the content"
+		newContent = content + "\n"
+	)
+
+	createBucket(t, s3client, bucket)
+	putFile(t, s3client, bucket, filename, content)
+
+	reader := bytes.NewBufferString(newContent)
+
+	dst := fmt.Sprintf("s3://%v/%v", bucket, filename)
+	cmd := s5cmd("--log=debug", "pipe", "-n", dst)
+	result := icmd.RunCmd(cmd, icmd.WithStdin(reader))
+
+	result.Assert(t, icmd.Success)
+
+	assertLines(t, result.Stdout(), map[int]compareFunc{
+		0: equals(`DEBUG "pipe s3://%v/%v": object already exists`, bucket, filename),
+	})
+
+	assertLines(t, result.Stderr(), map[int]compareFunc{})
+
+	// expect s3 object is not overridden
+	assert.Assert(t, ensureS3Object(s3client, bucket, filename, content))
+}
+
+// pipe s3://bucket/object (bucket/file exists)
+func TestUploadStdinToS3WithTheSameFilename(t *testing.T) {
+	t.Parallel()
+
+	bucket := s3BucketFromTestName(t)
+
+	s3client, s5cmd := setup(t)
+
+	const (
+		filename   = "testfile1.txt"
+		content    = "this is the content"
+		newContent = content + "\n"
+	)
+
+	createBucket(t, s3client, bucket)
+	putFile(t, s3client, bucket, filename, content)
+
+	reader := bytes.NewBufferString(newContent)
+
+	dst := fmt.Sprintf("s3://%v/%v", bucket, filename)
+
+	cmd := s5cmd("pipe", dst)
+	result := icmd.RunCmd(cmd, icmd.WithStdin(reader))
+
+	result.Assert(t, icmd.Success)
+
+	assertLines(t, result.Stdout(), map[int]compareFunc{
+		0: equals(`pipe %v`, dst),
+	})
+
+	// expect s3 object to be updated with new content
+	assert.Assert(t, ensureS3Object(s3client, bucket, filename, newContent))
+}
+
+func TestUploadStdinToS3WithAdjacentSlashes(t *testing.T) {
+	t.Parallel()
+
+	const (
+		filename = "index.txt"
+		content  = "this is a test file"
+	)
+
+	testcases := []struct {
+		name          string
+		dstpathprefix string
+	}{
+		{
+			name:          "pipe s3://bucket//a/b/",
+			dstpathprefix: "/a/b/",
+		},
+		{
+			name:          "pipe s3://bucket/a//b/",
+			dstpathprefix: "a//b/",
+		},
+		{
+			name:          "pipe s3://bucket/a/b//",
+			dstpathprefix: "a/b//",
+		},
+		{
+			name:          "pipe s3://bucket//a///b/",
+			dstpathprefix: "/a///b/",
+		},
+		{
+			name:          "pipe s3://bucket/a//b///",
+			dstpathprefix: "a//b///",
+		},
+		{
+			name:          "pipe s3://bucket/a//b//c//d///",
+			dstpathprefix: "a//b//c//d///",
+		},
+		{
+			name:          "pipe s3://bucket/bar/s3://",
+			dstpathprefix: "bar/s3://",
+		},
+	}
+	for _, tc := range testcases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			bucket := s3BucketFromTestName(t)
+
+			s3client, s5cmd := setup(t)
+
+			createBucket(t, s3client, bucket)
+
+			reader := bytes.NewBufferString(content)
+			objkey := fmt.Sprintf("%v/%v", tc.dstpathprefix, filename)
+			dstpath := fmt.Sprintf("s3://%v/%v", bucket, objkey)
+			cmd := s5cmd("pipe", dstpath)
+			result := icmd.RunCmd(cmd, icmd.WithStdin(reader))
+			result.Assert(t, icmd.Success)
+
+			assertLines(t, result.Stdout(), map[int]compareFunc{
+				0: suffix(`pipe %v`, dstpath),
+			})
+
+			// assert S3
+			assert.Assert(t, ensureS3Object(s3client, bucket, objkey, content))
+		})
+	}
+}
+
+// --json pipe s3://bucket/object
+func TestUploadStdinToS3JSON(t *testing.T) {
+	t.Parallel()
+
+	s3client, s5cmd := setup(t)
+
+	bucket := s3BucketFromTestName(t)
+	createBucket(t, s3client, bucket)
+
+	const (
+		filename = "testfile1.txt"
+		content  = "this is a test file"
+	)
+
+	reader := bytes.NewBufferString(content)
+
+	cmd := s5cmd("--json", "pipe", "s3://"+bucket+"/"+filename)
+	result := icmd.RunCmd(cmd, icmd.WithStdin(reader))
+
+	jsonText := `
+		{
+			"operation": "pipe",
+			"success": true,
+			"destination": "s3://%v/testfile1.txt",
+			"object": {
+				"type": "file"
+			}
+		}
+	`
+
+	result.Assert(t, icmd.Success)
+
+	assertLines(t, result.Stdout(), map[int]compareFunc{
+		0: json(jsonText, bucket),
+	}, jsonCheck(true))
+
+	// assert S3
+	assert.Assert(t, ensureS3Object(s3client, bucket, filename, content))
+}
+
+// pipe --storage-class=GLACIER s3://bucket/object
+func TestUploadStdinToS3WithStorageClassGlacier(t *testing.T) {
+	t.Parallel()
+
+	// storage class GLACIER does not exist in GCS.
+	skipTestIfGCS(t, "storage class GLACIER does not exist in GCS.")
+
+	s3client, s5cmd := setup(t)
+
+	bucket := s3BucketFromTestName(t)
+	createBucket(t, s3client, bucket)
+
+	const (
+		// make sure that Put reads the file header, not the extension
+		filename             = "index.txt"
+		content              = "content"
+		expectedStorageClass = "GLACIER"
+	)
+
+	reader := bytes.NewBufferString(content)
+
+	dstpath := fmt.Sprintf("s3://%v/%v", bucket, filename)
+
+	cmd := s5cmd("pipe", "--storage-class=GLACIER", dstpath)
+	result := icmd.RunCmd(cmd, icmd.WithStdin(reader))
+
+	result.Assert(t, icmd.Success)
+
+	assertLines(t, result.Stdout(), map[int]compareFunc{
+		0: suffix(`pipe %v`, dstpath),
+	})
+
+	// assert S3
+	assert.Assert(t, ensureS3Object(s3client, bucket, filename, content, ensureStorageClass(expectedStorageClass)))
+}
+
+// pipe --content-disposition inline s3://bucket/object
+func TestUploadStdinToToS3WithContentDisposition(t *testing.T) {
+	t.Parallel()
+
+	s3client, s5cmd := setup(t)
+
+	bucket := s3BucketFromTestName(t)
+	createBucket(t, s3client, bucket)
+
+	const (
+		// make sure that Put reads the file header and guess Content-Type correctly.
+		filename = "index.html"
+		content  = `
+<html lang="tr">
+	<head>
+	<meta charset="utf-8">
+	<body>
+		<header></header>
+		<main></main>
+		<footer></footer>
+	</body>
+</html>
+`
+		expectedContentType        = "text/html; charset=utf-8"
+		expectedContentDisposition = "inline"
+	)
+
+	dstpath := fmt.Sprintf("s3://%v/%v", bucket, filename)
+	reader := bytes.NewBufferString(content)
+
+	cmd := s5cmd("pipe", "--content-disposition", "inline", dstpath)
+	result := icmd.RunCmd(cmd, icmd.WithStdin(reader))
+
+	result.Assert(t, icmd.Success)
+
+	assertLines(t, result.Stdout(), map[int]compareFunc{
+		0: suffix(`pipe %v`, dstpath),
+	})
+
+	// assert S3
+	assert.Assert(t, ensureS3Object(s3client, bucket, filename, content, ensureContentType(expectedContentType), ensureContentDisposition(expectedContentDisposition)))
+}