RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
Single<MyComponent> myComponent$ returns a new instance, if you want the instance to be cached forever you can do myComponent$.cache() but what we actually want is to cache the last component until it is hidden for more than X time, so this is our current code:
static <T extends Component> Single<T> memoize(Single<T> src, int delay, TimeUnit unit) {
BehaviorSubject<Single<T>> cache = BehaviorSubject.create();
cache.onNext(memoize(src, delay, unit, cache));
return cache.flatMapSingle(Rx.identity()).firstOrError();
}
static <T extends Component> Single<T> memoize(Single<T> src, int delay, TimeUnit unit, Observer<Single<T>> cache) {
return src.doOnSuccess(c -> {
// Instantiating component…
var timeout = visible(c).switchMap(n -> n ? Observable.never() : timer(delay, unit)).firstOrError();
//noinspection ResultOfMethodCallIgnored component should hide eventually, so this will get disposed
timeout.subscribe(n -> {
// Discarding component…
cache.onNext(memoize(src, delay, unit, cache));
});
}).cache();
}Observable<Boolean>, it emmit the current state on subscription and changes when te component hides or reveal, the question or ask-for-suggestion is, how do you do this without a subject?
switchIfEmpty will only subscribe if maybeis empty
public void start() {
FixedSchedulerPool update = new FixedSchedulerPool(threadFactory, MAX_THREADS);
if (!pool.compareAndSet(NONE, update)) {
update.shutdown();
}
}
@Override
public void start() {
FixedSchedulerPool update = new FixedSchedulerPool(threadFactory, MAX_THREADS);
if (!pool.compareAndSet(NONE, update)) {
update.shutdown();
}
}
@Override
public void onMessage(WebSocket webSocket, String text) {
log.info(text);
final ConnectableFlux<String> replay = inputBus.getEvents().replay(1);
if (json == null && replay.autoConnect().hasElements().block()) {
json = replay.autoConnect().next().block();
}
outputBus.setObject(text);
webSocket.send(json);
}
But, It's not back pressured at map level with bounded size. I get an exception where the consumers are slow. Is there any workaround for this ? I can't use reactive pull, because I need to fit into the existing framework ( Map Reduce)
fileNames.flatMap(fn -> readFile(fn).retry(3)) or fileNames.flatMap(fn -> fn.readFile(fn)).toList().retry(3); but forEach is a terminal operator, do not even return an observable, so you should not use it if you want to apply retry the whole reading process
Hi, I have an Observable<T> and want to use it to construct another Observable<T> that emits a value whenever the original observable emits one, and emits another (constant) value if the original observable doesn’t emit anything within a certain timeframe. For example, notifications are emitted, and I want to emit a ‘hide notification’ event if there are no other notifications are emitted before X seconds. I’ve tried switchMapping the original stream with just and delay observables (shown below), but it doesn’t seem that idiomatic, and I’m not sure if it will behave correctly. Is there an obvious solution to this that I’m missing? Using Kotlin and RxJava2.
originalObs.switchMap {
Observable.concat(Observable.just(it), Observable.just(FALLBACK_VALUE).delay(30, TimeUnit.SECONDS))
}
Hello, what is the proper way to merge Observable/Singles?
public void testGetBlob() throws RequestException {
TestData.getNewApplication().flatMap(testApplication -> {
Client.initialize(testApplication.getAppId(), testApplication.getApiToken(), testApplication.getMasterKey());
assertNotNull(testApplication.getApiToken());
assertNotNull(testApplication.getAppId());
assertNotNull(testApplication.getMasterKey());
Entity entity = new Entity("Todo");
return entity.create();
}).flatMap(entity -> entity.setBlobProperty("text", "Hello world!".getBytes("UTF-8")))
.flatMap(isSuccess -> {
if(isSuccess) {
// need to access `entity` at this point
return Single.just(isSuccess);
} else {
return Single.just(false);
}
}).subscribe(success -> {
Browser.getWindow().getConsole().log("Blob created");
finishTest();
}, error -> {
Browser.getWindow().getConsole().error(error.getMessage());
fail();
});
delayTestFinish(5000);
}At the commented line, I need to be able to access the entity object, how can it be done?
I have 2 observable sources where fetchProductList() returns Observable<Datasource> and other canInvite.execute() takes a value and returns a Single<Boolean>. I need to compose the stream to feed every emission from first source to the second in a way that I can return all the emission from fetchProductList() with the true value as a list. I tried to compose it this way but it looks a bit clunky, Any better/simpler ways to do it?
fetchProductList()
.map { dataSource -> dataSource.data }
.flatMap{ data ->
Observable.from(data.roles.map { role ->
canInvite.execute(role.key).map { canInvite ->
role to canInvite
}
})
}
.compose { it -> Single.merge(it) }
.filter { it -> it.second == true}
.map { it -> it.first }
.toList()
.subscribe {
From 3.x, the as() methods have been removed and the to() methods now each work with their respective XConverer interfaces:
Flowable.to(Function<Flowable<T>, R>) is now Flowable.to(FlowableConverter<T, R>)
Observable.to(Function<Observable<T>, R>) is now Observable.to(ObservableConverter<T, R>)
Maybe.to(Function<Flowable<T>, R>) is now Maybe.to(MaybeConverter<T, R>)
Single.to(Function<Flowable<T>, R>) is now Maybe.to(SingleConverter<T, R>)
Completable.to(Function<Completable, R>) is now Completable.to(CompletableConverter<R>)
ParallelFlowable.to(Function<ParallelFlowable<T>, R) is now ParallelFlowable.to(ParallelFlowableConverter<T, R>)
return command.kafkaMessageListener.bus.getEvents().serialize()
.flatMap(x-> {
if(x == null) {
return Flux.error(new NoMessageReceivedException());
}
return Flux.just(x);
}).doOnError(x-> Flux.just(new NoMessageReceivedException()))
.subscribeOn(Schedulers.single())
.retryBackoff(3, Duration.ofSeconds(1), Duration.ofSeconds(60))
.publishOn(Schedulers.single())
.collectList().block().get(0);