RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
Flowable
.create(emitter -> {
int i = 0;
while (i++ < 10) {
emitter.onNext(RandomUtils.nextInt() + ":" + i);
Thread.sleep(100);
}
emitter.onComplete();
}, BackpressureStrategy.BUFFER)
//.subscribeOn(Schedulers.computation())
.observeOn(Schedulers.io())
.buffer(1, TimeUnit.SECONDS, Schedulers.trampoline())
.subscribe(list -> {
System.out.println("BlockedBufferRxTest.main:" + list.toString());
});
Thread.sleep(1000000);
Isn't there a backpressure problem here ?
Using RxJava 1.1.9, I have a problem as follows:
import rx.Observable;
public enum ZipWithTest {;
public static void main(String[] args) {
Observable
.just(1, 2, 3)
.doOnNext(next -> System.out.format("next: %d%n", next))
.zipWith(
Observable.just(10, 100, 1000),
(next, multiplier) -> next * multiplier)
.doOnNext(multiplied -> System.out.format("multiplied: %d%n", multiplied))
.toBlocking()
.last();
}
/*
next: 1
next: 2
next: 3
multiplied: 10
multiplied: 200
multiplied: 3000
*/
}I was expecting to get the following:
next: 1
multiplied: 10
next: 2
multiplied: 200
next: 3
multiplied: 3000What am I missing?
so that it invokes an action when it calls {@code onNext}
DoOnNext is called after ever onNext op from every element of source
@vy you should really upgrade to rxjava 2, but as rx1 observable has backpressure this might work as you said, but the problem is that zipWith has a buffer, so the 3 next call are juts filling the zip buffer, not sure if in rx1 exists the zipWith operator overload to indicate the buffer size, but you can do it in rx2, in that case you need to use Flowable instead of Observable to support backpressure
@ylecaillez you can use publish and merge the take(1) with a ignoreElements, so the source observable keep subscribe untils it completes (you can exgract the compose lamda to reuse the operator)
Observable.just(1, 2, 3).doOnNext(n -> out.println("emmit " + n))
.compose(o -> o.publish(p -> p.take(1)))
.subscribe(n -> out.println("final " + n));
// emmit 1 ⏎ final 1
Observable.just(1, 2, 3).doOnNext(n -> out.println("emmit " + n))
.compose(o -> o.publish(p -> p.take(1).mergeWith(p.ignoreElements())))
.subscribe(n -> out.println("final " + n));
// emmit 1 ⏎ final 1 ⏎ emmit 2 ⏎ emmit 3
'ello. i'm wondering if there's a recommended way to achieve the following on 1.x: i have a hot observable (a publishsubject) but i'd like to buffer the last produced item so that new subscribers always see the most recently published event when they subscribe (and then see any new events when i publish them after that)
i'm currently stuck on 1.x due to an incompatible third-party dependency in the project i'm working on, unfortunately
sorry, that should read "buffer the most recently produced value"
sorry for the noise!
I am trying to get a grasp of RxJava. How would i chain following situation in calls with RxJava: i retrieve a Single<User> from my database, loading a user by id for example, this should happen async. After loading the user object, i put some of his specific data in an arraylist. After that finished I want to listen on this array for changes, so that when an element gets added or removed I write to the database async to update the user. I have following code so far, also implemented the piece you have suggested
@igorbzin_gitlab You mean something like this? https://stackoverflow.com/q/28816691/61158
@akarnokd yes, something exactly like this, thank you. I am still not sure how to chain those asynchronous calls though, since i operate on the database.
Observable.just(uriObservableList)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<RxObservableList<Uri>>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "ONSUBSCRIBE ");
}
@Override
public void onNext(RxObservableList<Uri> uris) {
String s = makePathString(uris.getList());
AppExecutor.getInstance().diskIO().execute(() -> viewModel.updatePicturePaths(s)); // <-- THis async call should be performed on every onNext
mAdapter.updatePictures(uris.getList());
Log.d(TAG, "ONNEXT");
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "ONERROR " + e);
}
@Override
public void onComplete() {
Log.d(TAG, "ONCOMPLETE ");
}
});How do I construct my Subscription so that before every onNext call I can perform a write operation on the database? I have tried doOnNext which is supposed to be executed before onNext, but it is always called from the main thread
I have previously used MediatorLiveData to combine results from FCM, Room and Networking in my Repository. It worked well. However, now I want more complexity and some additional bonuses from RxJava. This is why I have decided to use RxJava this time. I can combine Room and Networking with Observable.concatArrayEager(my _observables) also I have used merge, but I want to get only latest changed value, not all observables, like it is done with MediatorLiveData However, I don't now how to do that after FCM pushes value, and how should I notify my main observable after new changes occur? No examples on this issue whatsoever. It is crucial part. I receive FCM in my BroadCastReceiver and then notified my repository's livedata, which notified my MediatorLiveData... How to do that with RxJava? Would really welcome any advice, because it is really important issue.
@akarnokd already tried it.
viewModel.getPlace().map(this::getPicturePaths)
.subscribeOn(Schedulers.io())
.toObservable()
.concatMap(new Function<ArrayList<Uri>, Observable<RxObservableList<Uri>>>() {
@Override
public Observable<RxObservableList<Uri>> apply(ArrayList<Uri> uriArrayList) throws Exception {
mAdapter = new PictureRvAdapter(MarkerActivity.this, uriArrayList, MarkerActivity.this);
LinearLayoutManager layoutManager = new LinearLayoutManager(MarkerActivity.this, LinearLayoutManager.VERTICAL, false);
DividerItemDecoration decoration = new DividerItemDecoration(getApplicationContext(), VERTICAL);
mPicturesRV.addItemDecoration(decoration);
mPicturesRV.setLayoutManager(layoutManager);
mPicturesRV.setAdapter(mAdapter);
beginTransition(constraintSetEnd, 850);
Log.d(TAG, "Adapter set, transition started");
uriObservableList.setList(uriArrayList);
return uriObservableList.getObservable();
//return Observable.just(uriObservableList);
}
})
.doOnNext(new Consumer<RxObservableList<Uri>>() {
@Override
public void accept(RxObservableList<Uri> uris) throws Exception {
String pathString = makePathString(uris.getList());
viewModel.updatePicturePaths(pathString);
Log.d(TAG, "Updated pictures in database");
}
})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<RxObservableList<Uri>>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "Onsubscribe called");
}
@Override
public void onNext(RxObservableList<Uri> uris) {
mAdapter.updatePictures(uris.getList());
Log.d(TAG, "Adapter pictures updated");
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "OnError : " + e);
}
@Override
public void onComplete() {
Log.d(TAG, "OnComplete called");
}
});still gives me this error: MarkerActivity: OnError : java.lang.IllegalStateException: Cannot access database on the main thread since it may potentially lock the UI for a long period of time.
@Flemingjp
Observable<String> stream = Observable.fromArray("a", "b", "c");
Observable<String> request = Single.just("1").toObservable();
stream.withLatestFrom(request, (streamItem, requestItem) -> streamItem + requestItem)
.forEach(System.out::println);
// output: a1 ⏎ b1 ⏎ c1 ⏎
Been studying the operators but can’t see an obvious way to do it
Schedulers.io() is un-bound thread pool, so when there is a blocked process,the number of threads will be increased unsafely.
@Test
public void threadCounter() throws Exception {
int j = 0;
while (j++ < 1000) {
Flowable
.fromCallable(() -> {
Thread.sleep(1000);
return RandomUtils.nextInt();
})
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.io())
.subscribe(c -> {
//System.out.println("c:" + c);
});
}
IoScheduler io = (IoScheduler) Schedulers.io();
int i = 0;
while (i++ < 10000) {
System.out.println("io size:" + io.size());
Thread.sleep(1000);
}
}
@nbransby How should that single emitted resulting item relate to the inputs?
@akarnokd my use case is keeping track of open tabs and the result should signal that all tabs have been closed, I have this but havent actually tested it yet
//close the window when all tabs are closed
tabDidOpen.buffer(tabDidClose)
.map { it.size - 1 }
.scan(0) { acc, current -> acc + current }
.filter { it == 0 }
.subscribe { closeWindow() }
@nbransby if you have a stream of “plus” and “minus” events, you can do something like this…
val lines = Observable.create<String> { while (true) it.onNext(readLine()!!) }
.subscribeOn(Schedulers.newThread()).share()
val plus = lines.filter { "+" == it }.map { { acc: Int -> acc + 1 } }
val minus = lines.filter { "-" == it }.map { { acc: Int -> acc - 1 } }
val calc = mergeArray<(Int) -> Int>(plus, minus).scan(0) { acc, fn -> fn(acc) }
calc.doOnNext(::println).blockingLast()
plus and minus are of type
Observable<(Int) -> Int>
@ibaca I want to merge the request into the stream, but not merge the two items together - almost inject the single value into the stream. I've tried using
mergeWith(...) but when the single terminates, the other stream also terminates. Im trying to find behaviour like this
Hello folks, I'm currently trying to port an application I had built in Electron using node.js to JavaFX. I relied heavily on RX in general for the initial effort (rxjs in particular of course) and I am having some trouble getting the semantics right for RxJava. Should mention I am using Kotlin, but I am not (intentionally, anyways) using RxKotlin per se.
I had written a lot of code to extend the Observable prototype in rxjs, and I am trying to translate these into extension methods in Kotlin.
function debounceOnMissedHeartbeat<TKey, TValue> (dueTime:number, onDebounceItemFactory:(key:TKey) => TValue, scheduler:Rx.Scheduler) {
let sources:Rx.GroupedObservable = this
return Rx.Observable.create(o => {
return sources.subscribe(innerSource => {
let key:TKey = innerSource.key
let debouncedStream = innerSource.debounceWithSelector(dueTime, () => onDebounceItemFactory(key), scheduler)
o.onNext(debouncedStream)
},
ex => o.onError(ex),
() => o.onCompleted()
)
})
}
Rx.Observable.prototype.debounceOnMissedHeartbeat = debounceOnMissedHeartbeatfunction debounceWithSelector (dueTime:number, itemSelector, scheduler:Rx.Scheduler) {
let source = this
return Rx.Observable.create(o => {
let disposables = new Rx.CompositeDisposable()
let debounceDisposable = new Rx.SerialDisposable()
disposables.add(debounceDisposable)
let debounce = () => {
debounceDisposable.setDisposable(
scheduler.scheduleFuture(
'',
dueTime,
() => {
let debouncedItem = itemSelector()
o.onNext(debouncedItem)
}
)
)
}
disposables.add(
source.subscribe(
item => {
debounce()
o.onNext(item)
},
ex => {
try {
o.onError(ex)
} catch (err1) {
}
},
() => o.onCompleted()
)
)
debounce()
return disposables
})
}
Rx.Observable.prototype.debounceWithSelector = debounceWithSelector