Merge branch '2.2.x'

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;
 
@@ -131,6 +136,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;
@@ -257,7 +267,7 @@ final class FluxFirstNonEmptyEmitting<T> extends Flux<T> implements SourceProduc
 	}
 
 	static final class FirstNonEmptyEmittingSubscriber<T> extends Operators.DeferredSubscription
-			implements InnerOperator<T, T> {
+			implements CoreSubscriber<T>, Scannable, Subscription {
 
 		final RaceCoordinator<T> parent;
 
@@ -276,14 +286,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
@@ -291,11 +300,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) {
@@ -325,6 +329,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;
@@ -140,8 +143,7 @@ public class FluxFirstNonEmptyEmittingTests {
 
 	@Test
 	public void scanSubscriber() {
-		CoreSubscriber<String> actual = new LambdaSubscriber<>(null, e -> {
-		}, null, null);
+		CoreSubscriber<String> actual = new TestSubscriber<>();
 		FluxFirstNonEmptyEmitting.RaceCoordinator<String> parent = new FluxFirstNonEmptyEmitting.RaceCoordinator<>(1);
 		FluxFirstNonEmptyEmitting.FirstNonEmptyEmittingSubscriber<String> test = new FluxFirstNonEmptyEmitting.FirstNonEmptyEmittingSubscriber<>(
 				actual, parent, 1);
@@ -157,8 +159,7 @@ public class FluxFirstNonEmptyEmittingTests {
 
 	@Test
 	public void scanRaceCoordinator() {
-		CoreSubscriber<String> actual = new LambdaSubscriber<>(null, e -> {
-		}, null, null);
+		CoreSubscriber<String> actual = new TestSubscriber<>();
 		FluxFirstNonEmptyEmitting.RaceCoordinator<String> parent = new FluxFirstNonEmptyEmitting.RaceCoordinator<>(1);
 		FluxFirstNonEmptyEmitting.FirstNonEmptyEmittingSubscriber<String> test = new FluxFirstNonEmptyEmitting.FirstNonEmptyEmittingSubscriber<>(
 				actual, parent, 1);
@@ -172,4 +173,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;
+		}
+
+	}
+
 }