lunny/git-lfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tq,transfer: move pkg "transfer" to "tq"

Browse Source
This commit is contained in:
Taylor Blau 2016-12-11 17:28:47 -07:00
parent c5ce4b7287
commit 891db97a42
16 changed files with 42 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
commands/commands.go
View File

@ -19,7 +19,7 @@ import (
"github.com/git-lfs/git-lfs/lfs"
"github.com/git-lfs/git-lfs/progress"
"github.com/git-lfs/git-lfs/tools"
"github.com/git-lfs/git-lfs/transfer"
"github.com/git-lfs/git-lfs/tq"
)
// Populate man pages
@ -41,10 +41,10 @@ var (
excludeArg string
)
// TransferManifest builds a transfer.Manifest from the commands package global
// TransferManifest builds a tq.Manifest from the commands package global
// cfg var.
func TransferManifest() *transfer.Manifest {
return transfer.ConfigureManifest(transfer.NewManifest(), cfg)
func TransferManifest() *tq.Manifest {
return tq.ConfigureManifest(tq.NewManifest(), cfg)
}
// Error prints a formatted message to Stderr. It also gets printed to the

5
lfs/download_queue.go
View File

@ -3,7 +3,6 @@ package lfs
import (
"github.com/git-lfs/git-lfs/api"
"github.com/git-lfs/git-lfs/tq"
"github.com/git-lfs/git-lfs/transfer"
)
type Downloadable struct {
@ -42,10 +41,10 @@ func NewDownloadable(p *WrappedPointer) *Downloadable {
// NewDownloadCheckQueue builds a checking queue, checks that objects are there but doesn't download
func NewDownloadCheckQueue(options ...tq.TransferQueueOption) *tq.TransferQueue {
return tq.NewTransferQueue(transfer.Download, options...)
return tq.NewTransferQueue(tq.Download, options...)
}
// NewDownloadQueue builds a DownloadQueue, allowing concurrent downloads.
func NewDownloadQueue(options ...tq.TransferQueueOption) *tq.TransferQueue {
return tq.NewTransferQueue(transfer.Download, options...)
return tq.NewTransferQueue(tq.Download, options...)
}

4
lfs/lfs.go
View File

@ -12,7 +12,7 @@ import (
"github.com/git-lfs/git-lfs/config"
"github.com/git-lfs/git-lfs/localstorage"
"github.com/git-lfs/git-lfs/tools"
"github.com/git-lfs/git-lfs/transfer"
"github.com/git-lfs/git-lfs/tq"
"github.com/rubyist/tracerx"
)
@ -63,7 +63,7 @@ func ObjectExistsOfSize(oid string, size int64) bool {
return tools.FileExistsOfSize(path, size)
}
func Environ(cfg *config.Configuration, manifest *transfer.Manifest) []string {
func Environ(cfg *config.Configuration, manifest *tq.Manifest) []string {
osEnviron := os.Environ()
env := make([]string, 0, len(osEnviron)+7)

4
lfs/pointer.go
View File

@ -13,7 +13,7 @@ import (
"github.com/git-lfs/git-lfs/errors"
"github.com/git-lfs/git-lfs/progress"
"github.com/git-lfs/git-lfs/transfer"
"github.com/git-lfs/git-lfs/tq"
)
var (
@ -60,7 +60,7 @@ func NewPointerExtension(name string, priority int, oid string) *PointerExtensio
return &PointerExtension{name, priority, oid, oidType}
}
func (p *Pointer) Smudge(writer io.Writer, workingfile string, download bool, manifest *transfer.Manifest, cb progress.CopyCallback) error {
func (p *Pointer) Smudge(writer io.Writer, workingfile string, download bool, manifest *tq.Manifest, cb progress.CopyCallback) error {
return PointerSmudge(writer, p, workingfile, download, manifest, cb)
}

14
lfs/pointer_smudge.go
View File

@ -8,7 +8,7 @@ import (
"github.com/cheggaaa/pb"
"github.com/git-lfs/git-lfs/tools"
"github.com/git-lfs/git-lfs/transfer"
"github.com/git-lfs/git-lfs/tq"
"github.com/git-lfs/git-lfs/api"
"github.com/git-lfs/git-lfs/config"
@ -17,7 +17,7 @@ import (
"github.com/rubyist/tracerx"
)
func PointerSmudgeToFile(filename string, ptr *Pointer, download bool, manifest *transfer.Manifest, cb progress.CopyCallback) error {
func PointerSmudgeToFile(filename string, ptr *Pointer, download bool, manifest *tq.Manifest, cb progress.CopyCallback) error {
os.MkdirAll(filepath.Dir(filename), 0755)
file, err := os.Create(filename)
if err != nil {
@ -37,7 +37,7 @@ func PointerSmudgeToFile(filename string, ptr *Pointer, download bool, manifest
return nil
}
func PointerSmudge(writer io.Writer, ptr *Pointer, workingfile string, download bool, manifest *transfer.Manifest, cb progress.CopyCallback) error {
func PointerSmudge(writer io.Writer, ptr *Pointer, workingfile string, download bool, manifest *tq.Manifest, cb progress.CopyCallback) error {
mediafile, err := LocalMediaPath(ptr.Oid)
if err != nil {
return err
@ -72,7 +72,7 @@ func PointerSmudge(writer io.Writer, ptr *Pointer, workingfile string, download
return nil
}
func downloadFile(writer io.Writer, ptr *Pointer, workingfile, mediafile string, manifest *transfer.Manifest, cb progress.CopyCallback) error {
func downloadFile(writer io.Writer, ptr *Pointer, workingfile, mediafile string, manifest *tq.Manifest, cb progress.CopyCallback) error {
fmt.Fprintf(os.Stderr, "Downloading %s (%s)\n", workingfile, pb.FormatBytes(ptr.Size))
xfers := manifest.GetDownloadAdapterNames()
@ -86,19 +86,19 @@ func downloadFile(writer io.Writer, ptr *Pointer, workingfile, mediafile string,
}
adapter := manifest.NewDownloadAdapter(adapterName)
var tcb transfer.TransferProgressCallback
var tcb tq.TransferProgressCallback
if cb != nil {
tcb = func(name string, totalSize, readSoFar int64, readSinceLast int) error {
return cb(totalSize, readSoFar, readSinceLast)
}
}
// Single download
adapterResultChan := make(chan transfer.TransferResult, 1)
adapterResultChan := make(chan tq.TransferResult, 1)
err = adapter.Begin(1, tcb, adapterResultChan)
if err != nil {
return err
}
adapter.Add(transfer.NewTransfer(filepath.Base(workingfile), obj, mediafile))
adapter.Add(tq.NewTransfer(filepath.Base(workingfile), obj, mediafile))
adapter.End()
res := <-adapterResultChan

3
lfs/upload_queue.go
View File

@ -9,7 +9,6 @@ import (
"github.com/git-lfs/git-lfs/config"
"github.com/git-lfs/git-lfs/errors"
"github.com/git-lfs/git-lfs/tq"
"github.com/git-lfs/git-lfs/transfer"
)
// Uploadable describes a file that can be uploaded.
@ -69,7 +68,7 @@ func NewUploadable(oid, filename string) (*Uploadable, error) {
// NewUploadQueue builds an UploadQueue, allowing `workers` concurrent uploads.
func NewUploadQueue(options ...tq.TransferQueueOption) *tq.TransferQueue {
return tq.NewTransferQueue(transfer.Upload, options...)
return tq.NewTransferQueue(tq.Upload, options...)
}
// ensureFile makes sure that the cleanPath exists before pushing it. If it

2
transfer/adapterbase.go → tq/adapterbase.go
View File

@ -1,4 +1,4 @@
package transfer
package tq
import (
"fmt"

2
transfer/basic_download.go → tq/basic_download.go
View File

@ -1,4 +1,4 @@
package transfer
package tq
import (
"fmt"

2
transfer/basic_upload.go → tq/basic_upload.go
View File

@ -1,4 +1,4 @@
package transfer
package tq
import (
"fmt"

2
transfer/custom.go → tq/custom.go
View File

@ -1,4 +1,4 @@
package transfer
package tq
import (
"bufio"

2
transfer/custom_test.go → tq/custom_test.go
View File

@ -1,4 +1,4 @@
package transfer
package tq
import (
"testing"

2
transfer/manifest.go → tq/manifest.go
View File

@ -1,4 +1,4 @@
package transfer
package tq
import (
"sync"

2
transfer/transfer.go → tq/transfer.go
View File

@ -1,6 +1,6 @@
// Package transfer collects together adapters for uploading and downloading LFS content
// NOTE: Subject to change, do not rely on this package from outside git-lfs source
package transfer
package tq
import "github.com/git-lfs/git-lfs/api"

31
tq/transfer_queue.go
View File

@ -8,7 +8,6 @@ import (
"github.com/git-lfs/git-lfs/config"
"github.com/git-lfs/git-lfs/errors"
"github.com/git-lfs/git-lfs/progress"
"github.com/git-lfs/git-lfs/transfer"
"github.com/rubyist/tracerx"
)
@ -112,8 +111,8 @@ func (b Batch) Swap(i, j int) { b[i], b[j] = b[j], b[i] }
// including calling the API, passing the actual transfer request to transfer
// adapters, and dealing with progress, errors and retries.
type TransferQueue struct {
direction transfer.Direction
adapter transfer.TransferAdapter
direction Direction
adapter TransferAdapter
adapterInProgress bool
adapterInitMutex sync.Mutex
dryRun bool
@ -133,7 +132,7 @@ type TransferQueue struct {
// once per unique OID on Add(), and is decremented when that transfer
// is marked as completed or failed, but not retried.
wait sync.WaitGroup
manifest *transfer.Manifest
manifest *Manifest
rc *retryCounter
}
@ -160,7 +159,7 @@ func WithBufferDepth(depth int) TransferQueueOption {
}
// NewTransferQueue builds a TransferQueue, direction and underlying mechanism determined by adapter
func NewTransferQueue(dir transfer.Direction, options ...TransferQueueOption) *TransferQueue {
func NewTransferQueue(dir Direction, options ...TransferQueueOption) *TransferQueue {
q := &TransferQueue{
batchSize: defaultBatchSize,
bufferDepth: defaultBatchSize,
@ -168,7 +167,7 @@ func NewTransferQueue(dir transfer.Direction, options ...TransferQueueOption) *T
errorc: make(chan error),
transferables: make(map[string]Transferable),
trMutex: &sync.Mutex{},
manifest: transfer.ConfigureManifest(transfer.NewManifest(), config.Config),
manifest: ConfigureManifest(NewManifest(), config.Config),
rc: newRetryCounter(config.Config),
}
@ -216,7 +215,7 @@ func (q *TransferQueue) Add(t Transferable) {
// b. If the read was a Transferable item, go to step 3.
// 3. Append the item to the batch.
// 4. Sort the batch by descending object size, make a batch API call, send
// the items to the `*transfer.adapterBase`.
// the items to the `*adapterBase`.
// 5. Process the worker results, incrementing and appending retries if
// possible.
// 6. If the `q.incoming` channel is open, go to step 2.
@ -299,7 +298,7 @@ func (q *TransferQueue) enqueueAndCollectRetriesFor(batch Batch) (Batch, error)
q.useAdapter(adapterName)
q.startProgress.Do(q.meter.Start)
toTransfer := make([]*transfer.Transfer, 0, len(objs))
toTransfer := make([]*Transfer, 0, len(objs))
for _, o := range objs {
if o.Error != nil {
@ -325,7 +324,7 @@ func (q *TransferQueue) enqueueAndCollectRetriesFor(batch Batch) (Batch, error)
t.SetObject(o)
q.meter.StartTransfer(t.Name())
toTransfer = append(toTransfer, transfer.NewTransfer(
toTransfer = append(toTransfer, NewTransfer(
t.Name(), t.Object(), t.Path(),
))
} else {
@ -367,7 +366,7 @@ func (q *TransferQueue) makeBatch() Batch { return make(Batch, 0, q.batchSize) }
// closed.
//
// addToAdapter returns immediately, and does not block.
func (q *TransferQueue) addToAdapter(pending []*transfer.Transfer) <-chan Transferable {
func (q *TransferQueue) addToAdapter(pending []*Transfer) <-chan Transferable {
retries := make(chan Transferable, len(pending))
if err := q.ensureAdapterBegun(); err != nil {
@ -385,7 +384,7 @@ func (q *TransferQueue) addToAdapter(pending []*transfer.Transfer) <-chan Transf
go func() {
defer close(retries)
var results <-chan transfer.TransferResult
var results <-chan TransferResult
if q.dryRun {
results = q.makeDryRunResults(pending)
} else {
@ -402,10 +401,10 @@ func (q *TransferQueue) addToAdapter(pending []*transfer.Transfer) <-chan Transf
// makeDryRunResults returns a channel populated immediately with "successful"
// results for all of the given transfers in "ts".
func (q *TransferQueue) makeDryRunResults(ts []*transfer.Transfer) <-chan transfer.TransferResult {
results := make(chan transfer.TransferResult, len(ts))
func (q *TransferQueue) makeDryRunResults(ts []*Transfer) <-chan TransferResult {
results := make(chan TransferResult, len(ts))
for _, t := range ts {
results <- transfer.TransferResult{t, nil}
results <- TransferResult{t, nil}
}
close(results)
@ -416,7 +415,7 @@ func (q *TransferQueue) makeDryRunResults(ts []*transfer.Transfer) <-chan transf
// handleTransferResult observes the transfer result, sending it on the retries
// channel if it was able to be retried.
func (q *TransferQueue) handleTransferResult(
res transfer.TransferResult, retries chan<- Transferable,
res TransferResult, retries chan<- Transferable,
) {
oid := res.Transfer.Object.Oid
@ -490,7 +489,7 @@ func (q *TransferQueue) Skip(size int64) {
}
func (q *TransferQueue) transferKind() string {
if q.direction == transfer.Download {
if q.direction == Download {
return "download"
} else {
return "upload"

2
transfer/transfer_test.go → tq/transfer_test.go
View File

@ -1,4 +1,4 @@
package transfer
package tq
import (
"testing"

2
transfer/tus_upload.go → tq/tus_upload.go
View File

@ -1,4 +1,4 @@
package transfer
package tq
import (
"fmt"