fixed bug

errorx/atmoic_error.go

@@ -0,0 +1,23 @@
+package errorx
+
+import "sync/atomic"
+
+// AtomicError defines an atomic error.
+type AtomicError struct {
+	err atomic.Value // error
+}
+
+// Set sets the error.
+func (ae *AtomicError) Set(err error) {
+	if err != nil {
+		ae.err.Store(err)
+	}
+}
+
+// Load returns the error.
+func (ae *AtomicError) Load() error {
+	if v := ae.err.Load(); v != nil {
+		return v.(error)
+	}
+	return nil
+}

errorx/batch_error.go

@@ -0,0 +1,50 @@
+package errorx
+
+import "bytes"
+
+type (
+	// A BatchError is an error that can hold multiple errors.
+	BatchError struct {
+		errs errorArray
+	}
+
+	errorArray []error
+)
+
+// Add adds err to be.
+func (be *BatchError) Add(err error) {
+	if err != nil {
+		be.errs = append(be.errs, err)
+	}
+}
+
+// Err returns an error that represents all errors.
+func (be *BatchError) Err() error {
+	switch len(be.errs) {
+	case 0:
+		return nil
+	case 1:
+		return be.errs[0]
+	default:
+		return be.errs
+	}
+}
+
+// NotNil checks if any error inside.
+func (be *BatchError) NotNil() bool {
+	return len(be.errs) > 0
+}
+
+// Error returns a string that represents inside errors.
+func (ea errorArray) Error() string {
+	var buf bytes.Buffer
+
+	for i := range ea {
+		if i > 0 {
+			buf.WriteByte('\n')
+		}
+		buf.WriteString(ea[i].Error())
+	}
+
+	return buf.String()
+}

errorx/callchain.go

@@ -0,0 +1,12 @@
+package errorx
+
+// Chain runs funs one by one until an error occurred.
+func Chain(fns ...func() error) error {
+	for _, fn := range fns {
+		if err := fn(); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

fx/parallel.go

@@ -0,0 +1,12 @@
+package fx
+
+import "github.com/abulo/ratel/v2/thread"
+
+// Parallel runs fns parallelly and waits for done.
+func Parallel(fns ...func()) {
+	group := thread.NewRoutineGroup()
+	for _, fn := range fns {
+		group.RunSafe(fn)
+	}
+	group.Wait()
+}

fx/retry.go

@@ -0,0 +1,46 @@
+package fx
+
+import "github.com/abulo/ratel/v2/errorx"
+
+const defaultRetryTimes = 3
+
+type (
+	// RetryOption defines the method to customize DoWithRetry.
+	RetryOption func(*retryOptions)
+
+	retryOptions struct {
+		times int
+	}
+)
+
+// DoWithRetry runs fn, and retries if failed. Default to retry 3 times.
+func DoWithRetry(fn func() error, opts ...RetryOption) error {
+	options := newRetryOptions()
+	for _, opt := range opts {
+		opt(options)
+	}
+
+	var berr errorx.BatchError
+	for i := 0; i < options.times; i++ {
+		if err := fn(); err != nil {
+			berr.Add(err)
+		} else {
+			return nil
+		}
+	}
+
+	return berr.Err()
+}
+
+// WithRetry customize a DoWithRetry call with given retry times.
+func WithRetry(times int) RetryOption {
+	return func(options *retryOptions) {
+		options.times = times
+	}
+}
+
+func newRetryOptions() *retryOptions {
+	return &retryOptions{
+		times: defaultRetryTimes,
+	}
+}

fx/ring.go

@@ -0,0 +1,52 @@
+package fx
+
+import "sync"
+
+// A Ring can be used as fixed size ring.
+type Ring struct {
+	elements []interface{}
+	index    int
+	lock     sync.Mutex
+}
+
+// NewRing returns a Ring object with the given size n.
+func NewRing(n int) *Ring {
+	if n < 1 {
+		panic("n should be greater than 0")
+	}
+
+	return &Ring{
+		elements: make([]interface{}, n),
+	}
+}
+
+// Add adds v into r.
+func (r *Ring) Add(v interface{}) {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	r.elements[r.index%len(r.elements)] = v
+	r.index++
+}
+
+// Take takes all items from r.
+func (r *Ring) Take() []interface{} {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	var size int
+	var start int
+	if r.index > len(r.elements) {
+		size = len(r.elements)
+		start = r.index % len(r.elements)
+	} else {
+		size = r.index
+	}
+
+	elements := make([]interface{}, size)
+	for i := 0; i < size; i++ {
+		elements[i] = r.elements[(start+i)%len(r.elements)]
+	}
+
+	return elements
+}