Refactor: Pipeline module (#417)

Author: Zweih

cmd/qp/fork.go

@@ -53,7 +53,7 @@ func rebuildCache(originName string) error {
 		wg.Add(1)
 		go func(targetDriver driver.Driver) {
 			defer wg.Done()
-			pipeline := phase.NewPipeline(targetDriver, cfg, false, cacheBasePath)
+			pipeline := phase.NewPipeline(targetDriver, cfg, cacheBasePath)
 
 			if storage.IsLockFileExists(pipeline.CachePath) {
 				return

cmd/qp/main.go

@@ -47,7 +47,7 @@ func mainWithConfig(configProvider config.ConfigProvider) error {
 
 	var pipelines []*phase.Pipeline
 	for _, driver := range origins.AvailableDrivers() {
-		p := phase.NewPipeline(driver, cfg, isInteractive, cacheBaseDir)
+		p := phase.NewPipeline(driver, cfg, cacheBaseDir)
 		pipelines = append(pipelines, p)
 	}
 

internal/pipeline/phase/phase.go

@@ -1,25 +1,15 @@
 package phase
 
 import (
-	"fmt"
 	"qp/internal/config"
-	out "qp/internal/display"
-	"qp/internal/pipeline/meta"
 	"qp/internal/pkgdata"
 	"sync"
 )
 
-type (
-	ProgressReporter = meta.ProgressReporter
-	ProgressMessage  = meta.ProgressMessage
-	PkgInfo          = pkgdata.PkgInfo
-)
-
 type Step func(
 	cfg *config.Config,
-	packages []*PkgInfo,
-	progressReporter meta.ProgressReporter,
-) ([]*PkgInfo, error)
+	pkgs []*pkgdata.PkgInfo,
+) ([]*pkgdata.PkgInfo, error)
 
 type PipelinePhase struct {
 	name string
@@ -28,75 +18,17 @@ type PipelinePhase struct {
 }
 
 func NewPhase(name string, step Step, wg *sync.WaitGroup) PipelinePhase {
-	return PipelinePhase{
-		name,
-		step,
-		wg,
-	}
+	return PipelinePhase{name, step, wg}
 }
 
 func (phase PipelinePhase) Run(
 	cfg *config.Config,
-	packages []*PkgInfo,
-	isInteractive bool,
-) ([]*PkgInfo, error) {
-	progressChan := phase.startProgress(isInteractive)
-	outputPackages, err := phase.step(
-		cfg,
-		packages,
-		phase.reportProgress(progressChan),
-	)
-	phase.stopProgress(progressChan)
-
+	pkgs []*pkgdata.PkgInfo,
+) ([]*pkgdata.PkgInfo, error) {
+	outputPackages, err := phase.step(cfg, pkgs)
 	if err != nil {
 		return nil, err
 	}
 
 	return outputPackages, nil
 }
-
-func (phase PipelinePhase) reportProgress(progressChan chan ProgressMessage) ProgressReporter {
-	if progressChan == nil {
-		return ProgressReporter(func(_ int, _ int, _ string) {})
-	}
-
-	return ProgressReporter(func(current int, total int, phaseName string) {
-		progressChan <- ProgressMessage{
-			Phase:       phaseName,
-			Progress:    (current * 100) / total,
-			Description: fmt.Sprintf(("%s is in progress..."), phase.name),
-		}
-	})
-}
-
-func (phase PipelinePhase) startProgress(isInteractive bool) chan ProgressMessage {
-	if !isInteractive {
-		return nil
-	}
-
-	progressChan := make(chan ProgressMessage)
-	phase.wg.Add(1)
-
-	go func() {
-		defer phase.wg.Done()
-		phase.displayProgress(progressChan)
-	}()
-
-	return progressChan
-}
-
-func (phase PipelinePhase) stopProgress(progressChan chan ProgressMessage) {
-	if progressChan != nil {
-		close(progressChan)
-		phase.wg.Wait()
-		out.ClearProgress()
-	}
-}
-
-func (phase PipelinePhase) displayProgress(progressChan chan ProgressMessage) {
-	for msg := range progressChan {
-		out.PrintProgress(msg.Phase, msg.Progress, msg.Description)
-	}
-
-	out.ClearProgress()
-}

internal/pipeline/phase/pipeline.go

@@ -10,28 +10,25 @@ import (
 )
 
 type Pipeline struct {
-	Origin        driver.Driver
-	Config        *config.Config
-	Pkgs          []*pkgdata.PkgInfo
-	IsInteractive bool
-	UsedCache     bool
-	CachePath     string
-	ModTime       int64
+	Origin    driver.Driver
+	Config    *config.Config
+	Pkgs      []*pkgdata.PkgInfo
+	UsedCache bool
+	CachePath string
+	ModTime   int64
 }
 
 func NewPipeline(
 	origin driver.Driver,
 	cfg *config.Config,
-	isInteractive bool,
 	baseCacheDir string,
 ) *Pipeline {
 	cachePath := filepath.Join(baseCacheDir, origin.Name())
 
 	return &Pipeline{
-		Origin:        origin,
-		Config:        cfg,
-		IsInteractive: isInteractive,
-		CachePath:     cachePath,
+		Origin:    origin,
+		Config:    cfg,
+		CachePath: cachePath,
 	}
 }
 
@@ -45,11 +42,11 @@ func (p *Pipeline) Run() ([]*pkgdata.PkgInfo, error) {
 		NewPhase("Save cache", p.saveCacheStep, &wg),
 	}
 
-	pkgs := []*PkgInfo{}
+	pkgs := []*pkgdata.PkgInfo{}
 	var err error
 
 	for _, ph := range phases {
-		pkgs, err = ph.Run(p.Config, pkgs, p.IsInteractive)
+		pkgs, err = ph.Run(p.Config, pkgs)
 		if err != nil {
 			return nil, fmt.Errorf("[%s] %w", ph.name, err)
 		}
@@ -66,11 +63,11 @@ func (p *Pipeline) RunCacheOnly() ([]*pkgdata.PkgInfo, error) {
 		NewPhase("Save cache", p.saveCacheStep, &wg),
 	}
 
-	pkgs := []*PkgInfo{}
+	pkgs := []*pkgdata.PkgInfo{}
 	var err error
 
 	for _, ph := range phases {
-		pkgs, err = ph.Run(p.Config, pkgs, p.IsInteractive)
+		pkgs, err = ph.Run(p.Config, pkgs)
 		if err != nil {
 			return nil, fmt.Errorf("[%s] %w", ph.name, err)
 		}

internal/pipeline/phase/steps.go

@@ -4,7 +4,6 @@ import (
 	"fmt"
 	"qp/internal/config"
 	out "qp/internal/display"
-	"qp/internal/pipeline/meta"
 	"qp/internal/pkgdata"
 	"qp/internal/storage"
 	"time"
@@ -13,7 +12,6 @@ import (
 func (p *Pipeline) loadCacheStep(
 	cfg *config.Config,
 	_ []*pkgdata.PkgInfo,
-	_ meta.ProgressReporter,
 ) ([]*pkgdata.PkgInfo, error) {
 	now := time.Now().Unix()
 	if cfg.RegenCache {
@@ -50,7 +48,6 @@ func (p *Pipeline) loadCacheStep(
 func (p *Pipeline) fetchStep(
 	_ *config.Config,
 	pkgs []*pkgdata.PkgInfo,
-	_ meta.ProgressReporter,
 ) ([]*pkgdata.PkgInfo, error) {
 	if p.UsedCache {
 		return pkgs, nil
@@ -70,7 +67,6 @@ func (p *Pipeline) fetchStep(
 func (p *Pipeline) resolveStep(
 	_ *config.Config,
 	pkgs []*pkgdata.PkgInfo,
-	reportProgress meta.ProgressReporter,
 ) ([]*pkgdata.PkgInfo, error) {
 	if p.UsedCache {
 		return pkgs, nil
@@ -87,7 +83,6 @@ func (p *Pipeline) resolveStep(
 func (p *Pipeline) saveCacheStep(
 	cfg *config.Config,
 	pkgs []*pkgdata.PkgInfo,
-	_ meta.ProgressReporter,
 ) ([]*pkgdata.PkgInfo, error) {
 	if cfg.NoCache || p.UsedCache {
 		return pkgs, nil