Skip to content

Commit d1cb196

Browse files
christopherlakeychristopherlakey
committed
Add support for rx.Observable and rx.Single return types to Hystrix module
1 parent 94ec7c2 commit d1cb196

6 files changed

Lines changed: 170 additions & 7 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,6 @@
1+
### Version 8.14
2+
* Add support for RxJava Observable and Single return types via the `HystrixFeign` builder.
3+
14
### Version 8.13
25
* Never expands >8kb responses into memory
36
* Bumps dependency versions, most notably Gson 2.5 and OkHttp 2.7

‎hystrix/README.md‎

Lines changed: 15 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -10,15 +10,23 @@ GitHub github = HystrixFeign.builder()
1010
.target(GitHub.class, "https://api.github.com");
1111
```
1212

13-
Methods that do *not* return [`HystrixCommand`](https://netflix.github.io/Hystrix/javadoc/com/netflix/hystrix/HystrixCommand.html) are still wrapped in a `HystrixCommand`, but `execute()` is automatically called for you.
13+
For asynchronous or reactive use, return `HystrixCommand<YourType>`.
1414

15-
For asynchronous or reactive use, return `HystrixCommand<YourType>` rather than just `YourType`.
15+
For RxJava compatibility, use `rx.Observable<YourType>` or `rx.Single<YourType>`. Rx types are <a href="http://reactivex.io/documentation/observable.html">cold</a>, which means a http call isn't made until there's a subscriber.
16+
17+
Methods that do *not* return [`HystrixCommand`](https://netflix.github.io/Hystrix/javadoc/com/netflix/hystrix/HystrixCommand.html), [`rx.Observable`](http://reactivex.io/RxJava/javadoc/rx/Observable.html) or [`rx.Single`] are still wrapped in a `HystrixCommand`, but `execute()` is automatically called for you.
1618

1719
```java
1820
interface YourApi {
1921
@RequestLine("GET /yourtype/{id}")
2022
HystrixCommand<YourType> getYourType(@Param("id") String id);
21-
23+
24+
@RequestLine("GET /yourtype/{id}")
25+
Observable<YourType> getYourTypeObservable(@Param("id") String id);
26+
27+
@RequestLine("GET /yourtype/{id}")
28+
Single<YourType> getYourTypeSingle(@Param("id") String id);
29+
2230
@RequestLine("GET /yourtype/{id}")
2331
YourType getYourTypeSynchronous(@Param("id") String id);
2432
}
@@ -27,7 +35,10 @@ YourApi api = HystrixFeign.builder()
2735
.target(YourApi.class, "https://example.com");
2836

2937
// for reactive
30-
api.getYourType("a").toObservable();
38+
api.getYourTypeObservable("a").toObservable
39+
40+
// or apply hystrix to RxJava methods
41+
api.getYourTypeObservable("a")
3142

3243
// for asynchronous
3344
api.getYourType("a").queue();

‎hystrix/src/main/java/feign/hystrix/HystrixDelegatingContract.java‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,12 +10,14 @@
1010

1111
import feign.Contract;
1212
import feign.MethodMetadata;
13+
import rx.Observable;
14+
import rx.Single;
1315

1416
/**
15-
* This special cases methods that return {@link HystrixCommand}, so that they
17+
* This special cases methods that return {@link HystrixCommand}, {@link Observable}, or {@link Single} so that they
1618
* are decoded properly.
1719
*
18-
* <p>For example, {@literal HystrixCommand<Foo>} will decode {@code Foo}.
20+
* <p>For example, {@literal HystrixCommand<Foo>} and {@literal Observable<Foo>} will decode {@code Foo}.
1921
*/
2022
// Visible for use in custom Hystrix invocation handlers
2123
public final class HystrixDelegatingContract implements Contract {
@@ -36,6 +38,12 @@ public List<MethodMetadata> parseAndValidatateMetadata(Class<?> targetType) {
3638
if (type instanceof ParameterizedType && ((ParameterizedType) type).getRawType().equals(HystrixCommand.class)) {
3739
Type actualType = resolveLastTypeParameter(type, HystrixCommand.class);
3840
metadata.returnType(actualType);
41+
} else if (type instanceof ParameterizedType && ((ParameterizedType) type).getRawType().equals(Observable.class)) {
42+
Type actualType = resolveLastTypeParameter(type, Observable.class);
43+
metadata.returnType(actualType);
44+
} else if (type instanceof ParameterizedType && ((ParameterizedType) type).getRawType().equals(Single.class)) {
45+
Type actualType = resolveLastTypeParameter(type, Single.class);
46+
metadata.returnType(actualType);
3947
}
4048
}
4149

‎hystrix/src/main/java/feign/hystrix/HystrixFeign.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
import feign.Feign;
77

88
/**
9-
* Allows Feign interfaces to return HystrixCommand objects.
9+
* Allows Feign interfaces to return HystrixCommand or rx.Observable or rx.Single objects.
1010
* Also decorates normal Feign methods with circuit breakers, but calls {@link HystrixCommand#execute()} directly.
1111
*/
1212
public final class HystrixFeign {

‎hystrix/src/main/java/feign/hystrix/HystrixInvocationHandler.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,8 @@
2828
import feign.InvocationHandlerFactory;
2929
import feign.InvocationHandlerFactory.MethodHandler;
3030
import feign.Target;
31+
import rx.Observable;
32+
import rx.Single;
3133

3234
final class HystrixInvocationHandler implements InvocationHandler {
3335

@@ -62,6 +64,12 @@ protected Object run() throws Exception {
6264

6365
if (HystrixCommand.class.isAssignableFrom(method.getReturnType())) {
6466
return hystrixCommand;
67+
} else if (Observable.class.isAssignableFrom(method.getReturnType())) {
68+
// Create a cold Observable
69+
return hystrixCommand.toObservable();
70+
} else if (Single.class.isAssignableFrom(method.getReturnType())) {
71+
// Create a cold Observable as a Single
72+
return hystrixCommand.toObservable().toSingle();
6573
}
6674
return hystrixCommand.execute();
6775
}

‎hystrix/src/test/java/feign/hystrix/HystrixBuilderTest.java‎

Lines changed: 133 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,10 @@
22

33
import static feign.assertj.MockWebServerAssertions.assertThat;
44

5+
import java.util.Arrays;
56
import java.util.List;
67

8+
import org.assertj.core.api.Assertions;
79
import org.junit.Rule;
810
import org.junit.Test;
911
import org.junit.rules.ExpectedException;
@@ -15,6 +17,9 @@
1517
import feign.Headers;
1618
import feign.RequestLine;
1719
import feign.gson.GsonDecoder;
20+
import rx.Observable;
21+
import rx.Single;
22+
import rx.observers.TestSubscriber;
1823

1924
public class HystrixBuilderTest {
2025

@@ -59,6 +64,109 @@ public void hystrixCommandList() {
5964
assertThat(command.execute()).hasSize(2).contains("foo", "bar");
6065
}
6166

67+
@Test
68+
public void rxObservable() {
69+
server.enqueue(new MockResponse().setBody("\"foo\""));
70+
71+
TestInterface api = target();
72+
73+
Observable<String> observable = api.observable();
74+
75+
assertThat(observable).isNotNull();
76+
assertThat(server.getRequestCount()).isEqualTo(0);
77+
78+
TestSubscriber<String> testSubscriber = new TestSubscriber<String>();
79+
observable.subscribe(testSubscriber);
80+
testSubscriber.awaitTerminalEvent();
81+
Assertions.assertThat(testSubscriber.getOnNextEvents().get(0)).isEqualTo("foo");
82+
}
83+
84+
@Test
85+
public void rxObservableInt() {
86+
server.enqueue(new MockResponse().setBody("1"));
87+
88+
TestInterface api = target();
89+
90+
Observable<Integer> observable = api.intObservable();
91+
92+
assertThat(observable).isNotNull();
93+
assertThat(server.getRequestCount()).isEqualTo(0);
94+
95+
TestSubscriber<Integer> testSubscriber = new TestSubscriber<Integer>();
96+
observable.subscribe(testSubscriber);
97+
testSubscriber.awaitTerminalEvent();
98+
Assertions.assertThat(testSubscriber.getOnNextEvents().get(0)).isEqualTo(new Integer(1));
99+
}
100+
101+
@Test
102+
public void rxObservableList() {
103+
server.enqueue(new MockResponse().setBody("[\"foo\",\"bar\"]"));
104+
105+
TestInterface api = target();
106+
107+
Observable<List<String>> observable = api.listObservable();
108+
109+
assertThat(observable).isNotNull();
110+
assertThat(server.getRequestCount()).isEqualTo(0);
111+
112+
113+
TestSubscriber<List<String>> testSubscriber = new TestSubscriber<List<String>>();
114+
observable.subscribe(testSubscriber);
115+
testSubscriber.awaitTerminalEvent();
116+
assertThat(testSubscriber.getOnNextEvents().get(0)).hasSize(2).contains("foo", "bar");
117+
}
118+
119+
@Test
120+
public void rxSingle() {
121+
server.enqueue(new MockResponse().setBody("\"foo\""));
122+
123+
TestInterface api = target();
124+
125+
Single<String> single = api.single();
126+
127+
assertThat(single).isNotNull();
128+
assertThat(server.getRequestCount()).isEqualTo(0);
129+
130+
TestSubscriber<String> testSubscriber = new TestSubscriber<String>();
131+
single.subscribe(testSubscriber);
132+
testSubscriber.awaitTerminalEvent();
133+
Assertions.assertThat(testSubscriber.getOnNextEvents().get(0)).isEqualTo("foo");
134+
}
135+
136+
@Test
137+
public void rxSingleInt() {
138+
server.enqueue(new MockResponse().setBody("1"));
139+
140+
TestInterface api = target();
141+
142+
Single<Integer> single = api.intSingle();
143+
144+
assertThat(single).isNotNull();
145+
assertThat(server.getRequestCount()).isEqualTo(0);
146+
147+
TestSubscriber<Integer> testSubscriber = new TestSubscriber<Integer>();
148+
single.subscribe(testSubscriber);
149+
testSubscriber.awaitTerminalEvent();
150+
Assertions.assertThat(testSubscriber.getOnNextEvents().get(0)).isEqualTo(new Integer(1));
151+
}
152+
153+
@Test
154+
public void rxSingleList() {
155+
server.enqueue(new MockResponse().setBody("[\"foo\",\"bar\"]"));
156+
157+
TestInterface api = target();
158+
159+
Single<List<String>> single = api.listSingle();
160+
161+
assertThat(single).isNotNull();
162+
assertThat(server.getRequestCount()).isEqualTo(0);
163+
164+
TestSubscriber<List<String>> testSubscriber = new TestSubscriber<List<String>>();
165+
single.subscribe(testSubscriber);
166+
testSubscriber.awaitTerminalEvent();
167+
assertThat(testSubscriber.getOnNextEvents().get(0)).hasSize(2).contains("foo", "bar");
168+
}
169+
62170
@Test
63171
public void plainString() {
64172
server.enqueue(new MockResponse().setBody("\"foo\""));
@@ -101,6 +209,31 @@ interface TestInterface {
101209
@Headers("Accept: application/json")
102210
HystrixCommand<Integer> intCommand();
103211

212+
@RequestLine("GET /")
213+
@Headers("Accept: application/json")
214+
Observable<List<String>> listObservable();
215+
216+
@RequestLine("GET /")
217+
@Headers("Accept: application/json")
218+
Observable<String> observable();
219+
220+
@RequestLine("GET /")
221+
@Headers("Accept: application/json")
222+
Single<Integer> intSingle();
223+
224+
@RequestLine("GET /")
225+
@Headers("Accept: application/json")
226+
Single<List<String>> listSingle();
227+
228+
@RequestLine("GET /")
229+
@Headers("Accept: application/json")
230+
Single<String> single();
231+
232+
@RequestLine("GET /")
233+
@Headers("Accept: application/json")
234+
Observable<Integer> intObservable();
235+
236+
104237
@RequestLine("GET /")
105238
@Headers("Accept: application/json")
106239
String get();

0 commit comments

Comments
 (0)