Improve tests in build: less verbosity, fix compiler warnings

build.gradle

@@ -174,16 +174,16 @@ configure(subprojects) { p ->
 	include '**/*Tests.*'
 	include '**/*Test.*'
 	include '**/*Spec.*'
-
   }
 
   //all test tasks will show FAILED/PASSED for each test method,
   // common exclusions, no scanning
   p.tasks.withType(Test).all {
 	testLogging {
-	  events "passed", "failed"
+	  events "FAILED"
 	  showExceptions true
-	  exceptionFormat "full"
+	  exceptionFormat "FULL"
+	  stackTraceFilters "ENTRY_POINT"
 	  maxGranularity 3
 	}
 	systemProperty("java.awt.headless", "true")
@@ -193,6 +193,17 @@ configure(subprojects) { p ->
 	scanForTestClasses = false
 	exclude '**/*Abstract*.*'
 	exclude '**/*OperatorTest*.*'
+
+	//allow re-run of failed tests only without special test tasks failing
+	// because the filter is too restrictive
+	filter.setFailOnNoMatchingTests(false)
+
+	//display intermediate results for special test tasks
+	afterSuite { desc, result ->
+	  if (!desc.parent) { // will match the outermost suite
+		println('\n' + "${desc} Results: ${result.resultType} (${result.testCount} tests, ${result.successfulTestCount} successes, ${result.failedTestCount} failures, ${result.skippedTestCount} skipped)")
+	  }
+	}
   }
 
   // now that kotlin-gradle-plugin 1.1.60 is out with fix for https://youtrack.jetbrains.com/issue/KT-17564
@@ -382,8 +393,6 @@ project('reactor-core') {
 	}
   }
 
-  sourceSets.test.resources.srcDirs = ["src/test/resources", "src/test/java"]
-
   if (!JavaVersion.current().isJava9Compatible()) {
 	test {
 	  jvmArgs = ["-Xbootclasspath/p:" + configurations.jsr166backport.asPath]

reactor-core/src/main/java/reactor/core/Scannable.java

@@ -226,8 +226,8 @@ T tryConvert(@Nullable Object o) {
 				return null;
 			}
 			if (safeConverter == null) {
-				//noinspection unchecked
-				return (T) o;
+				@SuppressWarnings("unchecked") T t = (T) o;
+				return t;
 			}
 			return safeConverter.apply(o);
 		}

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

@@ -4905,8 +4905,7 @@ public final Flux<T> hide() {
 	 * @return an indexed {@link Flux} with each source value combined with its 0-based index.
 	 */
 	public final Flux<Tuple2<Long, T>> index() {
-		//noinspection unchecked
-		return index(TUPLE2_BIFUNCTION);
+		return index(tuple2Function());
 	}
 
 	/**

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

@@ -94,8 +94,9 @@ static final class IndexFuseableSubscriber<I, T> implements InnerOperator<T, I>,
 		@Override
 		public void onSubscribe(Subscription s) {
 			if (Operators.validate(this.s, s)) {
-				//noinspection unchecked
-				this.s = (QueueSubscription<T>) s;
+				@SuppressWarnings("unchecked")
+				QueueSubscription<T> qs = (QueueSubscription<T>) s;
+				this.s = qs;
 
 				actual.onSubscribe(this);
 			}
@@ -239,8 +240,9 @@ static final class IndexFuseableConditionalSubscriber<I, T>
 		@Override
 		public void onSubscribe(Subscription s) {
 			if (Operators.validate(this.s, s)) {
-				//noinspection unchecked
-				this.s = (QueueSubscription<T>) s;
+				@SuppressWarnings("unchecked")
+				QueueSubscription<T> qs = (QueueSubscription<T>) s;
+				this.s = qs;
 				actual.onSubscribe(this);
 			}
 		}

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

@@ -148,8 +148,12 @@ static final class MergeOrderedMainProducer<T> implements InnerProducer<T> {
 				Comparator<? super T> comparator, int prefetch, int n) {
 			this.actual = actual;
 			this.comparator = comparator;
-			//noinspection unchecked
-			this.subscribers = new MergeOrderedInnerSubscriber[n];
+
+			@SuppressWarnings("unchecked")
+			MergeOrderedInnerSubscriber<T>[] mergeOrderedInnerSub =
+					new MergeOrderedInnerSubscriber[n];
+			this.subscribers = mergeOrderedInnerSub;
+
 			for (int i = 0; i < n; i++) {
 				this.subscribers[i] = new MergeOrderedInnerSubscriber<>(this, prefetch);
 			}
@@ -276,17 +280,19 @@ else if (innerDone) {
 						if (o != DONE) {
 							boolean smaller;
 							try {
-								//noinspection unchecked
-								smaller = min == null || comparator.compare(min, (T) o) > 0;
+								@SuppressWarnings("unchecked")
+								T t = (T) o;
+								smaller = min == null || comparator.compare(min, t) > 0;
 							} catch (Throwable ex) {
 								Exceptions.addThrowable(ERROR, this, ex);
 								cancel();
 								actual.onError(Exceptions.terminate(ERROR, this));
 								return;
 							}
 							if (smaller) {
-								//noinspection unchecked
-								min = (T) o;
+								@SuppressWarnings("unchecked")
+								T t = (T) o;
+								min = t;
 								minIndex = i;
 							}
 						}

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

@@ -301,8 +301,9 @@ void drainLoop() {
 					}
 
 					for (UnicastProcessor<T> w : ws) {
-						//noinspection unchecked
-						w.onNext((T) o);
+						@SuppressWarnings("unchecked")
+						T t = (T) o;
+						w.onNext(t);
 					}
 				}
 

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

@@ -51,14 +51,16 @@ class MonoCacheTime<T> extends MonoOperator<T, T> implements Runnable {
 		super(source);
 		this.ttl = ttl;
 		this.clock = clock;
-		//noinspection unchecked
-		this.state = (Signal<T>) EMPTY;
+		@SuppressWarnings("unchecked")
+		Signal<T> emptyState = (Signal<T>) EMPTY;
+		this.state = emptyState;
 	}
 
 	public void run() {
 		LOGGER.debug("expired {}", state);
-		//noinspection unchecked
-		state = (Signal<T>) EMPTY;
+		@SuppressWarnings("unchecked")
+		Signal<T> emptyState = (Signal<T>) EMPTY;
+		state = emptyState;
 	}
 
 	@Override
@@ -115,9 +117,9 @@ static final class CoordinatorSubscriber<T> implements InnerConsumer<T>, Signal<
 		static final AtomicReferenceFieldUpdater<CoordinatorSubscriber, Operators.MonoSubscriber[]> SUBSCRIBERS =
 				AtomicReferenceFieldUpdater.newUpdater(CoordinatorSubscriber.class, Operators.MonoSubscriber[].class, "subscribers");
 
+		@SuppressWarnings("unchecked")
 		CoordinatorSubscriber(MonoCacheTime<T> main) {
 			this.main = main;
-			//noinspection unchecked
 			this.subscribers = EMPTY;
 		}
 
@@ -173,7 +175,7 @@ final boolean add(Operators.MonoSubscriber<T, T> toAdd) {
 					return false;
 				}
 				int n = a.length;
-				//noinspection unchecked
+				@SuppressWarnings("unchecked")
 				Operators.MonoSubscriber<T, T>[] b = new Operators.MonoSubscriber[n + 1];
 				System.arraycopy(a, 0, b, 0, n);
 				b[n] = toAdd;
@@ -229,12 +231,12 @@ public void onSubscribe(Subscription s) {
 			}
 		}
 
+		@SuppressWarnings("unchecked")
 		private void signalCached(Signal<T> signal) {
 			if (STATE.compareAndSet(main, this, signal)) {
 				main.clock.schedule(main, main.ttl.toMillis(), TimeUnit.MILLISECONDS);
 			}
 
-			//noinspection unchecked
 			for (Operators.MonoSubscriber<T, T> inner : SUBSCRIBERS.getAndSet(this, TERMINATED)) {
 				if (signal.isOnNext()) {
 					inner.complete(signal.get());

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

@@ -556,7 +556,7 @@ FluxProcessor<O, O> getOrStart(){
 		return out;
 	}
 
-	@SuppressWarnings("rawtypes")
+	@SuppressWarnings({"rawtypes", "unchecked"})
     final static class NoopProcessor extends FluxProcessor {
 
 		@Override

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

@@ -240,11 +240,11 @@ final class LambdaOnNextErrorStrategy implements OnNextFailureStrategy {
 		private final BiFunction<? super Throwable, Object, ? extends Throwable> delegateProcessor;
 		private final BiPredicate<? super Throwable, Object> delegatePredicate;
 
+		@SuppressWarnings("unchecked")
 		public LambdaOnNextErrorStrategy(
 				BiFunction<? super Throwable, Object, ? extends Throwable> delegateProcessor) {
 			this.delegateProcessor = delegateProcessor;
 			if (delegateProcessor instanceof BiPredicate) {
-				//noinspection unchecked
 				this.delegatePredicate = (BiPredicate<? super Throwable, Object>) delegateProcessor;
 			}
 			else {

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

@@ -44,7 +44,7 @@ public interface Signal<T> extends Supplier<T>, Consumer<Subscriber<? super T>>
 	 *
 	 * @return an {@code OnCompleted} variety of {@code Signal}
 	 */
-	@SuppressWarnings("unchecked")
+	@SuppressWarnings({"deprecation", "unchecked"})
 	static <T> Signal<T> complete() {
 		return (Signal<T>) ImmutableSignal.ON_COMPLETE;
 	}
@@ -58,9 +58,9 @@ static <T> Signal<T> complete() {
 	 *
 	 * @return an {@code OnCompleted} variety of {@code Signal}
 	 */
+	@SuppressWarnings({"deprecation", "unchecked"})
 	static <T> Signal<T> complete(Context context) {
 		if (context.isEmpty()) {
-			//noinspection unchecked
 			return (Signal<T>) ImmutableSignal.ON_COMPLETE;
 		}
 		return new ImmutableSignal<>(context, SignalType.ON_COMPLETE, null, null, null);
@@ -158,6 +158,7 @@ static <T> Signal<T> subscribe(Subscription subscription, Context context) {
 	 * @param o the object to check
 	 * @return true if object represents the completion signal
 	 */
+	@SuppressWarnings("deprecation")
 	static boolean isComplete(Object o) {
 		return o == ImmutableSignal.ON_COMPLETE ||
 				(o instanceof Signal && ((Signal) o).getType() == SignalType.ON_COMPLETE);

reactor-core/src/test/java/reactor/core/ScannableTest.java

@@ -18,6 +18,7 @@
 
 import java.util.Collections;
 import java.util.List;
+import java.util.stream.Stream;
 
 import org.assertj.core.api.Condition;
 import org.junit.Test;
@@ -206,10 +207,12 @@ public void taggedFluxTest() {
 		                               .tag("2", "Two")
 		                               .hide();
 
-		assertThat(Scannable.from(tagged1).tags())
+		final Stream<Tuple2<String, String>> scannedTags1 = Scannable.from(tagged1).tags();
+		assertThat(scannedTags1.iterator())
 				.containsExactlyInAnyOrder(Tuples.of("1", "One"));
 
-		assertThat(Scannable.from(tagged2).tags())
+		final Stream<Tuple2<String, String>> scannedTags2 = Scannable.from(tagged2).tags();
+		assertThat(scannedTags2.iterator())
 				.containsExactlyInAnyOrder(Tuples.of("1", "One"), Tuples.of( "2", "Two"));
 	}
 
@@ -225,10 +228,12 @@ public void taggedHideFluxTest() {
 		                               .tag("2", "Two")
 		                               .hide();
 
-		assertThat(Scannable.from(tagged1).tags())
+		final Stream<Tuple2<String, String>> scannedTags1 = Scannable.from(tagged1).tags();
+		assertThat(scannedTags1.iterator())
 				.containsExactlyInAnyOrder(Tuples.of("1", "One"));
 
-		assertThat(Scannable.from(tagged2).tags())
+		final Stream<Tuple2<String, String>> scannedTags2 = Scannable.from(tagged2).tags();
+		assertThat(scannedTags2.iterator())
 				.containsExactlyInAnyOrder(Tuples.of("1", "One"), Tuples.of( "2", "Two"));
 	}
 
@@ -239,7 +244,8 @@ public void taggedAppendedFluxTest() {
 				    .tag("1", "One")
 				    .tag("2", "Two");
 
-		assertThat(Scannable.from(tagged1).tags())
+		final Stream<Tuple2<String, String>> scannedTags = Scannable.from(tagged1).tags();
+		assertThat(scannedTags.iterator())
 				.containsExactlyInAnyOrder(Tuples.of("1", "One"), Tuples.of( "2", "Two"));
 	}
 
@@ -251,7 +257,8 @@ public void taggedAppendedHideFluxTest() {
 				    .tag("1", "One")
 				    .tag("2", "Two");
 
-		assertThat(Scannable.from(tagged1).tags())
+		final Stream<Tuple2<String, String>> scannedTags = Scannable.from(tagged1).tags();
+		assertThat(scannedTags.iterator())
 				.containsExactlyInAnyOrder(Tuples.of("1", "One"), Tuples.of( "2", "Two"));
 	}
 
@@ -337,10 +344,12 @@ public void taggedMonoTest() {
 		                               .tag("2", "Two")
 		                               .hide();
 
-		assertThat(Scannable.from(tagged1).tags())
+		final Stream<Tuple2<String, String>> scannedTags1 = Scannable.from(tagged1).tags();
+		assertThat(scannedTags1.iterator())
 				.containsExactlyInAnyOrder(Tuples.of("1", "One"));
 
-		assertThat(Scannable.from(tagged2).tags())
+		final Stream<Tuple2<String, String>> scannedTags2 = Scannable.from(tagged2).tags();
+		assertThat(scannedTags2.iterator())
 				.containsExactlyInAnyOrder(Tuples.of("1", "One"), Tuples.of( "2", "Two"));
 	}
 
@@ -356,10 +365,12 @@ public void taggedHideMonoTest() {
 		                               .tag("2", "Two")
 		                               .hide();
 
-		assertThat(Scannable.from(tagged1).tags())
+		final Stream<Tuple2<String, String>> scannedTags1 = Scannable.from(tagged1).tags();
+		assertThat(scannedTags1.iterator())
 				.containsExactlyInAnyOrder(Tuples.of("1", "One"));
 
-		assertThat(Scannable.from(tagged2).tags())
+		final Stream<Tuple2<String, String>> scannedTags2 = Scannable.from(tagged2).tags();
+		assertThat(scannedTags2.iterator())
 				.containsExactlyInAnyOrder(Tuples.of("1", "One"), Tuples.of( "2", "Two"));
 	}
 
@@ -370,7 +381,8 @@ public void taggedAppendedMonoTest() {
 				    .tag("1", "One")
 				    .tag("2", "Two");
 
-		assertThat(Scannable.from(tagged1).tags())
+		final Stream<Tuple2<String, String>> scannedTags = Scannable.from(tagged1).tags();
+		assertThat(scannedTags.iterator())
 				.containsExactlyInAnyOrder(Tuples.of("1", "One"), Tuples.of( "2", "Two"));
 	}
 
@@ -382,7 +394,8 @@ public void taggedAppendedHideMonoTest() {
 				    .tag("1", "One")
 				    .tag("2", "Two");
 
-		assertThat(Scannable.from(tagged1).tags())
+		final Stream<Tuple2<String, String>> scannedTags = Scannable.from(tagged1).tags();
+		assertThat(scannedTags.iterator())
 				.containsExactlyInAnyOrder(Tuples.of("1", "One"), Tuples.of( "2", "Two"));
 	}
 
@@ -436,10 +449,11 @@ public void taggedParallelFluxTest() {
 		                               .tag("2", "Two")
 		                               .hide();
 
-		assertThat(Scannable.from(tagged1).tags())
-				.containsExactlyInAnyOrder(Tuples.of("1", "One"));
+		Stream<Tuple2<String, String>> scannedTags1 = Scannable.from(tagged1).tags();
+		assertThat(scannedTags1.iterator()).containsExactlyInAnyOrder(Tuples.of("1", "One"));
 
-		assertThat(Scannable.from(tagged2).tags())
+		Stream<Tuple2<String, String>> scannedTags2 = Scannable.from(tagged2).tags();
+		assertThat(scannedTags2.iterator())
 				.containsExactlyInAnyOrder(Tuples.of("1", "One"), Tuples.of( "2", "Two"));
 	}
 
@@ -450,7 +464,8 @@ public void taggedAppendedParallelFluxTest() {
 				    .tag("1", "One")
 				    .tag("2", "Two");
 
-		assertThat(Scannable.from(tagged1).tags())
+		final Stream<Tuple2<String, String>> scannedTags = Scannable.from(tagged1).tags();
+		assertThat(scannedTags.iterator())
 				.containsExactlyInAnyOrder(Tuples.of("1", "One"), Tuples.of( "2", "Two"));
 	}
 

reactor-core/src/test/java/reactor/core/publisher/EventLoopProcessorTest.java

@@ -143,6 +143,7 @@ public void awaitTerminationNanosLong() {
 				e.printStackTrace();
 			}
 		});
+		@SuppressWarnings("deprecation")
 		boolean shutResult = test.awaitAndShutdown(1000, TimeUnit.NANOSECONDS);
 		assertThat(test.executor)
 				.isNotNull()
@@ -172,6 +173,7 @@ public void awaitAndShutdownLongInterrupt() throws InterruptedException {
 				.when(executor).awaitTermination(anyLong(), any());
 		EventLoopProcessor<String> interruptingProcessor = initProcessor(executor);
 
+		@SuppressWarnings("deprecation")
 		boolean result = interruptingProcessor.awaitAndShutdown(100, TimeUnit.MILLISECONDS);
 
 		assertThat(Thread.currentThread().isInterrupted()).as("interrupted").isTrue();