Moves CloudFlux to spring cloud package.

FluxFirstNonEmptyEmitting and Tests used package private classes/interfaces from reactor. These were replaced with the public interfaces they were extending and any default methods copied to the implementation.

LambdaSubscriber was copied to the test where it was used and had similar treatment for its package protected interfaces.

Fixes gh-887

spring-cloud-commons/src/main/java/org/springframework/cloud/client/discovery/composite/reactive/ReactiveCompositeDiscoveryClient.java

@@ -19,11 +19,11 @@ package org.springframework.cloud.client.discovery.composite.reactive;
 import java.util.ArrayList;
 import java.util.List;
 
-import reactor.core.publisher.CloudFlux;
 import reactor.core.publisher.Flux;
 
 import org.springframework.cloud.client.ServiceInstance;
 import org.springframework.cloud.client.discovery.ReactiveDiscoveryClient;
+import org.springframework.cloud.commons.publisher.CloudFlux;
 import org.springframework.core.annotation.AnnotationAwareOrderComparator;
 
 /**

spring-cloud-commons/src/main/java/org/springframework/cloud/commons/publisher/CloudFlux.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2020 the original author or authors.
+ * Copyright 2013-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -14,13 +14,14 @@
  * limitations under the License.
  */
 
-package reactor.core.publisher;
+package org.springframework.cloud.commons.publisher;
 
 import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
 
 /**
  * INTERNAL USAGE ONLY. This functionality will be ported to reactor-core and will be
- * removed in a next release.
+ * removed in a future release.
  *
  * @author Tim Ysewyn
  */

spring-cloud-commons/src/main/java/org/springframework/cloud/commons/publisher/FluxFirstNonEmptyEmitting.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2020 the original author or authors.
+ * Copyright 2013-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -14,8 +14,9 @@
  * limitations under the License.
  */
 
-package reactor.core.publisher;
+package org.springframework.cloud.commons.publisher;
 
+import java.lang.reflect.Field;
 import java.util.Iterator;
 import java.util.Objects;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
@@ -25,12 +26,16 @@ import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
 import reactor.core.Scannable;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Operators;
 import reactor.util.annotation.Nullable;
 
+import org.springframework.util.ReflectionUtils;
+
 /**
  * @author Tim Ysewyn
  */
-final class FluxFirstNonEmptyEmitting<T> extends Flux<T> implements SourceProducer<T> {
+final class FluxFirstNonEmptyEmitting<T> extends Flux<T> implements Scannable, Publisher<T> {
 
 	final Publisher<? extends T>[] array;
 
@@ -137,6 +142,11 @@ final class FluxFirstNonEmptyEmitting<T> extends Flux<T> implements SourceProduc
 		return null; // no particular key to be represented, still useful in hooks
 	}
 
+	@Override
+	public String stepName() {
+		return "source(" + getClass().getSimpleName() + ")";
+	}
+
 	static final class RaceCoordinator<T> implements Subscription, Scannable {
 
 		final FirstNonEmptyEmittingSubscriber<T>[] subscribers;
@@ -264,8 +274,8 @@ final class FluxFirstNonEmptyEmitting<T> extends Flux<T> implements SourceProduc
 
 	}
 
-	static final class FirstNonEmptyEmittingSubscriber<T>
-			extends Operators.DeferredSubscription implements InnerOperator<T, T> {
+	static final class FirstNonEmptyEmittingSubscriber<T> extends Operators.DeferredSubscription
+			implements CoreSubscriber<T>, Scannable, Subscription {
 
 		final RaceCoordinator<T> parent;
 
@@ -285,14 +295,13 @@ final class FluxFirstNonEmptyEmitting<T> extends Flux<T> implements SourceProduc
 		@Override
 		@Nullable
 		public Object scanUnsafe(Attr key) {
-			if (key == Attr.PARENT) {
-				return s;
+			if (key == Attr.ACTUAL) {
+				return actual;
 			}
 			if (key == Attr.CANCELLED) {
 				return parent.cancelled;
 			}
-
-			return InnerOperator.super.scanUnsafe(key);
+			return super.scanUnsafe(key);
 		}
 
 		@Override
@@ -300,11 +309,6 @@ final class FluxFirstNonEmptyEmitting<T> extends Flux<T> implements SourceProduc
 			set(s);
 		}
 
-		@Override
-		public CoreSubscriber<? super T> actual() {
-			return actual;
-		}
-
 		@Override
 		public void onNext(T t) {
 			if (won) {
@@ -334,6 +338,11 @@ final class FluxFirstNonEmptyEmitting<T> extends Flux<T> implements SourceProduc
 			}
 		}
 
+		@Override
+		public String stepName() {
+			return "CloudFlux.firstNonEmpty";
+		}
+
 	}
 
 }

spring-cloud-commons/src/test/java/org/springframework/cloud/commons/publisher/FluxFirstNonEmptyEmittingTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2020 the original author or authors.
+ * Copyright 2013-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package reactor.core.publisher;
+package org.springframework.cloud.commons.publisher;
 
 import java.time.Duration;
 import java.util.Arrays;
@@ -24,6 +24,9 @@ import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
 import reactor.core.Scannable;
+import reactor.core.publisher.BaseSubscriber;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Operators;
 import reactor.test.StepVerifier;
 
 import static java.util.Collections.singletonList;
@@ -156,10 +159,8 @@ public class FluxFirstNonEmptyEmittingTests {
 
 	@Test
 	public void scanSubscriber() {
-		CoreSubscriber<String> actual = new LambdaSubscriber<>(null, e -> {
-		}, null, null);
-		FluxFirstNonEmptyEmitting.RaceCoordinator<String> parent = new FluxFirstNonEmptyEmitting.RaceCoordinator<>(
-				1);
+		CoreSubscriber<String> actual = new TestSubscriber<>();
+		FluxFirstNonEmptyEmitting.RaceCoordinator<String> parent = new FluxFirstNonEmptyEmitting.RaceCoordinator<>(1);
 		FluxFirstNonEmptyEmitting.FirstNonEmptyEmittingSubscriber<String> test = new FluxFirstNonEmptyEmitting.FirstNonEmptyEmittingSubscriber<>(
 				actual, parent, 1);
 		Subscription sub = Operators.emptySubscription();
@@ -174,10 +175,8 @@ public class FluxFirstNonEmptyEmittingTests {
 
 	@Test
 	public void scanRaceCoordinator() {
-		CoreSubscriber<String> actual = new LambdaSubscriber<>(null, e -> {
-		}, null, null);
-		FluxFirstNonEmptyEmitting.RaceCoordinator<String> parent = new FluxFirstNonEmptyEmitting.RaceCoordinator<>(
-				1);
+		CoreSubscriber<String> actual = new TestSubscriber<>();
+		FluxFirstNonEmptyEmitting.RaceCoordinator<String> parent = new FluxFirstNonEmptyEmitting.RaceCoordinator<>(1);
 		FluxFirstNonEmptyEmitting.FirstNonEmptyEmittingSubscriber<String> test = new FluxFirstNonEmptyEmitting.FirstNonEmptyEmittingSubscriber<>(
 				actual, parent, 1);
 		Subscription sub = Operators.emptySubscription();
@@ -190,4 +189,13 @@ public class FluxFirstNonEmptyEmittingTests {
 		assertThat(parent.scan(Scannable.Attr.CANCELLED)).isTrue();
 	}
 
+	static class TestSubscriber<T> extends BaseSubscriber<T> implements Scannable {
+
+		@Override
+		public Object scanUnsafe(Attr key) {
+			return null;
+		}
+
+	}
+
 }