fix: Split testing from publishing pipelines

.github/workflows/check.yml

@@ -0,0 +1,24 @@
+on: [push, pull_request]
+name: Test
+jobs:
+  test:
+    strategy:
+      matrix:
+        go-version: [1.19.x, 1.20.x]
+        os: [ubuntu-latest, macos-latest, windows-latest]
+    runs-on: ${{ matrix.os }}
+    steps:
+    - uses: actions/checkout@v3
+    - uses: actions/setup-go@v4
+      with:
+        go-version: ${{ matrix.go-version }}
+    - run: go test ./...
+
+  test-cache:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v3
+    - uses: actions/setup-go@v4
+      with:
+        go-version: 1.20.x
+    - run: go test -v -race ./...

.github/workflows/publish.yml

@@ -0,0 +1,13 @@
+on:
+  push:
+    branches: ["main"]
+name: Publish
+jobs:
+  publish:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v3
+    - uses: actions/setup-go@v4
+      with:
+        go-version: 1.18
+    - run: GOPROXY=proxy.golang.org go list -m github.com/junitechnology/jpipe

filter_test.go

@@ -40,7 +40,7 @@ func TestFilter(t *testing.T) {
 		start := time.Now()
 		channel := jpipe.FromSlice(pipeline, []int{1, 2, 3, 4, 5}).
 			Filter(func(i int) bool {
-				time.Sleep(10 * time.Millisecond)
+				time.Sleep(100 * time.Millisecond)
 				return i%2 == 1
 			}, jpipe.Concurrent(3))
 
@@ -49,7 +49,7 @@ func TestFilter(t *testing.T) {
 
 		slices.Sort(mappedValues) // The output order with concurrency is unpredictable
 		assert.Equal(t, []int{1, 3, 5}, mappedValues)
-		assert.Less(t, elapsed, 30*time.Millisecond) // It would have taken 50ms serially, but it takes about 20ms with 5 elements and concurrency 3
+		assert.Less(t, elapsed, 300*time.Millisecond) // It would have taken 500ms serially, but it takes about 200ms with 5 elements and concurrency 3
 		assertPipelineDone(t, pipeline, 10*time.Millisecond)
 	})
 
@@ -58,15 +58,15 @@ func TestFilter(t *testing.T) {
 		start := time.Now()
 		channel := jpipe.FromSlice(pipeline, []int{1, 2, 3, 4, 5}).
 			Filter(func(i int) bool {
-				time.Sleep(10 * time.Millisecond)
+				time.Sleep(100 * time.Millisecond)
 				return i%2 == 1
 			}, jpipe.Concurrent(3), jpipe.Ordered(3))
 
 		mappedValues := drainChannel(channel)
 		elapsed := time.Since(start)
 
 		assert.Equal(t, []int{1, 3, 5}, mappedValues)
-		assert.Less(t, elapsed, 30*time.Millisecond) // It would have taken 50ms serially, but it takes about 20ms with 5 elements and concurrency 3
+		assert.Less(t, elapsed, 300*time.Millisecond) // It would have taken 500ms serially, but it takes about 200ms with 5 elements and concurrency 3
 		assertPipelineDone(t, pipeline, 10*time.Millisecond)
 	})
 

sink_test.go

@@ -50,7 +50,7 @@ func TestForEach(t *testing.T) {
 		lock := sync.Mutex{}
 		start := time.Now()
 		<-channel.ForEach(func(value int) {
-			time.Sleep(10 * time.Millisecond)
+			time.Sleep(100 * time.Millisecond)
 			lock.Lock()
 			values = append(values, value)
 			lock.Unlock()
@@ -59,7 +59,7 @@ func TestForEach(t *testing.T) {
 
 		slices.Sort(values) // The output order with concurrency is unpredictable
 		assert.Equal(t, []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, values)
-		assert.Less(t, elapsed, 40*time.Millisecond) // It would have taken 100ms serially, but it takes about 20ms with 10 elements and concurrency 5
+		assert.Less(t, elapsed, 400*time.Millisecond) // It would have taken 1000ms serially, but it takes about 200ms with 10 elements and concurrency 5
 		assert.NoError(t, pipeline.Error())
 		assertPipelineDone(t, pipeline, 10*time.Millisecond)
 	})

transform_test.go

@@ -42,7 +42,7 @@ func TestMap(t *testing.T) {
 		channel := jpipe.FromSlice(pipeline, []int{1, 2, 3, 4, 5})
 		start := time.Now()
 		mappedChannel := jpipe.Map(channel, func(i int) string {
-			time.Sleep(10 * time.Millisecond)
+			time.Sleep(100 * time.Millisecond)
 			return fmt.Sprintf("%dA", i)
 		}, jpipe.Concurrent(3))
 
@@ -51,7 +51,7 @@ func TestMap(t *testing.T) {
 
 		slices.Sort(mappedValues) // The output order with concurrency is unpredictable
 		assert.Equal(t, []string{"1A", "2A", "3A", "4A", "5A"}, mappedValues)
-		assert.Less(t, elapsed, 30*time.Millisecond) // It would have taken 50ms serially, but it takes about 20ms with 5 elements and concurrency 3
+		assert.Less(t, elapsed, 300*time.Millisecond) // It would have taken 500ms serially, but it takes about 200ms with 5 elements and concurrency 3
 		assertPipelineDone(t, pipeline, 10*time.Millisecond)
 	})
 
@@ -60,15 +60,15 @@ func TestMap(t *testing.T) {
 		channel := jpipe.FromSlice(pipeline, []int{1, 2, 3, 4, 5})
 		start := time.Now()
 		mappedChannel := jpipe.Map(channel, func(i int) string {
-			time.Sleep(10 * time.Millisecond)
+			time.Sleep(100 * time.Millisecond)
 			return fmt.Sprintf("%dA", i)
 		}, jpipe.Concurrent(3), jpipe.Ordered(3))
 
 		mappedValues := drainChannel(mappedChannel)
 		elapsed := time.Since(start)
 
 		assert.Equal(t, []string{"1A", "2A", "3A", "4A", "5A"}, mappedValues)
-		assert.Less(t, elapsed, 30*time.Millisecond) // It would have taken 50ms serially, but it takes about 20ms with 5 elements and concurrency 3
+		assert.Less(t, elapsed, 300*time.Millisecond) // It would have taken 500ms serially, but it takes about 200ms with 5 elements and concurrency 3
 		assertPipelineDone(t, pipeline, 10*time.Millisecond)
 	})
 
@@ -124,7 +124,7 @@ func TestFlatMap(t *testing.T) {
 		channel := jpipe.FromSlice(pipeline, []int{1, 2, 3, 4, 5})
 		start := time.Now()
 		mappedChannel := jpipe.FlatMap(channel, func(i int) *jpipe.Channel[string] {
-			time.Sleep(10 * time.Millisecond)
+			time.Sleep(100 * time.Millisecond)
 			return jpipe.FromSlice(pipeline, []string{fmt.Sprintf("%dA", i), fmt.Sprintf("%dB", i)})
 		}, jpipe.Concurrent(3))
 
@@ -133,7 +133,7 @@ func TestFlatMap(t *testing.T) {
 
 		slices.Sort(mappedValues) // The output order with concurrency is unpredictable
 		assert.Equal(t, []string{"1A", "1B", "2A", "2B", "3A", "3B", "4A", "4B", "5A", "5B"}, mappedValues)
-		assert.Less(t, elapsed, 30*time.Millisecond) // It would have taken 50ms serially, but it takes about 20ms with 5 elements and concurrency 3
+		assert.Less(t, elapsed, 300*time.Millisecond) // It would have taken 500ms serially, but it takes about 200ms with 5 elements and concurrency 3
 		assertPipelineDone(t, pipeline, 10*time.Millisecond)
 	})
 }

utility_test.go

@@ -101,7 +101,7 @@ func TestTap(t *testing.T) {
 		start := time.Now()
 		channel := jpipe.FromSlice(pipeline, slice).
 			Tap(func(i int) {
-				time.Sleep(10 * time.Millisecond)
+				time.Sleep(100 * time.Millisecond)
 				valuesFromTap[i-1] = i // not using append to avoid the need for synchronization
 			}, jpipe.Concurrent(3))
 
@@ -112,7 +112,7 @@ func TestTap(t *testing.T) {
 		slices.Sort(outputValues) // The output order with concurrency is unpredictable
 		assert.Equal(t, slice, valuesFromTap)
 		assert.Equal(t, slice, outputValues)
-		assert.Less(t, elapsed, 30*time.Millisecond) // It would have taken 50ms serially, but it takes about 20ms with 5 elements and concurrency 3
+		assert.Less(t, elapsed, 300*time.Millisecond) // It would have taken 500ms serially, but it takes about 200ms with 5 elements and concurrency 3
 		assertPipelineDone(t, pipeline, 10*time.Millisecond)
 	})
 
@@ -123,7 +123,7 @@ func TestTap(t *testing.T) {
 		start := time.Now()
 		channel := jpipe.FromSlice(pipeline, slice).
 			Tap(func(i int) {
-				time.Sleep(10 * time.Millisecond)
+				time.Sleep(100 * time.Millisecond)
 				valuesFromTap[i-1] = i // not using append to avoid the need for synchronization
 			}, jpipe.Concurrent(3), jpipe.Ordered(3))
 
@@ -133,7 +133,7 @@ func TestTap(t *testing.T) {
 		slices.Sort(valuesFromTap) // The tap execution order with concurrency is unpredictable
 		assert.Equal(t, slice, valuesFromTap)
 		assert.Equal(t, slice, outputValues)
-		assert.Less(t, elapsed, 30*time.Millisecond) // It would have taken 50ms serially, but it takes about 20ms with 5 elements and concurrency 3
+		assert.Less(t, elapsed, 300*time.Millisecond) // It would have taken 500ms serially, but it takes about 200ms with 5 elements and concurrency 3
 		assertPipelineDone(t, pipeline, 10*time.Millisecond)
 	})