From 4932eef1c9ef9357788f404d4ebf18ce85ebd32c Mon Sep 17 00:00:00 2001
From: Spencer Gibb <spencer@gibb.us>
Date: Tue, 24 Oct 2017 14:46:41 -0400
Subject: [PATCH] Adds toObservable function to HystrixCommands.

Lets user have full control over how the Observable is created with
sane defaults for eager/non-eager cases.
---
 .../netflix/hystrix/HystrixCommands.java      | 66 +++++++++----------
 .../netflix/hystrix/HystrixCommandsTests.java | 12 ++++
 2 files changed, 44 insertions(+), 34 deletions(-)

diff --git a/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/hystrix/HystrixCommands.java b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/hystrix/HystrixCommands.java
index c6e1611b..41735616 100644
--- a/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/hystrix/HystrixCommands.java
+++ b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/hystrix/HystrixCommands.java
@@ -16,6 +16,8 @@
 
 package org.springframework.cloud.netflix.hystrix;
 
+import java.util.function.Function;
+
 import org.reactivestreams.Publisher;
 import org.springframework.util.StringUtils;
 
@@ -48,6 +50,7 @@ public class HystrixCommands {
 		private Publisher<T> fallback;
 		private Setter setter;
 		private boolean eager = false;
+		private Function<HystrixObservableCommand<T>, Observable<T>> toObservable;
 
 		public PublisherBuilder(Publisher<T> publisher) {
 			this.publisher = publisher;
@@ -78,10 +81,38 @@ public class HystrixCommands {
 			return this;
 		}
 
+		public PublisherBuilder<T> toObservable(Function<HystrixObservableCommand<T>, Observable<T>> toObservable) {
+			this.toObservable = toObservable;
+			return this;
+		}
+
 		public Publisher<T> build() {
 			if (!StringUtils.hasText(commandName) && setter == null) {
 				throw new IllegalStateException("commandName and setter can not both be empty");
 			}
+			Setter setterToUse = getSetter();
+
+			PublisherHystrixCommand<T> command = new PublisherHystrixCommand<>(setterToUse, this.publisher, this.fallback);
+
+			Observable<T> observable = getObservableFunction().apply(command);
+
+			return RxReactiveStreams.toPublisher(observable);
+		}
+
+		public Function<HystrixObservableCommand<T>, Observable<T>> getObservableFunction() {
+			Function<HystrixObservableCommand<T>, Observable<T>> observableFunc;
+
+			if (this.toObservable != null) {
+				observableFunc = this.toObservable;
+			} else if (this.eager) {
+				observableFunc = cmd -> cmd.observe();
+			} else { // apply a default onBackpressureBuffer if not eager
+				observableFunc = cmd -> cmd.toObservable().onBackpressureBuffer();
+			}
+			return observableFunc;
+		}
+
+		public Setter getSetter() {
 			Setter setterToUse;
 			if (this.setter != null) {
 				setterToUse = this.setter;
@@ -97,16 +128,7 @@ public class HystrixCommands {
 				HystrixCommandKey commandKey = HystrixCommandKey.Factory.asKey(this.commandName);
 				setterToUse = Setter.withGroupKey(groupKey).andCommandKey(commandKey);
 			}
-
-			PublisherHystrixCommand<T> command = new PublisherHystrixCommand<>(setterToUse, this.publisher, this.fallback);
-
-			Observable<T> observable;
-			if (this.eager) {
-				observable = command.observe();
-			} else {
-				observable = command.toObservable().onBackpressureBuffer();
-			}
-			return RxReactiveStreams.toPublisher(observable);
+			return setterToUse;
 		}
 
 		public Flux<T> toFlux() {
@@ -119,30 +141,6 @@ public class HystrixCommands {
 
 	}
 
-	public static <T> Mono<T> inject(String commandName, Mono<T> mono, Mono<T> fallback, boolean eager) {
-		String groupName = commandName + "group";
-		PublisherHystrixCommand<T> command = createHystrixCommand(commandName, groupName,
-				mono, fallback);
-		Observable<T> observable;
-		if (eager) {
-			observable = command.observe();
-		} else {
-			observable = command.toObservable();
-		}
-		return Mono.from(RxReactiveStreams.toPublisher(observable));
-	}
-
-	private static <T> PublisherHystrixCommand<T> createHystrixCommand(String commandName,
-			String groupName, Publisher<T> publisher, Publisher<T> fallback) {
-		HystrixCommandGroupKey groupKey = HystrixCommandGroupKey.Factory.asKey(groupName);
-		HystrixCommandKey commandKey = HystrixCommandKey.Factory.asKey(commandName);
-
-		Setter setter = Setter
-				.withGroupKey(groupKey).andCommandKey(commandKey);
-
-		return new PublisherHystrixCommand<>(setter, publisher, fallback);
-	}
-
 	private static class PublisherHystrixCommand<T> extends HystrixObservableCommand<T> {
 
 		private Publisher<T> publisher;
diff --git a/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/hystrix/HystrixCommandsTests.java b/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/hystrix/HystrixCommandsTests.java
index 2ceb332d..b06a7c3a 100644
--- a/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/hystrix/HystrixCommandsTests.java
+++ b/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/hystrix/HystrixCommandsTests.java
@@ -88,6 +88,18 @@ public class HystrixCommandsTests {
 		            .verifyComplete();
 	}
 
+	@Test
+	public void toObservableFunctionWorks() {
+		StepVerifier.create(HystrixCommands.from(Flux.just("1", "2"))
+				.commandName("multiflux")
+				.toObservable(cmd -> cmd.toObservable())
+				.build(), 1)
+				.expectNext("1")
+				.thenAwait(Duration.ofSeconds(1))
+				.thenRequest(1)
+				.expectError();
+	}
+
 	@Test
 	public void eagerFluxWorks() {
 		StepVerifier.create(HystrixCommands.from( Flux.just("1", "2"))