Espresso 等待 [RxIdlingResource] 空闲超时

Question

我是 Espresso 测试的新手。在我现有的应用程序中，我们使用 RxAndroid 来做一些网络。我们使用 RxBus 与应用程序的某些部分进行通信，否则这些部分看起来 "impossible".

我们导入了实现 IdlingResource 的 RxEspresso，因此我们可以使用我们的 RxAndroid 网络调用。

不幸的是 RxEspresso 不允许 RxBus 工作，因为它是一个 "hot observable" 并且永远不会关闭。所以它抛出 android.support.test.espresso.IdlingResourceTimeoutException: Wait for [RxIdlingResource] to become idle timed out

我制作了一个小型 Android 应用程序来证明我的观点。它有两个活动。第一个显示在 RecyclerView 中启动时通过网络调用检索到的一些项目。

点击它时，它会通过 RxBus 进行通信（我知道这有点矫枉过正，但纯粹是为了证明这一点）。 DetailActivity 然后显示数据。

我们如何编辑 RxEspresso 以便它与我们的 RxBus 一起工作？

RxIdlingResource 也检查 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();
        }
    }
}

接收总线

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();
    }
}

主要活动

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);
    }

Answer 1

由于我们没有得到这个问题的更好答案，我们假设 objects 通过 RxBus 发送是立即的，不需要计入 subscriptions.incrementAndGet();

我们简单地过滤掉了这一行之前的对象。在我们的例子中，对象是 class SerializedSubject 和 PublishSubject.

这是我们改变的方法。

@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;
}

Espresso 等待 [RxIdlingResource] 空闲超时

Espresso Wait for [RxIdlingResource] to become idle timed out

android

android-testing

android-espresso

rx-android