feat: Go and GoWait

Go performs a concurrent Effect by spinning up a goroutine for each item.

GoWait is like Go but it additionally accepts a sync.WaitGroup to which
it reports once the effect has completed.

Author: sharpvik

go.go

@@ -0,0 +1,17 @@
+package fungi
+
+import "sync"
+
+// Go performs a concurrent [Effect] by spinning up a goroutine for each item.
+func Go[T any](effect func(T)) StreamIdentity[T] {
+	return Effect(func(item T) { go effect(item) })
+}
+
+// GoWait is like [Go] but it additionally accepts a [sync.WaitGroup] to which
+// it reports once the effect has completed.
+func GoWait[T any](effect func(T), wg *sync.WaitGroup) StreamIdentity[T] {
+	return Effect(func(item T) {
+		wg.Add(1)
+		go func() { defer wg.Done(); effect(item) }()
+	})
+}

go_test.go

@@ -0,0 +1,50 @@
+package fungi
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestGo(t *testing.T) {
+	var wg sync.WaitGroup
+	var mutex sync.Mutex
+
+	total := 0
+	slice := []int{1, 2, 3, 4, 5}
+	stream := SliceStream(slice)
+
+	wg.Add(5)
+	count := Go(func(i int) {
+		defer wg.Done()
+		mutex.Lock()
+		defer mutex.Unlock()
+		total += i
+	})
+
+	consume := Loop(Nop[int])
+	require.NoError(t, consume(count(stream)))
+	wg.Wait()
+	require.Equal(t, 15, total)
+}
+
+func TestGoWait(t *testing.T) {
+	var wg sync.WaitGroup
+	var mutex sync.Mutex
+
+	total := 0
+	slice := []int{1, 2, 3, 4, 5}
+	stream := SliceStream(slice)
+
+	count := GoWait(func(i int) {
+		mutex.Lock()
+		defer mutex.Unlock()
+		total += i
+	}, &wg)
+
+	consume := Loop(Nop[int])
+	require.NoError(t, consume(count(stream)))
+	wg.Wait()
+	require.Equal(t, 15, total)
+}

unique.go

@@ -3,7 +3,7 @@ package fungi
 // Unique returns a stream of unique items. The incoming stream must be a stream
 // of comparable items.
 func Unique[T comparable](items Stream[T]) Stream[T] {
-	return UniqueBy(identity[T])(items)
+	return UniqueBy(Identity[T])(items)
 }
 
 // UniqueBy accepts a stream of any items and returns a stream of unique items.
@@ -35,4 +35,6 @@ func UniqueBy[T any, K comparable](id func(T) K) StreamIdentity[T] {
 	return StreamIdentity[T](unique)
 }
 
-func identity[T any](item T) T { return item }
+func Identity[T any](item T) T { return item }
+
+func Nop[T any](_ T) {}