greenrobot/objectboxrxjava

Optional RxJava 2 APIs for ObjectBox

objectboxrxjava's Issues

Can't use find on a query

With the current Rx implementation, it's not possible to use find for pagination.

Take RxQuery.single for example. Since we pass Query<T> and receive a Single<List<T>>, we can't apply that operator.

I really don't have a suggestion on how to fix this decently. Right now, I'm doing:

Single.just(box.query()
                .build()
                .find(from, to)
)

Truth is that I would prefer a somewhat different Rx implementation. Something along the ways of requery (https://github.com/requery/requery).

Thank you.

Put objects

Hello!
How can I save objects reactively?

Exceptions are swallowed

I've got an exception

io.objectbox.exception.DbException: Entity is expected to have a no-arg constructor: Item

but it didn't surface until I've debugged it and captured it at BoxStore.java:670

Deleting all entities from a box doesn't emit an item from a reactive query

When I subscribe to an observable such as RxQuery.observable(box.query().build()) and then delete all entities from the box via box.removeAll() then it doesn't emit an empty list of entities.

Subcribe on query with single value.

I have a query value.

Query<Student> query = mStudentBox.query().contains(Student_.name, searchText).build();
query.subscribe().on(AndroidScheduler.mainThread()).observer(new DataObserver<List<Student>>() {
            @Override
            public void onData(List<Student> data) {
                ......
            }
        });

But i don't need List, i need a single value, like this:

Query<Student> query = mStudentBox.query().contains(Student_.name, searchText).build();
query.findFirst();

How i can do this ? How i can subscribe on this query ? I didn't find the solution.
Thank you for your time

Move to objectbox-java project

NPE is when using with .single() query

I am getting this:

java.lang.NullPointerException: Attempt to invoke interface method 'void io.objectbox.reactive.DataSubscription.cancel()' on a null object reference
                      at io.objectbox.query.QueryPublisher.unsubscribe(QueryPublisher.java:105)
                      at io.objectbox.reactive.DataSubscriptionImpl.cancel(DataSubscriptionImpl.java:37)
                      at io.objectbox.rx.RxQuery$5$2.cancel(RxQuery.java:127)
                      at io.reactivex.internal.disposables.CancellableDisposable.dispose(CancellableDisposable.java:49)
                      at io.reactivex.internal.operators.single.SingleCreate$Emitter.onSuccess(SingleCreate.java:72)
                      at io.objectbox.rx.RxQuery$5$1.onData(RxQuery.java:120)
                      at io.objectbox.rx.RxQuery$5$1.onData(RxQuery.java:116)
                      at io.objectbox.query.QueryPublisher$2.run(QueryPublisher.java:84)
                      at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:457)
                      at java.util.concurrent.FutureTask.run(FutureTask.java:266)

The request looks like this:

        Query<Item> query = itemBox.query()
                .build();

        return RxQuery.single(query)
                .toObservable()
                .flatMap(items -> Observable.fromIterable(items));

Publish to jcenter

TL;DR: implementation 'io.objectbox:objectbox-rxjava:0.9.8' does not work

Can you guys update it on which version is available? Thanks!

'io.objectbox.reactive.DataSubscription.cancel()' on a null object reference

           java.lang.NullPointerException: Attempt to invoke interface method 'void io.objectbox.reactive.DataSubscription.cancel()' on a null object reference
                  at io.objectbox.query.QueryPublisher.unsubscribe(QueryPublisher.java:89)
                  at io.objectbox.reactive.DataSubscriptionImpl.cancel(DataSubscriptionImpl.java:21)
                  at io.objectbox.rx.RxQuery$5$2.cancel(RxQuery.java:127)
                  at io.reactivex.internal.disposables.CancellableDisposable.dispose(CancellableDisposable.java:49)
                  at io.reactivex.internal.disposables.DisposableHelper.set(DisposableHelper.java:54)
                  at io.reactivex.internal.operators.single.SingleCreate$Emitter.setDisposable(SingleCreate.java:109)
                  at io.reactivex.internal.operators.single.SingleCreate$Emitter.setCancellable(SingleCreate.java:114)
                  at io.objectbox.rx.RxQuery$5.subscribe(RxQuery.java:124)
                  at io.reactivex.internal.operators.single.SingleCreate.subscribeActual(SingleCreate.java:39)
                  at io.reactivex.Single.subscribe(Single.java:2707)
                  at io.reactivex.internal.operators.single.SingleSubscribeOn$SubscribeOnObserver.run(SingleSubscribeOn.java:89)
                  at io.reactivex.Scheduler$DisposeTask.run(Scheduler.java:452)
                  at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:61)
                  at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:52)
                  at java.util.concurrent.FutureTask.run(FutureTask.java:237)
                  at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:272)
                  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1133)
                  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:607)
                  at java.lang.Thread.run(Thread.java:761)

@Override
public synchronized void unsubscribe(DataObserver<List<T>> observer, @Nullable Object param) {
    DataPublisherUtils.removeObserverFromCopyOnWriteSet(observers, observer);
    if (observers.isEmpty()) {
        objectClassSubscription.cancel();  // this line
        objectClassSubscription = null;
    }
}

subscribe method in QueryPublisher never call  when i use Single , so objectClassSubscription  be null

override fun get(): Single<List<Airport>> = RxQuery.single(box.query()
        .equal(Airport_.deleted, 0)
        .build())

when using Observable instead of Single， all be right

    if(single) {
        if(onlyChanges) {
            throw new IllegalStateException("Illegal combination of single() and onlyChanges()");
        }
        publisher.publishSingle(observer, publisherParam);
    } else {
        publisher.subscribe(observer, publisherParam);
        if(!onlyChanges) {
            publisher.publishSingle(observer, publisherParam);
        }
    }