Method addToAdapter

tq/transfer_queue.go:711–749

addToAdapter adds the given "pending" transfers to the transfer adapters and returns a channel of Transfers that are to be retried in the next batch. After all of the items in the batch have been processed, the channel is closed. addToAdapter returns immediately, and does not block.

(e lfshttp.Endpoint, pending []*Transfer)

//
// addToAdapter returns immediately, and does not block.
func (q *TransferQueue) addToAdapter(e lfshttp.Endpoint, pending []*Transfer) <-chan *objectTuple {
	q.Upgrade()

	retries := make(chan *objectTuple, len(pending))

	if err := q.ensureAdapterBegun(e); err != nil {
		close(retries)

		q.errorc <- err
		for _, t := range pending {
			q.Skip(t.Size)
			q.wait.Done()
		}

		return retries
	}

	present, missingResults := q.partitionTransfers(pending)

	go func() {
		defer close(retries)

		var results <-chan TransferResult
		if q.dryRun {
			results = q.makeDryRunResults(present)
		} else {
			results = q.adapter.Add(present...)
		}

		for _, res := range missingResults {
			q.handleTransferResult(res, retries)
		}
		for res := range results {
			q.handleTransferResult(res, retries)
		}
	}()

	return retries
}

func (q *TransferQueue) partitionTransfers(transfers []*Transfer) (present []*Transfer, results []TransferResult) {
	q.Upgrade()