Ensure that Dispatcher can't be started more than one time (#632)

Just in case someone tries to call `.parallel()` on a _Stream_ instance returned by parallel collectors.

Author: pivovarit

src/main/java/com/pivovarit/collectors/Dispatcher.java

@@ -11,6 +11,7 @@
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
@@ -31,7 +32,8 @@ final class Dispatcher<T> {
     private final Executor executor;
     private final Semaphore limiter;
 
-    private volatile boolean started = false;
+    private final AtomicBoolean started = new AtomicBoolean(false);
+
     private volatile boolean shortCircuited = false;
 
     private Dispatcher(Executor executor, int permits) {
@@ -44,22 +46,23 @@ static <T> Dispatcher<T> of(Executor executor, int permits) {
     }
 
     void start() {
-        started = true;
-        dispatcher.execute(() -> {
-            try {
-                while (true) {
-                    Runnable task;
-                    if ((task = workingQueue.take()) != POISON_PILL) {
-                        limiter.acquire();
-                        executor.execute(withFinally(task, limiter::release));
-                    } else {
-                        break;
+        if (!started.getAndSet(true)) {
+            dispatcher.execute(() -> {
+                try {
+                    while (true) {
+                        Runnable task;
+                        if ((task = workingQueue.take()) != POISON_PILL) {
+                            limiter.acquire();
+                            executor.execute(withFinally(task, limiter::release));
+                        } else {
+                            break;
+                        }
                     }
+                } catch (Throwable e) {
+                    handle(e);
                 }
-            } catch (Throwable e) {
-                handle(e);
-            }
-        });
+            });
+        }
     }
 
     void stop() {
@@ -73,7 +76,7 @@ void stop() {
     }
 
     boolean isRunning() {
-        return started;
+        return started.get();
     }
 
     CompletableFuture<T> enqueue(Supplier<T> supplier) {