From 0cc41f46bece0024e1b24c8a24db5a375678816f Mon Sep 17 00:00:00 2001 From: Sergey Tyurin Date: Mon, 8 Jun 2026 14:50:09 -0700 Subject: [PATCH] Migrate to go.atoms.co/iox --- chanx/broadcaster.go | 2 +- chanx/chan.go | 2 +- chanx/chan_test.go | 4 +-- go.mod | 3 +- go.sum | 2 ++ iox/closer.go | 76 -------------------------------------------- iox/status.go | 61 ----------------------------------- metrics/metrics.go | 2 -- net/grpcx/stream.go | 4 +-- signalx/signal.go | 2 +- syncx/lock.go | 2 +- 11 files changed, 12 insertions(+), 148 deletions(-) delete mode 100644 iox/closer.go delete mode 100644 iox/status.go diff --git a/chanx/broadcaster.go b/chanx/broadcaster.go index 6ef62ad..52ffef8 100644 --- a/chanx/broadcaster.go +++ b/chanx/broadcaster.go @@ -5,7 +5,7 @@ import ( "fmt" "sync" - "go.atoms.co/lib/iox" + "go.atoms.co/iox" "go.atoms.co/lib/syncx" ) diff --git a/chanx/chan.go b/chanx/chan.go index e4aab87..6fd7bdd 100644 --- a/chanx/chan.go +++ b/chanx/chan.go @@ -6,7 +6,7 @@ import ( "sync" "time" - "go.atoms.co/lib/iox" + "go.atoms.co/iox" ) // NewFixed returns a new closed chan with the given elements. diff --git a/chanx/chan_test.go b/chanx/chan_test.go index 8440a59..cd46a30 100644 --- a/chanx/chan_test.go +++ b/chanx/chan_test.go @@ -6,9 +6,9 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "go.atoms.co/lib/testing/assertx" + "go.atoms.co/iox" "go.atoms.co/lib/chanx" - "go.atoms.co/lib/iox" + "go.atoms.co/lib/testing/assertx" ) func TestDrain(t *testing.T) { diff --git a/go.mod b/go.mod index add72e0..de3692c 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module go.atoms.co/lib -go 1.24.9 +go 1.24.11 require ( contrib.go.opencensus.io/exporter/prometheus v0.4.1 @@ -9,6 +9,7 @@ require ( github.com/google/uuid v1.6.0 github.com/prometheus/client_golang v1.18.0 github.com/stretchr/testify v1.11.1 + go.atoms.co/iox v1.0.0 go.atoms.co/slicex v1.1.0 go.opencensus.io v0.24.0 go.opentelemetry.io/otel v1.32.0 diff --git a/go.sum b/go.sum index e30efaf..b71716c 100644 --- a/go.sum +++ b/go.sum @@ -226,6 +226,8 @@ github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.1.32/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +go.atoms.co/iox v1.0.0 h1:tSCyphSPh17TnjhZn5OaGScJcO5aac2acOjO9M76skE= +go.atoms.co/iox v1.0.0/go.mod h1:2PpoGBhfToxA0U2UfXkDSskPMiUyQzKi9DZtANqolew= go.atoms.co/slicex v1.1.0 h1:kxlUQRtwvt/iDiYAPGz+drM3CmVw1WI9AZLQ/nvkoCU= go.atoms.co/slicex v1.1.0/go.mod h1:le4qkj6I+WTrEXXtTRyKF+or6wI5wOL9NW3Atk/zr5E= go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU= diff --git a/iox/closer.go b/iox/closer.go deleted file mode 100644 index dd08be2..0000000 --- a/iox/closer.go +++ /dev/null @@ -1,76 +0,0 @@ -// Package iox contains various io utilities. -package iox - -import ( - "context" - "sync/atomic" -) - -// AsyncCloser is an async closer that supports a quit chan as a close notification mechanism. Thread-safe. -type AsyncCloser interface { - RAsyncCloser - WAsyncCloser -} - -// RAsyncCloser is an async closer that supports a read-only quit chan as a close notification mechanism. Thread-safe. -type RAsyncCloser interface { - // IsClosed returns true iff the instance is closed. - IsClosed() bool - // Closed returns a quit chan that is closed iff the instance is closed. - Closed() <-chan struct{} -} - -// WAsyncCloser is an async closer that supports a quit chan as a close mechanism. Thread-safe. -type WAsyncCloser interface { - // Close closes the instance. No error is returned as it is usually called in a defer. - Close() -} - -type asyncCloser struct { - quit chan struct{} - closed atomic.Bool -} - -// NewAsyncCloser returns a new open closer. -func NewAsyncCloser() AsyncCloser { - return &asyncCloser{quit: make(chan struct{})} -} - -func (c *asyncCloser) IsClosed() bool { - return c.closed.Load() -} - -func (c *asyncCloser) Closed() <-chan struct{} { - return c.quit -} - -func (c *asyncCloser) Close() { - if c.closed.CompareAndSwap(false, true) { - close(c.quit) - } -} - -// WithCancel closes the closer on context closure. Returns original closure for convenience. -func WithCancel(ctx context.Context, closer AsyncCloser) AsyncCloser { - return WithQuit(ctx.Done(), closer) -} - -// WithQuit closes the closer, if the quit channel if closed. Returns original closure for convenience. -func WithQuit(quit <-chan struct{}, closer AsyncCloser) AsyncCloser { - go func() { - select { - case <-quit: - closer.Close() - case <-closer.Closed(): - } - }() - return closer -} - -// WhenClosed closes the child closer, when the parent closes -func WhenClosed(parent RAsyncCloser, child WAsyncCloser) { - go func() { - <-parent.Closed() - child.Close() - }() -} diff --git a/iox/status.go b/iox/status.go deleted file mode 100644 index 19bb112..0000000 --- a/iox/status.go +++ /dev/null @@ -1,61 +0,0 @@ -package iox - -import ( - "fmt" - "time" -) - -// Failure represents an I/O operation failure that can only be retried after a given time. Value type. -type Failure struct { - Err error - Retry time.Time -} - -func (f Failure) String() string { - return fmt.Sprintf("[err=%v, retry=%v]", f.Err, f.Retry.Unix()) -} - -// Status represents the status of a pending, retryable I/O operation. The default value is -// an operation that has yet to be attempted. -type Status struct { - Last *Failure - Attempts int - Inflight bool -} - -func (s Status) ToInflight() Status { - return Status{ - Last: s.Last, - Attempts: s.Attempts, - Inflight: true, - } -} - -func (s Status) ToFailed(err error, retry time.Time) Status { - return Status{ - Last: &Failure{Err: err, Retry: retry}, - Attempts: s.Attempts + 1, - Inflight: false, - } -} - -func (s Status) IsReady(now time.Time) bool { - if s.Inflight { - return false - } - if s.Last != nil { - return now.After(s.Last.Retry) - } - return true -} - -func (s Status) String() string { - status := "pending" - if s.Inflight { - status = "inflight" - } - if s.Last == nil { - return status - } - return fmt.Sprintf("%v[attempts=%v, last=%v]", status, s.Attempts, *s.Last) -} diff --git a/metrics/metrics.go b/metrics/metrics.go index 2a71c9d..df55c3d 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -18,7 +18,6 @@ const ( Exponential Distribution = iota Uniform UserDefined - version = "1.0.1" ) type UnitType string @@ -177,7 +176,6 @@ func Init(appName string) error { // set the default app value for all metrics. initAppName(appName) - err := view.Register(ocgrpc.DefaultServerViews...) if err != nil { return err diff --git a/net/grpcx/stream.go b/net/grpcx/stream.go index 37637d3..667c98f 100644 --- a/net/grpcx/stream.go +++ b/net/grpcx/stream.go @@ -6,10 +6,10 @@ import ( "google.golang.org/grpc" - "go.atoms.co/lib/log" + "go.atoms.co/iox" "go.atoms.co/lib/chanx" "go.atoms.co/lib/contextx" - "go.atoms.co/lib/iox" + "go.atoms.co/lib/log" ) const ( diff --git a/signalx/signal.go b/signalx/signal.go index f49678c..9059780 100644 --- a/signalx/signal.go +++ b/signalx/signal.go @@ -5,7 +5,7 @@ import ( "os/signal" "syscall" - "go.atoms.co/lib/iox" + "go.atoms.co/iox" ) // InterruptChan returns a channel that receives a signal when the process receives either syscall.SIGTERM or syscall.SIGINT diff --git a/syncx/lock.go b/syncx/lock.go index cc1a97a..4105dad 100644 --- a/syncx/lock.go +++ b/syncx/lock.go @@ -3,7 +3,7 @@ package syncx import ( "sync/atomic" - "go.atoms.co/lib/iox" + "go.atoms.co/iox" ) // Lock is a chan-based closeable semaphore with N concurrent locks granted. If N=1, it acts as a mutex.