diff --git a/spring-cloud-netflix-core/pom.xml b/spring-cloud-netflix-core/pom.xml index 5107bd0b..e269c330 100644 --- a/spring-cloud-netflix-core/pom.xml +++ b/spring-cloud-netflix-core/pom.xml @@ -39,6 +39,16 @@ spring-boot-starter-web true + + io.projectreactor + reactor-core + true + + + io.reactivex + rxjava-reactive-streams + true + org.springframework.boot spring-boot-starter-aop 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 new file mode 100644 index 00000000..08af3d9d --- /dev/null +++ b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/hystrix/HystrixCommands.java @@ -0,0 +1,96 @@ +/* + * 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 Flux wrap(String commandName, Flux flux) { + return wrap(commandName, flux, null); + } + + public static Flux wrap(String commandName, Flux flux, Flux fallback) { + String groupName = commandName + "group"; + PublisherHystrixCommand command = createHystrixCommand(commandName, groupName, + flux, fallback); + return Flux.from(RxReactiveStreams.toPublisher(command.toObservable())); + } + + public static Mono wrap(String commandName, Mono mono) { + return wrap(commandName, mono, null); + } + + public static Mono wrap(String commandName, Mono mono, Mono fallback) { + String groupName = commandName + "group"; + PublisherHystrixCommand command = createHystrixCommand(commandName, groupName, + mono, fallback); + return Mono.from(RxReactiveStreams.toPublisher(command.toObservable())); + } + + private static PublisherHystrixCommand createHystrixCommand(String commandName, + String groupName, Publisher publisher, Publisher 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 extends HystrixObservableCommand { + + private Publisher publisher; + private Publisher fallback; + + protected PublisherHystrixCommand(Setter setter, Publisher publisher, + Publisher fallback) { + super(setter); + this.publisher = publisher; + this.fallback = fallback; + } + + @Override + protected Observable construct() { + return RxReactiveStreams.toObservable(publisher); + } + + @Override + protected Observable resumeWithFallback() { + if (this.fallback != null) { + return RxReactiveStreams.toObservable(this.fallback); + } + return super.resumeWithFallback(); + } + } +} diff --git a/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/HystrixCommandsTests.java b/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/HystrixCommandsTests.java new file mode 100644 index 00000000..cba75473 --- /dev/null +++ b/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/HystrixCommandsTests.java @@ -0,0 +1,83 @@ +/* + * 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 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 list = HystrixCommands.wrap("multiflux", Flux.error(new Exception()), Flux.just("a", "b")).collectList().block(); + assertThat(list).hasSize(2).contains("a", "b"); + } + +}