Preference { public Observable<T> asObservable() { return values; } Preference(SharedPreferences preferences, final String key, T defValue, Proxy<T> proxy, Observable<String> keyChanges); T asValue(); Observable<T> asObservable(); Action1<? super T> asAction();  }
@Test public void asObservable() { Preference<String> preference = rxPreferences.getString("key1", "defValue"); TestSubscriber<String> o = new TestSubscriber<>(); Subscription subscription = preference.asObservable().subscribe(o); o.assertValues("defValue"); rxPreferences.putString("key1", "value1"); o.assertValues("defValue", "value1"); rxPreferences.remove("key1"); o.assertValues("defValue", "value1", "defValue"); subscription.unsubscribe(); rxPreferences.putString("key1", "foo"); o.assertValues("defValue", "value1", "defValue"); }
@Test public void asObservableHonorsBackpressure() { Preference<String> preference = rxPreferences.getString("key1", "defValue"); TestSubscriber<String> o = new TestSubscriber<>(2); preference.asObservable().subscribe(o); o.assertValues("defValue"); rxPreferences.putString("key1", "value1"); o.assertValues("defValue", "value1"); rxPreferences.putString("key1", "value2"); o.assertValues("defValue", "value1"); o.requestMore(1); o.assertValues("defValue", "value1", "value2"); for (int i = 0; i < 1000; i++) { rxPreferences.putString("key1", "value" + i); } o.assertValues("defValue", "value1", "value2"); o.requestMore(Long.MAX_VALUE); o.assertValues("defValue", "value1", "value2", "value999"); }