functions: error check should check twice

tychoish · Aug 31, 2023 · e0c854f · e0c854f
1 parent b1a1df7
commit e0c854f
Show file tree

Hide file tree

Showing 4 changed files with 126 additions and 1 deletion.
diff --git a/process_test.go b/process_test.go
@@ -198,6 +198,47 @@ func TestProcess(t *testing.T) {
 			check.NotError(t, e)
 			check.Equal(t, 1, called)
 		})
+		t.Run("Multi", func(t *testing.T) {
+			called := 0
+			hfcall := 0
+			var hf Future[error] = func() error {
+				hfcall++
+				switch hfcall {
+				case 1, 2, 3:
+					return nil
+				case 4, 5:
+					return ers.ErrCurrentOpAbort
+				}
+				return errors.New("unexpected error")
+			}
+			wf := MakeProcessor(func(in int) error {
+				called++
+				check.Equal(t, in, 42)
+				if hfcall == 3 {
+					return io.EOF
+				}
+				return nil
+			})
+			ecpf := wf.WithErrorCheck(hf)
+			e := ecpf.Wait(42)
+			check.NotError(t, e)
+			check.Equal(t, 1, called)
+			check.Equal(t, 2, hfcall)
+
+			e = ecpf.Wait(42)
+			check.Error(t, e)
+			check.Equal(t, 2, called)
+			check.Equal(t, 4, hfcall)
+			check.ErrorIs(t, e, ers.ErrCurrentOpAbort)
+			check.ErrorIs(t, e, io.EOF)
+
+			e = ecpf.Wait(42)
+			check.Error(t, e)
+			check.Equal(t, 2, called)
+			check.Equal(t, 5, hfcall)
+			check.ErrorIs(t, e, ers.ErrCurrentOpAbort)
+			check.NotErrorIs(t, e, io.EOF)
+		})
 	})
 	t.Run("Worker", func(t *testing.T) {
 		ctx, cancel := context.WithCancel(context.Background())

diff --git a/producer.go b/producer.go
@@ -215,7 +215,7 @@ func (pf Producer[T]) CheckForce() (T, bool)               { o, e := pf.Wait();
 // original producer returns.
 func (pf Producer[T]) Launch(ctx context.Context) Producer[T] {
 	eh, ef := HF.ErrorCollector()
-	pipe := Blocking(make(chan T, 1))
+	pipe := Blocking(make(chan T))
 	pipe.Send().Processor().
 		ReadAll(pf).
 		Operation(eh).

diff --git a/producer_test.go b/producer_test.go
@@ -604,6 +604,50 @@ func TestProducer(t *testing.T) {
 			check.Equal(t, 42, out)
 			check.Equal(t, 1, called)
 		})
+		t.Run("Multi", func(t *testing.T) {
+			called := 0
+			hfcall := 0
+			var hf Future[error] = func() error {
+				hfcall++
+				switch hfcall {
+				case 1, 2, 3:
+					return nil
+				case 4, 5:
+					return ers.ErrCurrentOpAbort
+				}
+				return errors.New("unexpected error")
+			}
+			wf := MakeProducer(func() (int, error) {
+				called++
+				if hfcall == 3 {
+					return 0, io.EOF
+				}
+				return 42, nil
+			})
+			ecpf := wf.WithErrorCheck(hf)
+			out, e := ecpf.Wait()
+			check.NotError(t, e)
+			check.Equal(t, out, 42)
+			check.Equal(t, 1, called)
+			check.Equal(t, 2, hfcall)
+
+			out, e = ecpf.Wait()
+			check.Equal(t, out, 0)
+			check.Error(t, e)
+			check.Equal(t, 2, called)
+			check.Equal(t, 4, hfcall)
+			check.ErrorIs(t, e, ers.ErrCurrentOpAbort)
+			check.ErrorIs(t, e, io.EOF)
+
+			out, e = ecpf.Wait()
+			check.Error(t, e)
+			check.Equal(t, out, 0)
+			check.Equal(t, 2, called)
+			check.Equal(t, 5, hfcall)
+			check.ErrorIs(t, e, ers.ErrCurrentOpAbort)
+			check.NotErrorIs(t, e, io.EOF)
+		})
+
 	})
 	t.Run("Limit", func(t *testing.T) {
 		t.Run("Serial", func(t *testing.T) {

diff --git a/worker_test.go b/worker_test.go
@@ -490,6 +490,46 @@ func TestWorker(t *testing.T) {
 			check.NotError(t, e)
 			check.Equal(t, 1, called)
 		})
+		t.Run("Multi", func(t *testing.T) {
+			called := 0
+			hfcall := 0
+			var hf Future[error] = func() error {
+				hfcall++
+				switch hfcall {
+				case 1, 2, 3:
+					return nil
+				case 4, 5:
+					return ers.ErrCurrentOpAbort
+				}
+				return errors.New("unexpected error")
+			}
+			wf := Worker(func(ctx context.Context) error {
+				called++
+				if hfcall == 3 {
+					return io.EOF
+				}
+				return nil
+			})
+			ecpf := wf.WithErrorCheck(hf)
+			e := ecpf.Wait()
+			check.NotError(t, e)
+			check.Equal(t, 1, called)
+			check.Equal(t, 2, hfcall)
+
+			e = ecpf.Wait()
+			check.Error(t, e)
+			check.Equal(t, 2, called)
+			check.Equal(t, 4, hfcall)
+			check.ErrorIs(t, e, ers.ErrCurrentOpAbort)
+			check.ErrorIs(t, e, io.EOF)
+
+			e = ecpf.Wait()
+			check.Error(t, e)
+			check.Equal(t, 2, called)
+			check.Equal(t, 5, hfcall)
+			check.ErrorIs(t, e, ers.ErrCurrentOpAbort)
+			check.NotErrorIs(t, e, io.EOF)
+		})
 	})
 	t.Run("Delay", func(t *testing.T) {
 		t.Run("Basic", func(t *testing.T) {