refactor: simplify reader with manager and messenger

Signed-off-by: Terry Howe <[email protected]>

Author: TerryHowe

cmd/oras/internal/display/status/progress/manager.go

@@ -45,19 +45,23 @@ type manager struct {
 	status       []*status
 	statusLock   sync.RWMutex
 	console      *console.Console
+	actionPrompt string
+	donePrompt   string
 	updating     sync.WaitGroup
 	renderDone   chan struct{}
 	renderClosed chan struct{}
 }
 
 // NewManager initialized a new progress manager.
-func NewManager(f *os.File) (Manager, error) {
-	c, err := console.New(f)
+func NewManager(actionPrompt string, donePrompt string, tty *os.File) (Manager, error) {
+	c, err := console.New(tty)
 	if err != nil {
 		return nil, err
 	}
 	m := &manager{
 		console:      c,
+		actionPrompt: actionPrompt,
+		donePrompt:   donePrompt,
 		renderDone:   make(chan struct{}),
 		renderClosed: make(chan struct{}),
 	}

cmd/oras/internal/display/status/track/reader.go

@@ -17,56 +17,44 @@ package track
 
 import (
 	"io"
-	"os"
 
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
 	"oras.land/oras/cmd/oras/internal/display/status/progress"
 )
 
+type Reader interface {
+	io.Reader
+	Done()
+	Close()
+	Start()
+}
+
 type reader struct {
 	base         io.Reader
 	offset       int64
 	actionPrompt string
 	donePrompt   string
 	descriptor   ocispec.Descriptor
-	manager      progress.Manager
 	messenger    *progress.Messenger
+	done         bool
 }
 
 // NewReader returns a new reader with tracked progress.
-func NewReader(r io.Reader, descriptor ocispec.Descriptor, actionPrompt string, donePrompt string, tty *os.File) (*reader, error) {
-	manager, err := progress.NewManager(tty)
-	if err != nil {
-		return nil, err
+func NewReader(r io.Reader, descriptor ocispec.Descriptor, messenger *progress.Messenger) Reader {
+	tr := reader{
+		base:       r,
+		descriptor: descriptor,
+		messenger:  messenger,
 	}
-	return managedReader(r, descriptor, manager, actionPrompt, donePrompt)
-}
-
-func managedReader(r io.Reader, descriptor ocispec.Descriptor, manager progress.Manager, actionPrompt string, donePrompt string) (*reader, error) {
-	messenger, err := manager.Add()
-	if err != nil {
-		return nil, err
-	}
-
-	return &reader{
-		base:         r,
-		descriptor:   descriptor,
-		actionPrompt: actionPrompt,
-		donePrompt:   donePrompt,
-		manager:      manager,
-		messenger:    messenger,
-	}, nil
-}
-
-// StopManager stops the messenger channel and related manager.
-func (r *reader) StopManager() {
-	r.Close()
-	_ = r.manager.Close()
+	return &tr
 }
 
 // Done sends message to mark the tracked progress as complete.
 func (r *reader) Done() {
-	r.messenger.Send(r.donePrompt, r.descriptor, r.descriptor.Size)
+	if !r.done {
+		r.messenger.Send(r.donePrompt, r.descriptor, r.descriptor.Size)
+		r.done = true
+	}
 	r.messenger.Stop()
 }
 

cmd/oras/internal/display/status/track/target.go

@@ -35,9 +35,7 @@ type GraphTarget interface {
 
 type graphTarget struct {
 	oras.GraphTarget
-	manager      progress.Manager
-	actionPrompt string
-	donePrompt   string
+	manager progress.Manager
 }
 
 type referenceGraphTarget struct {
@@ -46,15 +44,13 @@ type referenceGraphTarget struct {
 
 // NewTarget creates a new tracked Target.
 func NewTarget(t oras.GraphTarget, actionPrompt, donePrompt string, tty *os.File) (GraphTarget, error) {
-	manager, err := progress.NewManager(tty)
+	manager, err := progress.NewManager(actionPrompt, donePrompt, tty)
 	if err != nil {
 		return nil, err
 	}
 	gt := &graphTarget{
-		GraphTarget:  t,
-		manager:      manager,
-		actionPrompt: actionPrompt,
-		donePrompt:   donePrompt,
+		GraphTarget: t,
+		manager:     manager,
 	}
 
 	if _, ok := t.(registry.ReferencePusher); ok {
@@ -74,7 +70,13 @@ func (t *graphTarget) Mount(ctx context.Context, desc ocispec.Descriptor, fromRe
 
 // Push pushes the content to the base oras.GraphTarget with tracking.
 func (t *graphTarget) Push(ctx context.Context, expected ocispec.Descriptor, content io.Reader) error {
-	r, err := managedReader(content, expected, t.manager, t.actionPrompt, t.donePrompt)
+	messenger, err := t.manager.Add()
+	if err != nil {
+		return err
+	}
+	defer messenger.Stop()
+
+	r := NewReader(content, expected, messenger)
 	if err != nil {
 		return err
 	}
@@ -89,7 +91,13 @@ func (t *graphTarget) Push(ctx context.Context, expected ocispec.Descriptor, con
 
 // PushReference pushes the content to the base oras.GraphTarget with tracking.
 func (rgt *referenceGraphTarget) PushReference(ctx context.Context, expected ocispec.Descriptor, content io.Reader, reference string) error {
-	r, err := managedReader(content, expected, rgt.manager, rgt.actionPrompt, rgt.donePrompt)
+	messenger, err := rgt.manager.Add()
+	if err != nil {
+		return err
+	}
+	defer messenger.Stop()
+
+	r := NewReader(content, expected, messenger)
 	if err != nil {
 		return err
 	}

cmd/oras/root/blob/fetch.go

@@ -28,6 +28,7 @@ import (
 	"oras.land/oras-go/v2/registry/remote"
 	"oras.land/oras/cmd/oras/internal/argument"
 	"oras.land/oras/cmd/oras/internal/command"
+	"oras.land/oras/cmd/oras/internal/display/status/progress"
 	"oras.land/oras/cmd/oras/internal/display/status/track"
 	oerrors "oras.land/oras/cmd/oras/internal/errors"
 	"oras.land/oras/cmd/oras/internal/option"
@@ -170,12 +171,20 @@ func (opts *fetchBlobOptions) doFetch(ctx context.Context, src oras.ReadOnlyTarg
 			return ocispec.Descriptor{}, err
 		}
 	} else {
-		// TTY output
-		trackedReader, err := track.NewReader(vr, desc, "Downloading", "Downloaded ", opts.TTY)
+		manager, err := progress.NewManager("Downloading", "Downloaded ", opts.TTY)
 		if err != nil {
-			return ocispec.Descriptor{}, err
+			return desc, err
 		}
-		defer trackedReader.StopManager()
+		defer manager.Close()
+
+		messenger, err := manager.Add()
+		if err != nil {
+			return desc, err
+		}
+		defer messenger.Stop()
+
+		// TTY output
+		trackedReader := track.NewReader(vr, desc, messenger)
 		trackedReader.Start()
 		if _, err = io.Copy(writer, trackedReader); err != nil {
 			return ocispec.Descriptor{}, err

cmd/oras/root/blob/push.go

@@ -19,6 +19,7 @@ import (
 	"context"
 	"errors"
 	"io"
+	"oras.land/oras/cmd/oras/internal/display/status/progress"
 	"os"
 
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
@@ -154,12 +155,20 @@ func (opts *pushBlobOptions) doPush(ctx context.Context, printer *output.Printer
 		return printer.PrintStatus(desc, "Uploaded ")
 	}
 
-	// TTY output
-	trackedReader, err := track.NewReader(r, desc, "Uploading", "Uploaded ", opts.TTY)
+	manager, err := progress.NewManager("Downloading", "Downloaded ", opts.TTY)
 	if err != nil {
 		return err
 	}
-	defer trackedReader.StopManager()
+	defer manager.Close()
+
+	messenger, err := manager.Add()
+	if err != nil {
+		return err
+	}
+	defer messenger.Stop()
+
+	// TTY output
+	trackedReader := track.NewReader(r, desc, messenger)
 	trackedReader.Start()
 	r = trackedReader
 	if err := t.Push(ctx, desc, r); err != nil {