Operations progress

[nncp.git] / src / tx.go

diff --git a/src/tx.go b/src/tx.go
index 88d4e9af4b4dfd9fe3f570fbca88b6d65296a951..67bb659a672426895919fcd5961633c75d455df9 100644 (file)
--- a/src/tx.go
+++ b/src/tx.go
@@ -39,6 +39,8 @@ import (
 )
 
 const (
+       MaxFileSize = 1 << 62
+
        TarBlockSize = 512
        TarExt       = ".tar"
 )
@@ -49,11 +51,8 @@ func (ctx *Ctx) Tx(
        nice uint8,
        size, minSize int64,
        src io.Reader,
+       pktName string,
 ) (*Node, error) {
-       tmp, err := ctx.NewTmpFileWHash()
-       if err != nil {
-               return nil, err
-       }
        hops := make([]*Node, 0, 1+len(node.Via))
        hops = append(hops, node)
        lastNode := node
@@ -69,14 +68,22 @@ func (ctx *Ctx) Tx(
        if padSize < 0 {
                padSize = 0
        }
+       if !ctx.IsEnoughSpace(size + padSize) {
+               return nil, errors.New("is not enough space")
+       }
+       tmp, err := ctx.NewTmpFileWHash()
+       if err != nil {
+               return nil, err
+       }
+
        errs := make(chan error)
        curSize := size
        pipeR, pipeW := io.Pipe()
        go func(size int64, src io.Reader, dst io.WriteCloser) {
                ctx.LogD("tx", SDS{
                        "node": hops[0].Id,
-                       "nice": strconv.Itoa(int(nice)),
-                       "size": strconv.FormatInt(size, 10),
+                       "nice": int(nice),
+                       "size": size,
                }, "wrote")
                errs <- PktEncWrite(ctx.Self, hops[0], pkt, nice, size, padSize, src, dst)
                dst.Close()
@@ -91,8 +98,8 @@ func (ctx *Ctx) Tx(
                go func(node *Node, pkt *Pkt, size int64, src io.Reader, dst io.WriteCloser) {
                        ctx.LogD("tx", SDS{
                                "node": node.Id,
-                               "nice": strconv.Itoa(int(nice)),
-                               "size": strconv.FormatInt(size, 10),
+                               "nice": int(nice),
+                               "size": size,
                        }, "trns wrote")
                        errs <- PktEncWrite(ctx.Self, node, pkt, nice, size, 0, src, dst)
                        dst.Close()
@@ -100,7 +107,11 @@ func (ctx *Ctx) Tx(
                curSize = PktEncOverhead + PktSizeOverhead + sizeWithTags(PktOverhead+curSize)
        }
        go func() {
-               _, err := io.Copy(tmp.W, pipeR)
+               _, err := CopyProgressed(
+                       tmp.W, pipeR,
+                       SDS{"xx": string(TTx), "pkt": pktName, "fullsize": curSize},
+                       ctx.ShowPrgrs,
+               )
                errs <- err
        }()
        for i := 0; i <= len(hops); i++ {
@@ -275,58 +286,12 @@ func prepareTxFile(srcPath string) (reader io.Reader, closer io.Closer, fileSize
        return
 }
 
-func (ctx *Ctx) TxFile(node *Node, nice uint8, srcPath, dstPath string, minSize int64) error {
-       dstPathSpecified := false
-       if dstPath == "" {
-               if srcPath == "-" {
-                       return errors.New("Must provide destination filename")
-               }
-               dstPath = filepath.Base(srcPath)
-       } else {
-               dstPathSpecified = true
-       }
-       dstPath = filepath.Clean(dstPath)
-       if filepath.IsAbs(dstPath) {
-               return errors.New("Relative destination path required")
-       }
-       reader, closer, fileSize, archived, err := prepareTxFile(srcPath)
-       if closer != nil {
-               defer closer.Close()
-       }
-       if err != nil {
-               return err
-       }
-       if archived && !dstPathSpecified {
-               dstPath += TarExt
-       }
-       pkt, err := NewPkt(PktTypeFile, nice, []byte(dstPath))
-       if err != nil {
-               return err
-       }
-       _, err = ctx.Tx(node, pkt, nice, fileSize, minSize, reader)
-       sds := SDS{
-               "type": "file",
-               "node": node.Id,
-               "nice": strconv.Itoa(int(nice)),
-               "src":  srcPath,
-               "dst":  dstPath,
-               "size": strconv.FormatInt(fileSize, 10),
-       }
-       if err == nil {
-               ctx.LogI("tx", sds, "sent")
-       } else {
-               sds["err"] = err
-               ctx.LogE("tx", sds, "sent")
-       }
-       return err
-}
-
-func (ctx *Ctx) TxFileChunked(
+func (ctx *Ctx) TxFile(
        node *Node,
        nice uint8,
        srcPath, dstPath string,
-       minSize int64,
        chunkSize int64,
+       minSize, maxSize int64,
 ) error {
        dstPathSpecified := false
        if dstPath == "" {
@@ -348,6 +313,9 @@ func (ctx *Ctx) TxFileChunked(
        if err != nil {
                return err
        }
+       if fileSize > maxSize {
+               return errors.New("Too big than allowed")
+       }
        if archived && !dstPathSpecified {
                dstPath += TarExt
        }
@@ -357,20 +325,19 @@ func (ctx *Ctx) TxFileChunked(
                if err != nil {
                        return err
                }
-               _, err = ctx.Tx(node, pkt, nice, fileSize, minSize, reader)
+               _, err = ctx.Tx(node, pkt, nice, fileSize, minSize, reader, dstPath)
                sds := SDS{
                        "type": "file",
                        "node": node.Id,
-                       "nice": strconv.Itoa(int(nice)),
+                       "nice": int(nice),
                        "src":  srcPath,
                        "dst":  dstPath,
-                       "size": strconv.FormatInt(fileSize, 10),
+                       "size": fileSize,
                }
                if err == nil {
                        ctx.LogI("tx", sds, "sent")
                } else {
-                       sds["err"] = err
-                       ctx.LogE("tx", sds, "sent")
+                       ctx.LogE("tx", sds, err, "sent")
                }
                return err
        }
@@ -413,20 +380,20 @@ func (ctx *Ctx) TxFileChunked(
                        sizeToSend,
                        minSize,
                        io.TeeReader(reader, hsh),
+                       path,
                )
                sds := SDS{
                        "type": "file",
                        "node": node.Id,
-                       "nice": strconv.Itoa(int(nice)),
+                       "nice": int(nice),
                        "src":  srcPath,
                        "dst":  path,
-                       "size": strconv.FormatInt(sizeToSend, 10),
+                       "size": sizeToSend,
                }
                if err == nil {
                        ctx.LogI("tx", sds, "sent")
                } else {
-                       sds["err"] = err
-                       ctx.LogE("tx", sds, "sent")
+                       ctx.LogE("tx", sds, err, "sent")
                        return err
                }
                hsh.Sum(metaPkt.Checksums[chunkNum][:0])
@@ -447,20 +414,19 @@ func (ctx *Ctx) TxFileChunked(
                return err
        }
        metaPktSize := int64(metaBuf.Len())
-       _, err = ctx.Tx(node, pkt, nice, metaPktSize, minSize, &metaBuf)
+       _, err = ctx.Tx(node, pkt, nice, metaPktSize, minSize, &metaBuf, path)
        sds := SDS{
                "type": "file",
                "node": node.Id,
-               "nice": strconv.Itoa(int(nice)),
+               "nice": int(nice),
                "src":  srcPath,
                "dst":  path,
-               "size": strconv.FormatInt(metaPktSize, 10),
+               "size": metaPktSize,
        }
        if err == nil {
                ctx.LogI("tx", sds, "sent")
        } else {
-               sds["err"] = err
-               ctx.LogE("tx", sds, "sent")
+               ctx.LogE("tx", sds, err, "sent")
        }
        return err
 }
@@ -484,20 +450,19 @@ func (ctx *Ctx) TxFreq(
        }
        src := strings.NewReader(dstPath)
        size := int64(src.Len())
-       _, err = ctx.Tx(node, pkt, nice, size, minSize, src)
+       _, err = ctx.Tx(node, pkt, nice, size, minSize, src, srcPath)
        sds := SDS{
                "type":      "freq",
                "node":      node.Id,
-               "nice":      strconv.Itoa(int(nice)),
-               "replynice": strconv.Itoa(int(replyNice)),
+               "nice":      int(nice),
+               "replynice": int(replyNice),
                "src":       srcPath,
                "dst":       dstPath,
        }
        if err == nil {
                ctx.LogI("tx", sds, "sent")
        } else {
-               sds["err"] = err
-               ctx.LogE("tx", sds, "sent")
+               ctx.LogE("tx", sds, err, "sent")
        }
        return err
 }
@@ -533,20 +498,19 @@ func (ctx *Ctx) TxExec(
                return err
        }
        size := int64(compressed.Len())
-       _, err = ctx.Tx(node, pkt, nice, size, minSize, &compressed)
+       _, err = ctx.Tx(node, pkt, nice, size, minSize, &compressed, handle)
        sds := SDS{
                "type":      "exec",
                "node":      node.Id,
-               "nice":      strconv.Itoa(int(nice)),
-               "replynice": strconv.Itoa(int(replyNice)),
+               "nice":      int(nice),
+               "replynice": int(replyNice),
                "dst":       strings.Join(append([]string{handle}, args...), " "),
-               "size":      strconv.FormatInt(size, 10),
+               "size":      size,
        }
        if err == nil {
                ctx.LogI("tx", sds, "sent")
        } else {
-               sds["err"] = err
-               ctx.LogE("tx", sds, "sent")
+               ctx.LogE("tx", sds, err, "sent")
        }
        return err
 }
@@ -555,15 +519,24 @@ func (ctx *Ctx) TxTrns(node *Node, nice uint8, size int64, src io.Reader) error
        sds := SDS{
                "type": "trns",
                "node": node.Id,
-               "nice": strconv.Itoa(int(nice)),
-               "size": strconv.FormatInt(size, 10),
+               "nice": int(nice),
+               "size": size,
        }
        ctx.LogD("tx", sds, "taken")
+       if !ctx.IsEnoughSpace(size) {
+               err := errors.New("is not enough space")
+               ctx.LogE("tx", sds, err, err.Error())
+               return err
+       }
        tmp, err := ctx.NewTmpFileWHash()
        if err != nil {
                return err
        }
-       if _, err = io.Copy(tmp.W, src); err != nil {
+       if _, err = CopyProgressed(tmp.W, src, SDS{
+               "xx":       string(TTx),
+               "pkt":      node.Id.String(),
+               "fullsize": size,
+       }, ctx.ShowPrgrs); err != nil {
                return err
        }
        nodePath := filepath.Join(ctx.Spool, node.Id.String())
@@ -571,8 +544,7 @@ func (ctx *Ctx) TxTrns(node *Node, nice uint8, size int64, src io.Reader) error
        if err == nil {
                ctx.LogI("tx", sds, "sent")
        } else {
-               sds["err"] = err
-               ctx.LogI("tx", sds, "sent")
+               ctx.LogI("tx", SdsAdd(sds, SDS{"err": err}), "sent")
        }
        os.Symlink(nodePath, filepath.Join(ctx.Spool, node.Name))
        return err