Adding drain and any methods for handling channel drains and any conversions

Author: benjivesterby

.golangci.yml

@@ -17,19 +17,19 @@ linters-settings:
   goconst:
     min-len: 2
     min-occurrences: 2
-  gocritic:
-    enabled-tags:
-      - diagnostic
-      - experimental
-      - opinionated
-      - performance
-      - style
-    disabled-checks:
-      - dupImport # https://github.com/go-critic/go-critic/issues/845
-      - ifElseChain
-      - octalLiteral
-      - whyNoLint
-      - wrapperFunc
+  # gocritic:
+  #   enabled-tags:
+  #     - diagnostic
+  #     - experimental
+  #     - opinionated
+  #     - performance
+  #     - style
+  #   disabled-checks:
+  #     - dupImport # https://github.com/go-critic/go-critic/issues/845
+  #     - ifElseChain
+  #     - octalLiteral
+  #     - whyNoLint
+  #     - wrapperFunc
   gocyclo:
     min-complexity: 15
   goimports:
@@ -73,7 +73,7 @@ linters:
     - exhaustive
     - funlen
     - goconst
-    - gocritic
+    # - gocritic
     - gocyclo
     - gofmt
     - goimports
@@ -143,4 +143,4 @@ run:
 service:
   golangci-lint-version: 1.23.x # use the fixed version to not introduce new linters unexpectedly
   prepare:
-    - echo "here I can run custom commands, but no preparation needed for this repo"
+    - echo "here I can run custom commands, but no preparation needed for this repo"

.pre-commit-config.yaml

@@ -8,23 +8,23 @@ repos:
         entry: detect-secrets-hook
         args: ['--baseline', '.secrets.baseline']
 -   repo: https://github.com/golangci/golangci-lint
-    rev: v1.43.0
+    rev: v1.49.0
     hooks:
     -   id: golangci-lint
 -   repo: https://github.com/Bahjat/pre-commit-golang    
-    rev: c3086eea8af86847dbdff2e46b85a5fe3c9d9656 # pragma: allowlist secret
+    rev: v1.0.3 # pragma: allowlist secret
     hooks:
       - id: go-unit-tests
 -   repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.0.1
+    rev: v4.3.0
     hooks:
     -   id: check-json
     -   id: check-added-large-files
     -   id: pretty-format-json
     -   id: check-merge-conflict
     -   id: check-yaml
 -   repo: https://github.com/igorshubovych/markdownlint-cli
-    rev: v0.30.0
+    rev: v0.32.2
     hooks:
     -   id: markdownlint-fix
 -   repo: https://github.com/koalaman/shellcheck-precommit

scaler_test.go

@@ -35,6 +35,7 @@ func ScalerTest[U ~[]T, T comparable](
 			}
 
 			// Test that the scaler can be used with a nil context.
+			//nolint:staticcheck
 			out, err := s.Exec(nil, testdata.Chan(ctx))
 			if err != nil {
 				t.Errorf("expected no error, got %v", err)
@@ -83,6 +84,7 @@ func Test_Scaler_Exec(t *testing.T) {
 func Test_Scaler_NilFn(t *testing.T) {
 	s := Scaler[any, any]{}
 
+	//nolint:staticcheck
 	_, err := s.Exec(nil, nil)
 	if err == nil {
 		t.Error("Expected error, got nil")
@@ -106,6 +108,7 @@ func Test_Scaler_NilCtx(t *testing.T) {
 	cancel()
 
 	// Test that the scaler can be used with a nil context.
+	//nolint:staticcheck
 	out, err := s.Exec(nil, nil)
 	if err != nil {
 		t.Errorf("expected no error, got %v", err)
@@ -130,6 +133,7 @@ func Test_Scaler_CloseIn(t *testing.T) {
 	close(in)
 
 	// Test that the scaler can be used with a nil context.
+	//nolint:staticcheck
 	out, err := s.Exec(nil, in)
 	if err != nil {
 		t.Errorf("expected no error, got %v", err)

stream.go

@@ -3,7 +3,7 @@
 //
 // It is recommended to use the package via the following import:
 //
-//     import . "go.atomizer.io/stream"
+//	import . "go.atomizer.io/stream"
 //
 // Using the `.` import allows for functions to be called directly as if
 // the functions were in the same namespace without the need to append
@@ -18,13 +18,23 @@ import (
 	"go.structs.dev/gen"
 )
 
+type readonly[T any] interface {
+	<-chan T | chan T
+}
+
+type writeonly[T any] interface {
+	chan<- T | chan T
+}
+
 // Pipe accepts an incoming data channel and pipes it to the supplied
 // outgoing data channel.
 //
 // NOTE: Execute the Pipe function in a goroutine if parallel execution is
-// desired. Cancelling the context or closing the incoming channel is important
+// desired. Canceling the context or closing the incoming channel is important
 // to ensure that the goroutine is properly terminated.
-func Pipe[T any](ctx context.Context, in <-chan T, out chan<- T) {
+func Pipe[In readonly[T], Out writeonly[T], T any](
+	ctx context.Context, in In, out Out,
+) {
 	ctx = _ctx(ctx)
 
 	for {
@@ -51,10 +61,10 @@ type InterceptFunc[T, U any] func(context.Context, T) (U, bool)
 // accepts the incoming data and returns data of the same type and a boolean
 // indicating whether the data should be forwarded to the output channel.
 // The function is executed for each data item in the incoming channel as long
-// as the context is not cancelled or the incoming channel remains open.
-func Intercept[T, U any](
+// as the context is not canceled or the incoming channel remains open.
+func Intercept[In readonly[T], T, U any](
 	ctx context.Context,
-	in <-chan T,
+	in In,
 	fn InterceptFunc[T, U],
 ) <-chan U {
 	ctx = _ctx(ctx)
@@ -100,9 +110,9 @@ func Intercept[T, U any](
 // that receives all the data from the supplied channels.
 //
 // NOTE: The transfer takes place in a goroutine for each channel
-// so ensuring that the context is cancelled or the incoming channels
+// so ensuring that the context is canceled or the incoming channels
 // are closed is important to ensure that the goroutine is terminated.
-func FanIn[T any](ctx context.Context, in ...<-chan T) <-chan T {
+func FanIn[In readonly[T], T any](ctx context.Context, in ...In) <-chan T {
 	ctx = _ctx(ctx)
 	out := make(chan T)
 
@@ -135,9 +145,11 @@ func FanIn[T any](ctx context.Context, in ...<-chan T) <-chan T {
 // supplied outgoing data channels.
 //
 // NOTE: Execute the FanOut function in a goroutine if parallel execution is
-// desired. Cancelling the context or closing the incoming channel is important
+// desired. Canceling the context or closing the incoming channel is important
 // to ensure that the goroutine is properly terminated.
-func FanOut[T any](ctx context.Context, in <-chan T, out ...chan<- T) {
+func FanOut[In readonly[T], Out writeonly[T], T any](
+	ctx context.Context, in In, out ...Out,
+) {
 	ctx = _ctx(ctx)
 
 	if len(out) == 0 {
@@ -178,25 +190,26 @@ func FanOut[T any](ctx context.Context, in <-chan T, out ...chan<- T) {
 			for len(selectCases) > 1 {
 				chosen, _, _ := reflect.Select(selectCases)
 
-				// The context was cancelled.
+				// The context was canceled.
 				if chosen == 0 {
 					return
 				}
 
 				selectCases = gen.Exclude(selectCases, selectCases[chosen])
 			}
 		}
-
 	}
 }
 
 // Distribute accepts an incoming data channel and distributes the data among
 // the supplied outgoing data channels using a dynamic select statement.
 //
 // NOTE: Execute the Distribute function in a goroutine if parallel execution is
-// desired. Cancelling the context or closing the incoming channel is important
+// desired. Canceling the context or closing the incoming channel is important
 // to ensure that the goroutine is properly terminated.
-func Distribute[T any](ctx context.Context, in <-chan T, out ...chan<- T) {
+func Distribute[In readonly[T], Out writeonly[T], T any](
+	ctx context.Context, in In, out ...Out,
+) {
 	ctx = _ctx(ctx)
 
 	if len(out) == 0 {
@@ -226,6 +239,32 @@ func Distribute[T any](ctx context.Context, in <-chan T, out ...chan<- T) {
 			})
 			_, _, _ = reflect.Select(selectCases)
 		}
-
 	}
 }
+
+// Drain accepts a channel and drains the channel until the channel is closed
+// or the context is canceled.
+func Drain[U readonly[T], T any](ctx context.Context, in U) {
+	ctx = _ctx(ctx)
+
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case _, ok := <-in:
+				if !ok {
+					return
+				}
+			}
+		}
+	}()
+}
+
+// Any accepts an incoming data channel and converts the channel to a readonly
+// channel of the `any` type.
+func Any[U readonly[T], T any](ctx context.Context, in U) <-chan any {
+	return Intercept(ctx, in, func(_ context.Context, in T) (any, bool) {
+		return in, true
+	})
+}

stream_bench_test.go

@@ -19,7 +19,7 @@ func Benchmark_Pipe(b *testing.B) {
 	for n := 0; n < b.N; n++ {
 		select {
 		case <-ctx.Done():
-			b.Fatal("context cancelled")
+			b.Fatal("context canceled")
 		case c1 <- value:
 		case out, ok := <-c2:
 			if !ok {
@@ -47,7 +47,7 @@ func Benchmark_Intercept(b *testing.B) {
 	for n := 0; n < b.N; n++ {
 		select {
 		case <-ctx.Done():
-			b.Fatal("context cancelled")
+			b.Fatal("context canceled")
 		case in <- value:
 		case out, ok := <-out:
 			if !ok {
@@ -75,7 +75,7 @@ func Benchmark_FanIn(b *testing.B) {
 		for i := 0; i < 2; i++ {
 			select {
 			case <-ctx.Done():
-				b.Fatal("context cancelled")
+				b.Fatal("context canceled")
 			case _, ok := <-out:
 				if !ok {
 					b.Fatal("out closed prematurely")
@@ -134,6 +134,7 @@ func Benchmark_Scaler(b *testing.B) {
 
 	for n := 0; n < b.N; n++ {
 		// Test that the scaler can be used with a nil context.
+		//nolint:staticcheck
 		out, err := s.Exec(nil, testdata.Chan(ctx))
 		if err != nil {
 			b.Errorf("expected no error, got %v", err)