From 632882e31983045b5e05f2998d540da811468a0c Mon Sep 17 00:00:00 2001 From: Shiwei Zhang Date: Wed, 27 Nov 2024 16:51:41 +0800 Subject: [PATCH 01/15] feat: new progress package Signed-off-by: Shiwei Zhang --- progress/interface.go | 63 +++++++++++++++++++++++++++++++++++++++++++ progress/reader.go | 56 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 119 insertions(+) create mode 100644 progress/interface.go create mode 100644 progress/reader.go diff --git a/progress/interface.go b/progress/interface.go new file mode 100644 index 00000000..1a7fe790 --- /dev/null +++ b/progress/interface.go @@ -0,0 +1,63 @@ +package progress + +import ( + "io" + + ocispec "github.com/opencontainers/image-spec/specs-go/v1" +) + +// State represents the state of a descriptor. +type State int + +// Registered states. +const ( + StateUnknown State = iota + StateInitialized + StateTransmitting + StateTransmitted + StateExists + StateSkipped + StateMounted +) + +// Status represents the status of a descriptor. +type Status struct { + // State represents the state of the descriptor. + State State + + // Offset represents the current offset of the descriptor. + // Offset is discarded if set to a negative value. + Offset int64 +} + +// Tracker updates the status of a descriptor. +type Tracker interface { + io.Closer + + // Update updates the status of the descriptor. + Update(status Status) error + + // Fail marks the descriptor as failed. + Fail(err error) error +} + +// Manager tracks the progress of multiple descriptors. +type Manager interface { + io.Closer + + // Track starts tracking the progress of a descriptor. + Track(desc ocispec.Descriptor) (Tracker, error) +} + +// Record adds the progress of a descriptor as a single entry. +func Record(m Manager, desc ocispec.Descriptor, status Status) error { + tracker, err := m.Track(desc) + if err != nil { + return err + } + err = tracker.Update(status) + if err != nil { + return err + } + return tracker.Close() +} diff --git a/progress/reader.go b/progress/reader.go new file mode 100644 index 00000000..1adfea92 --- /dev/null +++ b/progress/reader.go @@ -0,0 +1,56 @@ +package progress + +import "io" + +// ReadTracker tracks the transmission based on the read operation. +type ReadTracker struct { + base io.Reader + tracker Tracker + offset int64 +} + +// NewReadTracker attaches a tracker to a reader. +func NewReadTracker(track Tracker, r io.Reader) *ReadTracker { + return &ReadTracker{ + base: r, + tracker: track, + } +} + +// Read reads from the base reader and updates the status. +func (rt *ReadTracker) Read(p []byte) (n int, err error) { + n, err = rt.base.Read(p) + rt.offset += int64(n) + _ = rt.tracker.Update(Status{ + State: StateTransmitting, + Offset: rt.offset, + }) + if err != nil && err != io.EOF { + _ = rt.tracker.Fail(err) + } + return n, err +} + +// Close closes the tracker. +func (rt *ReadTracker) Close() error { + return rt.tracker.Close() +} + +// Start starts tracking the transmission. +func (rt *ReadTracker) Start() error { + return rt.tracker.Update(Status{ + State: StateInitialized, + Offset: -1, + }) +} + +// Done marks the transmission as complete. +// Done should be called after the transmission is complete. +// Note: Reading all content from the reader does not imply the transmission is +// complete. +func (rt *ReadTracker) Done() error { + return rt.tracker.Update(Status{ + State: StateTransmitted, + Offset: -1, + }) +} From b20b75902284b280264db779bcfea382467e6d23 Mon Sep 17 00:00:00 2001 From: Shiwei Zhang Date: Wed, 27 Nov 2024 18:49:33 +0800 Subject: [PATCH 02/15] refactor: split files Signed-off-by: Shiwei Zhang --- progress/interface.go | 63 ------------------------------------------- progress/manager.go | 28 +++++++++++++++++++ progress/reader.go | 19 ------------- progress/status.go | 25 +++++++++++++++++ progress/tracker.go | 33 +++++++++++++++++++++++ 5 files changed, 86 insertions(+), 82 deletions(-) delete mode 100644 progress/interface.go create mode 100644 progress/manager.go create mode 100644 progress/status.go create mode 100644 progress/tracker.go diff --git a/progress/interface.go b/progress/interface.go deleted file mode 100644 index 1a7fe790..00000000 --- a/progress/interface.go +++ /dev/null @@ -1,63 +0,0 @@ -package progress - -import ( - "io" - - ocispec "github.com/opencontainers/image-spec/specs-go/v1" -) - -// State represents the state of a descriptor. -type State int - -// Registered states. -const ( - StateUnknown State = iota - StateInitialized - StateTransmitting - StateTransmitted - StateExists - StateSkipped - StateMounted -) - -// Status represents the status of a descriptor. -type Status struct { - // State represents the state of the descriptor. - State State - - // Offset represents the current offset of the descriptor. - // Offset is discarded if set to a negative value. - Offset int64 -} - -// Tracker updates the status of a descriptor. -type Tracker interface { - io.Closer - - // Update updates the status of the descriptor. - Update(status Status) error - - // Fail marks the descriptor as failed. - Fail(err error) error -} - -// Manager tracks the progress of multiple descriptors. -type Manager interface { - io.Closer - - // Track starts tracking the progress of a descriptor. - Track(desc ocispec.Descriptor) (Tracker, error) -} - -// Record adds the progress of a descriptor as a single entry. -func Record(m Manager, desc ocispec.Descriptor, status Status) error { - tracker, err := m.Track(desc) - if err != nil { - return err - } - err = tracker.Update(status) - if err != nil { - return err - } - return tracker.Close() -} diff --git a/progress/manager.go b/progress/manager.go new file mode 100644 index 00000000..b5c289ea --- /dev/null +++ b/progress/manager.go @@ -0,0 +1,28 @@ +package progress + +import ( + "io" + + ocispec "github.com/opencontainers/image-spec/specs-go/v1" +) + +// Manager tracks the progress of multiple descriptors. +type Manager interface { + io.Closer + + // Track starts tracking the progress of a descriptor. + Track(desc ocispec.Descriptor) (Tracker, error) +} + +// Record adds the progress of a descriptor as a single entry. +func Record(m Manager, desc ocispec.Descriptor, status Status) error { + tracker, err := m.Track(desc) + if err != nil { + return err + } + err = tracker.Update(status) + if err != nil { + return err + } + return tracker.Close() +} diff --git a/progress/reader.go b/progress/reader.go index 1adfea92..0b35271b 100644 --- a/progress/reader.go +++ b/progress/reader.go @@ -35,22 +35,3 @@ func (rt *ReadTracker) Read(p []byte) (n int, err error) { func (rt *ReadTracker) Close() error { return rt.tracker.Close() } - -// Start starts tracking the transmission. -func (rt *ReadTracker) Start() error { - return rt.tracker.Update(Status{ - State: StateInitialized, - Offset: -1, - }) -} - -// Done marks the transmission as complete. -// Done should be called after the transmission is complete. -// Note: Reading all content from the reader does not imply the transmission is -// complete. -func (rt *ReadTracker) Done() error { - return rt.tracker.Update(Status{ - State: StateTransmitted, - Offset: -1, - }) -} diff --git a/progress/status.go b/progress/status.go new file mode 100644 index 00000000..35e702c8 --- /dev/null +++ b/progress/status.go @@ -0,0 +1,25 @@ +package progress + +// State represents the state of a descriptor. +type State int + +// Registered states. +const ( + StateUnknown State = iota // unknown state + StateInitialized // progress initialized + StateTransmitting // transmitting content + StateTransmitted // content transmitted + StateExists // content exists + StateSkipped // content skipped + StateMounted // content mounted +) + +// Status represents the status of a descriptor. +type Status struct { + // State represents the state of the descriptor. + State State + + // Offset represents the current offset of the descriptor. + // Offset is discarded if set to a negative value. + Offset int64 +} diff --git a/progress/tracker.go b/progress/tracker.go new file mode 100644 index 00000000..bdd769ce --- /dev/null +++ b/progress/tracker.go @@ -0,0 +1,33 @@ +package progress + +import "io" + +// Tracker updates the status of a descriptor. +type Tracker interface { + io.Closer + + // Update updates the status of the descriptor. + Update(status Status) error + + // Fail marks the descriptor as failed. + Fail(err error) error +} + +// Start starts tracking the transmission. +func Start(t Tracker) error { + return t.Update(Status{ + State: StateInitialized, + Offset: -1, + }) +} + +// Done marks the transmission as complete. +// Done should be called after the transmission is complete. +// Note: Reading all content from the reader does not imply the transmission is +// complete. +func Done(t Tracker) error { + return t.Update(Status{ + State: StateTransmitted, + Offset: -1, + }) +} From 235743dc36f26fa27a8cd30d5f99cdc83b4ec66d Mon Sep 17 00:00:00 2001 From: Shiwei Zhang Date: Wed, 27 Nov 2024 18:54:37 +0800 Subject: [PATCH 03/15] refactor: unexport ReadTracker Signed-off-by: Shiwei Zhang --- progress/reader.go | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/progress/reader.go b/progress/reader.go index 0b35271b..bde03da6 100644 --- a/progress/reader.go +++ b/progress/reader.go @@ -2,23 +2,23 @@ package progress import "io" -// ReadTracker tracks the transmission based on the read operation. -type ReadTracker struct { +// TrackReader bind a reader with a tracker. +func TrackReader(t Tracker, r io.Reader) io.ReadCloser { + return &readTracker{ + base: r, + tracker: t, + } +} + +// readTracker tracks the transmission based on the read operation. +type readTracker struct { base io.Reader tracker Tracker offset int64 } -// NewReadTracker attaches a tracker to a reader. -func NewReadTracker(track Tracker, r io.Reader) *ReadTracker { - return &ReadTracker{ - base: r, - tracker: track, - } -} - // Read reads from the base reader and updates the status. -func (rt *ReadTracker) Read(p []byte) (n int, err error) { +func (rt *readTracker) Read(p []byte) (n int, err error) { n, err = rt.base.Read(p) rt.offset += int64(n) _ = rt.tracker.Update(Status{ @@ -32,6 +32,6 @@ func (rt *ReadTracker) Read(p []byte) (n int, err error) { } // Close closes the tracker. -func (rt *ReadTracker) Close() error { +func (rt *readTracker) Close() error { return rt.tracker.Close() } From 68442f2eec1588a06f9c010f949eebe7ead27f9b Mon Sep 17 00:00:00 2001 From: Shiwei Zhang Date: Wed, 27 Nov 2024 18:56:10 +0800 Subject: [PATCH 04/15] refactor: merge reader to tracker Signed-off-by: Shiwei Zhang --- progress/reader.go | 37 ------------------------------------- progress/tracker.go | 34 ++++++++++++++++++++++++++++++++++ 2 files changed, 34 insertions(+), 37 deletions(-) delete mode 100644 progress/reader.go diff --git a/progress/reader.go b/progress/reader.go deleted file mode 100644 index bde03da6..00000000 --- a/progress/reader.go +++ /dev/null @@ -1,37 +0,0 @@ -package progress - -import "io" - -// TrackReader bind a reader with a tracker. -func TrackReader(t Tracker, r io.Reader) io.ReadCloser { - return &readTracker{ - base: r, - tracker: t, - } -} - -// readTracker tracks the transmission based on the read operation. -type readTracker struct { - base io.Reader - tracker Tracker - offset int64 -} - -// Read reads from the base reader and updates the status. -func (rt *readTracker) Read(p []byte) (n int, err error) { - n, err = rt.base.Read(p) - rt.offset += int64(n) - _ = rt.tracker.Update(Status{ - State: StateTransmitting, - Offset: rt.offset, - }) - if err != nil && err != io.EOF { - _ = rt.tracker.Fail(err) - } - return n, err -} - -// Close closes the tracker. -func (rt *readTracker) Close() error { - return rt.tracker.Close() -} diff --git a/progress/tracker.go b/progress/tracker.go index bdd769ce..abe61f0a 100644 --- a/progress/tracker.go +++ b/progress/tracker.go @@ -31,3 +31,37 @@ func Done(t Tracker) error { Offset: -1, }) } + +// TrackReader bind a reader with a tracker. +func TrackReader(t Tracker, r io.Reader) io.ReadCloser { + return &readTracker{ + base: r, + tracker: t, + } +} + +// readTracker tracks the transmission based on the read operation. +type readTracker struct { + base io.Reader + tracker Tracker + offset int64 +} + +// Read reads from the base reader and updates the status. +func (rt *readTracker) Read(p []byte) (n int, err error) { + n, err = rt.base.Read(p) + rt.offset += int64(n) + _ = rt.tracker.Update(Status{ + State: StateTransmitting, + Offset: rt.offset, + }) + if err != nil && err != io.EOF { + _ = rt.tracker.Fail(err) + } + return n, err +} + +// Close closes the tracker. +func (rt *readTracker) Close() error { + return rt.tracker.Close() +} From fb37103b4ad66e1ffe9a3ab6428fc46f139f813a Mon Sep 17 00:00:00 2001 From: Shiwei Zhang Date: Wed, 27 Nov 2024 19:12:33 +0800 Subject: [PATCH 05/15] feat: support WriteTo Signed-off-by: Shiwei Zhang --- progress/tracker.go | 54 ++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 49 insertions(+), 5 deletions(-) diff --git a/progress/tracker.go b/progress/tracker.go index abe61f0a..67487ea9 100644 --- a/progress/tracker.go +++ b/progress/tracker.go @@ -34,10 +34,14 @@ func Done(t Tracker) error { // TrackReader bind a reader with a tracker. func TrackReader(t Tracker, r io.Reader) io.ReadCloser { - return &readTracker{ + rt := readTracker{ base: r, tracker: t, } + if _, ok := r.(io.WriterTo); ok { + return &readTrackerWriteTo{rt} + } + return &rt } // readTracker tracks the transmission based on the read operation. @@ -51,10 +55,12 @@ type readTracker struct { func (rt *readTracker) Read(p []byte) (n int, err error) { n, err = rt.base.Read(p) rt.offset += int64(n) - _ = rt.tracker.Update(Status{ - State: StateTransmitting, - Offset: rt.offset, - }) + if n > 0 { + _ = rt.tracker.Update(Status{ + State: StateTransmitting, + Offset: rt.offset, + }) + } if err != nil && err != io.EOF { _ = rt.tracker.Fail(err) } @@ -65,3 +71,41 @@ func (rt *readTracker) Read(p []byte) (n int, err error) { func (rt *readTracker) Close() error { return rt.tracker.Close() } + +// readTrackerWriteTo is readTracker with WriteTo support. +type readTrackerWriteTo struct { + readTracker +} + +// WriteTo writes to the base writer and updates the status. +func (rt *readTrackerWriteTo) WriteTo(w io.Writer) (n int64, err error) { + wt := &writeTracker{ + base: w, + tracker: rt.tracker, + offset: rt.offset, + } + return rt.base.(io.WriterTo).WriteTo(wt) +} + +// writeTracker tracks the transmission based on the write operation. +type writeTracker struct { + base io.Writer + tracker Tracker + offset int64 +} + +// Write writes to the base writer and updates the status. +func (wt *writeTracker) Write(p []byte) (n int, err error) { + n, err = wt.base.Write(p) + wt.offset += int64(n) + if n > 0 { + _ = wt.tracker.Update(Status{ + State: StateTransmitting, + Offset: wt.offset, + }) + } + if err != nil { + _ = wt.tracker.Fail(err) + } + return n, err +} From 3eed9166411351cdee6347a0b343177b9efa0b16 Mon Sep 17 00:00:00 2001 From: Shiwei Zhang Date: Thu, 28 Nov 2024 15:22:16 +0800 Subject: [PATCH 06/15] feat: add utility functions Signed-off-by: Shiwei Zhang --- progress/manager.go | 17 +++++++++++++++++ progress/tracker.go | 20 ++++++++++++++++++++ 2 files changed, 37 insertions(+) diff --git a/progress/manager.go b/progress/manager.go index b5c289ea..4140339b 100644 --- a/progress/manager.go +++ b/progress/manager.go @@ -14,6 +14,23 @@ type Manager interface { Track(desc ocispec.Descriptor) (Tracker, error) } +// ManagerFunc is an adapter to allow the use of ordinary functions as Managers. +// If f is a function with the appropriate signature, ManagerFunc(f) is a +// [Manager] that calls f. +type ManagerFunc func(ocispec.Descriptor, Status, error) error + +// Track starts tracking the progress of a descriptor. +func (f ManagerFunc) Track(desc ocispec.Descriptor) (Tracker, error) { + return TrackerFunc(func(status Status, err error) error { + return f(desc, status, err) + }), nil +} + +// Close closes the manager. +func (f ManagerFunc) Close() error { + return nil +} + // Record adds the progress of a descriptor as a single entry. func Record(m Manager, desc ocispec.Descriptor, status Status) error { tracker, err := m.Track(desc) diff --git a/progress/tracker.go b/progress/tracker.go index 67487ea9..578868b1 100644 --- a/progress/tracker.go +++ b/progress/tracker.go @@ -13,6 +13,26 @@ type Tracker interface { Fail(err error) error } +// TrackerFunc is an adapter to allow the use of ordinary functions as Trackers. +// If f is a function with the appropriate signature, TrackerFunc(f) is a +// [Tracker] that calls f. +type TrackerFunc func(Status, error) error + +// Update updates the status of the descriptor. +func (f TrackerFunc) Update(status Status) error { + return f(status, nil) +} + +// Fail marks the descriptor as failed. +func (f TrackerFunc) Fail(err error) error { + return f(Status{}, err) +} + +// Close closes the tracker. +func (f TrackerFunc) Close() error { + return nil +} + // Start starts tracking the transmission. func Start(t Tracker) error { return t.Update(Status{ From 81e4b149380378958492ff172baea71d0753602c Mon Sep 17 00:00:00 2001 From: Shiwei Zhang Date: Thu, 28 Nov 2024 15:29:12 +0800 Subject: [PATCH 07/15] fix: fix tracker offset Signed-off-by: Shiwei Zhang --- progress/tracker.go | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/progress/tracker.go b/progress/tracker.go index 578868b1..52cfbad1 100644 --- a/progress/tracker.go +++ b/progress/tracker.go @@ -72,8 +72,8 @@ type readTracker struct { } // Read reads from the base reader and updates the status. -func (rt *readTracker) Read(p []byte) (n int, err error) { - n, err = rt.base.Read(p) +func (rt *readTracker) Read(p []byte) (int, error) { + n, err := rt.base.Read(p) rt.offset += int64(n) if n > 0 { _ = rt.tracker.Update(Status{ @@ -98,13 +98,15 @@ type readTrackerWriteTo struct { } // WriteTo writes to the base writer and updates the status. -func (rt *readTrackerWriteTo) WriteTo(w io.Writer) (n int64, err error) { +func (rt *readTrackerWriteTo) WriteTo(w io.Writer) (int64, error) { wt := &writeTracker{ base: w, tracker: rt.tracker, offset: rt.offset, } - return rt.base.(io.WriterTo).WriteTo(wt) + n, err := rt.base.(io.WriterTo).WriteTo(wt) + rt.offset = wt.offset + return n, err } // writeTracker tracks the transmission based on the write operation. @@ -115,8 +117,8 @@ type writeTracker struct { } // Write writes to the base writer and updates the status. -func (wt *writeTracker) Write(p []byte) (n int, err error) { - n, err = wt.base.Write(p) +func (wt *writeTracker) Write(p []byte) (int, error) { + n, err := wt.base.Write(p) wt.offset += int64(n) if n > 0 { _ = wt.tracker.Update(Status{ From 2c11aca492d8e485224a61a5619acebe311a4e65 Mon Sep 17 00:00:00 2001 From: Shiwei Zhang Date: Thu, 28 Nov 2024 15:32:36 +0800 Subject: [PATCH 08/15] docs: add licenses Signed-off-by: Shiwei Zhang --- progress/manager.go | 16 ++++++++++++++++ progress/status.go | 15 +++++++++++++++ progress/tracker.go | 15 +++++++++++++++ 3 files changed, 46 insertions(+) diff --git a/progress/manager.go b/progress/manager.go index 4140339b..15ec67da 100644 --- a/progress/manager.go +++ b/progress/manager.go @@ -1,3 +1,19 @@ +/* +Copyright The ORAS Authors. +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Package progress tracks the status of descriptors being processed. package progress import ( diff --git a/progress/status.go b/progress/status.go index 35e702c8..cc17d9c8 100644 --- a/progress/status.go +++ b/progress/status.go @@ -1,3 +1,18 @@ +/* +Copyright The ORAS Authors. +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + package progress // State represents the state of a descriptor. diff --git a/progress/tracker.go b/progress/tracker.go index 52cfbad1..4d164b53 100644 --- a/progress/tracker.go +++ b/progress/tracker.go @@ -1,3 +1,18 @@ +/* +Copyright The ORAS Authors. +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + package progress import "io" From 2d3b978a4820d158cf97ea4aa8d81bcfa3b399bd Mon Sep 17 00:00:00 2001 From: Shiwei Zhang Date: Thu, 28 Nov 2024 16:00:17 +0800 Subject: [PATCH 09/15] fix: fix error handling Signed-off-by: Shiwei Zhang --- progress/tracker.go | 21 ++++++++++++++------- 1 file changed, 14 insertions(+), 7 deletions(-) diff --git a/progress/tracker.go b/progress/tracker.go index 4d164b53..8dc6c238 100644 --- a/progress/tracker.go +++ b/progress/tracker.go @@ -25,6 +25,7 @@ type Tracker interface { Update(status Status) error // Fail marks the descriptor as failed. + // Fail should return nil on successful failure marking. Fail(err error) error } @@ -91,13 +92,17 @@ func (rt *readTracker) Read(p []byte) (int, error) { n, err := rt.base.Read(p) rt.offset += int64(n) if n > 0 { - _ = rt.tracker.Update(Status{ + if updateErr := rt.tracker.Update(Status{ State: StateTransmitting, Offset: rt.offset, - }) + }); updateErr != nil { + return n, updateErr + } } if err != nil && err != io.EOF { - _ = rt.tracker.Fail(err) + if failErr := rt.tracker.Fail(err); failErr != nil { + return n, failErr + } } return n, err } @@ -136,13 +141,15 @@ func (wt *writeTracker) Write(p []byte) (int, error) { n, err := wt.base.Write(p) wt.offset += int64(n) if n > 0 { - _ = wt.tracker.Update(Status{ + if updateErr := wt.tracker.Update(Status{ State: StateTransmitting, Offset: wt.offset, - }) + }); updateErr != nil { + return n, updateErr + } } if err != nil { - _ = wt.tracker.Fail(err) + return n, wt.tracker.Fail(err) } - return n, err + return n, nil } From e5bd985c66ea9f4b3688725ef7d4d84b90d57cda Mon Sep 17 00:00:00 2001 From: Shiwei Zhang Date: Thu, 28 Nov 2024 17:11:35 +0800 Subject: [PATCH 10/15] fix: tracked reader should not be a closer Signed-off-by: Shiwei Zhang --- progress/tracker.go | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/progress/tracker.go b/progress/tracker.go index 8dc6c238..518ad257 100644 --- a/progress/tracker.go +++ b/progress/tracker.go @@ -69,7 +69,7 @@ func Done(t Tracker) error { } // TrackReader bind a reader with a tracker. -func TrackReader(t Tracker, r io.Reader) io.ReadCloser { +func TrackReader(t Tracker, r io.Reader) io.Reader { rt := readTracker{ base: r, tracker: t, @@ -107,11 +107,6 @@ func (rt *readTracker) Read(p []byte) (int, error) { return n, err } -// Close closes the tracker. -func (rt *readTracker) Close() error { - return rt.tracker.Close() -} - // readTrackerWriteTo is readTracker with WriteTo support. type readTrackerWriteTo struct { readTracker From 6114ae1ebb6d17cab443133f4d9f8adf1945e2c8 Mon Sep 17 00:00:00 2001 From: Shiwei Zhang Date: Thu, 28 Nov 2024 17:17:00 +0800 Subject: [PATCH 11/15] docs: add example for TrackReader Signed-off-by: Shiwei Zhang --- progress/example_test.go | 83 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 83 insertions(+) create mode 100644 progress/example_test.go diff --git a/progress/example_test.go b/progress/example_test.go new file mode 100644 index 00000000..809c56d4 --- /dev/null +++ b/progress/example_test.go @@ -0,0 +1,83 @@ +/* +Copyright The ORAS Authors. +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package progress_test + +import ( + "crypto/rand" + "fmt" + "io" + + "oras.land/oras-go/v2/progress" +) + +// ExampleTrackReader demonstrates how to track the transmission progress of a +// reader. +func ExampleTrackReader() { + // Set up a progress tracker. + total := int64(11) + tracker := progress.TrackerFunc(func(status progress.Status, err error) error { + if err != nil { + fmt.Printf("Error: %v\n", err) + return nil + } + switch status.State { + case progress.StateInitialized: + fmt.Println("Start reading content") + case progress.StateTransmitting: + fmt.Printf("Progress: %d/%d bytes\n", status.Offset, total) + case progress.StateTransmitted: + fmt.Println("Finish reading content") + default: + // Ignore other states. + } + return nil + }) + // Close takes no effect for TrackerFunc but should be called for general + // Tracker implementations. + defer tracker.Close() + + // Wrap a reader of a random content generator with the progress tracker. + r := io.LimitReader(rand.Reader, total) + rc := progress.TrackReader(tracker, r) + + // Start tracking the transmission. + if err := progress.Start(tracker); err != nil { + panic(err) + } + + // Read from the random content generator and discard the content, while + // tracking the progress. + // Note: io.Discard is wrapped with a io.MultiWriter for dropping + // the io.ReadFrom interface for demonstration purposes. + buf := make([]byte, 3) + w := io.MultiWriter(io.Discard) + if _, err := io.CopyBuffer(w, rc, buf); err != nil { + panic(err) + } + + // Finish tracking the transmission. + if err := progress.Done(tracker); err != nil { + panic(err) + } + + // Output: + // Start reading content + // Progress: 3/11 bytes + // Progress: 6/11 bytes + // Progress: 9/11 bytes + // Progress: 11/11 bytes + // Finish reading content +} From 02bbed026da90f080a11d51b3b121738331ea195 Mon Sep 17 00:00:00 2001 From: Shiwei Zhang Date: Mon, 2 Dec 2024 11:12:58 +0800 Subject: [PATCH 12/15] chore: rearrange code Signed-off-by: Shiwei Zhang --- progress/manager.go | 12 ++++++------ progress/tracker.go | 12 ++++++------ 2 files changed, 12 insertions(+), 12 deletions(-) diff --git a/progress/manager.go b/progress/manager.go index 15ec67da..2e1f9b6d 100644 --- a/progress/manager.go +++ b/progress/manager.go @@ -33,7 +33,12 @@ type Manager interface { // ManagerFunc is an adapter to allow the use of ordinary functions as Managers. // If f is a function with the appropriate signature, ManagerFunc(f) is a // [Manager] that calls f. -type ManagerFunc func(ocispec.Descriptor, Status, error) error +type ManagerFunc func(desc ocispec.Descriptor, status Status, err error) error + +// Close closes the manager. +func (f ManagerFunc) Close() error { + return nil +} // Track starts tracking the progress of a descriptor. func (f ManagerFunc) Track(desc ocispec.Descriptor) (Tracker, error) { @@ -42,11 +47,6 @@ func (f ManagerFunc) Track(desc ocispec.Descriptor) (Tracker, error) { }), nil } -// Close closes the manager. -func (f ManagerFunc) Close() error { - return nil -} - // Record adds the progress of a descriptor as a single entry. func Record(m Manager, desc ocispec.Descriptor, status Status) error { tracker, err := m.Track(desc) diff --git a/progress/tracker.go b/progress/tracker.go index 518ad257..c67cde37 100644 --- a/progress/tracker.go +++ b/progress/tracker.go @@ -32,7 +32,12 @@ type Tracker interface { // TrackerFunc is an adapter to allow the use of ordinary functions as Trackers. // If f is a function with the appropriate signature, TrackerFunc(f) is a // [Tracker] that calls f. -type TrackerFunc func(Status, error) error +type TrackerFunc func(status Status, err error) error + +// Close closes the tracker. +func (f TrackerFunc) Close() error { + return nil +} // Update updates the status of the descriptor. func (f TrackerFunc) Update(status Status) error { @@ -44,11 +49,6 @@ func (f TrackerFunc) Fail(err error) error { return f(Status{}, err) } -// Close closes the tracker. -func (f TrackerFunc) Close() error { - return nil -} - // Start starts tracking the transmission. func Start(t Tracker) error { return t.Update(Status{ From e7a5326f53451f9c31cbb28cb0b44e64ad48bb0a Mon Sep 17 00:00:00 2001 From: Shiwei Zhang Date: Mon, 2 Dec 2024 11:15:06 +0800 Subject: [PATCH 13/15] refactor!: remove Record util Signed-off-by: Shiwei Zhang --- progress/manager.go | 13 ------------- 1 file changed, 13 deletions(-) diff --git a/progress/manager.go b/progress/manager.go index 2e1f9b6d..4d783122 100644 --- a/progress/manager.go +++ b/progress/manager.go @@ -46,16 +46,3 @@ func (f ManagerFunc) Track(desc ocispec.Descriptor) (Tracker, error) { return f(desc, status, err) }), nil } - -// Record adds the progress of a descriptor as a single entry. -func Record(m Manager, desc ocispec.Descriptor, status Status) error { - tracker, err := m.Track(desc) - if err != nil { - return err - } - err = tracker.Update(status) - if err != nil { - return err - } - return tracker.Close() -} From 7bf1484ee974ad825fbb686f9e84040e59c04d08 Mon Sep 17 00:00:00 2001 From: Shiwei Zhang Date: Mon, 2 Dec 2024 15:49:45 +0800 Subject: [PATCH 14/15] fix: covert CRLF to LF Signed-off-by: Shiwei Zhang --- progress/example_test.go | 166 +++++++++++----------- progress/manager.go | 96 ++++++------- progress/status.go | 80 +++++------ progress/tracker.go | 300 +++++++++++++++++++-------------------- 4 files changed, 321 insertions(+), 321 deletions(-) diff --git a/progress/example_test.go b/progress/example_test.go index 809c56d4..2f7754dd 100644 --- a/progress/example_test.go +++ b/progress/example_test.go @@ -1,83 +1,83 @@ -/* -Copyright The ORAS Authors. -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - -http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package progress_test - -import ( - "crypto/rand" - "fmt" - "io" - - "oras.land/oras-go/v2/progress" -) - -// ExampleTrackReader demonstrates how to track the transmission progress of a -// reader. -func ExampleTrackReader() { - // Set up a progress tracker. - total := int64(11) - tracker := progress.TrackerFunc(func(status progress.Status, err error) error { - if err != nil { - fmt.Printf("Error: %v\n", err) - return nil - } - switch status.State { - case progress.StateInitialized: - fmt.Println("Start reading content") - case progress.StateTransmitting: - fmt.Printf("Progress: %d/%d bytes\n", status.Offset, total) - case progress.StateTransmitted: - fmt.Println("Finish reading content") - default: - // Ignore other states. - } - return nil - }) - // Close takes no effect for TrackerFunc but should be called for general - // Tracker implementations. - defer tracker.Close() - - // Wrap a reader of a random content generator with the progress tracker. - r := io.LimitReader(rand.Reader, total) - rc := progress.TrackReader(tracker, r) - - // Start tracking the transmission. - if err := progress.Start(tracker); err != nil { - panic(err) - } - - // Read from the random content generator and discard the content, while - // tracking the progress. - // Note: io.Discard is wrapped with a io.MultiWriter for dropping - // the io.ReadFrom interface for demonstration purposes. - buf := make([]byte, 3) - w := io.MultiWriter(io.Discard) - if _, err := io.CopyBuffer(w, rc, buf); err != nil { - panic(err) - } - - // Finish tracking the transmission. - if err := progress.Done(tracker); err != nil { - panic(err) - } - - // Output: - // Start reading content - // Progress: 3/11 bytes - // Progress: 6/11 bytes - // Progress: 9/11 bytes - // Progress: 11/11 bytes - // Finish reading content -} +/* +Copyright The ORAS Authors. +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package progress_test + +import ( + "crypto/rand" + "fmt" + "io" + + "oras.land/oras-go/v2/progress" +) + +// ExampleTrackReader demonstrates how to track the transmission progress of a +// reader. +func ExampleTrackReader() { + // Set up a progress tracker. + total := int64(11) + tracker := progress.TrackerFunc(func(status progress.Status, err error) error { + if err != nil { + fmt.Printf("Error: %v\n", err) + return nil + } + switch status.State { + case progress.StateInitialized: + fmt.Println("Start reading content") + case progress.StateTransmitting: + fmt.Printf("Progress: %d/%d bytes\n", status.Offset, total) + case progress.StateTransmitted: + fmt.Println("Finish reading content") + default: + // Ignore other states. + } + return nil + }) + // Close takes no effect for TrackerFunc but should be called for general + // Tracker implementations. + defer tracker.Close() + + // Wrap a reader of a random content generator with the progress tracker. + r := io.LimitReader(rand.Reader, total) + rc := progress.TrackReader(tracker, r) + + // Start tracking the transmission. + if err := progress.Start(tracker); err != nil { + panic(err) + } + + // Read from the random content generator and discard the content, while + // tracking the progress. + // Note: io.Discard is wrapped with a io.MultiWriter for dropping + // the io.ReadFrom interface for demonstration purposes. + buf := make([]byte, 3) + w := io.MultiWriter(io.Discard) + if _, err := io.CopyBuffer(w, rc, buf); err != nil { + panic(err) + } + + // Finish tracking the transmission. + if err := progress.Done(tracker); err != nil { + panic(err) + } + + // Output: + // Start reading content + // Progress: 3/11 bytes + // Progress: 6/11 bytes + // Progress: 9/11 bytes + // Progress: 11/11 bytes + // Finish reading content +} diff --git a/progress/manager.go b/progress/manager.go index 4d783122..439b90d1 100644 --- a/progress/manager.go +++ b/progress/manager.go @@ -1,48 +1,48 @@ -/* -Copyright The ORAS Authors. -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - -http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -// Package progress tracks the status of descriptors being processed. -package progress - -import ( - "io" - - ocispec "github.com/opencontainers/image-spec/specs-go/v1" -) - -// Manager tracks the progress of multiple descriptors. -type Manager interface { - io.Closer - - // Track starts tracking the progress of a descriptor. - Track(desc ocispec.Descriptor) (Tracker, error) -} - -// ManagerFunc is an adapter to allow the use of ordinary functions as Managers. -// If f is a function with the appropriate signature, ManagerFunc(f) is a -// [Manager] that calls f. -type ManagerFunc func(desc ocispec.Descriptor, status Status, err error) error - -// Close closes the manager. -func (f ManagerFunc) Close() error { - return nil -} - -// Track starts tracking the progress of a descriptor. -func (f ManagerFunc) Track(desc ocispec.Descriptor) (Tracker, error) { - return TrackerFunc(func(status Status, err error) error { - return f(desc, status, err) - }), nil -} +/* +Copyright The ORAS Authors. +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Package progress tracks the status of descriptors being processed. +package progress + +import ( + "io" + + ocispec "github.com/opencontainers/image-spec/specs-go/v1" +) + +// Manager tracks the progress of multiple descriptors. +type Manager interface { + io.Closer + + // Track starts tracking the progress of a descriptor. + Track(desc ocispec.Descriptor) (Tracker, error) +} + +// ManagerFunc is an adapter to allow the use of ordinary functions as Managers. +// If f is a function with the appropriate signature, ManagerFunc(f) is a +// [Manager] that calls f. +type ManagerFunc func(desc ocispec.Descriptor, status Status, err error) error + +// Close closes the manager. +func (f ManagerFunc) Close() error { + return nil +} + +// Track starts tracking the progress of a descriptor. +func (f ManagerFunc) Track(desc ocispec.Descriptor) (Tracker, error) { + return TrackerFunc(func(status Status, err error) error { + return f(desc, status, err) + }), nil +} diff --git a/progress/status.go b/progress/status.go index cc17d9c8..e6c4d1cb 100644 --- a/progress/status.go +++ b/progress/status.go @@ -1,40 +1,40 @@ -/* -Copyright The ORAS Authors. -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - -http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package progress - -// State represents the state of a descriptor. -type State int - -// Registered states. -const ( - StateUnknown State = iota // unknown state - StateInitialized // progress initialized - StateTransmitting // transmitting content - StateTransmitted // content transmitted - StateExists // content exists - StateSkipped // content skipped - StateMounted // content mounted -) - -// Status represents the status of a descriptor. -type Status struct { - // State represents the state of the descriptor. - State State - - // Offset represents the current offset of the descriptor. - // Offset is discarded if set to a negative value. - Offset int64 -} +/* +Copyright The ORAS Authors. +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package progress + +// State represents the state of a descriptor. +type State int + +// Registered states. +const ( + StateUnknown State = iota // unknown state + StateInitialized // progress initialized + StateTransmitting // transmitting content + StateTransmitted // content transmitted + StateExists // content exists + StateSkipped // content skipped + StateMounted // content mounted +) + +// Status represents the status of a descriptor. +type Status struct { + // State represents the state of the descriptor. + State State + + // Offset represents the current offset of the descriptor. + // Offset is discarded if set to a negative value. + Offset int64 +} diff --git a/progress/tracker.go b/progress/tracker.go index c67cde37..58ee8f62 100644 --- a/progress/tracker.go +++ b/progress/tracker.go @@ -1,150 +1,150 @@ -/* -Copyright The ORAS Authors. -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - -http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package progress - -import "io" - -// Tracker updates the status of a descriptor. -type Tracker interface { - io.Closer - - // Update updates the status of the descriptor. - Update(status Status) error - - // Fail marks the descriptor as failed. - // Fail should return nil on successful failure marking. - Fail(err error) error -} - -// TrackerFunc is an adapter to allow the use of ordinary functions as Trackers. -// If f is a function with the appropriate signature, TrackerFunc(f) is a -// [Tracker] that calls f. -type TrackerFunc func(status Status, err error) error - -// Close closes the tracker. -func (f TrackerFunc) Close() error { - return nil -} - -// Update updates the status of the descriptor. -func (f TrackerFunc) Update(status Status) error { - return f(status, nil) -} - -// Fail marks the descriptor as failed. -func (f TrackerFunc) Fail(err error) error { - return f(Status{}, err) -} - -// Start starts tracking the transmission. -func Start(t Tracker) error { - return t.Update(Status{ - State: StateInitialized, - Offset: -1, - }) -} - -// Done marks the transmission as complete. -// Done should be called after the transmission is complete. -// Note: Reading all content from the reader does not imply the transmission is -// complete. -func Done(t Tracker) error { - return t.Update(Status{ - State: StateTransmitted, - Offset: -1, - }) -} - -// TrackReader bind a reader with a tracker. -func TrackReader(t Tracker, r io.Reader) io.Reader { - rt := readTracker{ - base: r, - tracker: t, - } - if _, ok := r.(io.WriterTo); ok { - return &readTrackerWriteTo{rt} - } - return &rt -} - -// readTracker tracks the transmission based on the read operation. -type readTracker struct { - base io.Reader - tracker Tracker - offset int64 -} - -// Read reads from the base reader and updates the status. -func (rt *readTracker) Read(p []byte) (int, error) { - n, err := rt.base.Read(p) - rt.offset += int64(n) - if n > 0 { - if updateErr := rt.tracker.Update(Status{ - State: StateTransmitting, - Offset: rt.offset, - }); updateErr != nil { - return n, updateErr - } - } - if err != nil && err != io.EOF { - if failErr := rt.tracker.Fail(err); failErr != nil { - return n, failErr - } - } - return n, err -} - -// readTrackerWriteTo is readTracker with WriteTo support. -type readTrackerWriteTo struct { - readTracker -} - -// WriteTo writes to the base writer and updates the status. -func (rt *readTrackerWriteTo) WriteTo(w io.Writer) (int64, error) { - wt := &writeTracker{ - base: w, - tracker: rt.tracker, - offset: rt.offset, - } - n, err := rt.base.(io.WriterTo).WriteTo(wt) - rt.offset = wt.offset - return n, err -} - -// writeTracker tracks the transmission based on the write operation. -type writeTracker struct { - base io.Writer - tracker Tracker - offset int64 -} - -// Write writes to the base writer and updates the status. -func (wt *writeTracker) Write(p []byte) (int, error) { - n, err := wt.base.Write(p) - wt.offset += int64(n) - if n > 0 { - if updateErr := wt.tracker.Update(Status{ - State: StateTransmitting, - Offset: wt.offset, - }); updateErr != nil { - return n, updateErr - } - } - if err != nil { - return n, wt.tracker.Fail(err) - } - return n, nil -} +/* +Copyright The ORAS Authors. +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package progress + +import "io" + +// Tracker updates the status of a descriptor. +type Tracker interface { + io.Closer + + // Update updates the status of the descriptor. + Update(status Status) error + + // Fail marks the descriptor as failed. + // Fail should return nil on successful failure marking. + Fail(err error) error +} + +// TrackerFunc is an adapter to allow the use of ordinary functions as Trackers. +// If f is a function with the appropriate signature, TrackerFunc(f) is a +// [Tracker] that calls f. +type TrackerFunc func(status Status, err error) error + +// Close closes the tracker. +func (f TrackerFunc) Close() error { + return nil +} + +// Update updates the status of the descriptor. +func (f TrackerFunc) Update(status Status) error { + return f(status, nil) +} + +// Fail marks the descriptor as failed. +func (f TrackerFunc) Fail(err error) error { + return f(Status{}, err) +} + +// Start starts tracking the transmission. +func Start(t Tracker) error { + return t.Update(Status{ + State: StateInitialized, + Offset: -1, + }) +} + +// Done marks the transmission as complete. +// Done should be called after the transmission is complete. +// Note: Reading all content from the reader does not imply the transmission is +// complete. +func Done(t Tracker) error { + return t.Update(Status{ + State: StateTransmitted, + Offset: -1, + }) +} + +// TrackReader bind a reader with a tracker. +func TrackReader(t Tracker, r io.Reader) io.Reader { + rt := readTracker{ + base: r, + tracker: t, + } + if _, ok := r.(io.WriterTo); ok { + return &readTrackerWriteTo{rt} + } + return &rt +} + +// readTracker tracks the transmission based on the read operation. +type readTracker struct { + base io.Reader + tracker Tracker + offset int64 +} + +// Read reads from the base reader and updates the status. +func (rt *readTracker) Read(p []byte) (int, error) { + n, err := rt.base.Read(p) + rt.offset += int64(n) + if n > 0 { + if updateErr := rt.tracker.Update(Status{ + State: StateTransmitting, + Offset: rt.offset, + }); updateErr != nil { + return n, updateErr + } + } + if err != nil && err != io.EOF { + if failErr := rt.tracker.Fail(err); failErr != nil { + return n, failErr + } + } + return n, err +} + +// readTrackerWriteTo is readTracker with WriteTo support. +type readTrackerWriteTo struct { + readTracker +} + +// WriteTo writes to the base writer and updates the status. +func (rt *readTrackerWriteTo) WriteTo(w io.Writer) (int64, error) { + wt := &writeTracker{ + base: w, + tracker: rt.tracker, + offset: rt.offset, + } + n, err := rt.base.(io.WriterTo).WriteTo(wt) + rt.offset = wt.offset + return n, err +} + +// writeTracker tracks the transmission based on the write operation. +type writeTracker struct { + base io.Writer + tracker Tracker + offset int64 +} + +// Write writes to the base writer and updates the status. +func (wt *writeTracker) Write(p []byte) (int, error) { + n, err := wt.base.Write(p) + wt.offset += int64(n) + if n > 0 { + if updateErr := wt.tracker.Update(Status{ + State: StateTransmitting, + Offset: wt.offset, + }); updateErr != nil { + return n, updateErr + } + } + if err != nil { + return n, wt.tracker.Fail(err) + } + return n, nil +} From fc11b2fcdbcad8f12c29b3d81740c5b00460c299 Mon Sep 17 00:00:00 2001 From: Shiwei Zhang Date: Tue, 31 Dec 2024 18:17:45 +0800 Subject: [PATCH 15/15] fix: fix tracker and add tests Signed-off-by: Shiwei Zhang --- progress/tracker.go | 30 ++- progress/tracker_test.go | 414 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 437 insertions(+), 7 deletions(-) create mode 100644 progress/tracker_test.go diff --git a/progress/tracker.go b/progress/tracker.go index 58ee8f62..0431a767 100644 --- a/progress/tracker.go +++ b/progress/tracker.go @@ -88,6 +88,8 @@ type readTracker struct { } // Read reads from the base reader and updates the status. +// On partial read, the tracker treats it as two reads: a successful read with +// status update and a failed read with failure report. func (rt *readTracker) Read(p []byte) (int, error) { n, err := rt.base.Read(p) rt.offset += int64(n) @@ -96,7 +98,7 @@ func (rt *readTracker) Read(p []byte) (int, error) { State: StateTransmitting, Offset: rt.offset, }); updateErr != nil { - return n, updateErr + err = updateErr } } if err != nil && err != io.EOF { @@ -113,6 +115,8 @@ type readTrackerWriteTo struct { } // WriteTo writes to the base writer and updates the status. +// On partial write, the tracker treats it as two writes: a successful write +// with status update and a failed write with failure report. func (rt *readTrackerWriteTo) WriteTo(w io.Writer) (int64, error) { wt := &writeTracker{ base: w, @@ -121,17 +125,25 @@ func (rt *readTrackerWriteTo) WriteTo(w io.Writer) (int64, error) { } n, err := rt.base.(io.WriterTo).WriteTo(wt) rt.offset = wt.offset + if err != nil && wt.trackerErr == nil { + if failErr := rt.tracker.Fail(err); failErr != nil { + return n, failErr + } + } return n, err } // writeTracker tracks the transmission based on the write operation. type writeTracker struct { - base io.Writer - tracker Tracker - offset int64 + base io.Writer + tracker Tracker + offset int64 + trackerErr error } // Write writes to the base writer and updates the status. +// On partial write, the tracker treats it as two writes: a successful write +// with status update and a failed write with failure report. func (wt *writeTracker) Write(p []byte) (int, error) { n, err := wt.base.Write(p) wt.offset += int64(n) @@ -140,11 +152,15 @@ func (wt *writeTracker) Write(p []byte) (int, error) { State: StateTransmitting, Offset: wt.offset, }); updateErr != nil { - return n, updateErr + wt.trackerErr = updateErr + err = updateErr } } if err != nil { - return n, wt.tracker.Fail(err) + if failErr := wt.tracker.Fail(err); failErr != nil { + wt.trackerErr = failErr + return n, failErr + } } - return n, nil + return n, err } diff --git a/progress/tracker_test.go b/progress/tracker_test.go new file mode 100644 index 00000000..f4190cc0 --- /dev/null +++ b/progress/tracker_test.go @@ -0,0 +1,414 @@ +/* +Copyright The ORAS Authors. +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package progress + +import ( + "bytes" + "errors" + "io" + "testing" +) + +func TestTrackerFunc_Close(t *testing.T) { + var f TrackerFunc + if err := f.Close(); err != nil { + t.Errorf("TrackerFunc.Close() error = %v, wantErr false", err) + } +} + +func TestTrackerFunc_Update(t *testing.T) { + wantStatus := Status{ + State: StateTransmitted, + Offset: 42, + } + var wantErr error + tracker := TrackerFunc(func(status Status, err error) error { + if status != wantStatus { + t.Errorf("TrackerFunc status = %v, want %v", status, wantStatus) + } + if err != nil { + t.Errorf("TrackerFunc err = %v, want nil", err) + } + return wantErr + }) + + if err := tracker.Update(wantStatus); err != wantErr { + t.Errorf("TrackerFunc.Update() error = %v, want %v", err, wantErr) + } + + wantErr = errors.New("fail to track") + if err := tracker.Update(wantStatus); err != wantErr { + t.Errorf("TrackerFunc.Update() error = %v, want %v", err, wantErr) + } +} + +func TestTrackerFunc_Fail(t *testing.T) { + reportErr := errors.New("fail to process") + var wantStatus Status + var wantErr error + tracker := TrackerFunc(func(status Status, err error) error { + if status != wantStatus { + t.Errorf("TrackerFunc status = %v, want %v", status, wantStatus) + } + if err != reportErr { + t.Errorf("TrackerFunc err = %v, want %v", err, reportErr) + } + return wantErr + }) + + if err := tracker.Fail(reportErr); err != wantErr { + t.Errorf("TrackerFunc.Fail() error = %v, want %v", err, wantErr) + } + + wantErr = errors.New("fail to track") + if err := tracker.Fail(reportErr); err != wantErr { + t.Errorf("TrackerFunc.Fail() error = %v, want %v", err, wantErr) + } +} + +func TestStart(t *testing.T) { + tests := []struct { + name string + t Tracker + wantErr bool + }{ + { + name: "successful report initialization", + t: TrackerFunc(func(status Status, err error) error { + if status.State != StateInitialized { + t.Errorf("expected state to be StateInitialized, got %v", status.State) + } + return nil + }), + }, + { + name: "fail to report initialization", + t: TrackerFunc(func(status Status, err error) error { + return errors.New("fail to track") + }), + wantErr: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if err := Start(tt.t); (err != nil) != tt.wantErr { + t.Errorf("Start() error = %v, wantErr %v", err, tt.wantErr) + } + }) + } +} + +func TestDone(t *testing.T) { + tests := []struct { + name string + t Tracker + wantErr bool + }{ + { + name: "successful report initialization", + t: TrackerFunc(func(status Status, err error) error { + if status.State != StateTransmitted { + t.Errorf("expected state to be StateTransmitted, got %v", status.State) + } + return nil + }), + }, + { + name: "fail to report initialization", + t: TrackerFunc(func(status Status, err error) error { + return errors.New("fail to track") + }), + wantErr: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if err := Done(tt.t); (err != nil) != tt.wantErr { + t.Errorf("Done() error = %v, wantErr %v", err, tt.wantErr) + } + }) + } +} + +func TestTrackReader(t *testing.T) { + const bufSize = 6 + content := []byte("hello world") + t.Run("track io.Reader", func(t *testing.T) { + var wantStatus Status + tracker := TrackerFunc(func(status Status, err error) error { + if status != wantStatus { + t.Errorf("TrackerFunc status = %v, want %v", status, wantStatus) + } + if err != nil { + t.Errorf("TrackerFunc err = %v, want nil", err) + } + return nil + }) + var reader io.Reader = bytes.NewReader(content) + reader = io.LimitReader(reader, int64(len(content))) // remove the io.WriterTo interface + gotReader := TrackReader(tracker, reader) + if _, ok := gotReader.(*readTracker); !ok { + t.Fatalf("TrackReader() = %v, want *readTracker", gotReader) + } + + wantStatus = Status{ + State: StateTransmitting, + Offset: bufSize, + } + buf := make([]byte, bufSize) + n, err := gotReader.Read(buf) + if err != nil { + t.Fatalf("TrackReader() error = %v, want nil", err) + } + if n != bufSize { + t.Fatalf("TrackReader() n = %v, want %v", n, bufSize) + } + if want := content[:bufSize]; !bytes.Equal(buf, want) { + t.Fatalf("TrackReader() buf = %v, want %v", buf, want) + } + + wantStatus = Status{ + State: StateTransmitting, + Offset: int64(len(content)), + } + n, err = gotReader.Read(buf) + if err != nil { + t.Fatalf("TrackReader() error = %v, want nil", err) + } + if want := len(content) - bufSize; n != want { + t.Fatalf("TrackReader() n = %v, want %v", n, want) + } + buf = buf[:n] + if want := content[bufSize:]; !bytes.Equal(buf, want) { + t.Fatalf("TrackReader() buf = %v, want %v", buf, want) + } + }) + + t.Run("track io.Reader + io.WriterTo", func(t *testing.T) { + var wantStatus Status + tracker := TrackerFunc(func(status Status, err error) error { + if status != wantStatus { + t.Errorf("TrackerFunc status = %v, want %v", status, wantStatus) + } + if err != nil { + t.Errorf("TrackerFunc err = %v, want nil", err) + } + return nil + }) + var reader io.Reader = bytes.NewReader(content) + gotReader := TrackReader(tracker, reader) + if _, ok := gotReader.(*readTrackerWriteTo); !ok { + t.Fatalf("TrackReader() = %v, want *readTrackerWriteTo", gotReader) + } + + wantStatus = Status{ + State: StateTransmitting, + Offset: bufSize, + } + buf := make([]byte, bufSize) + n, err := gotReader.Read(buf) + if err != nil { + t.Fatalf("TrackReader() error = %v, want nil", err) + } + if n != bufSize { + t.Fatalf("TrackReader() n = %v, want %v", n, bufSize) + } + if want := content[:bufSize]; !bytes.Equal(buf, want) { + t.Fatalf("TrackReader() buf = %v, want %v", buf, want) + } + + wantStatus = Status{ + State: StateTransmitting, + Offset: int64(len(content)), + } + writeBuf := bytes.NewBuffer(nil) + wn, err := gotReader.(io.WriterTo).WriteTo(writeBuf) + if err != nil { + t.Fatalf("TrackReader() error = %v, want nil", err) + } + if want := len(content) - bufSize; wn != int64(want) { + t.Fatalf("TrackReader() n = %v, want %v", wn, want) + } + buf = writeBuf.Bytes() + if want := content[bufSize:]; !bytes.Equal(buf, want) { + t.Fatalf("TrackReader() buf = %v, want %v", buf, want) + } + }) + + t.Run("empty io.Reader", func(t *testing.T) { + tracker := TrackerFunc(func(status Status, err error) error { + t.Errorf("TrackerFunc should not be called for empty read") + return nil + }) + gotReader := TrackReader(tracker, bytes.NewReader(nil)) + + buf := make([]byte, bufSize) + n, err := gotReader.Read(buf) + if want := io.EOF; err != want { + t.Fatalf("TrackReader() error = %v, want %v", err, want) + } + if want := 0; n != want { + t.Fatalf("TrackReader() n = %v, want %v", n, want) + } + + writeBuf := bytes.NewBuffer(nil) + wn, err := gotReader.(io.WriterTo).WriteTo(writeBuf) + if err != nil { + t.Fatalf("TrackReader() error = %v, want nil", err) + } + if want := int64(0); wn != want { + t.Fatalf("TrackReader() n = %v, want %v", wn, want) + } + buf = writeBuf.Bytes() + if want := []byte{}; !bytes.Equal(buf, want) { + t.Fatalf("TrackReader() buf = %v, want %v", buf, want) + } + }) + + t.Run("report failure", func(t *testing.T) { + var wantStatus Status + wantErr := errors.New("fail to track") + trackerMockStage := 0 + tracker := TrackerFunc(func(status Status, err error) error { + defer func() { + trackerMockStage++ + }() + switch trackerMockStage { + case 0: + if status != wantStatus { + t.Errorf("TrackerFunc status = %v, want %v", status, wantStatus) + } + if err != nil { + t.Errorf("TrackerFunc err = %v, want nil", err) + } + return wantErr + case 1: + var emptyStatus Status + if wantStatus := emptyStatus; status != wantStatus { + t.Errorf("TrackerFunc status = %v, want %v", status, wantStatus) + } + if err != wantErr { + t.Errorf("TrackerFunc err = %v, want %v", err, wantErr) + } + return nil + default: + t.Errorf("TrackerFunc should not be called") + return nil + } + }) + gotReader := TrackReader(tracker, bytes.NewReader(content)) + + wantStatus = Status{ + State: StateTransmitting, + Offset: bufSize, + } + buf := make([]byte, bufSize) + n, err := gotReader.Read(buf) + if err != wantErr { + t.Fatalf("TrackReader() error = %v, want %v", err, wantErr) + } + if n != bufSize { + t.Fatalf("TrackReader() n = %v, want %v", n, bufSize) + } + if want := content[:bufSize]; !bytes.Equal(buf, want) { + t.Fatalf("TrackReader() buf = %v, want %v", buf, want) + } + + wantStatus = Status{ + State: StateTransmitting, + Offset: int64(len(content)), + } + trackerMockStage = 0 + writeBuf := bytes.NewBuffer(nil) + wn, err := gotReader.(io.WriterTo).WriteTo(writeBuf) + if err != wantErr { + t.Fatalf("TrackReader() error = %v, want %v", err, wantErr) + } + if want := len(content) - bufSize; wn != int64(want) { + t.Fatalf("TrackReader() n = %v, want %v", wn, want) + } + buf = writeBuf.Bytes() + if want := content[bufSize:]; !bytes.Equal(buf, want) { + t.Fatalf("TrackReader() buf = %v, want %v", buf, want) + } + }) + + t.Run("process failure", func(t *testing.T) { + reportErr := io.ErrClosedPipe + var wantStatus Status + var wantErr error + tracker := TrackerFunc(func(status Status, err error) error { + if status != wantStatus { + t.Errorf("TrackerFunc status = %v, want %v", status, wantStatus) + } + if err != reportErr { + t.Errorf("TrackerFunc err = %v, want %v", err, reportErr) + } + return wantErr + }) + pipeReader, pipeWriter := io.Pipe() + pipeReader.Close() + pipeWriter.Close() + gotReader := TrackReader(tracker, pipeReader) + + buf := make([]byte, bufSize) + n, err := gotReader.Read(buf) + if err != reportErr { + t.Fatalf("TrackReader() error = %v, want %v", err, reportErr) + } + if want := 0; n != want { + t.Fatalf("TrackReader() n = %v, want %v", n, want) + } + + wantErr = errors.New("fail to track") + n, err = gotReader.Read(buf) + if err != wantErr { + t.Fatalf("TrackReader() error = %v, want %v", err, wantErr) + } + if want := 0; n != want { + t.Fatalf("TrackReader() n = %v, want %v", n, want) + } + + gotReader = TrackReader(tracker, io.MultiReader(pipeReader)) // wrap io.WriteTo + wantErr = nil + writeBuf := bytes.NewBuffer(nil) + wn, err := gotReader.(io.WriterTo).WriteTo(writeBuf) + if err != reportErr { + t.Fatalf("TrackReader() error = %v, want %v", err, reportErr) + } + if want := int64(0); wn != want { + t.Fatalf("TrackReader() n = %v, want %v", wn, want) + } + buf = writeBuf.Bytes() + if want := []byte{}; !bytes.Equal(buf, want) { + t.Fatalf("TrackReader() buf = %v, want %v", buf, want) + } + + gotReader = TrackReader(tracker, io.MultiReader(pipeReader)) // wrap io.WriteTo + wantErr = errors.New("fail to track") + wn, err = gotReader.(io.WriterTo).WriteTo(writeBuf) + if err != wantErr { + t.Fatalf("TrackReader() error = %v, want %v", err, wantErr) + } + if want := int64(0); wn != want { + t.Fatalf("TrackReader() n = %v, want %v", wn, want) + } + buf = writeBuf.Bytes() + if want := []byte{}; !bytes.Equal(buf, want) { + t.Fatalf("TrackReader() buf = %v, want %v", buf, want) + } + }) +}