Issue
I'm new to Espresso testing. In my existing application we are using RxAndroid to do some networking. We use a RxBus to communicate to parts of our application that would otherwise seem "impossible".
We imported RxEspresso which implements IdlingResource
so we could use our RxAndroid network calls.
Unfortunately RxEspresso does not allow RxBus to work since it's a "hot observable" and never closes. So it throws android.support.test.espresso.IdlingResourceTimeoutException: Wait for [RxIdlingResource] to become idle timed out
I made a small Android application demonstrating my point.
It has two activities. The first displays some items retrieved through a network call on startup in a RecyclerView
.
When clicked upon it communicates through the RxBus (I know it's overkill, but purely to demonstrate the point). The DetailActivity
then shows the data.
How can we edit RxEspresso so it will work with our RxBus?
RxIdlingResource also check RxEspresso
/**
* Provides the hooks for both RxJava and Espresso so that Espresso knows when to wait
* until RxJava subscriptions have completed.
*/
public final class RxIdlingResource extends RxJavaObservableExecutionHook implements IdlingResource {
public static final String TAG = "RxIdlingResource";
static LogLevel LOG_LEVEL = NONE;
private final AtomicInteger subscriptions = new AtomicInteger(0);
private static RxIdlingResource INSTANCE;
private ResourceCallback resourceCallback;
private RxIdlingResource() {
//private
}
public static RxIdlingResource get() {
if (INSTANCE == null) {
INSTANCE = new RxIdlingResource();
Espresso.registerIdlingResources(INSTANCE);
}
return INSTANCE;
}
/* ======================== */
/* IdlingResource Overrides */
/* ======================== */
@Override
public String getName() {
return TAG;
}
@Override
public boolean isIdleNow() {
int activeSubscriptionCount = subscriptions.get();
boolean isIdle = activeSubscriptionCount == 0;
if (LOG_LEVEL.atOrAbove(DEBUG)) {
Log.d(TAG, "activeSubscriptionCount: " + activeSubscriptionCount);
Log.d(TAG, "isIdleNow: " + isIdle);
}
return isIdle;
}
@Override
public void registerIdleTransitionCallback(ResourceCallback resourceCallback) {
if (LOG_LEVEL.atOrAbove(DEBUG)) {
Log.d(TAG, "registerIdleTransitionCallback");
}
this.resourceCallback = resourceCallback;
}
/* ======================================= */
/* RxJavaObservableExecutionHook Overrides */
/* ======================================= */
@Override
public <T> Observable.OnSubscribe<T> onSubscribeStart(Observable<? extends T> observableInstance,
final Observable.OnSubscribe<T> onSubscribe) {
int activeSubscriptionCount = subscriptions.incrementAndGet();
if (LOG_LEVEL.atOrAbove(DEBUG)) {
if (LOG_LEVEL.atOrAbove(VERBOSE)) {
Log.d(TAG, onSubscribe + " - onSubscribeStart: " + activeSubscriptionCount, new Throwable());
} else {
Log.d(TAG, onSubscribe + " - onSubscribeStart: " + activeSubscriptionCount);
}
}
onSubscribe.call(new Subscriber<T>() {
@Override
public void onCompleted() {
onFinally(onSubscribe, "onCompleted");
}
@Override
public void onError(Throwable e) {
onFinally(onSubscribe, "onError");
}
@Override
public void onNext(T t) {
//nothing
}
});
return onSubscribe;
}
private <T> void onFinally(Observable.OnSubscribe<T> onSubscribe, final String finalizeCaller) {
int activeSubscriptionCount = subscriptions.decrementAndGet();
if (LOG_LEVEL.atOrAbove(DEBUG)) {
Log.d(TAG, onSubscribe + " - " + finalizeCaller + ": " + activeSubscriptionCount);
}
if (activeSubscriptionCount == 0) {
Log.d(TAG, "onTransitionToIdle");
resourceCallback.onTransitionToIdle();
}
}
}
RxBus
public class RxBus {
//private final PublishSubject<Object> _bus = PublishSubject.create();
// If multiple threads are going to emit events to this
// then it must be made thread-safe like this instead
private final Subject<Object, Object> _bus = new SerializedSubject<>(PublishSubject.create());
public void send(Object o) {
_bus.onNext(o);
}
public Observable<Object> toObserverable() {
return _bus;
}
public boolean hasObservers() {
return _bus.hasObservers();
}
}
MainActivity
public class MainActivity extends AppCompatActivity {
@Bind(R.id.rv)
RecyclerView RV;
private List<NewsItem> newsItems;
private RecyclerViewAdapter adapter;
private Observable<List<NewsItem>> newsItemsObservable;
private CompositeSubscription subscriptions = new CompositeSubscription();
private RxBus rxBus;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
ButterKnife.bind(this);
//Subscribe to RxBus
rxBus = new RxBus();
subscriptions.add(rxBus.toObserverable()
.subscribe(new Action1<Object>() {
@Override
public void call(Object event) {
//2.
NewsItem myClickNewsItem = (NewsItem) event;
startActivity(new Intent(MainActivity.this, DetailActivity.class).putExtra("text", myClickNewsItem.getBodyText()));
}
}));
//Set the adapter
adapter = new RecyclerViewAdapter(this);
//Set onClickListener on the list
ItemClickSupport.addTo(RV).setOnItemClickListener(new ItemClickSupport.OnItemClickListener() {
@Override
public void onItemClicked(RecyclerView recyclerView, int position, View v) {
//Send the clicked item over the RxBus.
//Receives it in 2.
rxBus.send(newsItems.get(position));
}
});
RV.setLayoutManager(new LinearLayoutManager(this));
RestAdapter retrofit = new RestAdapter.Builder()
.setEndpoint("http://URL.com/json")
.build();
ServiceAPI api = retrofit.create(ServiceAPI.class);
newsItemsObservable = api.listNewsItems(); //onComplete goes to setNewsItems
}
@Override
protected void onPostCreate(Bundle savedInstanceState) {
super.onPostCreate(savedInstanceState);
NewsItemObserver observer = new NewsItemObserver(this);
newsItemsObservable.delaySubscription(1, TimeUnit.SECONDS).observeOn(AndroidSchedulers.mainThread()).subscribeOn(Schedulers.io()).subscribe(observer);
}
public void setNewsItems(List<NewsItem> newsItems) {
this.newsItems = newsItems;
adapter.setNewsItems(newsItems);
RV.setAdapter(adapter);
}
Solution
Since we didn't obtain any better answer to this problem we assumed objects
send through the RxBus were immediate and didn't need to be counted in the subscriptions.incrementAndGet();
We simply filtered the objects out before this line. In our case the objects were of the class SerializedSubject
and PublishSubject
.
Here is the method we changed.
@Override
public <T> Observable.OnSubscribe<T> onSubscribeStart(Observable<? extends T> observableInstance, final Observable.OnSubscribe<T> onSubscribe) {
int activeSubscriptionCount = 0;
if (observableInstance instanceof SerializedSubject || observableInstance instanceof PublishSubject) {
Log.d(TAG, "Observable we won't register: " + observableInstance.toString());
} else {
activeSubscriptionCount = subscriptions.incrementAndGet();
}
if (LOG_LEVEL.atOrAbove(DEBUG)) {
if (LOG_LEVEL.atOrAbove(VERBOSE)) {
Log.d(TAG, onSubscribe + " - onSubscribeStart: " + activeSubscriptionCount, new Throwable());
} else {
Log.d(TAG, onSubscribe + " - onSubscribeStart: " + activeSubscriptionCount);
}
}
onSubscribe.call(new Subscriber<T>() {
@Override
public void onCompleted() {
onFinally(onSubscribe, "onCompleted");
}
@Override
public void onError(Throwable e) {
onFinally(onSubscribe, "onError");
}
@Override
public void onNext(T t) {
Log.d(TAG, "onNext:: " + t.toString());
//nothing
}
});
return onSubscribe;
}
Answered By - timr
0 comments:
Post a Comment
Note: Only a member of this blog may post a comment.