Adds HystrixCommands helper object.

spring-cloud-netflix-core/pom.xml

@@ -40,6 +40,16 @@
 			<optional>true</optional>
 		</dependency>
 		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-core</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>io.reactivex</groupId>
+			<artifactId>rxjava-reactive-streams</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-aop</artifactId>
 			<optional>true</optional>

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

+/*
+ * Copyright 2013-2017 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
+ *
+ *     http://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 org.springframework.cloud.netflix.hystrix;
+
+import org.reactivestreams.Publisher;
+
+import com.netflix.hystrix.HystrixCommandGroupKey;
+import com.netflix.hystrix.HystrixCommandKey;
+import com.netflix.hystrix.HystrixObservableCommand;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import rx.Observable;
+import rx.RxReactiveStreams;
+
+/**
+ * Utility class to wrap a {@see Publisher} in a {@see HystrixObservableCommand}. Good for
+ * use in a Spring WebFlux application. Allows more flexibility than the @HystrixCommand
+ * annotation.
+ * @author Spencer Gibb
+ */
+public class HystrixCommands {
+
+	public static <T> Flux<T> wrap(String commandName, Flux<T> flux) {
+		return wrap(commandName, flux, null);
+	}
+
+	public static <T> Flux<T> wrap(String commandName, Flux<T> flux, Flux<T> fallback) {
+		String groupName = commandName + "group";
+		PublisherHystrixCommand<T> command = createHystrixCommand(commandName, groupName,
+				flux, fallback);
+		return Flux.from(RxReactiveStreams.toPublisher(command.toObservable()));
+	}
+
+	public static <T> Mono<T> wrap(String commandName, Mono<T> mono) {
+		return wrap(commandName, mono, null);
+	}
+
+	public static <T> Mono<T> wrap(String commandName, Mono<T> mono, Mono<T> fallback) {
+		String groupName = commandName + "group";
+		PublisherHystrixCommand<T> command = createHystrixCommand(commandName, groupName,
+				mono, fallback);
+		return Mono.from(RxReactiveStreams.toPublisher(command.toObservable()));
+	}
+
+	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);
+
+		HystrixObservableCommand.Setter setter = HystrixObservableCommand.Setter
+				.withGroupKey(groupKey).andCommandKey(commandKey);
+
+		return new PublisherHystrixCommand<>(setter, publisher, fallback);
+	}
+
+	private static class PublisherHystrixCommand<T> extends HystrixObservableCommand<T> {
+
+		private Publisher<T> publisher;
+		private Publisher<T> fallback;
+
+		protected PublisherHystrixCommand(Setter setter, Publisher<T> publisher,
+				Publisher<T> fallback) {
+			super(setter);
+			this.publisher = publisher;
+			this.fallback = fallback;
+		}
+
+		@Override
+		protected Observable<T> construct() {
+			return RxReactiveStreams.toObservable(publisher);
+		}
+
+		@Override
+		protected Observable<T> resumeWithFallback() {
+			if (this.fallback != null) {
+				return RxReactiveStreams.toObservable(this.fallback);
+			}
+			return super.resumeWithFallback();
+		}
+	}
+}

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

+/*
+ * Copyright 2013-2017 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
+ *
+ *     http://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 org.springframework.cloud.netflix;
+
+import com.netflix.hystrix.exception.HystrixRuntimeException;
+import org.junit.Ignore;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.springframework.cloud.netflix.hystrix.HystrixCommands;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class HystrixCommandsTests {
+
+	@Rule
+	public ExpectedException exception = ExpectedException.none();
+
+	@Test
+	public void monoWorks() {
+		String result = HystrixCommands.wrap("testworks", Mono.just("works")).block();
+		assertThat(result).isEqualTo("works");
+	}
+
+	@Test
+	public void monoTimesOut() {
+		exception.expect(HystrixRuntimeException.class);
+		HystrixCommands.wrap("failcmd", Mono.fromCallable(() -> {
+			Thread.sleep(1500);
+			return "timeout";
+		})).block();
+	}
+
+	@Test
+	public void monoFallbackWorks() {
+		String result = HystrixCommands.wrap("failcmd", Mono.error(new Exception()), Mono.just("fallback")).block();
+		assertThat(result).isEqualTo("fallback");
+	}
+
+	@Test
+	public void fluxWorks() {
+		List<String> list = HystrixCommands.wrap("multiflux", Flux.just("1", "2")).collectList().block();
+		assertThat(list).hasSize(2).contains("1", "2");
+	}
+
+	@Test
+	// @Ignore
+	public void fluxTimesOut() {
+		exception.expect(HystrixRuntimeException.class);
+		HystrixCommands.wrap("failcmd", Flux.from(s -> {
+			try {
+				Thread.sleep(1500);
+			} catch (InterruptedException e) {
+				throw new RuntimeException(e);
+			}
+		})).blockFirst();
+	}
+
+	@Test
+	public void fluxFallbackWorks() {
+		List<String> list = HystrixCommands.wrap("multiflux", Flux.error(new Exception()), Flux.just("a", "b")).collectList().block();
+		assertThat(list).hasSize(2).contains("a", "b");
+	}
+
+}