Skip to content

Commit 271903e

Browse files
greenrobot-teamgreenrobot-team
committed
Migrate to RxJava 3, update mockito [2.28.2->3.3.3] as Java 8 now available.
1 parent 4f1a176 commit 271903e

File tree

6 files changed

+50
-103
lines changed

6 files changed

+50
-103
lines changed

‎build.gradle‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ buildscript {
2020
ob_native_dep = "io.objectbox:objectbox-$objectboxPlatform:$nativeVersion"
2121

2222
junit_version = '4.13'
23+
mockito_version = '3.3.3'
2324

2425
println "version=$ob_version"
2526
println "objectboxNativeDependency=$ob_native_dep"

‎objectbox-rxjava/build.gradle‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,8 +8,7 @@ dependencies {
88
compile 'io.reactivex.rxjava2:rxjava:2.2.18'
99

1010
testCompile "junit:junit:$junit_version"
11-
// Mockito 3.x requires Java 8.
12-
testCompile 'org.mockito:mockito-core:2.28.2'
11+
testCompile "org.mockito:mockito-core:$mockito_version"
1312
}
1413

1514
task javadocJar(type: Jar, dependsOn: javadoc) {

‎objectbox-rxjava3/build.gradle‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -5,11 +5,10 @@ targetCompatibility = JavaVersion.VERSION_1_8
55

66
dependencies {
77
compile project(':objectbox-java')
8-
compile 'io.reactivex.rxjava2:rxjava:2.2.18'
8+
compile 'io.reactivex.rxjava3:rxjava:3.0.1'
99

1010
testCompile "junit:junit:$junit_version"
11-
// Mockito 3.x requires Java 8.
12-
testCompile 'org.mockito:mockito-core:2.28.2'
11+
testCompile "org.mockito:mockito-core:$mockito_version"
1312
}
1413

1514
task javadocJar(type: Jar, dependsOn: javadoc) {

‎objectbox-rxjava3/src/main/java/io/objectbox/rx/RxBoxStore.java‎

Lines changed: 8 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -17,12 +17,8 @@
1717
package io.objectbox.rx;
1818

1919
import io.objectbox.BoxStore;
20-
import io.objectbox.reactive.DataObserver;
2120
import io.objectbox.reactive.DataSubscription;
22-
import io.reactivex.Observable;
23-
import io.reactivex.ObservableEmitter;
24-
import io.reactivex.ObservableOnSubscribe;
25-
import io.reactivex.functions.Cancellable;
21+
import io.reactivex.rxjava3.core.Observable;
2622

2723
/**
2824
* Static methods to Rx-ify ObjectBox queries.
@@ -33,24 +29,13 @@ public abstract class RxBoxStore {
3329
* Once a transaction is committed, you will get info on classes with changed Objects.
3430
*/
3531
public static <T> Observable<Class> observable(final BoxStore boxStore) {
36-
return Observable.create(new ObservableOnSubscribe<Class>() {
37-
@Override
38-
public void subscribe(final ObservableEmitter<Class> emitter) throws Exception {
39-
final DataSubscription dataSubscription = boxStore.subscribe().observer(new DataObserver<Class>() {
40-
@Override
41-
public void onData(Class data) {
42-
if (!emitter.isDisposed()) {
43-
emitter.onNext(data);
44-
}
45-
}
46-
});
47-
emitter.setCancellable(new Cancellable() {
48-
@Override
49-
public void cancel() throws Exception {
50-
dataSubscription.cancel();
51-
}
52-
});
53-
}
32+
return Observable.create(emitter -> {
33+
final DataSubscription dataSubscription = boxStore.subscribe().observer(data -> {
34+
if (!emitter.isDisposed()) {
35+
emitter.onNext(data);
36+
}
37+
});
38+
emitter.setCancellable(dataSubscription::cancel);
5439
});
5540
}
5641

‎objectbox-rxjava3/src/main/java/io/objectbox/rx/RxQuery.java‎

Lines changed: 29 additions & 67 deletions
Original file line numberDiff line numberDiff line change
@@ -19,19 +19,12 @@
1919
import java.util.List;
2020

2121
import io.objectbox.query.Query;
22-
import io.objectbox.reactive.DataObserver;
2322
import io.objectbox.reactive.DataSubscription;
24-
import io.reactivex.BackpressureStrategy;
25-
import io.reactivex.Flowable;
26-
import io.reactivex.FlowableEmitter;
27-
import io.reactivex.FlowableOnSubscribe;
28-
import io.reactivex.Observable;
29-
import io.reactivex.ObservableEmitter;
30-
import io.reactivex.ObservableOnSubscribe;
31-
import io.reactivex.Single;
32-
import io.reactivex.SingleEmitter;
33-
import io.reactivex.SingleOnSubscribe;
34-
import io.reactivex.functions.Cancellable;
23+
import io.reactivex.rxjava3.core.BackpressureStrategy;
24+
import io.reactivex.rxjava3.core.Flowable;
25+
import io.reactivex.rxjava3.core.FlowableEmitter;
26+
import io.reactivex.rxjava3.core.Observable;
27+
import io.reactivex.rxjava3.core.Single;
3528

3629
/**
3730
* Static methods to Rx-ify ObjectBox queries.
@@ -50,37 +43,23 @@ public static <T> Flowable<T> flowableOneByOne(final Query<T> query) {
5043
* Uses given BackpressureStrategy.
5144
*/
5245
public static <T> Flowable<T> flowableOneByOne(final Query<T> query, BackpressureStrategy strategy) {
53-
return Flowable.create(new FlowableOnSubscribe<T>() {
54-
@Override
55-
public void subscribe(final FlowableEmitter<T> emitter) throws Exception {
56-
createListItemEmitter(query, emitter);
57-
}
58-
59-
}, strategy);
46+
return Flowable.create(emitter -> createListItemEmitter(query, emitter), strategy);
6047
}
6148

6249
static <T> void createListItemEmitter(final Query<T> query, final FlowableEmitter<T> emitter) {
63-
final DataSubscription dataSubscription = query.subscribe().observer(new DataObserver<List<T>>() {
64-
@Override
65-
public void onData(List<T> data) {
66-
for (T datum : data) {
67-
if (emitter.isCancelled()) {
68-
return;
69-
} else {
70-
emitter.onNext(datum);
71-
}
72-
}
73-
if (!emitter.isCancelled()) {
74-
emitter.onComplete();
50+
final DataSubscription dataSubscription = query.subscribe().observer(data -> {
51+
for (T datum : data) {
52+
if (emitter.isCancelled()) {
53+
return;
54+
} else {
55+
emitter.onNext(datum);
7556
}
7657
}
77-
});
78-
emitter.setCancellable(new Cancellable() {
79-
@Override
80-
public void cancel() throws Exception {
81-
dataSubscription.cancel();
58+
if (!emitter.isCancelled()) {
59+
emitter.onComplete();
8260
}
8361
});
62+
emitter.setCancellable(dataSubscription::cancel);
8463
}
8564

8665
/**
@@ -89,44 +68,27 @@ public void cancel() throws Exception {
8968
* (see {@link Query#subscribe()} for details).
9069
*/
9170
public static <T> Observable<List<T>> observable(final Query<T> query) {
92-
return Observable.create(new ObservableOnSubscribe<List<T>>() {
93-
@Override
94-
public void subscribe(final ObservableEmitter<List<T>> emitter) throws Exception {
95-
final DataSubscription dataSubscription = query.subscribe().observer(new DataObserver<List<T>>() {
96-
@Override
97-
public void onData(List<T> data) {
98-
if (!emitter.isDisposed()) {
99-
emitter.onNext(data);
100-
}
101-
}
102-
});
103-
emitter.setCancellable(new Cancellable() {
104-
@Override
105-
public void cancel() throws Exception {
106-
dataSubscription.cancel();
107-
}
108-
});
109-
}
71+
return Observable.create(emitter -> {
72+
final DataSubscription dataSubscription = query.subscribe().observer(data -> {
73+
if (!emitter.isDisposed()) {
74+
emitter.onNext(data);
75+
}
76+
});
77+
emitter.setCancellable(dataSubscription::cancel);
11078
});
11179
}
11280

11381
/**
11482
* The returned Single emits one Query result as a List.
11583
*/
11684
public static <T> Single<List<T>> single(final Query<T> query) {
117-
return Single.create(new SingleOnSubscribe<List<T>>() {
118-
@Override
119-
public void subscribe(final SingleEmitter<List<T>> emitter) throws Exception {
120-
query.subscribe().single().observer(new DataObserver<List<T>>() {
121-
@Override
122-
public void onData(List<T> data) {
123-
if (!emitter.isDisposed()) {
124-
emitter.onSuccess(data);
125-
}
126-
}
127-
});
128-
// no need to cancel, single never subscribes
129-
}
85+
return Single.create(emitter -> {
86+
query.subscribe().single().observer(data -> {
87+
if (!emitter.isDisposed()) {
88+
emitter.onSuccess(data);
89+
}
90+
});
91+
// no need to cancel, single never subscribes
13092
});
13193
}
13294
}

‎objectbox-rxjava3/src/test/java/io/objectbox/rx/QueryObserverTest.java‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -31,14 +31,15 @@
3131

3232
import io.objectbox.query.FakeQueryPublisher;
3333
import io.objectbox.query.MockQuery;
34-
import io.reactivex.Flowable;
35-
import io.reactivex.Observable;
36-
import io.reactivex.Observer;
37-
import io.reactivex.Single;
38-
import io.reactivex.SingleObserver;
39-
import io.reactivex.annotations.NonNull;
40-
import io.reactivex.disposables.Disposable;
41-
import io.reactivex.functions.Consumer;
34+
import io.reactivex.rxjava3.annotations.NonNull;
35+
import io.reactivex.rxjava3.core.Flowable;
36+
import io.reactivex.rxjava3.core.Observable;
37+
import io.reactivex.rxjava3.core.Observer;
38+
import io.reactivex.rxjava3.core.Single;
39+
import io.reactivex.rxjava3.core.SingleObserver;
40+
import io.reactivex.rxjava3.disposables.Disposable;
41+
import io.reactivex.rxjava3.functions.Consumer;
42+
4243

4344
import static org.junit.Assert.*;
4445

0 commit comments

Comments
 (0)