I have some basic project that has like four calls to some external resource, that in current version runs synchronously. What I would like to achieve is to wrap that calls into HystrixObservableCommand and then call it asynchronously.
From what I have read, after calling .observe() at the HystrixObservableCommand object, the wrapped logic should be called immediately and asynchronously. However I am doing something wrong, because it works synchronously.
In the example code, the output is Void, because I'm not interested in output (for now). That is also why I did not assigned the Observable to any object, just called constructor.observe().
@Component
public class LoggerProducer {
private static final Logger LOGGER = Logger.getLogger(LoggerProducer.class);
@Autowired
SimpMessagingTemplate template;
private void push(Iterable<Message> messages, String topic) throws Exception {
template.convertAndSend("/messages/"+topic, messages);
}
public void splitAndPush(Iterable<Message> messages) {
Map<MessageTypeEnum, List<Message>> groupByMessageType = StreamSupport.stream(messages.spliterator(), true)
.collect(Collectors.groupingBy(Message::getType));
//should be async - it's not
new CommandPushToBrowser(groupByMessageType.get(MessageTypeEnum.INFO),
MessageTypeEnum.INFO.toString().toLowerCase()).observe();
new CommandPushToBrowser(groupByMessageType.get(MessageTypeEnum.WARN),
MessageTypeEnum.WARN.toString().toLowerCase()).observe();
new CommandPushToBrowser(groupByMessageType.get(MessageTypeEnum.ERROR),
MessageTypeEnum.ERROR.toString().toLowerCase()).observe();
}
class CommandPushToBrowser extends HystrixObservableCommand<Void> {
private Iterable<Message> messages;
private String messageTypeName;
public CommandPushToBrowser(Iterable<Message> messages, String messageTypeName) {
super(HystrixCommandGroupKey.Factory.asKey("Messages"));
this.messageTypeName = messageTypeName;
this.messages = messages;
}
@Override
protected Observable<Void> construct() {
return Observable.create(new Observable.OnSubscribe<Void>() {
@Override
public void call(Subscriber<? super Void> observer) {
try {
for (int i = 0 ; i < 50 ; i ++ ) {
LOGGER.info("Count: " + i + " messageType " + messageTypeName);
}
if (null != messages) {
push(messages, messageTypeName);
LOGGER.info("Message type: " + messageTypeName + " pushed: " + messages);
}
if (!observer.isUnsubscribed()) {
observer.onCompleted();
}
} catch (Exception e) {
e.printStackTrace();
observer.onError(e);
}
}
});
}
}
}
There are some pure "test" code fragments there, as I was trying to figure out the problem, just ignore the logic, main focus is to make it run async with .observe(). I do know that I may achieve that with standard HystrixCommand, but this is not the goal.
Hope someone helps :) Regards,
Answer was found:
"Observables do not add concurrency automatically. If you are modeling synchronous, blocking execution with an Observable, then they will execute synchronously.
You can easily make it asynchronous by scheduling on a thread using subscribeOn(Schedulers.io()). Here is a simply example for wrapping a blocking call with an Observable: https://speakerdeck.com/benjchristensen/applying-reactive-programming-with-rxjava-at-goto-chicago-2015?slide=33
However, if you are wrapping blocking calls, you should just stick with using HystrixCommand as that’s what it’s built for and it defaults to running everything in a separate thread. Using HystrixCommand.observe() will give you the concurrent, async composition you’re looking for.
HystrixObservableCommand is intended for wrapping around async, non-blocking Observables that don’t need extra threads."
-- Ben Christensen - Netflix Edge Engineering
Source: https://groups.google.com/forum/#!topic/hystrixoss/g7ZLIudE8Rs