Mandatory cancellation support, coverage tests

akarnokd · Jan 1, 2018 · 6158001 · 6158001
1 parent 3741210
commit 6158001
Show file tree

Hide file tree

Showing 29 changed files with 218 additions and 15 deletions.
diff --git a/README.md b/README.md
@@ -9,7 +9,7 @@ Prototype Java 9 library based on the asynchronous enumerable concept (where mov
 ### Gradle
 
 ```groovy
-compile "com.github.akarnokd:async-enumerable:0.5.0"
+compile "com.github.akarnokd:async-enumerable:0.6.0"
 ```
 
 ### Getting started

diff --git a/gradle.properties b/gradle.properties
@@ -1,3 +1,3 @@
-version=0.5.0
+version=0.6.0
 org.gradle.jvmargs=-XX:+IgnoreUnrecognizedVMOptions --permit-illegal-access --show-version
 
diff --git a/src/main/java/hu/akarnokd/asyncenum/AsyncConcatArray.java b/src/main/java/hu/akarnokd/asyncenum/AsyncConcatArray.java
@@ -17,7 +17,7 @@
 package hu.akarnokd.asyncenum;
 
 import java.util.concurrent.*;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.*;
 import java.util.function.BiConsumer;
 
 final class AsyncConcatArray<T> implements AsyncEnumerable<T> {
@@ -38,33 +38,36 @@ static final class ConcatArrayEnumerator<T> extends AtomicInteger
 
         final AsyncEnumerable<T>[] sources;
 
-        AsyncEnumerator<T> currentEnumerator;
+        final AtomicReference<AsyncEnumerator<T>> currentEnumerator;
 
         CompletableFuture<Boolean> currentStage;
 
         int index;
 
         ConcatArrayEnumerator(AsyncEnumerable<T>[] sources) {
             this.sources = sources;
+            this.currentEnumerator = new AtomicReference<>();
         }
 
         @Override
         public CompletionStage<Boolean> moveNext() {
-            if (currentEnumerator == null) {
+            if (currentEnumerator.get() == null) {
                 if (index == sources.length) {
                     return FALSE;
                 }
-                currentEnumerator = sources[index++].enumerator();
+                if (!AsyncEnumeratorHelper.replace(currentEnumerator, sources[index++].enumerator())) {
+                    return CANCELLED;
+                }
             }
 
             currentStage = new CompletableFuture<>();
-            currentEnumerator.moveNext().whenComplete(this);
+            currentEnumerator.getPlain().moveNext().whenComplete(this);
             return currentStage;
         }
 
         @Override
         public T current() {
-            return currentEnumerator.current();
+            return currentEnumerator.getPlain().current();
         }
 
         @Override
@@ -82,11 +85,20 @@ public void accept(Boolean aBoolean, Throwable throwable) {
                             currentStage.complete(false);
                             break;
                         }
-                        currentEnumerator = sources[index++].enumerator();
-                        currentEnumerator.moveNext().whenComplete(this);
+                        AsyncEnumerator<T> en = sources[index++].enumerator();
+                        if (AsyncEnumeratorHelper.replace(currentEnumerator, en)) {
+                            en.moveNext().whenComplete(this);
+                        } else {
+                            break;
+                        }
                     } while (decrementAndGet() != 0);
                 }
             }
         }
+
+        @Override
+        public void cancel() {
+            AsyncEnumeratorHelper.cancel(currentEnumerator);
+        }
     }
 }
diff --git a/src/main/java/hu/akarnokd/asyncenum/AsyncEmpty.java b/src/main/java/hu/akarnokd/asyncenum/AsyncEmpty.java
@@ -40,4 +40,9 @@ public CompletionStage<Boolean> moveNext() {
     public Object current() {
         return null;
     }
+
+    @Override
+    public void cancel() {
+        // No action, consumer should stop calling moveNext().
+    }
 }
diff --git a/src/main/java/hu/akarnokd/asyncenum/AsyncEnumerator.java b/src/main/java/hu/akarnokd/asyncenum/AsyncEnumerator.java
@@ -51,8 +51,5 @@ public interface AsyncEnumerator<T> {
      * Instructs the AsyncEnumerator to cancel any outstanding async activity and
      * release resources associated with it.
      */
-    // FIXME make mandatory
-    default void cancel() {
-
-    }
+    void cancel();
 }
diff --git a/src/main/java/hu/akarnokd/asyncenum/AsyncEnumeratorHelper.java b/src/main/java/hu/akarnokd/asyncenum/AsyncEnumeratorHelper.java
@@ -33,6 +33,11 @@ public Object current() {
         return null;
     }
 
+    @Override
+    public void cancel() {
+        // No action, consumer should stop calling moveNext().
+    }
+
     @SuppressWarnings("unchecked")
     static <T> boolean cancel(AtomicReference<AsyncEnumerator<T>> target) {
         AsyncEnumerator<?> current = target.getAndSet((AsyncEnumerator<T>)CANCELLED);

diff --git a/src/main/java/hu/akarnokd/asyncenum/AsyncError.java b/src/main/java/hu/akarnokd/asyncenum/AsyncError.java
@@ -40,4 +40,9 @@ public CompletionStage<Boolean> moveNext() {
     public T current() {
         return null;
     }
+
+    @Override
+    public void cancel() {
+        // No action, consumer should stop calling moveNext().
+    }
 }
diff --git a/src/main/java/hu/akarnokd/asyncenum/AsyncFromArray.java b/src/main/java/hu/akarnokd/asyncenum/AsyncFromArray.java
@@ -59,5 +59,10 @@ public CompletionStage<Boolean> moveNext() {
         public T current() {
             return current;
         }
+
+        @Override
+        public void cancel() {
+            // No action, consumer should stop calling moveNext().
+        }
     }
 }
diff --git a/src/main/java/hu/akarnokd/asyncenum/AsyncFromCallable.java b/src/main/java/hu/akarnokd/asyncenum/AsyncFromCallable.java
@@ -62,5 +62,10 @@ public CompletionStage<Boolean> moveNext() {
         public T current() {
             return result;
         }
+
+        @Override
+        public void cancel() {
+            // No action, consumer should stop calling moveNext().
+        }
     }
 }
diff --git a/src/main/java/hu/akarnokd/asyncenum/AsyncFromCharSequence.java b/src/main/java/hu/akarnokd/asyncenum/AsyncFromCharSequence.java
@@ -59,5 +59,10 @@ public CompletionStage<Boolean> moveNext() {
         public Integer current() {
             return current;
         }
+
+        @Override
+        public void cancel() {
+            // No action, consumer should stop calling moveNext().
+        }
     }
 }
diff --git a/src/main/java/hu/akarnokd/asyncenum/AsyncFromCompletionStage.java b/src/main/java/hu/akarnokd/asyncenum/AsyncFromCompletionStage.java
@@ -74,5 +74,10 @@ public void accept(T t, Throwable throwable) {
                 completable.complete(true);
             }
         }
+
+        @Override
+        public void cancel() {
+            // No action, consumer should stop calling moveNext().
+        }
     }
 }
diff --git a/src/main/java/hu/akarnokd/asyncenum/AsyncFromIterable.java b/src/main/java/hu/akarnokd/asyncenum/AsyncFromIterable.java
@@ -56,5 +56,10 @@ public CompletionStage<Boolean> moveNext() {
         public T current() {
             return current;
         }
+
+        @Override
+        public void cancel() {
+            // No action, consumer should stop calling moveNext().
+        }
     }
 }
diff --git a/src/main/java/hu/akarnokd/asyncenum/AsyncJust.java b/src/main/java/hu/akarnokd/asyncenum/AsyncJust.java
@@ -54,5 +54,10 @@ public CompletionStage<Boolean> moveNext() {
         public T current() {
             return value;
         }
+
+        @Override
+        public void cancel() {
+            // No action, consumer should stop calling moveNext().
+        }
     }
 }
diff --git a/src/main/java/hu/akarnokd/asyncenum/AsyncMax.java b/src/main/java/hu/akarnokd/asyncenum/AsyncMax.java
@@ -52,6 +52,8 @@ static final class SumLongEnumerator<T> extends AtomicInteger
 
         CompletableFuture<Boolean> completable;
 
+        volatile boolean cancelled;
+
         SumLongEnumerator(AsyncEnumerator<T> source, Comparator<? super T> comparator) {
             this.source = source;
             this.comparator = comparator;
@@ -76,6 +78,9 @@ public T current() {
         void collectSource() {
             if (getAndIncrement() == 0) {
                 do {
+                    if (cancelled) {
+                        return;
+                    }
                     source.moveNext().whenComplete(this);
                 } while (decrementAndGet() != 0);
             }
@@ -110,5 +115,11 @@ public void accept(Boolean aBoolean, Throwable throwable) {
                 }
             }
         }
+
+        @Override
+        public void cancel() {
+            cancelled = true;
+            source.cancel();
+        }
     }
 }
diff --git a/src/main/java/hu/akarnokd/asyncenum/AsyncNever.java b/src/main/java/hu/akarnokd/asyncenum/AsyncNever.java
@@ -41,4 +41,9 @@ public CompletionStage<Boolean> moveNext() {
     public Object current() {
         return null;
     }
+
+    @Override
+    public void cancel() {
+        // No action, consumer should stop calling moveNext().
+    }
 }
diff --git a/src/main/java/hu/akarnokd/asyncenum/AsyncObserveOn.java b/src/main/java/hu/akarnokd/asyncenum/AsyncObserveOn.java
@@ -57,6 +57,10 @@ public T current() {
             return source.current();
         }
 
+        @Override
+        public void cancel() {
+            source.cancel();
+        }
     }
 
 }
diff --git a/src/main/java/hu/akarnokd/asyncenum/AsyncRange.java b/src/main/java/hu/akarnokd/asyncenum/AsyncRange.java
@@ -64,5 +64,10 @@ public CompletionStage<Boolean> moveNext() {
         public Integer current() {
             return current;
         }
+
+        @Override
+        public void cancel() {
+            // No action, consumer should stop calling moveNext().
+        }
     }
 }
diff --git a/src/main/java/hu/akarnokd/asyncenum/AsyncRepeatCallable.java b/src/main/java/hu/akarnokd/asyncenum/AsyncRepeatCallable.java
@@ -56,5 +56,10 @@ public CompletionStage<Boolean> moveNext() {
         public T current() {
             return result;
         }
+
+        @Override
+        public void cancel() {
+            // No action, consumer should stop calling moveNext().
+        }
     }
 }
diff --git a/src/main/java/hu/akarnokd/asyncenum/AsyncRepeatItem.java b/src/main/java/hu/akarnokd/asyncenum/AsyncRepeatItem.java
@@ -40,4 +40,9 @@ public CompletionStage<Boolean> moveNext() {
     public T current() {
         return item;
     }
+
+    @Override
+    public void cancel() {
+        // No action, consumer should stop calling moveNext().
+    }
 }
diff --git a/src/main/java/hu/akarnokd/asyncenum/AsyncSubscribeOn.java b/src/main/java/hu/akarnokd/asyncenum/AsyncSubscribeOn.java
@@ -65,7 +65,20 @@ public T current() {
 
         @Override
         public void run() {
-            source.complete(upstream.enumerator());
+            AsyncEnumerator<T> en = upstream.enumerator();
+            if (!source.complete(en)) {
+                en.cancel();
+            }
+        }
+
+        @Override
+        public void cancel() {
+            if (!source.completeExceptionally(new CancellationException())) {
+                AsyncEnumerator<T> en = source.getNow(null);
+                if (en != null) {
+                    en.cancel();
+                }
+            }
         }
     }
 }
diff --git a/src/main/java/hu/akarnokd/asyncenum/AsyncSumInt.java b/src/main/java/hu/akarnokd/asyncenum/AsyncSumInt.java
@@ -51,6 +51,8 @@ static final class SumIntEnumerator<T> extends AtomicInteger
 
         CompletableFuture<Boolean> cf;
 
+        volatile boolean cancelled;
+
         SumIntEnumerator(AsyncEnumerator<T> source, Function<? super T, ? extends Number> selector) {
             this.source = source;
             this.selector = selector;
@@ -75,6 +77,9 @@ public Integer current() {
         void collectSource() {
             if (getAndIncrement() == 0) {
                 do {
+                    if (cancelled) {
+                        return;
+                    }
                     source.moveNext().whenComplete(this);
                 } while (decrementAndGet() != 0);
             }
@@ -102,5 +107,11 @@ public void accept(Boolean aBoolean, Throwable throwable) {
                 }
             }
         }
+
+        @Override
+        public void cancel() {
+            cancelled = true;
+            source.cancel();
+        }
     }
 }
diff --git a/src/main/java/hu/akarnokd/asyncenum/AsyncSumLong.java b/src/main/java/hu/akarnokd/asyncenum/AsyncSumLong.java
@@ -51,6 +51,8 @@ static final class SumLongEnumerator<T> extends AtomicInteger
 
         CompletableFuture<Boolean> cf;
 
+        volatile boolean cancelled;
+
         SumLongEnumerator(AsyncEnumerator<T> source, Function<? super T, ? extends Number> selector) {
             this.source = source;
             this.selector = selector;
@@ -75,6 +77,9 @@ public Long current() {
         void collectSource() {
             if (getAndIncrement() == 0) {
                 do {
+                    if (cancelled) {
+                        return;
+                    }
                     source.moveNext().whenComplete(this);
                 } while (decrementAndGet() != 0);
             }
@@ -102,5 +107,11 @@ public void accept(Boolean aBoolean, Throwable throwable) {
                 }
             }
         }
+
+        @Override
+        public void cancel() {
+            cancelled = true;
+            source.cancel();
+        }
     }
 }
diff --git a/src/test/java/hu/akarnokd/asyncenum/AsyncConcatArrayTest.java b/src/test/java/hu/akarnokd/asyncenum/AsyncConcatArrayTest.java
@@ -21,6 +21,7 @@
 import java.util.*;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
 
 public class AsyncConcatArrayTest {
 
@@ -82,4 +83,16 @@ public void concatWith() {
     public void emptyArray() {
         TestHelper.assertResult(AsyncEnumerable.concatArray());
     }
+
+    @Test
+    public void cancelThenMove() {
+        TestHelper.withExecutor(executor -> {
+            for (int i = 0; i < 10000; i++) {
+                AsyncEnumerator<Integer> en = AsyncEnumerable.concatArray(AsyncEnumerable.range(1, 5))
+                        .enumerator();
+
+                TestHelper.race(en::cancel, en::moveNext, executor);
+            }
+        });
+    }
 }
diff --git a/src/test/java/hu/akarnokd/asyncenum/AsyncEnumeratorHelperTest.java b/src/test/java/hu/akarnokd/asyncenum/AsyncEnumeratorHelperTest.java
@@ -77,4 +77,9 @@ public void cancelledCurrentNull() {
     public void cancelledMoveNext() throws InterruptedException {
         assertSame(AsyncEnumerable.CANCELLED, AsyncEnumeratorHelper.CANCELLED.moveNext());
     }
+
+    @Test
+    public void cancelledCancel() {
+        AsyncEnumeratorHelper.CANCELLED.cancel();
+    }
 }