fix overflow issue with hystrix timeout operator (#2391)

- The fix adds a necessary buffer between the rx-rs adapter and
the timeout operator introduced in AbstractCommand. Unfortunately,
the hystrix operator does not comply with RS backpressure requirements.
It will always request an unbounded demand to a source regardless of
the command consumer demand. With this commit we protect RS adapter and
downstream flow from overflow. Ideally a rework of the operator should
be planned.

spring-cloud-netflix-core/pom.xml

@@ -44,6 +44,11 @@
 			<artifactId>reactor-core</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-test</artifactId>
+			<scope>test</scope>
+		</dependency>
 		<dependency>
 			<groupId>io.reactivex</groupId>
 			<artifactId>rxjava-reactive-streams</artifactId>

spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/hystrix/HystrixCommands.java

@@ -104,7 +104,7 @@ public class HystrixCommands {
 			if (this.eager) {
 				observable = command.observe();
 			} else {
-				observable = command.toObservable();
+				observable = command.toObservable().onBackpressureBuffer();
 			}
 			return RxReactiveStreams.toPublisher(observable);
 		}

spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/hystrix/HystrixCommandsTests.java

@@ -16,18 +16,19 @@
 
 package org.springframework.cloud.netflix.hystrix;
 
+import java.time.Duration;
 import java.util.List;
 
+import com.netflix.hystrix.exception.HystrixRuntimeException;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
-
-import com.netflix.hystrix.exception.HystrixRuntimeException;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+import reactor.test.StepVerifier;
+
+import static org.assertj.core.api.Assertions.assertThat;
 
 public class HystrixCommandsTests {
 
@@ -75,6 +76,18 @@ public class HystrixCommandsTests {
 		assertThat(list).hasSize(2).contains("1", "2");
 	}
 
+	@Test
+	public void fluxWorksDeferredRequest() {
+		StepVerifier.create(HystrixCommands.from(Flux.just("1", "2"))
+		                                   .commandName("multiflux")
+		                                   .build(), 1)
+		            .expectNext("1")
+		            .thenAwait(Duration.ofSeconds(1))
+		            .thenRequest(1)
+		            .expectNext("2")
+		            .verifyComplete();
+	}
+
 	@Test
 	public void eagerFluxWorks() {
 		List<String> list = HystrixCommands.from( Flux.just("1", "2"))