Take first non-empty flux into consideration for composite reactive service discovery

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

@@ -19,6 +19,7 @@ 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;
@@ -55,7 +56,7 @@ public class ReactiveCompositeDiscoveryClient implements ReactiveDiscoveryClient
 		for (ReactiveDiscoveryClient discoveryClient : discoveryClients) {
 			serviceInstances.add(discoveryClient.getInstances(serviceId));
 		}
-		return Flux.first(serviceInstances);
+		return CloudFlux.firstNonEmpty(serviceInstances);
 	}
 
 	@Override

spring-cloud-commons/src/main/java/reactor/core/publisher/CloudFlux.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2019-2019 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://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.publisher;
+
+import org.reactivestreams.Publisher;
+
+/**
+ * INTERNAL USAGE ONLY. This functionality will be ported to reactor-core and will be
+ * removed in a next release.
+ *
+ * @author Tim Ysewyn
+ */
+public abstract class CloudFlux<T> extends Flux<T> {
+
+	/**
+	 * Pick the first {@link Publisher} to emit an onNext/onError signal and replay all
+	 * signals from that {@link Publisher}, effectively behaving like the fastest of these
+	 * competing sources. If all the sources complete empty, a single completion signal is
+	 * sent. Note that if all the sources are empty (never emit an element, ie. no onNext)
+	 * AND at least one is also infinite (no onComplete/onError signal), the resulting
+	 * {@link Flux} will be infinite and empty (like {@link Flux#never()}).
+	 * @param sources The competing source publishers
+	 * @param <I> The type of values in both source and output sequences
+	 * @return a new {@link Flux} behaving like the fastest of its sources
+	 */
+	@SafeVarargs
+	public static <I> Flux<I> firstNonEmpty(Publisher<? extends I>... sources) {
+		return onAssembly(new FluxFirstNonEmptyEmitting<>(sources));
+	}
+
+	/**
+	 * Pick the first {@link Publisher} to emit an onNext/onError signal and replay all
+	 * signals from that {@link Publisher}, effectively behaving like the fastest of these
+	 * competing sources. If all the sources complete empty, a single completion signal is
+	 * sent. Note that if all the sources are empty (never emit an element, ie. no onNext)
+	 * AND at least one is also infinite (no onComplete/onError signal), the resulting
+	 * {@link Flux} will be infinite and empty (like {@link Flux#never()}).
+	 * @param sources The competing source publishers
+	 * @param <I> The type of values in both source and output sequences
+	 * @return a new {@link reactor.core.publisher.Flux} behaving like the fastest of its
+	 * sources
+	 */
+	public static <I> Flux<I> firstNonEmpty(
+			Iterable<? extends Publisher<? extends I>> sources) {
+		return onAssembly(new FluxFirstNonEmptyEmitting<>(sources));
+	}
+
+}

spring-cloud-commons/src/main/java/reactor/core/publisher/FluxFirstNonEmptyEmitting.java

@@ -0,0 +1,339 @@
+/*
+ * Copyright 2019-2019 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://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.publisher;
+
+import java.util.Iterator;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.stream.Stream;
+
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscription;
+import reactor.core.CoreSubscriber;
+import reactor.core.Scannable;
+import reactor.util.annotation.Nullable;
+
+/**
+ * @author Tim Ysewyn
+ */
+final class FluxFirstNonEmptyEmitting<T> extends Flux<T> implements SourceProducer<T> {
+
+	final Publisher<? extends T>[] array;
+
+	final Iterable<? extends Publisher<? extends T>> iterable;
+
+	@SafeVarargs
+	FluxFirstNonEmptyEmitting(Publisher<? extends T>... array) {
+		this.array = Objects.requireNonNull(array, "array");
+		this.iterable = null;
+	}
+
+	FluxFirstNonEmptyEmitting(Iterable<? extends Publisher<? extends T>> iterable) {
+		this.array = null;
+		this.iterable = Objects.requireNonNull(iterable);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public void subscribe(CoreSubscriber<? super T> actual) {
+		Publisher<? extends T>[] a = array;
+		int n;
+		if (a == null) {
+			n = 0;
+			a = new Publisher[8];
+
+			Iterator<? extends Publisher<? extends T>> it;
+
+			try {
+				it = Objects.requireNonNull(iterable.iterator(),
+						"The iterator returned is null");
+			}
+			catch (Throwable e) {
+				Operators.error(actual,
+						Operators.onOperatorError(e, actual.currentContext()));
+				return;
+			}
+
+			for (;;) {
+
+				boolean b;
+
+				try {
+					b = it.hasNext();
+				}
+				catch (Throwable e) {
+					Operators.error(actual,
+							Operators.onOperatorError(e, actual.currentContext()));
+					return;
+				}
+
+				if (!b) {
+					break;
+				}
+
+				Publisher<? extends T> p;
+
+				try {
+					p = Objects.requireNonNull(it.next(),
+							"The Publisher returned by the iterator is null");
+				}
+				catch (Throwable e) {
+					Operators.error(actual,
+							Operators.onOperatorError(e, actual.currentContext()));
+					return;
+				}
+
+				if (n == a.length) {
+					Publisher<? extends T>[] c = new Publisher[n + (n >> 2)];
+					System.arraycopy(a, 0, c, 0, n);
+					a = c;
+				}
+				a[n++] = p;
+			}
+
+		}
+		else {
+			n = a.length;
+		}
+
+		if (n == 0) {
+			Operators.complete(actual);
+			return;
+		}
+		if (n == 1) {
+			Publisher<? extends T> p = a[0];
+
+			if (p == null) {
+				Operators.error(actual,
+						new NullPointerException("The single source Publisher is null"));
+			}
+			else {
+				p.subscribe(actual);
+			}
+			return;
+		}
+
+		RaceCoordinator<T> coordinator = new RaceCoordinator<>(n);
+
+		coordinator.subscribe(a, n, actual);
+	}
+
+	@Override
+	public Object scanUnsafe(Attr key) {
+		return null; // no particular key to be represented, still useful in hooks
+	}
+
+	static final class RaceCoordinator<T> implements Subscription, Scannable {
+
+		final FirstNonEmptyEmittingSubscriber<T>[] subscribers;
+
+		volatile boolean cancelled;
+
+		volatile int wip;
+
+		volatile int competingSubscribers;
+
+		@SuppressWarnings("rawtypes")
+		static final AtomicIntegerFieldUpdater<RaceCoordinator> WIP = AtomicIntegerFieldUpdater
+				.newUpdater(RaceCoordinator.class, "wip");
+
+		static final AtomicIntegerFieldUpdater<RaceCoordinator> COMPETING_SUBSCRIBERS = AtomicIntegerFieldUpdater
+				.newUpdater(RaceCoordinator.class, "competingSubscribers");
+
+		@SuppressWarnings("unchecked")
+		RaceCoordinator(int n) {
+			subscribers = new FirstNonEmptyEmittingSubscriber[n];
+			wip = Integer.MIN_VALUE;
+			competingSubscribers = n;
+		}
+
+		@Override
+		public Stream<? extends Scannable> inners() {
+			return Stream.of(subscribers);
+		}
+
+		@Override
+		@Nullable
+		public Object scanUnsafe(Attr key) {
+			if (key == Attr.CANCELLED) {
+				return cancelled;
+			}
+
+			return null;
+		}
+
+		void subscribe(Publisher<? extends T>[] sources, int n,
+				CoreSubscriber<? super T> actual) {
+			FirstNonEmptyEmittingSubscriber<T>[] a = subscribers;
+
+			for (int i = 0; i < n; i++) {
+				a[i] = new FirstNonEmptyEmittingSubscriber<>(actual, this, i);
+			}
+
+			actual.onSubscribe(this);
+
+			for (int i = 0; i < n; i++) {
+				if (cancelled || wip != Integer.MIN_VALUE) {
+					return;
+				}
+
+				Publisher<? extends T> p = sources[i];
+
+				if (p == null) {
+					if (WIP.compareAndSet(this, Integer.MIN_VALUE, -1)) {
+						actual.onError(new NullPointerException(
+								"The " + i + " th Publisher source is null"));
+					}
+					return;
+				}
+
+				p.subscribe(a[i]);
+			}
+
+		}
+
+		@Override
+		public void request(long n) {
+			if (Operators.validate(n)) {
+				int w = wip;
+				if (w >= 0) {
+					subscribers[w].request(n);
+				}
+				else {
+					for (FirstNonEmptyEmittingSubscriber<T> s : subscribers) {
+						s.request(n);
+					}
+				}
+			}
+		}
+
+		@Override
+		public void cancel() {
+			if (cancelled) {
+				return;
+			}
+			cancelled = true;
+
+			int w = wip;
+			if (w >= 0) {
+				subscribers[w].cancel();
+			}
+			else {
+				for (FirstNonEmptyEmittingSubscriber<T> s : subscribers) {
+					s.cancel();
+				}
+			}
+		}
+
+		boolean tryWin(int index) {
+			if (wip == Integer.MIN_VALUE) {
+				if (WIP.compareAndSet(this, Integer.MIN_VALUE, index)) {
+
+					FirstNonEmptyEmittingSubscriber<T>[] a = subscribers;
+					int n = a.length;
+
+					for (int i = 0; i < n; i++) {
+						if (i != index) {
+							a[i].cancel();
+						}
+					}
+
+					return true;
+				}
+			}
+			return false;
+		}
+
+		int resignFromRace() {
+			return COMPETING_SUBSCRIBERS.decrementAndGet(this);
+		}
+
+	}
+
+	static final class FirstNonEmptyEmittingSubscriber<T>
+			extends Operators.DeferredSubscription implements InnerOperator<T, T> {
+
+		final RaceCoordinator<T> parent;
+
+		final CoreSubscriber<? super T> actual;
+
+		final int index;
+
+		boolean won;
+
+		FirstNonEmptyEmittingSubscriber(CoreSubscriber<? super T> actual,
+				RaceCoordinator<T> parent, int index) {
+			this.actual = actual;
+			this.parent = parent;
+			this.index = index;
+		}
+
+		@Override
+		@Nullable
+		public Object scanUnsafe(Attr key) {
+			if (key == Attr.PARENT) {
+				return s;
+			}
+			if (key == Attr.CANCELLED) {
+				return parent.cancelled;
+			}
+
+			return InnerOperator.super.scanUnsafe(key);
+		}
+
+		@Override
+		public void onSubscribe(Subscription s) {
+			set(s);
+		}
+
+		@Override
+		public CoreSubscriber<? super T> actual() {
+			return actual;
+		}
+
+		@Override
+		public void onNext(T t) {
+			if (won) {
+				actual.onNext(t);
+			}
+			else if (parent.tryWin(index)) {
+				won = true;
+				actual.onNext(t);
+			}
+		}
+
+		@Override
+		public void onError(Throwable t) {
+			if (won) {
+				actual.onError(t);
+			}
+			else if (parent.tryWin(index)) {
+				won = true;
+				actual.onError(t);
+			}
+		}
+
+		@Override
+		public void onComplete() {
+			if (won || parent.resignFromRace() == 0) {
+				actual.onComplete();
+			}
+		}
+
+	}
+
+}

spring-cloud-commons/src/test/java/reactor/core/publisher/FluxFirstNonEmptyEmittingTests.java

@@ -0,0 +1,193 @@
+/*
+ * Copyright 2019-2019 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://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.publisher;
+
+import java.time.Duration;
+import java.util.Arrays;
+
+import org.junit.Test;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscription;
+import reactor.core.CoreSubscriber;
+import reactor.core.Scannable;
+import reactor.test.StepVerifier;
+
+import static java.util.Collections.singletonList;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNullPointerException;
+
+/**
+ * @author Tim Ysewyn
+ */
+public class FluxFirstNonEmptyEmittingTests {
+
+	@Test
+	public void arrayNull() {
+		assertThatNullPointerException()
+				.isThrownBy(() -> CloudFlux.firstNonEmpty((Publisher<Integer>[]) null));
+	}
+
+	@Test
+	public void iterableNull() {
+		assertThatNullPointerException().isThrownBy(
+				() -> CloudFlux.firstNonEmpty((Iterable<Publisher<Integer>>) null));
+	}
+
+	@Test
+	public void firstWinner() {
+		StepVerifier
+				.create(CloudFlux.firstNonEmpty(Flux.range(1, 10), Flux.range(11, 10)))
+				.expectNext(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).verifyComplete();
+	}
+
+	@Test
+	public void firstWinnerSecondEmpty() {
+		StepVerifier.create(CloudFlux.firstNonEmpty(Flux.range(1, 10), Flux.empty()))
+				.expectNext(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).verifyComplete();
+	}
+
+	@Test
+	public void firstWinnerBackpressured() {
+		StepVerifier
+				.create(CloudFlux.firstNonEmpty(Flux.range(1, 10), Flux.range(11, 10)))
+				.thenRequest(5).expectNext(1, 2, 3, 4, 5).thenCancel()
+				.verifyThenAssertThat().hasNotDiscardedElements().hasNotDroppedElements()
+				.hasNotDroppedErrors();
+	}
+
+	@Test
+	public void secondWinner() {
+		StepVerifier
+				.create(CloudFlux.firstNonEmpty(Flux.never(), Flux.range(11, 10).log()))
+				.expectNext(11, 12, 13, 14, 15, 16, 17, 18, 19, 20).verifyComplete();
+	}
+
+	@Test
+	public void secondWinnerFirstEmpty() {
+		StepVerifier
+				.create(CloudFlux.firstNonEmpty(Flux.empty(), Flux.range(11, 10).log()))
+				.expectNext(11, 12, 13, 14, 15, 16, 17, 18, 19, 20).verifyComplete();
+	}
+
+	@Test
+	public void bothEmpty() {
+		StepVerifier.create(CloudFlux.firstNonEmpty(Flux.empty(), Flux.empty()))
+				.expectComplete().verifyThenAssertThat().hasNotDiscardedElements()
+				.hasNotDroppedElements().hasNotDroppedErrors();
+	}
+
+	@Test
+	public void neverAndEmpty() {
+		StepVerifier
+				.withVirtualTime(
+						() -> CloudFlux.firstNonEmpty(Flux.never(), Flux.empty()))
+				.expectSubscription().expectNoEvent(Duration.ofDays(1)).thenCancel()
+				.verifyThenAssertThat().hasNotDiscardedElements().hasNotDroppedElements()
+				.hasNotDroppedErrors();
+	}
+
+	@Test
+	public void firstEmitsError() {
+		RuntimeException ex = new RuntimeException("forced failure");
+		StepVerifier
+				.create(CloudFlux.firstNonEmpty(Flux.<Integer>error(ex), Flux.empty()))
+				.expectErrorMessage("forced failure").verifyThenAssertThat()
+				.hasNotDiscardedElements().hasNotDroppedElements().hasNotDroppedErrors();
+	}
+
+	@Test
+	public void secondEmitsError() {
+		RuntimeException ex = new RuntimeException("forced failure");
+		StepVerifier
+				.create(CloudFlux.firstNonEmpty(Flux.empty(), Flux.<Integer>error(ex)))
+				.expectErrorMessage("forced failure").verifyThenAssertThat()
+				.hasNotDiscardedElements().hasNotDroppedElements().hasNotDroppedErrors();
+	}
+
+	@Test
+	public void neverAndSecondEmitsError() {
+		RuntimeException ex = new RuntimeException("forced failure");
+		StepVerifier
+				.create(CloudFlux.firstNonEmpty(Flux.never(), Flux.<Integer>error(ex)))
+				.expectErrorMessage("forced failure").verifyThenAssertThat()
+				.hasNotDiscardedElements().hasNotDroppedElements().hasNotDroppedErrors();
+	}
+
+	@Test
+	public void singleArrayNullSource() {
+		StepVerifier.create(CloudFlux.firstNonEmpty((Publisher<Object>) null))
+				.expectError(NullPointerException.class).verify();
+	}
+
+	@Test
+	public void arrayOneIsNullSource() {
+		StepVerifier.create(CloudFlux.firstNonEmpty(Flux.never(), null, Flux.never()))
+				.expectError(NullPointerException.class).verify();
+	}
+
+	@Test
+	public void singleIterableNullSource() {
+		StepVerifier
+				.create(CloudFlux.firstNonEmpty(singletonList((Publisher<Object>) null)))
+				.expectError(NullPointerException.class).verify();
+	}
+
+	@Test
+	public void iterableOneIsNullSource() {
+		StepVerifier
+				.create(CloudFlux.firstNonEmpty(Arrays.asList(Flux.never(),
+						(Publisher<Object>) null, Flux.never())))
+				.expectError(NullPointerException.class).verify();
+	}
+
+	@Test
+	public void scanSubscriber() {
+		CoreSubscriber<String> actual = new LambdaSubscriber<>(null, e -> {
+		}, null, null);
+		FluxFirstNonEmptyEmitting.RaceCoordinator<String> parent = new FluxFirstNonEmptyEmitting.RaceCoordinator<>(
+				1);
+		FluxFirstNonEmptyEmitting.FirstNonEmptyEmittingSubscriber<String> test = new FluxFirstNonEmptyEmitting.FirstNonEmptyEmittingSubscriber<>(
+				actual, parent, 1);
+		Subscription sub = Operators.emptySubscription();
+		test.onSubscribe(sub);
+
+		assertThat(test.scan(Scannable.Attr.PARENT)).isSameAs(sub);
+		assertThat(test.scan(Scannable.Attr.ACTUAL)).isSameAs(actual);
+		assertThat(test.scan(Scannable.Attr.CANCELLED)).isFalse();
+		parent.cancelled = true;
+		assertThat(test.scan(Scannable.Attr.CANCELLED)).isTrue();
+	}
+
+	@Test
+	public void scanRaceCoordinator() {
+		CoreSubscriber<String> actual = new LambdaSubscriber<>(null, e -> {
+		}, null, null);
+		FluxFirstNonEmptyEmitting.RaceCoordinator<String> parent = new FluxFirstNonEmptyEmitting.RaceCoordinator<>(
+				1);
+		FluxFirstNonEmptyEmitting.FirstNonEmptyEmittingSubscriber<String> test = new FluxFirstNonEmptyEmitting.FirstNonEmptyEmittingSubscriber<>(
+				actual, parent, 1);
+		Subscription sub = Operators.emptySubscription();
+		test.onSubscribe(sub);
+
+		assertThat(test.scan(Scannable.Attr.PARENT)).isSameAs(sub);
+		assertThat(test.scan(Scannable.Attr.ACTUAL)).isSameAs(actual);
+		assertThat(parent.scan(Scannable.Attr.CANCELLED)).isFalse();
+		parent.cancelled = true;
+		assertThat(parent.scan(Scannable.Attr.CANCELLED)).isTrue();
+	}
+
+}

spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/CachingServiceInstanceListSupplier.java

@@ -77,7 +77,8 @@ public class CachingServiceInstanceListSupplier implements ServiceInstanceListSu
 									.getCache(SERVICE_INSTANCE_CACHE_NAME);
 							if (cache == null) {
 								if (log.isErrorEnabled()) {
-									log.error("Unable to find cache for writing: " + SERVICE_INSTANCE_CACHE_NAME);
+									log.error("Unable to find cache for writing: "
+											+ SERVICE_INSTANCE_CACHE_NAME);
 								}
 							}
 							else {