44import android .databinding .ObservableField ;
55import android .support .annotation .NonNull ;
66
7+ import rx .AsyncEmitter ;
78import rx .Observable ;
8- import rx .Subscriber ;
9- import rx .functions .Action0 ;
10- import rx .subscriptions .Subscriptions ;
119
1210import static android .databinding .Observable .OnPropertyChangedCallback ;
1311
@@ -17,29 +15,21 @@ private RxUtils() {
1715 }
1816
1917 public static <T > Observable <T > toObservable (@ NonNull final ObservableField <T > observableField ) {
20- return Observable .create (new Observable .OnSubscribe <T >() {
21- @ Override
22- public void call (final Subscriber <? super T > subscriber ) {
23- subscriber .onNext (observableField .get ());
24- 25- final OnPropertyChangedCallback callback = new OnPropertyChangedCallback () {
26- @ Override
27- public void onPropertyChanged (android .databinding .Observable dataBindingObservable , int propertyId ) {
28- if (dataBindingObservable == observableField ) {
29- subscriber .onNext (observableField .get ());
30- }
18+ return Observable .fromEmitter (asyncEmitter -> {
19+ 20+ final OnPropertyChangedCallback callback = new OnPropertyChangedCallback () {
21+ @ Override
22+ public void onPropertyChanged (android .databinding .Observable dataBindingObservable , int propertyId ) {
23+ if (dataBindingObservable == observableField ) {
24+ asyncEmitter .onNext (observableField .get ());
3125 }
32- };
26+ }
27+ };
3328
34- observableField .addOnPropertyChangedCallback (callback );
29+ observableField .addOnPropertyChangedCallback (callback );
3530
36- subscriber .add (Subscriptions .create (new Action0 () {
37- @ Override
38- public void call () {
39- observableField .removeOnPropertyChangedCallback (callback );
40- }
41- }));
42- }
43- });
31+ asyncEmitter .setCancellation (() -> observableField .removeOnPropertyChangedCallback (callback ));
32+ 33+ }, AsyncEmitter .BackpressureMode .LATEST );
4434 }
4535}
0 commit comments