Add BlockHound integration

build.gradle

@@ -17,6 +17,8 @@ buildscript {
   ext.jfrogExtractorVersion = "4.5.2"
   ext.bndVersion = "4.2.0"
   ext.checkstyleVersion = "6.19"
+  ext.blockhoundVersion = "1.0.1.RELEASE"
+  ext.testsetsVersion = "2.2.1"
 
   // --------------------------------------
 
@@ -34,6 +36,7 @@ buildscript {
     classpath "com.jfrog.bintray.gradle:gradle-bintray-plugin:$bintrayVersion"
     classpath "org.jfrog.buildinfo:build-info-extractor-gradle:$jfrogExtractorVersion"
     classpath "biz.aQute.bnd:biz.aQute.bnd.gradle:$bndVersion"
+    classpath "org.unbroken-dome.gradle-plugins:gradle-testsets-plugin:$testsetsVersion"
   }
 }
 
@@ -65,6 +68,7 @@ apply plugin: "com.github.hierynomus.license"
 apply plugin: "com.jfrog.bintray"
 apply plugin: "com.jfrog.artifactory"
 apply plugin: "eclipse"
+apply plugin: "org.unbroken-dome.test-sets"
 
 sourceCompatibility = JavaVersion.VERSION_1_6
 targetCompatibility = JavaVersion.VERSION_1_6
@@ -73,18 +77,26 @@ repositories {
   mavenCentral()
 }
 
+testSets {
+    blockHoundTest
+}
+
 dependencies {
     signature "org.codehaus.mojo.signature:java16:1.1@signature"
 
     api "org.reactivestreams:reactive-streams:$reactiveStreamsVersion"
     jmh "org.reactivestreams:reactive-streams:$reactiveStreamsVersion"
 
+    compileOnly "io.projectreactor.tools:blockhound:$blockhoundVersion"
+
     testImplementation "junit:junit:$junitVersion"
     testImplementation "org.mockito:mockito-core:$mockitoVersion"
 
     testImplementation "org.reactivestreams:reactive-streams-tck:$reactiveStreamsVersion"
     testImplementation "org.testng:testng:$testNgVersion"
     testImplementation "com.google.guava:guava:$guavaVersion"
+
+    blockHoundTestImplementation "io.projectreactor.tools:blockhound:$blockhoundVersion"
 }
 
 javadoc {

src/blockHoundTest/java/io/reactivex/rxjava3/BlockHoundIntegrationTest.java

@@ -0,0 +1,63 @@
+/**
+ * Copyright (c) 2019-present, RxJava Contributors.
+ *
+ * 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 io.reactivex.rxjava3;
+
+import io.reactivex.rxjava3.internal.schedulers.RxThreadFactory;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import reactor.blockhound.BlockHound;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+public class BlockHoundIntegrationTest {
+
+    @BeforeClass
+    public static void beforeClass() {
+        BlockHound.install();
+    }
+
+    @Test
+    public void blockingCallInNonBlockingThread() throws Exception {
+        FutureTask<Void> task = new FutureTask<Void>(new Callable<Void>() {
+            @Override
+            public Void call() throws Exception {
+                Thread.sleep(0);
+                return null;
+            }
+        });
+        Thread thread = new RxThreadFactory("test-", 1, true).newThread(task);
+        thread.start();
+
+        try {
+            task.get(5, TimeUnit.SECONDS);
+            Assert.fail("Should fail");
+        } catch (ExecutionException e) {
+            Throwable throwable = e.getCause();
+            assertNotNull("An exception was thrown", throwable);
+            assertTrue("Blocking call was reported", throwable.getMessage().contains("Blocking call"));
+        } finally {
+            thread.interrupt();
+        }
+    }
+}

src/main/java/io/reactivex/rxjava3/internal/schedulers/RxJavaBlockHoundIntegration.java

@@ -0,0 +1,52 @@
+/**
+ * Copyright (c) 2019-present, RxJava Contributors.
+ *
+ * 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 io.reactivex.rxjava3.internal.schedulers;
+
+import io.reactivex.rxjava3.annotations.Experimental;
+import reactor.blockhound.BlockHound;
+import reactor.blockhound.integration.BlockHoundIntegration;
+
+import java.util.function.Function;
+import java.util.function.Predicate;
+
+/**
+ * This class integrates RxJava with BlockHound.
+ * <p>
+ * It is public but only because of the ServiceLoader's limitations
+ * and SHOULD NOT be considered a public API.
+ */
+@SuppressWarnings("Since15")
+@Experimental
+public final class RxJavaBlockHoundIntegration implements BlockHoundIntegration {
+
+    @Override
+    public void applyTo(BlockHound.Builder builder) {
+        builder.nonBlockingThreadPredicate(new Function<Predicate<Thread>, Predicate<Thread>>() {
+            @Override
+            public Predicate<Thread> apply(Predicate<Thread> p) {
+                return p.or(new Predicate<Thread>() {
+                    @Override
+                    public boolean test(Thread obj) {
+                        return obj instanceof NonBlockingThread;
+                    }
+                });
+            }
+        });
+    }
+
+    @Override
+    public int compareTo(BlockHoundIntegration o) {
+        return 0;
+    }
+}

src/main/resources/META-INF/services/reactor.blockhound.integration.BlockHoundIntegration

@@ -0,0 +1,14 @@
+# Copyright (c) 2019-present, RxJava Contributors.
+#
+# 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.
+io.reactivex.rxjava3.internal.schedulers.RxJavaBlockHoundIntegration