In the below code the subscriber stops recieving data whenever there is a timeout exception. How can I make sure that the subscriber does not stop when there is exception.
public class ReactiveDataService
{
private static String[] quotes = {"ITEM1", "ITEM2", "ITEM3"};
public Observable<Notification<String>> getStreamData()
{
return Observable.create(subscriber -> {
if(!subscriber.isUnsubscribed())
{
Stream<String> streams = Arrays.stream(quotes);
streams.map(quote -> quote.toString()).filter(quote -> quote!=null)
.forEach(q -> {
subscriber.onNext(Notification.createOnNext(q));
try
{
Random rand = new Random();
Integer i = (rand.nextInt(5)+1)*1000;
Thread.sleep(i);
}
catch (InterruptedException e)
{
e.printStackTrace();
}
});
}
subscriber.onCompleted();
});
}
}
public class ReactiveResource
{
public static void main(String args[])
{
Observable<Notification<String>> watcher = new ReactiveResource().getData()
.timeout(4, TimeUnit.SECONDS)
.doOnError(failure -> System.out.println("Error:" + failure.getCause()))
.onErrorResumeNext(th -> {
return Observable.just(Notification.createOnError(new TimeoutException("Timed Out!")));
});
watcher.subscribe(
ReactiveResource::callBack,
ReactiveResource::errorCallBack,
ReactiveResource::completeCallBack
);
}
public static Action1 callBack(Notification<String> data)
{
System.out.println(data.getValue());
return null;
}
public static void errorCallBack(Throwable throwable)
{
System.out.println(throwable instanceof TimeoutException);
System.out.println(throwable);
}
public static void completeCallBack()
{
System.out.println("On completed successfully");
}
private Observable<Notification<String>> getData()
{
return new ReactiveDataService().getStreamData();
}