Merge pull request #6 from reugn/develop

v0.5.0

Author: reugn

.github/workflows/build.yml

@@ -1,21 +1,30 @@
 name: Build
 
-on: [push, pull_request]
+on:
+  push:
+    branches:
+      - '**'
+  pull_request:
+    branches:
+      - master
 
 jobs:
   build:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        go-version: [1.18.x]
+        go-version: [1.19.x]
     steps:
     - name: Setup Go
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v3
       with:
         go-version: ${{ matrix.go-version }}
+
     - name: Checkout code
-      uses: actions/checkout@v2
+      uses: actions/checkout@v3
+
     - name: Run coverage
       run: go test . -coverprofile=coverage.out -covermode=atomic
+      
     - name: Upload coverage to Codecov
       run: bash <(curl -s https://codecov.io/bash)

.github/workflows/golangci-lint.yml

@@ -10,6 +10,8 @@ jobs:
     name: lint
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - name: Checkout code
+        uses: actions/checkout@v3
+
       - name: golangci-lint
-        uses: golangci/golangci-lint-action@v2
+        uses: golangci/golangci-lint-action@v3

.golangci.yml

@@ -5,7 +5,6 @@ run:
 linters:
   disable-all: true
   enable:
-    - deadcode
     - dupl
     - errcheck
     - exportloopref
@@ -15,15 +14,16 @@ linters:
     - gocyclo
     - gofmt
     - goimports
+    - gosimple
     - govet
     - ineffassign
     - lll
     - misspell
     - prealloc
     - revive
+    - staticcheck
     - typecheck
     - unconvert
-    - varcheck
 
 issues:
   exclude-rules:

README.md

@@ -6,19 +6,18 @@
   <a href="https://codecov.io/gh/reugn/async"><img src="https://codecov.io/gh/reugn/async/branch/master/graph/badge.svg"></a>
 </div>
 <br/>
-Async provides synchronization and asynchronous computation utilities for Go.
-
-The implemented patterns were taken from Scala and Java.
+Async is a synchronization and asynchronous computation package for Go.
 
 ## Overview
 * **Future** - A placeholder object for a value that may not yet exist.
 * **Promise** - While futures are defined as a type of read-only placeholder object created for a result which doesn’t yet exist, a promise can be thought of as a writable, single-assignment container, which completes a future.
 * **Task** - A data type for controlling possibly lazy and asynchronous computations.
+* **WaitGroupContext** - A WaitGroup with the `context.Context` support for graceful unblocking.
 * **Reentrant Lock** - Mutex that allows goroutines to enter into the lock on a resource more than once.
 * **Optimistic Lock** - Mutex that allows optimistic reading. Could be retried or switched to RLock in case of failure. Significantly improves performance in case of frequent reads and short writes. See [benchmarks](./benchmarks/README.md).
 
 ## Examples
 Can be found in the examples directory/tests.
 
 ## License
-Licensed under the MIT License.
+Licensed under the MIT License.

examples/future/main.go

@@ -9,7 +9,7 @@ import (
 
 func main() {
 	future := asyncAction()
-	result, err := future.Get()
+	result, err := future.Join()
 	if err != nil {
 		log.Fatal(err)
 	}

future.go

@@ -1,6 +1,10 @@
 package async
 
-import "sync"
+import (
+	"fmt"
+	"sync"
+	"time"
+)
 
 // Future represents a value which may or may not currently be available,
 // but will be available at some point, or an error if that value could not be made available.
@@ -13,8 +17,12 @@ type Future[T any] interface {
 	// this Future.
 	FlatMap(func(T) (Future[T], error)) Future[T]
 
-	// Get blocks until the Future is completed and returns either a result or an error.
-	Get() (T, error)
+	// Join blocks until the Future is completed and returns either a result or an error.
+	Join() (T, error)
+
+	// Get blocks for at most the given time duration for this Future to complete
+	// and returns either a result or an error.
+	Get(time.Duration) (T, error)
 
 	// Recover handles any error that this Future might contain using a resolver function.
 	Recover(func() (T, error)) Future[T]
@@ -49,16 +57,35 @@ func NewFuture[T any]() Future[T] {
 // accept blocks once, until the Future result is available.
 func (fut *FutureImpl[T]) accept() {
 	fut.acceptOnce.Do(func() {
-		sig := <-fut.done
-		switch v := sig.(type) {
-		case error:
-			fut.err = v
-		default:
-			fut.value = v.(T)
+		result := <-fut.done
+		fut.setResult(result)
+	})
+}
+
+// acceptTimeout blocks once, until the Future result is available or until the timeout expires.
+func (fut *FutureImpl[T]) acceptTimeout(timeout time.Duration) {
+	fut.acceptOnce.Do(func() {
+		timer := time.NewTimer(timeout)
+		defer timer.Stop()
+		select {
+		case result := <-fut.done:
+			fut.setResult(result)
+		case <-timer.C:
+			fut.err = fmt.Errorf("Future timeout after %s", timeout)
 		}
 	})
 }
 
+// setResult assigns a value to the Future instance.
+func (fut *FutureImpl[T]) setResult(result interface{}) {
+	switch value := result.(type) {
+	case error:
+		fut.err = value
+	default:
+		fut.value = value.(T)
+	}
+}
+
 // Map creates a new Future by applying a function to the successful result of this Future
 // and returns the result of the function as a new Future.
 func (fut *FutureImpl[T]) Map(f func(T) (T, error)) Future[T] {
@@ -90,19 +117,26 @@ func (fut *FutureImpl[T]) FlatMap(f func(T) (Future[T], error)) Future[T] {
 				var nilT T
 				next.complete(nilT, terr)
 			} else {
-				next.complete(tfut.Get())
+				next.complete(tfut.Join())
 			}
 		}
 	}()
 	return next
 }
 
-// Get blocks until the Future is completed and returns either a result or an error.
-func (fut *FutureImpl[T]) Get() (T, error) {
+// Join blocks until the Future is completed and returns either a result or an error.
+func (fut *FutureImpl[T]) Join() (T, error) {
 	fut.accept()
 	return fut.value, fut.err
 }
 
+// Get blocks for at most the given time duration for this Future to complete
+// and returns either a result or an error.
+func (fut *FutureImpl[T]) Get(timeout time.Duration) (T, error) {
+	fut.acceptTimeout(timeout)
+	return fut.value, fut.err
+}
+
 // Recover handles any error that this Future might contain using a given resolver function.
 // Returns the result as a new Future.
 func (fut *FutureImpl[T]) Recover(f func() (T, error)) Future[T] {
@@ -125,7 +159,7 @@ func (fut *FutureImpl[T]) RecoverWith(rf Future[T]) Future[T] {
 	go func() {
 		fut.accept()
 		if fut.err != nil {
-			next.complete(rf.Get())
+			next.complete(rf.Join())
 		} else {
 			next.complete(fut.value, nil)
 		}

future_test.go

@@ -14,10 +14,10 @@ func TestFuture(t *testing.T) {
 		time.Sleep(time.Millisecond * 100)
 		p.Success(true)
 	}()
-	v, e := p.Future().Get()
+	res, err := p.Future().Join()
 
-	internal.AssertEqual(t, v, true)
-	internal.AssertEqual(t, e, nil)
+	internal.AssertEqual(t, res, true)
+	internal.AssertEqual(t, err, nil)
 }
 
 func TestFutureUtils(t *testing.T) {
@@ -34,7 +34,7 @@ func TestFutureUtils(t *testing.T) {
 	}()
 	arr := []Future[int]{p1.Future(), p2.Future(), p3.Future()}
 	res := []interface{}{1, 2, 3}
-	futRes, _ := FutureSeq(arr).Get()
+	futRes, _ := FutureSeq(arr).Join()
 
 	internal.AssertEqual(t, res, futRes)
 }
@@ -46,7 +46,7 @@ func TestFutureFirstCompleted(t *testing.T) {
 		p.Success(true)
 	}()
 	timeout := FutureTimer[bool](time.Millisecond * 100)
-	futRes, futErr := FutureFirstCompletedOf(p.Future(), timeout).Get()
+	futRes, futErr := FutureFirstCompletedOf(p.Future(), timeout).Join()
 
 	internal.AssertEqual(t, false, futRes)
 	if futErr == nil {
@@ -60,7 +60,7 @@ func TestFutureTransform(t *testing.T) {
 		time.Sleep(time.Millisecond * 100)
 		p1.Success(1)
 	}()
-	res, _ := p1.Future().Map(func(v int) (int, error) {
+	future := p1.Future().Map(func(v int) (int, error) {
 		return v + 1, nil
 	}).FlatMap(func(v int) (Future[int], error) {
 		nv := v + 1
@@ -69,8 +69,12 @@ func TestFutureTransform(t *testing.T) {
 		return p2.Future(), nil
 	}).Recover(func() (int, error) {
 		return 5, nil
-	}).Get()
+	})
 
+	res, _ := future.Get(time.Second * 5)
+	internal.AssertEqual(t, 3, res)
+
+	res, _ = future.Join()
 	internal.AssertEqual(t, 3, res)
 }
 
@@ -83,7 +87,22 @@ func TestFutureFailure(t *testing.T) {
 		time.Sleep(time.Millisecond * 200)
 		p2.Success(2)
 	}()
-	res, _ := p1.Future().RecoverWith(p2.Future()).Get()
+	res, _ := p1.Future().RecoverWith(p2.Future()).Join()
 
 	internal.AssertEqual(t, 2, res)
 }
+
+func TestFutureTimeout(t *testing.T) {
+	p := NewPromise[bool]()
+	go func() {
+		time.Sleep(time.Millisecond * 200)
+		p.Success(true)
+	}()
+	future := p.Future()
+
+	_, err := future.Get(time.Millisecond * 50)
+	internal.AssertErrorContains(t, err, "timeout")
+
+	_, err = future.Join()
+	internal.AssertErrorContains(t, err, "timeout")
+}

future_utils.go

@@ -10,12 +10,12 @@ func FutureSeq[T any](futures []Future[T]) Future[[]interface{}] {
 	next := NewFuture[[]interface{}]()
 	go func() {
 		seq := make([]interface{}, len(futures))
-		for i, f := range futures {
-			v, e := f.Get()
-			if e != nil {
-				seq[i] = e
+		for i, future := range futures {
+			res, err := future.Join()
+			if err != nil {
+				seq[i] = err
 			} else {
-				seq[i] = v
+				seq[i] = res
 			}
 		}
 		next.complete(seq, nil)
@@ -30,7 +30,7 @@ func FutureFirstCompletedOf[T any](futures ...Future[T]) Future[T] {
 	go func() {
 		for _, f := range futures {
 			go func(future Future[T]) {
-				next.complete(future.Get())
+				next.complete(future.Join())
 			}(f)
 		}
 	}()
@@ -45,7 +45,7 @@ func FutureTimer[T any](d time.Duration) Future[T] {
 		timer := time.NewTimer(d)
 		<-timer.C
 		var nilT T
-		next.complete(nilT, fmt.Errorf("FutureTimer %v timeout", d))
+		next.complete(nilT, fmt.Errorf("FutureTimer %s timeout", d))
 	}()
 	return next
 }

internal/test_utils.go

@@ -2,12 +2,32 @@ package internal
 
 import (
 	"reflect"
+	"strings"
 	"testing"
 )
 
 // AssertEqual verifies equality of two objects.
-func AssertEqual[T any](t *testing.T, a T, b T) {
+func AssertEqual[T any](t *testing.T, a, b T) {
 	if !reflect.DeepEqual(a, b) {
 		t.Fatalf("%v != %v", a, b)
 	}
 }
+
+// AssertErrorContains checks whether the given error contains the specified string.
+func AssertErrorContains(t *testing.T, err error, str string) {
+	if err == nil {
+		t.Fatalf("Error is nil")
+	} else if !strings.Contains(err.Error(), str) {
+		t.Fatalf("Error doen't contain string: %s", str)
+	}
+}
+
+// AssertPanic checks whether the given function panics.
+func AssertPanic(t *testing.T, f func()) {
+	defer func() {
+		if r := recover(); r == nil {
+			t.Errorf("The function did not panic")
+		}
+	}()
+	f()
+}