func (r *Manager) ReflectAll() (*RunStats, error) {
	start := time.Now()

	logger.Log().Infof("starting reflection")

	pendingFilenames := []string{}
	stats := &RunStats{}

	logger.Log().Debugf("checking %v for blobs...", r.blobsPath)
	f, err := os.Open(r.blobsPath)
	if err != nil {
		return nil, errors.Err(err)
	}

	entries, err := f.Readdir(-1)
	if err != nil {
		return nil, errors.Err(err)
	}
	err = f.Close()
	if err != nil {
		return nil, errors.Err(err)
	}

	for _, file := range entries {
		if !file.IsDir() {
			pendingFilenames = append(pendingFilenames, path.Join(r.blobsPath, file.Name()))
		}
	}
	stats.TotalBlobs = len(pendingFilenames)
	logger.Log().Debugf("%v blobs found", stats.TotalBlobs)

	var wg sync.WaitGroup
	wgSize := 10

	for i, f := range pendingFilenames {
		select {
		case <-r.abortUploadChan:
			break
		default:
		}

		wg.Add(1)
		go func(wf string) {
			rc := reflector.Client{}
			err := rc.Connect(r.reflector)
			if err != nil {
				return
			}
			b, err := ioutil.ReadFile(wf)
			if err != nil {
				stats.Lock()
				stats.Errors = append(stats.Errors, ReflError{wf, err})
				stats.Unlock()
			}

			err = rc.SendBlob(b)
			if errors.Is(err, reflector.ErrBlobExists) || err == nil {
				stats.Lock()
				stats.ReflectedBlobs++
				stats.Unlock()
				if err := os.Remove(wf); err != nil {
					stats.Lock()
					stats.Errors = append(stats.Errors, ReflError{wf, err})
					stats.Unlock()
				}
			} else {
				stats.Lock()
				stats.Errors = append(stats.Errors, ReflError{wf, err})
				stats.Unlock()
			}
			wg.Done()
		}(f)
		if (i%wgSize) == 0 || len(pendingFilenames)-1-i == 0 {
			wg.Wait()
		}
	}

	logger.Log().Infof("reflection run complete in %.2f minutes", time.Since(start).Minutes())

	return stats, nil
}
// ReflectAll uploads and then deletes all blobs in the blob directory.
ReflectAll uploads and then deletes all blobs in the blob directory.
// ReflectAll uploads and then deletes all blobs in the blob directory .
func (r *Manager) ReflectAll() (*RunStats, error) {
	start := time.Now()
	logger.Log().Infof("starting reflection")
	pendingFilenames := []string{}
	stats := &RunStats{}
	logger.Log().Debugf("checking %v for blobs...", r.blobsPath)
	f, err := os.Open(r.blobsPath)
	if err != nil {
		return nil, errors.Err(err)
	}
	entries, err := f.Readdir(-1)
	if err != nil {
		return nil, errors.Err(err)
	}
	err = f.Close()
	if err != nil {
		return nil, errors.Err(err)
	}
	for _, file := range entries {
		if !file.IsDir() {
			pendingFilenames = append(pendingFilenames, path.Join(r.blobsPath, file.Name()))
		}
	}
	stats.TotalBlobs = len(pendingFilenames)
	logger.Log().Debugf("%v blobs found", stats.TotalBlobs)
	var wg sync.WaitGroup
	wgSize := 10
	for i, f := range pendingFilenames {
		select {
		case <-r.abortUploadChan:
			break
		default:
		}
		wg.Add(1)
		go func(wf string) {
			rc := reflector.Client{}
			err := rc.Connect(r.reflector)
			if err != nil {
				return
			}
			b, err := ioutil.ReadFile(wf)
			if err != nil {
				stats.Lock()
				stats.Errors = append(stats.Errors, ReflError{wf, err})
				stats.Unlock()
			}
			err = rc.SendBlob(b)
			if errors.Is(err, reflector.ErrBlobExists) || err == nil {
				stats.Lock()
				stats.ReflectedBlobs++
				stats.Unlock()
				if err := os.Remove(wf); err != nil {
					stats.Lock()
					stats.Errors = append(stats.Errors, ReflError{wf, err})
					stats.Unlock()
				}
			} else {
				stats.Lock()
				stats.Errors = append(stats.Errors, ReflError{wf, err})
				stats.Unlock()
			}
			wg.Done()
		}(f)
		if (i%wgSize) == 0 || len(pendingFilenames)-1-i == 0 {
			wg.Wait()
		}
	}
	logger.Log().Infof("reflection run complete in %.2f minutes", time.Since(start).Minutes())
	return stats, nil
}
func ( r * Manager ) ReflectAll ( ) ( * RunStats , error ) { start := time . Now ( ) logger . Log ( ) . Infof ( " " ) pendingFilenames := [ ] string { } stats := & RunStats { } logger . Log ( ) . Debugf ( " " , r . blobsPath ) f , err := os . Open ( r . blobsPath ) if err != nil { return nil , errors . Err ( err ) } entries , err := f . Readdir ( - 1 ) if err != nil { return nil , errors . Err ( err ) } err = f . Close ( ) if err != nil { return nil , errors . Err ( err ) } for _ , file := range entries { if ! file . IsDir ( ) { pendingFilenames = append ( pendingFilenames , path . Join ( r . blobsPath , file . Name ( ) ) ) } } stats . TotalBlobs = len ( pendingFilenames ) logger . Log ( ) . Debugf ( " " , stats . TotalBlobs ) var wg sync . WaitGroup wgSize := 10 for i , f := range pendingFilenames { select { case <- r . abortUploadChan : break default : } wg . Add ( 1 ) go func ( wf string ) { rc := reflector . Client { } err := rc . Connect ( r . reflector ) if err != nil { return } b , err := ioutil . ReadFile ( wf ) if err != nil { stats . Lock ( ) stats . Errors = append ( stats . Errors , ReflError { wf , err } ) stats . Unlock ( ) } err = rc . SendBlob ( b ) if errors . Is ( err , reflector . ErrBlobExists ) || err == nil { stats . Lock ( ) stats . ReflectedBlobs ++ stats . Unlock ( ) if err := os . Remove ( wf ) ; err != nil { stats . Lock ( ) stats . Errors = append ( stats . Errors , ReflError { wf , err } ) stats . Unlock ( ) } } else { stats . Lock ( ) stats . Errors = append ( stats . Errors , ReflError { wf , err } ) stats . Unlock ( ) } wg . Done ( ) } ( f ) if ( i % wgSize ) == 0 || len ( pendingFilenames ) - 1 - i == 0 { wg . Wait ( ) } } logger . Log ( ) . Infof ( " " , time . Since ( start ) . Minutes ( ) ) return stats , nil }
ReflectAll uploads and then deletes all blobs in the blob directory.
ReflectAll uploads and then deletes all blobs in the blob directory .