-
Notifications
You must be signed in to change notification settings - Fork 154
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
support of hystrix added (in progress)
- Loading branch information
skarpenko
committed
Mar 6, 2018
1 parent
39c6e45
commit cae258c
Showing
15 changed files
with
568 additions
and
345 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
97 changes: 97 additions & 0 deletions
97
feign-reactive-cloud/src/main/java/feign/HystrixMethodHandler.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,97 @@ | ||
package feign; | ||
|
||
import com.netflix.hystrix.HystrixObservableCommand; | ||
import feign.reactive.ReactiveMethodHandlerFactory; | ||
import org.reactivestreams.Publisher; | ||
import org.springframework.lang.Nullable; | ||
import reactor.core.publisher.Flux; | ||
import reactor.core.publisher.Mono; | ||
import rx.Observable; | ||
import rx.RxReactiveStreams; | ||
|
||
import java.lang.reflect.ParameterizedType; | ||
import java.lang.reflect.Type; | ||
import java.util.function.Function; | ||
|
||
import static feign.Util.checkNotNull; | ||
|
||
/** | ||
* @author Sergii Karpenko | ||
*/ | ||
public class HystrixMethodHandler implements ReactiveMethodHandler { | ||
|
||
private final MethodMetadata methodMetadata; | ||
private final Type returnPublisherType; | ||
private final ReactiveMethodHandler methodHandler; | ||
private final Function<Throwable, Object> fallbackFactory; | ||
private HystrixObservableCommand.Setter hystrixObservableCommandSetter; | ||
|
||
private HystrixMethodHandler( | ||
MethodMetadata methodMetadata, | ||
ReactiveMethodHandler methodHandler, | ||
@Nullable | ||
Function<Throwable, Object> fallbackFactory, | ||
HystrixObservableCommand.Setter hystrixObservableCommandSetter) { | ||
this.methodMetadata = checkNotNull(methodMetadata, "methodMetadata must be not null"); | ||
returnPublisherType = ((ParameterizedType) methodMetadata.returnType()).getRawType(); | ||
this.methodHandler = checkNotNull(methodHandler, "methodHandler must be not null"); | ||
this.fallbackFactory = fallbackFactory; | ||
this.hystrixObservableCommandSetter = checkNotNull(hystrixObservableCommandSetter, "hystrixObservableCommandSetter must be not null"); | ||
} | ||
|
||
@Override | ||
@SuppressWarnings("unchecked") | ||
public Publisher invoke(final Object[] argv) { | ||
|
||
Observable<Object> observable = new HystrixObservableCommand<Object>(hystrixObservableCommandSetter){ | ||
@Override | ||
protected Observable<Object> construct() { | ||
return RxReactiveStreams.toObservable((Publisher)methodHandler.invoke(argv)); | ||
} | ||
|
||
@Override | ||
protected Observable<Object> resumeWithFallback() { | ||
if(fallbackFactory != null){ | ||
Object fallback = fallbackFactory.apply(getExecutionException()); | ||
Object fallbackValue = getFallbackValue(fallback, argv); | ||
return Observable.just(fallbackValue); | ||
} else { | ||
return super.resumeWithFallback(); | ||
} | ||
} | ||
}.toObservable(); | ||
|
||
Publisher<Object> publisher = RxReactiveStreams.toPublisher(observable); | ||
|
||
return returnPublisherType == Mono.class ? Mono.from(publisher) : Flux.from(publisher); | ||
} | ||
|
||
protected Object getFallbackValue(Object fallback, final Object[] argv){ | ||
//TODO implement | ||
return null; | ||
} | ||
|
||
static class Factory implements ReactiveMethodHandlerFactory { | ||
private final ReactiveMethodHandlerFactory methodHandlerFactory; | ||
private final Function<Throwable, Object> fallbackFactory; | ||
private final HystrixObservableCommand.Setter hystrixObservableCommandSetter; | ||
|
||
Factory(ReactiveMethodHandlerFactory methodHandlerFactory, | ||
@Nullable | ||
Function<Throwable, Object> fallbackFactory, | ||
HystrixObservableCommand.Setter hystrixObservableCommandSetter) { | ||
this.methodHandlerFactory = checkNotNull(methodHandlerFactory, "methodHandlerFactory must not be null"); | ||
this.fallbackFactory = fallbackFactory; | ||
this.hystrixObservableCommandSetter = checkNotNull(hystrixObservableCommandSetter, "hystrixObservableCommandSetter must not be null"); | ||
} | ||
|
||
@Override | ||
public ReactiveMethodHandler create(final Target target, final MethodMetadata metadata) { | ||
return new HystrixMethodHandler( | ||
metadata, | ||
methodHandlerFactory.create(target, metadata), | ||
fallbackFactory, | ||
hystrixObservableCommandSetter); | ||
} | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.