Adding spring cloud circuitbreaker hystrix implementation

docs/src/main/asciidoc/spring-cloud-circuitbreaker-hystrix.adoc

@@ -0,0 +1,64 @@
+=== Configuring Hystrix Circuit Breakers
+
+==== Default Configuration
+
+To provide a default configuration for all of your circuit breakers create a `Customize` bean that is passed a
+`HystrixCircuitBreakerFactory` or `ReactiveHystrixCircuitBreakerFactory`.
+The `configureDefault` method can be used to provide a default configuration.
+
+====
+[source,java]
+----
+@Bean
+public Customizer<HystrixCircuitBreakerFactory> defaultConfig() {
+	return factory -> factory.configureDefault(id -> HystrixCommand.Setter
+			.withGroupKey(HystrixCommandGroupKey.Factory.asKey(id))
+			.andCommandPropertiesDefaults(HystrixCommandProperties.Setter()
+			.withExecutionTimeoutInMilliseconds(4000)));
+}
+----
+====
+
+===== Reactive Example
+
+====
+[source,java]
+----
+@Bean
+public Customizer<ReactiveHystrixCircuitBreakerFactory> defaultConfig() {
+	return factory -> factory.configureDefault(id -> HystrixObservableCommand.Setter
+			.withGroupKey(HystrixCommandGroupKey.Factory.asKey(id))
+			.andCommandPropertiesDefaults(HystrixCommandProperties.Setter()
+					.withExecutionTimeoutInMilliseconds(4000)));
+}
+----
+====
+
+==== Specific Circuit Breaker Configuration
+
+Similarly to providing a default configuration, you can create a `Customize` bean this is passed a
+`HystrixCircuitBreakerFactory`
+
+====
+[source,java]
+----
+@Bean
+public Customizer<HystrixCircuitBreakerFactory> customizer() {
+	return factory -> factory.configure(builder -> builder.commandProperties(
+					HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(2000)), "foo", "bar");
+}
+----
+====
+
+===== Reactive Example
+
+====
+[source,java]
+----
+@Bean
+public Customizer<ReactiveHystrixCircuitBreakerFactory> customizer() {
+	return factory -> factory.configure(builder -> builder.commandProperties(
+					HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(2000)), "foo", "bar");
+}
+----
+====

docs/src/main/asciidoc/spring-cloud-netflix.adoc

@@ -554,6 +554,10 @@ when running a Eureka server you must include these dependencies in your POM or
 </dependency>
 ----
 
+== Circuit Breaker: Spring Cloud Circuit Breaker With Hystrix
+
+include::spring-cloud-circuitbreaker-hystrix.adoc[]
+
 == Circuit Breaker: Hystrix Clients
 
 Netflix has created a library called https://github.com/Netflix/Hystrix[Hystrix] that implements the https://martinfowler.com/bliki/CircuitBreaker.html[circuit breaker pattern].

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

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2013-2018 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 org.springframework.cloud.netflix.hystrix;
+
+import com.netflix.hystrix.HystrixCommandGroupKey;
+import com.netflix.hystrix.HystrixCommandKey;
+import com.netflix.hystrix.HystrixCommandProperties;
+
+import org.springframework.cloud.client.circuitbreaker.ConfigBuilder;
+import org.springframework.util.StringUtils;
+
+/**
+ * @author Ryan Baxter
+ */
+public abstract class AbstractHystrixConfigBuilder<CONFB>
+		implements ConfigBuilder<CONFB> {
+
+	private final String commandName;
+
+	protected String groupName;
+
+	protected HystrixCommandProperties.Setter commandProperties;
+
+	public AbstractHystrixConfigBuilder(String id) {
+		this.commandName = id;
+	}
+
+	public AbstractHystrixConfigBuilder groupName(String groupName) {
+		this.groupName = groupName;
+		return this;
+	}
+
+	public AbstractHystrixConfigBuilder commandProperties(
+			HystrixCommandProperties.Setter commandProperties) {
+		this.commandProperties = commandProperties;
+		return this;
+	}
+
+	protected HystrixCommandGroupKey getGroupKey() {
+		String groupNameToUse;
+		if (StringUtils.hasText(this.groupName)) {
+			groupNameToUse = this.groupName;
+		}
+		else {
+			groupNameToUse = commandName + "group";
+		}
+		return HystrixCommandGroupKey.Factory.asKey(groupNameToUse);
+	}
+
+	protected HystrixCommandKey getCommandKey() {
+		return HystrixCommandKey.Factory.asKey(this.commandName);
+	}
+
+	protected HystrixCommandProperties.Setter getCommandPropertiesSetter() {
+		return this.commandProperties != null ? this.commandProperties
+				: HystrixCommandProperties.Setter();
+	}
+
+}

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

@@ -16,6 +16,11 @@
 
 package org.springframework.cloud.netflix.hystrix;
 
+import java.util.ArrayList;
+import java.util.List;
+
+import javax.annotation.PostConstruct;
+
 import com.netflix.hystrix.Hystrix;
 import com.netflix.hystrix.contrib.javanica.aop.aspectj.HystrixCommandAspect;
 import com.netflix.hystrix.contrib.metrics.eventstream.HystrixMetricsStreamServlet;
@@ -26,6 +31,7 @@ import org.reactivestreams.Publisher;
 import rx.Observable;
 import rx.RxReactiveStreams;
 
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.actuate.autoconfigure.endpoint.condition.ConditionalOnEnabledEndpoint;
 import org.springframework.boot.actuate.autoconfigure.health.ConditionalOnEnabledHealthIndicator;
 import org.springframework.boot.actuate.autoconfigure.health.HealthIndicatorAutoConfiguration;
@@ -33,10 +39,14 @@ import org.springframework.boot.actuate.health.HealthIndicator;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.client.actuator.HasFeatures;
+import org.springframework.cloud.client.circuitbreaker.CircuitBreakerFactory;
+import org.springframework.cloud.client.circuitbreaker.Customizer;
+import org.springframework.cloud.client.circuitbreaker.ReactiveCircuitBreakerFactory;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.web.reactive.DispatcherHandler;
@@ -62,6 +72,20 @@ public class HystrixAutoConfiguration {
 		return new HystrixHealthIndicator();
 	}
 
+	@Bean
+	@ConditionalOnMissingBean(CircuitBreakerFactory.class)
+	public CircuitBreakerFactory hystrixCircuitBreakerFactory() {
+		return new HystrixCircuitBreakerFactory();
+	}
+
+	@Bean
+	@ConditionalOnMissingBean(ReactiveCircuitBreakerFactory.class)
+	@ConditionalOnClass(
+			name = { "reactor.core.publisher.Mono", "reactor.core.publisher.Flux" })
+	public ReactiveHystrixCircuitBreakerFactory reactiveHystrixCircuitBreakerFactory() {
+		return new ReactiveHystrixCircuitBreakerFactory();
+	}
+
 	@Configuration
 	@ConditionalOnProperty(value = "management.metrics.binders.hystrix.enabled",
 			matchIfMissing = true)
@@ -127,4 +151,38 @@ public class HystrixAutoConfiguration {
 
 	}
 
+	@Configuration
+	protected static class HystrixCircuitBreakerCustomizerConfiguration {
+
+		@Autowired(required = false)
+		private List<Customizer<HystrixCircuitBreakerFactory>> customizers = new ArrayList<>();
+
+		@Autowired(required = false)
+		private HystrixCircuitBreakerFactory factory;
+
+		@PostConstruct
+		public void init() {
+			customizers.forEach(customizer -> customizer.customize(factory));
+		}
+
+	}
+
+	@Configuration
+	@ConditionalOnClass(
+			name = { "reactor.core.publisher.Mono", "reactor.core.publisher.Flux" })
+	protected static class ReactiveHystrixCircuitBreakerCustomizerConfiguration {
+
+		@Autowired(required = false)
+		private List<Customizer<ReactiveHystrixCircuitBreakerFactory>> customizers = new ArrayList<>();
+
+		@Autowired(required = false)
+		private ReactiveHystrixCircuitBreakerFactory factory;
+
+		@PostConstruct
+		public void init() {
+			customizers.forEach(customizer -> customizer.customize(factory));
+		}
+
+	}
+
 }

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

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2013-2018 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 org.springframework.cloud.netflix.hystrix;
+
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import com.netflix.hystrix.HystrixCommand;
+
+import org.springframework.cloud.client.circuitbreaker.CircuitBreaker;
+
+/**
+ * Hystrix implementation of {@link CircuitBreaker}.
+ *
+ * @author Ryan Baxter
+ */
+public class HystrixCircuitBreaker implements CircuitBreaker {
+
+	private HystrixCommand.Setter setter;
+
+	public HystrixCircuitBreaker(HystrixCommand.Setter setter) {
+		this.setter = setter;
+	}
+
+	@Override
+	public <T> T run(Supplier<T> toRun, Function<Throwable, T> fallback) {
+
+		HystrixCommand<T> command = new HystrixCommand<T>(setter) {
+			@Override
+			protected T run() throws Exception {
+				return toRun.get();
+			}
+
+			@Override
+			protected T getFallback() {
+				return fallback.apply(getExecutionException());
+			}
+		};
+		return command.execute();
+	}
+
+}

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

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2013-2018 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 org.springframework.cloud.netflix.hystrix;
+
+import java.util.function.Function;
+
+import com.netflix.hystrix.HystrixCommand;
+import com.netflix.hystrix.HystrixCommandGroupKey;
+
+import org.springframework.cloud.client.circuitbreaker.CircuitBreakerFactory;
+import org.springframework.util.Assert;
+
+/**
+ * Builds Hystrix circuit breakers.
+ *
+ * @author Ryan Baxter
+ */
+public class HystrixCircuitBreakerFactory extends
+		CircuitBreakerFactory<HystrixCommand.Setter, HystrixCircuitBreakerFactory.HystrixConfigBuilder> {
+
+	private Function<String, HystrixCommand.Setter> defaultConfiguration = id -> HystrixCommand.Setter
+			.withGroupKey(HystrixCommandGroupKey.Factory.asKey(id));
+
+	public void configureDefault(
+			Function<String, HystrixCommand.Setter> defaultConfiguration) {
+		this.defaultConfiguration = defaultConfiguration;
+	}
+
+	public HystrixConfigBuilder configBuilder(String id) {
+		return new HystrixConfigBuilder(id);
+	}
+
+	public HystrixCircuitBreaker create(String id) {
+		Assert.hasText(id, "A CircuitBreaker must have an id.");
+		HystrixCommand.Setter setter = getConfigurations().computeIfAbsent(id,
+				defaultConfiguration);
+		return new HystrixCircuitBreaker(setter);
+	}
+
+	public static class HystrixConfigBuilder
+			extends AbstractHystrixConfigBuilder<HystrixCommand.Setter> {
+
+		public HystrixConfigBuilder(String id) {
+			super(id);
+		}
+
+		@Override
+		public HystrixCommand.Setter build() {
+			return HystrixCommand.Setter.withGroupKey(getGroupKey())
+					.andCommandKey(getCommandKey())
+					.andCommandPropertiesDefaults(getCommandPropertiesSetter());
+		}
+
+	}
+
+}

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

@@ -0,0 +1,84 @@
+/*
+ * Copyright 2013-2018 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 org.springframework.cloud.netflix.hystrix;
+
+import java.util.function.Function;
+
+import com.netflix.hystrix.HystrixObservableCommand;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import rx.Observable;
+import rx.RxReactiveStreams;
+import rx.Subscription;
+
+import org.springframework.cloud.client.circuitbreaker.ReactiveCircuitBreaker;
+
+/**
+ * @author Ryan Baxter
+ */
+public class ReactiveHystrixCircuitBreaker implements ReactiveCircuitBreaker {
+
+	private HystrixObservableCommand.Setter setter;
+
+	public ReactiveHystrixCircuitBreaker(HystrixObservableCommand.Setter setter) {
+		this.setter = setter;
+	}
+
+	@Override
+	public <T> Mono<T> run(Mono<T> toRun, Function<Throwable, Mono<T>> fallback) {
+		HystrixObservableCommand<T> command = createCommand(toRun, fallback);
+
+		return Mono.create(s -> {
+			Subscription sub = command.toObservable().subscribe(s::success, s::error,
+					s::success);
+			s.onCancel(sub::unsubscribe);
+		});
+	}
+
+	@Override
+	public <T> Flux<T> run(Flux<T> toRun, Function<Throwable, Flux<T>> fallback) {
+		HystrixObservableCommand<T> command = createCommand(toRun, fallback);
+
+		return Flux.create(s -> {
+			Subscription sub = command.toObservable().subscribe(s::next, s::error,
+					s::complete);
+			s.onCancel(sub::unsubscribe);
+		});
+	}
+
+	private <T> HystrixObservableCommand<T> createCommand(Publisher<T> toRun,
+			Function fallback) {
+		HystrixObservableCommand<T> command = new HystrixObservableCommand<T>(setter) {
+			@Override
+			protected Observable<T> construct() {
+				return RxReactiveStreams.toObservable(toRun);
+			}
+
+			@Override
+			protected Observable<T> resumeWithFallback() {
+				if (fallback == null) {
+					super.resumeWithFallback();
+				}
+				return RxReactiveStreams.toObservable(
+						(Publisher) fallback.apply(this.getExecutionException()));
+			}
+		};
+		return command;
+	}
+
+}

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

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2013-2018 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 org.springframework.cloud.netflix.hystrix;
+
+import java.util.function.Function;
+
+import com.netflix.hystrix.HystrixCommandGroupKey;
+import com.netflix.hystrix.HystrixObservableCommand;
+
+import org.springframework.cloud.client.circuitbreaker.ReactiveCircuitBreaker;
+import org.springframework.cloud.client.circuitbreaker.ReactiveCircuitBreakerFactory;
+import org.springframework.util.Assert;
+
+/**
+ * @author Ryan Baxter
+ */
+public class ReactiveHystrixCircuitBreakerFactory extends
+		ReactiveCircuitBreakerFactory<HystrixObservableCommand.Setter, ReactiveHystrixCircuitBreakerFactory.ReactiveHystrixConfigBuilder> {
+
+	private Function<String, HystrixObservableCommand.Setter> defaultConfiguration = id -> HystrixObservableCommand.Setter
+			.withGroupKey(HystrixCommandGroupKey.Factory.asKey(id));
+
+	@Override
+	protected ReactiveHystrixConfigBuilder configBuilder(String id) {
+		return new ReactiveHystrixConfigBuilder(id);
+	}
+
+	@Override
+	public void configureDefault(
+			Function<String, HystrixObservableCommand.Setter> defaultConfiguration) {
+		this.defaultConfiguration = defaultConfiguration;
+	}
+
+	@Override
+	public ReactiveCircuitBreaker create(String id) {
+		Assert.hasText(id, "A CircuitBreaker must have an id.");
+		HystrixObservableCommand.Setter setter = getConfigurations().computeIfAbsent(id,
+				defaultConfiguration);
+		return new ReactiveHystrixCircuitBreaker(setter);
+	}
+
+	public static class ReactiveHystrixConfigBuilder
+			extends AbstractHystrixConfigBuilder<HystrixObservableCommand.Setter> {
+
+		public ReactiveHystrixConfigBuilder(String id) {
+			super(id);
+		}
+
+		@Override
+		public HystrixObservableCommand.Setter build() {
+			return HystrixObservableCommand.Setter.withGroupKey(getGroupKey())
+					.andCommandKey(getCommandKey())
+					.andCommandPropertiesDefaults(getCommandPropertiesSetter());
+		}
+
+	}
+
+}

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

@@ -0,0 +1,129 @@
+/*
+ * Copyright 2013-2018 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 org.springframework.cloud.netflix.hystrix;
+
+import com.netflix.hystrix.HystrixCommand;
+import com.netflix.hystrix.HystrixCommandGroupKey;
+import com.netflix.hystrix.HystrixCommandProperties;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.client.TestRestTemplate;
+import org.springframework.cloud.client.circuitbreaker.CircuitBreakerFactory;
+import org.springframework.cloud.client.circuitbreaker.Customizer;
+import org.springframework.cloud.netflix.test.NoSecurityConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.stereotype.Service;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT;
+
+/**
+ * @author Ryan Baxter
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest(webEnvironment = RANDOM_PORT,
+		classes = HystrixCircuitBreakerIntegrationTest.Application.class)
+@DirtiesContext
+@Import(NoSecurityConfiguration.class)
+public class HystrixCircuitBreakerIntegrationTest {
+
+	@Autowired
+	Application.DemoControllerService service;
+
+	@Test
+	public void testSlow() {
+		assertThat(service.slow()).isEqualTo("fallback");
+	}
+
+	@Test
+	public void testNormal() {
+		assertThat(service.normal()).isEqualTo("normal");
+	}
+
+	@Configuration
+	@EnableAutoConfiguration
+	@RestController
+	protected static class Application {
+
+		@RequestMapping("/slow")
+		public String slow() throws InterruptedException {
+			Thread.sleep(3000);
+			return "slow";
+		}
+
+		@GetMapping("/normal")
+		public String normal() {
+			return "normal";
+		}
+
+		@Bean
+		public Customizer<HystrixCircuitBreakerFactory> customizer() {
+			return factory -> factory
+					.configure(
+							builder -> builder.commandProperties(HystrixCommandProperties
+									.Setter().withExecutionTimeoutInMilliseconds(2000)),
+							"slow");
+		}
+
+		@Bean
+		public Customizer<HystrixCircuitBreakerFactory> defaultConfig() {
+			return factory -> factory.configureDefault(id -> HystrixCommand.Setter
+					.withGroupKey(HystrixCommandGroupKey.Factory.asKey(id))
+					.andCommandPropertiesDefaults(HystrixCommandProperties.Setter()
+							.withExecutionTimeoutInMilliseconds(4000)));
+		}
+
+		@Service
+		public static class DemoControllerService {
+
+			private TestRestTemplate rest;
+
+			private CircuitBreakerFactory cbFactory;
+
+			DemoControllerService(TestRestTemplate rest,
+					CircuitBreakerFactory cbBuilder) {
+				this.rest = rest;
+				this.cbFactory = cbBuilder;
+			}
+
+			public String slow() {
+				return cbFactory.create("slow").run(
+						() -> rest.getForObject("/slow", String.class), t -> "fallback");
+			}
+
+			public String normal() {
+				return cbFactory.create("normal").run(
+						() -> rest.getForObject("/normal", String.class),
+						t -> "fallback");
+			}
+
+		}
+
+	}
+
+}

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

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2013-2018 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 org.springframework.cloud.netflix.hystrix;
+
+import org.junit.Test;
+
+import org.springframework.cloud.client.circuitbreaker.CircuitBreaker;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Ryan Baxter
+ */
+public class HystrixCircuitBreakerTest {
+
+	@Test
+	public void run() {
+		CircuitBreaker cb = new HystrixCircuitBreakerFactory().create("foo");
+		String s = cb.run(() -> "foobar", t -> "fallback");
+		assertThat(cb.run(() -> "foobar", t -> "fallback")).isEqualTo("foobar");
+	}
+
+	@Test
+	public void fallback() {
+		CircuitBreaker cb = new HystrixCircuitBreakerFactory().create("foo");
+		assertThat((String) cb.run(() -> {
+			throw new RuntimeException("Boom");
+		}, t -> "fallback")).isEqualTo("fallback");
+	}
+
+}

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

@@ -0,0 +1,151 @@
+/*
+ * Copyright 2013-2018 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 org.springframework.cloud.netflix.hystrix;
+
+import java.time.Duration;
+
+import com.netflix.hystrix.HystrixCommandGroupKey;
+import com.netflix.hystrix.HystrixCommandProperties;
+import com.netflix.hystrix.HystrixObservableCommand;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import reactor.core.publisher.Mono;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.web.server.LocalServerPort;
+import org.springframework.cloud.client.circuitbreaker.Customizer;
+import org.springframework.cloud.client.circuitbreaker.ReactiveCircuitBreakerFactory;
+import org.springframework.cloud.netflix.test.NoSecurityConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.stereotype.Service;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.reactive.function.client.WebClient;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT;
+
+/**
+ * @author Ryan Baxter
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest(webEnvironment = RANDOM_PORT,
+		classes = ReactiveHystrixCircuitBreakerIntegrationTest.Application.class)
+@DirtiesContext
+@Import(NoSecurityConfiguration.class)
+public class ReactiveHystrixCircuitBreakerIntegrationTest {
+
+	@LocalServerPort
+	int port = 0;
+
+	@Autowired
+	ReactiveHystrixCircuitBreakerIntegrationTest.Application.DemoControllerService service;
+
+	@Before
+	public void setup() {
+		service.setPort(port);
+	}
+
+	@Test
+	public void testSlow() {
+		assertThat(service.slow().block()).isEqualTo("fallback");
+	}
+
+	@Test
+	public void testNormal() {
+		assertThat(service.normal().block()).isEqualTo("normal");
+	}
+
+	@Configuration
+	@EnableAutoConfiguration
+	@RestController
+	protected static class Application {
+
+		@RequestMapping("/slow")
+		public Mono<String> slow() {
+			return Mono.just("slow").delayElement(Duration.ofSeconds(3));
+		}
+
+		@GetMapping("/normal")
+		public Mono<String> normal() {
+			return Mono.just("normal");
+		}
+
+		@Bean
+		public Customizer<ReactiveHystrixCircuitBreakerFactory> customizer() {
+			return factory -> factory
+					.configure(
+							builder -> builder.commandProperties(HystrixCommandProperties
+									.Setter().withExecutionTimeoutInMilliseconds(2000)),
+							"slow");
+		}
+
+		@Bean
+		public Customizer<ReactiveHystrixCircuitBreakerFactory> defaultConfig() {
+			return factory -> factory
+					.configureDefault(id -> HystrixObservableCommand.Setter
+							.withGroupKey(HystrixCommandGroupKey.Factory.asKey(id))
+							.andCommandPropertiesDefaults(HystrixCommandProperties
+									.Setter().withExecutionTimeoutInMilliseconds(4000)));
+		}
+
+		@Service
+		public static class DemoControllerService {
+
+			private int port = 0;
+
+			private ReactiveCircuitBreakerFactory cbFactory;
+
+			DemoControllerService(ReactiveCircuitBreakerFactory cbBuilder) {
+				this.cbFactory = cbBuilder;
+			}
+
+			public Mono<String> slow() {
+				return WebClient.builder().baseUrl("http://localhost:" + port).build()
+						.get().uri("/slow").retrieve().bodyToMono(String.class)
+						.transform(it -> cbFactory.create("slow").run(it, t -> {
+							t.printStackTrace();
+							return Mono.just("fallback");
+						}));
+			}
+
+			public Mono<String> normal() {
+				return WebClient.builder().baseUrl("http://localhost:" + port).build()
+						.get().uri("/normal").retrieve().bodyToMono(String.class)
+						.transform(it -> cbFactory.create("normal").run(it, t -> {
+							t.printStackTrace();
+							return Mono.just("fallback");
+						}));
+			}
+
+			public void setPort(int port) {
+				this.port = port;
+			}
+
+		}
+
+	}
+
+}

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

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2013-2018 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 org.springframework.cloud.netflix.hystrix;
+
+import org.assertj.core.util.Arrays;
+import org.junit.Test;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.client.circuitbreaker.ReactiveCircuitBreaker;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Ryan Baxter
+ */
+public class ReactiveHystrixCircuitBreakerTest {
+
+	@Test
+	public void monoRun() {
+		ReactiveCircuitBreaker cb = new ReactiveHystrixCircuitBreakerFactory()
+				.create("foo");
+		Mono<String> s = Mono.just("foobar")
+				.transform(it -> cb.run(it, t -> Mono.just("fallback")));
+		assertThat(s.block()).isEqualTo("foobar");
+	}
+
+	@Test
+	public void monoFallback() {
+		ReactiveCircuitBreaker cb = new ReactiveHystrixCircuitBreakerFactory()
+				.create("foo");
+		assertThat(Mono.error(new RuntimeException("boom"))
+				.transform(it -> cb.run(it, t -> Mono.just("fallback"))).block())
+						.isEqualTo("fallback");
+	}
+
+	@Test
+	public void fluxRun() {
+		ReactiveCircuitBreaker cb = new ReactiveHystrixCircuitBreakerFactory()
+				.create("foo");
+		Flux<String> s = Flux.just("foobar", "hello world")
+				.transform(it -> cb.run(it, t -> Flux.just("fallback")));
+		assertThat(s.collectList().block())
+				.isEqualTo(Arrays.asList(new String[] { "foobar", "hello world" }));
+	}
+
+	@Test
+	public void fluxFallback() {
+		ReactiveCircuitBreaker cb = new ReactiveHystrixCircuitBreakerFactory()
+				.create("foo");
+		assertThat(Flux.error(new RuntimeException("boom"))
+				.transform(it -> cb.run(it, t -> Flux.just("fallback"))).collectList()
+				.block()).isEqualTo(Arrays.asList(new String[] { "fallback" }));
+	}
+
+}