Netflix Hystrix Reactive RxJava Execution Command Example
In this tutorial we show you how to execute a reactive Hystrix command. We use the HystrixObservableCommand which is used to wrap code that will execute potentially risky functionality. Typically meaning a service call over the network with fault and latency tolerance.
Project Structure
Let’s start by looking at the project structure.
Maven Dependencies
We use Apache Maven to manage our project dependencies. Make sure the following dependencies reside on the class-path.
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.memorynotfound.netflix.hystrix</groupId>
<artifactId>reactive-execution</artifactId>
<version>1.0.0-SNAPSHOT</version>
<url>https://memorynotfound.com</url>
<name>Netflix Hystrix - ${project.artifactId}</name>
<dependencies>
<dependency>
<groupId>com.netflix.hystrix</groupId>
<artifactId>hystrix-core</artifactId>
<version>1.5.3</version>
</dependency>
<!-- logging -->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.25</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-simple</artifactId>
<version>1.7.25</version>
</dependency>
<!-- testing -->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.7.0</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
</plugins>
</build>
</project>Netflix Hystrix Reactive RxJava Execution Command Example
This example uses the HystrixObservableCommand which is used to wrap code that will execute potentially risky functionality (typically meaning a service call over the network) with fault and latency tolerance, statistics and performance metics capture, circuit breaker and bulkhead functionality. This command should be used for a purely non-blocking all pattern. The caller of this command will be subscribed to the Observable<R> returned by the run() method.
The construct() method will be executed when the observe() or toObservable() methods are invoked. This’ll produce a non blocking method invocation and returns a result in a reactive java programming way.
package com.memorynotfound.netflix.hystrix;
import com.netflix.hystrix.HystrixCommandGroupKey;
import com.netflix.hystrix.HystrixObservableCommand;
import rx.Observable;
import rx.Subscriber;
import rx.schedulers.Schedulers;
public class GreetingCommand extends HystrixObservableCommand<String> {
private final String name;
public GreetingCommand(String name) {
super(HystrixCommandGroupKey.Factory.asKey("GreetingGroup"));
this.name = name;
}
@Override
protected Observable<String> construct() {
return Observable.create(new Observable.OnSubscribe<String>(){
@Override
public void call(Subscriber<? super String> observer) {
try {
if (!observer.isUnsubscribed()){
if (name == null){
throw new IllegalArgumentException("name cannot be null");
}
observer.onNext("Hello " + name + "!");
observer.onCompleted();
}
} catch (Exception e){
observer.onError(e);
}
}
}).subscribeOn(Schedulers.io());
}
@Override
protected Observable<String> resumeWithFallback() {
return Observable.create((Observable.OnSubscribe<String>) observer -> {
if (!observer.isUnsubscribed()){
observer.onNext("Hello Guest!");
observer.onCompleted();
}
}).subscribeOn(Schedulers.io());
}
}Writing a Reactive RxJava Fallback
If observe() or toObservable() fails in any way then the resumeWithFallback() method will be invoked to provide an opportunity to return a fallback response. This should typically not require network transport. In other words, this should be a static, stubbed or cached result that can immediately be returned upon failure. If network traffic is wanted for fallback (such as going to MemCache or Redis) then the fallback implementation should invoke another HystrixObservableCommand instance that protects against that network access and possibly has another level of fallback that does not involve network access.
Unit Test Netflix Hystrix Reactive RxJava Execution with JUnit
You can execute a HystrixObservableCommand reactively by using the observe() or toObservable() method. Which is used for asynchronous execution of command with a callback by subscribing to the Observable. This eagerly starts execution of the command the same as HystrixCommand.queue() and HystrixCommand.execute().
In the testBlockingReactiveGreeting JUnit test, we create a blocking execution using the Observable.toBlocking() method. This creates a BlockingObservable which has blocking capabilities. The single() method is used to retrieve the item after the result is emitted.
The following testNonBlockingReactiveGreeting and executeNonBlockingReactiveGreeting we execute non-blocking reactive programming hystrix commands.
The last testNonBlockingReactiveGreetingFallBack JUnit test demonstrates the resumeWithFallback method from the HystrixObservableCommand.
package com.memorynotfound.netflix.hystrix.test;
import com.memorynotfound.netflix.hystrix.GreetingCommand;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rx.Observable;
import rx.Observer;
import rx.functions.Action1;
import rx.observers.TestSubscriber;
import rx.schedulers.Schedulers;
import rx.schedulers.TestScheduler;
import java.util.concurrent.TimeUnit;
import static org.junit.Assert.*;
public class GreetingCommandTest {
private static final Logger log = LoggerFactory.getLogger(GreetingCommandTest.class);
// blocking
@Test
public void testBlockingReactiveGreeting() throws Exception {
GreetingCommand command1 = new GreetingCommand("World");
GreetingCommand command2 = new GreetingCommand("John");
assertEquals("Hello World!", command1.observe().toBlocking().single());
assertEquals("Hello John!", command2.observe().toBlocking().single());
assertFalse(command1.isResponseFromFallback());
assertFalse(command2.isResponseFromFallback());
}
// non-blocking
@Test
public void testNonBlockingReactiveGreeting() throws Exception {
GreetingCommand command1 = new GreetingCommand("World");
GreetingCommand command2 = new GreetingCommand("John");
Observable<String> oWorld = command1.observe();
Observable<String> oJohn = command2.observe();
TestScheduler testScheduler = Schedulers.test();
TestSubscriber<String> worldSubscriber = new TestSubscriber<>();
TestSubscriber<String> johnSubscriber = new TestSubscriber<>();
oWorld.subscribe(worldSubscriber);
oJohn.subscribe(johnSubscriber);
testScheduler.advanceTimeBy(100L, TimeUnit.SECONDS);
assertEquals("Hello World!", worldSubscriber.getOnNextEvents().get(0));
assertEquals("Hello John!", johnSubscriber.getOnNextEvents().get(0));
assertFalse(command1.isResponseFromFallback());
assertFalse(command2.isResponseFromFallback());
}
@Test
public void executeNonBlockingReactiveGreeting() {
GreetingCommand command1 = new GreetingCommand("World");
GreetingCommand command2 = new GreetingCommand("John");
Observable<String> oWorld = command1.observe();
Observable<String> oJohn = command2.observe();
// non-blocking
// - this is a verbose anonymous inner-class approach and doesn't do assertions
oWorld.subscribe(new Observer<String>() {
@Override
public void onCompleted() {
// nothing needed here
}
@Override
public void onError(Throwable e) {
log.error("", e);
}
@Override
public void onNext(String greeting) {
log.debug(greeting);
}
});
// non-blocking
// - also verbose anonymous inner-class
// - ignore errors and onCompleted signal
oJohn.subscribe(log::debug);
oJohn.subscribe(greeting -> log.debug(greeting));
oJohn.subscribe(new Action1<String>() {
@Override
public void call(String greeting) {
log.debug(greeting);
}
});
}
@Test
public void testBlockingReactiveGreetingFallBack() throws Exception {
try {
GreetingCommand command = new GreetingCommand(null);
assertEquals("Hello Guest!", command.observe().toBlocking().single());
assertTrue(command.isResponseFromFallback());
} catch (Exception e){
fail("we should not get an exception as we return a fallback");
}
}
@Test
public void testNonBlockingReactiveGreetingFallBack() throws Exception {
try {
GreetingCommand command = new GreetingCommand(null);
Observable<String> oNull = command.observe();
TestScheduler testScheduler = Schedulers.test();
TestSubscriber<String> nullSubscriber = new TestSubscriber<>();
oNull.subscribe(nullSubscriber);
testScheduler.advanceTimeBy(100L, TimeUnit.MILLISECONDS);
assertEquals("Hello Guest!", nullSubscriber.getOnNextEvents().get(0));
assertTrue(command.isResponseFromFallback());
} catch (Exception e){
fail("we should not get an exception as we return a fallback");
}
}
}JUnit Test Results
When we run all of our JUnit tests, we receive the following output.
References
- Netflix Hystrix Documentation
- Netflix Hystrix JavaDoc API
- HystrixObservableCommand JavaDoc
- rx.Observable JavaDoc
- rx.BlockingObservable JavaDoc
