func (u *Uploader) worker(pathChan chan string) {
	for {
		select {
		case <-u.stopper.Ch():
			return
		case filepath, ok := <-pathChan:
			if !ok {
				return
			}
			err := u.uploadBlob(filepath)
			if err != nil {
				log.Errorln(err)
			} else if u.deleteBlobsAfterUpload {
				err = os.Remove(filepath)
				if err != nil {
					log.Errorln(errors.Prefix("deleting blob", err))
				}
			}
		}
	}
}
worker reads paths from a channel,  uploads them, and optionally deletes them
package reflector
import (
	"io/ioutil"
	"os"
	"path"
	"sync"
	"time"
	"github.com/lbryio/reflector.go/db"
	"github.com/lbryio/reflector.go/internal/metrics"
	"github.com/lbryio/reflector.go/store"
	"github.com/lbryio/lbry.go/v2/extras/errors"
	"github.com/lbryio/lbry.go/v2/extras/stop"
	log "github.com/sirupsen/logrus"
)
type increment int
const (
	sdInc increment = iota + 1
	blobInc
	errInc
)
type Summary struct {
	Total, AlreadyStored, Sd, Blob, Err int
}
type Uploader struct {
	db                     *db.SQL
	store                  *store.DBBackedStore  
	workers                int
	skipExistsCheck        bool
	deleteBlobsAfterUpload bool
	stopper                *stop.Group
	countChan              chan increment
	count Summary
}
func NewUploader(db *db.SQL, store *store.DBBackedStore, workers int, skipExistsCheck, deleteBlobsAfterUpload bool) *Uploader {
	return &Uploader{
		db:                     db,
		store:                  store,
		workers:                workers,
		skipExistsCheck:        skipExistsCheck,
		deleteBlobsAfterUpload: deleteBlobsAfterUpload,
		stopper:                stop.New(),
		countChan:              make(chan increment),
	}
}
func (u *Uploader) Stop() {
	log.Infoln("stopping uploader")
	u.stopper.StopAndWait()
}
func (u *Uploader) Upload(dirOrFilePath string) error {
	paths, err := getPaths(dirOrFilePath)
	if err != nil {
		return err
	}
	u.count.Total = len(paths)
	hashes := make([]string, len(paths))
	for i, p := range paths {
		hashes[i] = path.Base(p)
	}
	log.Debug("checking for existing blobs")
	var exists map[string]bool
	if !u.skipExistsCheck {
		exists, err = u.db.HasBlobs(hashes, false)
		if err != nil {
			return err
		}
		u.count.AlreadyStored = len(exists)
	}
	log.Debugf("%d new blobs to upload", u.count.Total-u.count.AlreadyStored)
	workerWG := sync.WaitGroup{}
	pathChan := make(chan string)
	for i := 0; i < u.workers; i++ {
		workerWG.Add(1)
		metrics.RoutinesQueue.WithLabelValues("reflector", "upload").Inc()
		go func(i int) {
			defer metrics.RoutinesQueue.WithLabelValues("reflector", "upload").Dec()
			defer workerWG.Done()
			defer func(i int) { log.Debugf("worker %d quitting", i) }(i)
			u.worker(pathChan)
		}(i)
	}
	countWG := sync.WaitGroup{}
	countWG.Add(1)
	metrics.RoutinesQueue.WithLabelValues("reflector", "uploader").Inc()
	go func() {
		defer metrics.RoutinesQueue.WithLabelValues("reflector", "uploader").Dec()
		defer countWG.Done()
		u.counter()
	}()
Upload:
	for _, f := range paths {
		if exists != nil && exists[path.Base(f)] {
			continue
		}
		select {
		case pathChan <- f:
		case <-u.stopper.Ch():
			break Upload
		}
	}
	close(pathChan)
	workerWG.Wait()
	close(u.countChan)
	countWG.Wait()
	u.stopper.Stop()
	log.Debugf(
		"upload stats: %d blobs total, %d already stored, %d SD blobs uploaded, %d content blobs uploaded, %d errors",
		u.count.Total, u.count.AlreadyStored, u.count.Sd, u.count.Blob, u.count.Err,
	)
	return nil
}