tq/basic_upload.go

   1 package tq
   2
   3 import (
   4         "io"
   5         "io/ioutil"
   6         "os"
   7         "path/filepath"
   8         "strconv"
   9         "strings"
  10
  11         "github.com/git-lfs/git-lfs/errors"
  12         "github.com/git-lfs/git-lfs/lfsapi"
  13         "github.com/git-lfs/git-lfs/tools"
  14 )
  15
  16 const (
  17         BasicAdapterName = "basic"
  18 )
  19
  20 // Adapter for basic uploads (non resumable)
  21 type basicUploadAdapter struct {
  22         *adapterBase
  23 }
  24
  25 func (a *basicUploadAdapter) ClearTempStorage() error {
  26         // Should be empty already but also remove dir
  27         return os.RemoveAll(a.tempDir())
  28 }
  29
  30 func (a *basicUploadAdapter) tempDir() string {
  31         // Must be dedicated to this adapter as deleted by ClearTempStorage
  32         d := filepath.Join(os.TempDir(), "git-lfs-basic-temp")
  33         if err := os.MkdirAll(d, 0755); err != nil {
  34                 return os.TempDir()
  35         }
  36         return d
  37 }
  38
  39 func (a *basicUploadAdapter) WorkerStarting(workerNum int) (interface{}, error) {
  40         return nil, nil
  41 }
  42 func (a *basicUploadAdapter) WorkerEnding(workerNum int, ctx interface{}) {
  43 }
  44
  45 func (a *basicUploadAdapter) DoTransfer(ctx interface{}, t *Transfer, cb ProgressCallback, authOkFunc func()) error {
  46         rel, err := t.Rel("upload")
  47         if err != nil {
  48                 return err
  49         }
  50         if rel == nil {
  51                 return errors.Errorf("No upload action for object: %s", t.Oid)
  52         }
  53
  54         req, err := a.newHTTPRequest("PUT", rel)
  55         if err != nil {
  56                 return err
  57         }
  58
  59         if len(req.Header.Get("Content-Type")) == 0 {
  60                 req.Header.Set("Content-Type", "application/octet-stream")
  61         }
  62
  63         if req.Header.Get("Transfer-Encoding") == "chunked" {
  64                 req.TransferEncoding = []string{"chunked"}
  65         } else {
  66                 req.Header.Set("Content-Length", strconv.FormatInt(t.Size, 10))
  67         }
  68
  69         req.ContentLength = t.Size
  70
  71         f, err := os.OpenFile(t.Path, os.O_RDONLY, 0644)
  72         if err != nil {
  73                 return errors.Wrap(err, "basic upload")
  74         }
  75         defer f.Close()
  76
  77         // Ensure progress callbacks made while uploading
  78         // Wrap callback to give name context
  79         ccb := func(totalSize int64, readSoFar int64, readSinceLast int) error {
  80                 if cb != nil {
  81                         return cb(t.Name, totalSize, readSoFar, readSinceLast)
  82                 }
  83                 return nil
  84         }
  85
  86         cbr := tools.NewBodyWithCallback(f, t.Size, ccb)
  87         var reader lfsapi.ReadSeekCloser = cbr
  88
  89         // Signal auth was ok on first read; this frees up other workers to start
  90         if authOkFunc != nil {
  91                 reader = newStartCallbackReader(reader, func() error {
  92                         authOkFunc()
  93                         return nil
  94                 })
  95         }
  96
  97         req.Body = reader
  98
  99         req = a.apiClient.LogRequest(req, "lfs.data.upload")
 100         res, err := a.doHTTP(t, req)
 101         if err != nil {
 102                 // We're about to return a retriable error, meaning that this
 103                 // transfer will either be retried, or it will fail.
 104                 //
 105                 // Either way, let's decrement the number of bytes that we've
 106                 // read _so far_, so that the next iteration doesn't re-transfer
 107                 // those bytes, according to the progress meter.
 108                 if perr := cbr.ResetProgress(); perr != nil {
 109                         err = errors.Wrap(err, perr.Error())
 110                 }
 111
 112                 return errors.NewRetriableError(err)
 113         }
 114
 115         // A status code of 403 likely means that an authentication token for the
 116         // upload has expired. This can be safely retried.
 117         if res.StatusCode == 403 {
 118                 err = errors.New("http: received status 403")
 119                 return errors.NewRetriableError(err)
 120         }
 121
 122         if res.StatusCode > 299 {
 123                 return errors.Wrapf(nil, "Invalid status for %s %s: %d",
 124                         req.Method,
 125                         strings.SplitN(req.URL.String(), "?", 2)[0],
 126                         res.StatusCode,
 127                 )
 128         }
 129
 130         io.Copy(ioutil.Discard, res.Body)
 131         res.Body.Close()
 132
 133         return verifyUpload(a.apiClient, a.remote, t)
 134 }
 135
 136 // startCallbackReader is a reader wrapper which calls a function as soon as the
 137 // first Read() call is made. This callback is only made once
 138 type startCallbackReader struct {
 139         cb     func() error
 140         cbDone bool
 141         lfsapi.ReadSeekCloser
 142 }
 143
 144 func (s *startCallbackReader) Read(p []byte) (n int, err error) {
 145         if !s.cbDone && s.cb != nil {
 146                 if err := s.cb(); err != nil {
 147                         return 0, err
 148                 }
 149                 s.cbDone = true
 150         }
 151         return s.ReadSeekCloser.Read(p)
 152 }
 153 func newStartCallbackReader(r lfsapi.ReadSeekCloser, cb func() error) *startCallbackReader {
 154         return &startCallbackReader{
 155                 ReadSeekCloser: r,
 156                 cb:             cb,
 157         }
 158 }
 159
 160 func configureBasicUploadAdapter(m *Manifest) {
 161         m.RegisterNewAdapterFunc(BasicAdapterName, Upload, func(name string, dir Direction) Adapter {
 162                 switch dir {
 163                 case Upload:
 164                         bu := &basicUploadAdapter{newAdapterBase(m.fs, name, dir, nil)}
 165                         // self implements impl
 166                         bu.transferImpl = bu
 167                         return bu
 168                 case Download:
 169                         panic("Should never ask this func for basic download")
 170                 }
 171                 return nil
 172         })
 173 }