From 7033a9877f4f7ed879c6cd5dbb5dbfbe81118368 Mon Sep 17 00:00:00 2001 From: Tim Ysewyn Date: Wed, 23 Oct 2019 20:34:15 +0200 Subject: [PATCH] Take first non-empty flux into consideration for composite reactive service discovery --- .../ReactiveCompositeDiscoveryClient.java | 3 +- .../reactor/core/publisher/CloudFlux.java | 62 ++++ .../publisher/FluxFirstNonEmptyEmitting.java | 339 ++++++++++++++++++ .../FluxFirstNonEmptyEmittingTests.java | 193 ++++++++++ .../CachingServiceInstanceListSupplier.java | 3 +- 5 files changed, 598 insertions(+), 2 deletions(-) create mode 100644 spring-cloud-commons/src/main/java/reactor/core/publisher/CloudFlux.java create mode 100644 spring-cloud-commons/src/main/java/reactor/core/publisher/FluxFirstNonEmptyEmitting.java create mode 100644 spring-cloud-commons/src/test/java/reactor/core/publisher/FluxFirstNonEmptyEmittingTests.java diff --git a/spring-cloud-commons/src/main/java/org/springframework/cloud/client/discovery/composite/reactive/ReactiveCompositeDiscoveryClient.java b/spring-cloud-commons/src/main/java/org/springframework/cloud/client/discovery/composite/reactive/ReactiveCompositeDiscoveryClient.java index 01889c09..c93eeeb1 100644 --- a/spring-cloud-commons/src/main/java/org/springframework/cloud/client/discovery/composite/reactive/ReactiveCompositeDiscoveryClient.java +++ b/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 diff --git a/spring-cloud-commons/src/main/java/reactor/core/publisher/CloudFlux.java b/spring-cloud-commons/src/main/java/reactor/core/publisher/CloudFlux.java new file mode 100644 index 00000000..ec1b1bb0 --- /dev/null +++ b/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 extends Flux { + + /** + * 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 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 Flux firstNonEmpty(Publisher... 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 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 Flux firstNonEmpty( + Iterable> sources) { + return onAssembly(new FluxFirstNonEmptyEmitting<>(sources)); + } + +} diff --git a/spring-cloud-commons/src/main/java/reactor/core/publisher/FluxFirstNonEmptyEmitting.java b/spring-cloud-commons/src/main/java/reactor/core/publisher/FluxFirstNonEmptyEmitting.java new file mode 100644 index 00000000..abe375dd --- /dev/null +++ b/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 extends Flux implements SourceProducer { + + final Publisher[] array; + + final Iterable> iterable; + + @SafeVarargs + FluxFirstNonEmptyEmitting(Publisher... array) { + this.array = Objects.requireNonNull(array, "array"); + this.iterable = null; + } + + FluxFirstNonEmptyEmitting(Iterable> iterable) { + this.array = null; + this.iterable = Objects.requireNonNull(iterable); + } + + @SuppressWarnings("unchecked") + @Override + public void subscribe(CoreSubscriber actual) { + Publisher[] a = array; + int n; + if (a == null) { + n = 0; + a = new Publisher[8]; + + Iterator> 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 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[] 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 p = a[0]; + + if (p == null) { + Operators.error(actual, + new NullPointerException("The single source Publisher is null")); + } + else { + p.subscribe(actual); + } + return; + } + + RaceCoordinator 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 implements Subscription, Scannable { + + final FirstNonEmptyEmittingSubscriber[] subscribers; + + volatile boolean cancelled; + + volatile int wip; + + volatile int competingSubscribers; + + @SuppressWarnings("rawtypes") + static final AtomicIntegerFieldUpdater WIP = AtomicIntegerFieldUpdater + .newUpdater(RaceCoordinator.class, "wip"); + + static final AtomicIntegerFieldUpdater 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 inners() { + return Stream.of(subscribers); + } + + @Override + @Nullable + public Object scanUnsafe(Attr key) { + if (key == Attr.CANCELLED) { + return cancelled; + } + + return null; + } + + void subscribe(Publisher[] sources, int n, + CoreSubscriber actual) { + FirstNonEmptyEmittingSubscriber[] 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 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 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 s : subscribers) { + s.cancel(); + } + } + } + + boolean tryWin(int index) { + if (wip == Integer.MIN_VALUE) { + if (WIP.compareAndSet(this, Integer.MIN_VALUE, index)) { + + FirstNonEmptyEmittingSubscriber[] 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 + extends Operators.DeferredSubscription implements InnerOperator { + + final RaceCoordinator parent; + + final CoreSubscriber actual; + + final int index; + + boolean won; + + FirstNonEmptyEmittingSubscriber(CoreSubscriber actual, + RaceCoordinator 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 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(); + } + } + + } + +} diff --git a/spring-cloud-commons/src/test/java/reactor/core/publisher/FluxFirstNonEmptyEmittingTests.java b/spring-cloud-commons/src/test/java/reactor/core/publisher/FluxFirstNonEmptyEmittingTests.java new file mode 100644 index 00000000..adbf508a --- /dev/null +++ b/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[]) null)); + } + + @Test + public void iterableNull() { + assertThatNullPointerException().isThrownBy( + () -> CloudFlux.firstNonEmpty((Iterable>) 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.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.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.error(ex))) + .expectErrorMessage("forced failure").verifyThenAssertThat() + .hasNotDiscardedElements().hasNotDroppedElements().hasNotDroppedErrors(); + } + + @Test + public void singleArrayNullSource() { + StepVerifier.create(CloudFlux.firstNonEmpty((Publisher) 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) null))) + .expectError(NullPointerException.class).verify(); + } + + @Test + public void iterableOneIsNullSource() { + StepVerifier + .create(CloudFlux.firstNonEmpty(Arrays.asList(Flux.never(), + (Publisher) null, Flux.never()))) + .expectError(NullPointerException.class).verify(); + } + + @Test + public void scanSubscriber() { + CoreSubscriber actual = new LambdaSubscriber<>(null, e -> { + }, null, null); + FluxFirstNonEmptyEmitting.RaceCoordinator parent = new FluxFirstNonEmptyEmitting.RaceCoordinator<>( + 1); + FluxFirstNonEmptyEmitting.FirstNonEmptyEmittingSubscriber 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 actual = new LambdaSubscriber<>(null, e -> { + }, null, null); + FluxFirstNonEmptyEmitting.RaceCoordinator parent = new FluxFirstNonEmptyEmitting.RaceCoordinator<>( + 1); + FluxFirstNonEmptyEmitting.FirstNonEmptyEmittingSubscriber 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(); + } + +} diff --git a/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/CachingServiceInstanceListSupplier.java b/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/CachingServiceInstanceListSupplier.java index 716d914f..7ad619e4 100644 --- a/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/CachingServiceInstanceListSupplier.java +++ b/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 {