for #1102 Detect some blocking APIs inside parallel and single Scheduler

`block`, `blockOptional`, `blockFirst`, `blockLast`, `toIterable` and `toStream`
all check if the executing thread is NonBlocking and throw an
`IllegalStateException` if it is.

reactor-core/src/main/java/reactor/core/publisher/BlockingOptionalMonoSubscriber.java

@@ -39,6 +39,7 @@
 import org.reactivestreams.Subscription;
 import reactor.core.Disposable;
 import reactor.core.Exceptions;
+import reactor.core.scheduler.Schedulers;
 import reactor.util.annotation.Nullable;
 
 /**
@@ -109,6 +110,9 @@ public final void dispose() {
 	 * @return an Optional representing the first value (or empty Optional if the source is empty)
 	 */
 	final Optional<T> blockingGet() {
+		if (Schedulers.isInNonBlockingThread()) {
+			throw new IllegalStateException("blockOptional() is blocking, which is not supported in thread " + Thread.currentThread().getName());
+		}
 		if (getCount() != 0) {
 			try {
 				await();
@@ -142,6 +146,9 @@ final Optional<T> blockingGet() {
 	 * @return an Optional representing the first value (or empty Optional if the source is empty)
 	 */
 	final Optional<T> blockingGet(long timeout, TimeUnit unit) {
+		if (Schedulers.isInNonBlockingThread()) {
+			throw new IllegalStateException("blockOptional() is blocking, which is not supported in thread " + Thread.currentThread().getName());
+		}
 		if (getCount() != 0) {
 			try {
 				if (!await(timeout, unit)) {

reactor-core/src/main/java/reactor/core/publisher/BlockingSingleSubscriber.java

@@ -22,6 +22,7 @@
 import org.reactivestreams.Subscription;
 import reactor.core.Disposable;
 import reactor.core.Exceptions;
+import reactor.core.scheduler.Schedulers;
 import reactor.util.annotation.Nullable;
 
 /**
@@ -72,6 +73,9 @@ public final void dispose() {
 	 */
 	@Nullable
 	final T blockingGet() {
+		if (Schedulers.isInNonBlockingThread()) {
+			throw new IllegalStateException("block()/blockFirst()/blockLast() are blocking, which is not supported in thread " + Thread.currentThread().getName());
+		}
 		if (getCount() != 0) {
 			try {
 				await();
@@ -103,6 +107,9 @@ final T blockingGet() {
 	 */
 	@Nullable
 	final T blockingGet(long timeout, TimeUnit unit) {
+		if (Schedulers.isInNonBlockingThread()) {
+			throw new IllegalStateException("block()/blockFirst()/blockLast() are blocking, which is not supported in thread " + Thread.currentThread().getName());
+		}
 		if (getCount() != 0) {
 			try {
 				if (!await(timeout, unit)) {

reactor-core/src/main/java/reactor/core/publisher/Flux.java

@@ -7751,6 +7751,9 @@ public final Iterable<T> toIterable(int batchSize) {
 	 */
 	public final Iterable<T> toIterable(int batchSize, @Nullable Supplier<Queue<T>>
 			queueProvider) {
+		if (Schedulers.isInNonBlockingThread()) {
+			throw new IllegalStateException("toIterable() is blocking, which is not supported in thread " + Thread.currentThread().getName());
+		}
 		final Supplier<Queue<T>> provider;
 		if(queueProvider == null){
 			provider = Queues.get(batchSize);
@@ -7786,6 +7789,9 @@ public final Stream<T> toStream() {
 	 * @return a {@link Stream} of unknown size with onClose attached to {@link Subscription#cancel()}
 	 */
 	public final Stream<T> toStream(int batchSize) {
+		if (Schedulers.isInNonBlockingThread()) {
+			throw new IllegalStateException("toStream() is blocking, which is not supported in thread " + Thread.currentThread().getName());
+		}
 		final Supplier<Queue<T>> provider;
 		provider = Queues.get(batchSize);
 		return new BlockingIterable<>(this, batchSize, provider).stream();

reactor-core/src/main/java/reactor/core/scheduler/ElasticScheduler.java

@@ -186,8 +186,8 @@ public Disposable schedulePeriodically(Runnable task, long initialDelay, long pe
 	public String toString() {
 		StringBuilder ts = new StringBuilder(Schedulers.ELASTIC)
 				.append('(');
-		if (factory instanceof Schedulers.SchedulerThreadFactory) {
-			ts.append('\"').append(((Schedulers.SchedulerThreadFactory) factory).get()).append('\"');
+		if (factory instanceof ReactorThreadFactory) {
+			ts.append('\"').append(((ReactorThreadFactory) factory).get()).append('\"');
 		}
 		ts.append(')');
 		return ts.toString();

reactor-core/src/main/java/reactor/core/scheduler/NonBlocking.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) 2011-2018 Pivotal Software Inc, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package reactor.core.scheduler;
+
+import java.util.concurrent.ThreadFactory;
+
+/**
+ * A marker interface that is detected on {@link Thread Threads} while executing Reactor
+ * blocking APIs, resulting in these calls throwing an exception.
+ * <p>
+ * Extend {@link AbstractReactorThreadFactory} for a {@link ThreadFactory} that can easily
+ * create such threads, and optionally name them and further configure them.
+ * <p>
+ * See {@link Schedulers#isBlockingCurrentThreadOk()} and
+ * {@link Schedulers#isBlockingCurrentThreadOk(Thread)} for a check that includes detecting
+ * this marker interface.
+ *
+ * @author Simon Baslé
+ */
+public interface NonBlocking { }

reactor-core/src/main/java/reactor/core/scheduler/ParallelScheduler.java

@@ -174,8 +174,8 @@ public Disposable schedulePeriodically(Runnable task,
     public String toString() {
         StringBuilder ts = new StringBuilder(Schedulers.PARALLEL)
                 .append('(').append(n);
-        if (factory instanceof Schedulers.SchedulerThreadFactory) {
-            ts.append(",\"").append(((Schedulers.SchedulerThreadFactory) factory).get()).append('\"');
+        if (factory instanceof ReactorThreadFactory) {
+            ts.append(",\"").append(((ReactorThreadFactory) factory).get()).append('\"');
         }
         ts.append(')');
         return ts.toString();

reactor-core/src/main/java/reactor/core/scheduler/ReactorThreadFactory.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright (c) 2011-2018 Pivotal Software Inc, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package reactor.core.scheduler;
+
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BiConsumer;
+import java.util.function.Supplier;
+
+import org.jetbrains.annotations.NotNull;
+import reactor.util.annotation.Nullable;
+
+/**
+ * The standard Reactor {@link ThreadFactory Thread factories} to be used by {@link Scheduler},
+ * creating {@link Thread} with a prefix (which can be retrieved with the {@link #get()} method).
+ *
+ * @author Simon Baslé
+ */
+class ReactorThreadFactory implements ThreadFactory,
+                                      Supplier<String>,
+                                      Thread.UncaughtExceptionHandler {
+
+	final private String                        name;
+	final private AtomicLong                    counterReference;
+	final private boolean                       daemon;
+	final private boolean                       rejectBlocking;
+
+	@Nullable
+	final private BiConsumer<Thread, Throwable> uncaughtExceptionHandler;
+
+	ReactorThreadFactory(String name,
+			AtomicLong counterReference,
+			boolean daemon,
+			boolean rejectBlocking,
+			@Nullable BiConsumer<Thread, Throwable> uncaughtExceptionHandler) {
+		this.name = name;
+		this.counterReference = counterReference;
+		this.daemon = daemon;
+		this.rejectBlocking = rejectBlocking;
+		this.uncaughtExceptionHandler = uncaughtExceptionHandler;
+	}
+
+	@Override
+	public final Thread newThread(@NotNull Runnable runnable) {
+		String newThreadName = name + "-" + counterReference.incrementAndGet();
+		Thread t = rejectBlocking
+				? new NonBlockingThread(runnable, newThreadName)
+				: new Thread(runnable, newThreadName);
+		if (daemon) {
+			t.setDaemon(true);
+		}
+		if (uncaughtExceptionHandler != null) {
+			t.setUncaughtExceptionHandler(this);
+		}
+		return t;
+	}
+
+	@Override
+	public void uncaughtException(Thread t, Throwable e) {
+		if (uncaughtExceptionHandler == null) {
+			return;
+		}
+
+		uncaughtExceptionHandler.accept(t,e);
+	}
+
+	/**
+	 * Get the prefix used for new {@link Thread Threads} created by this {@link ThreadFactory}.
+	 * The factory can also be seen as a {@link Supplier Supplier&lt;String&gt;}.
+	 *
+	 * @return the thread name prefix
+	 */
+	@Override
+	public final String get() {
+		return name;
+	}
+
+	static final class NonBlockingThread extends Thread implements NonBlocking {
+
+		public NonBlockingThread(Runnable target, String name) {
+			super(target, name);
+		}
+	}
+}