android - RxJava 同时删除操作

Question

我在尝试以 react 方式理解对同一个可观察对象的同时操作应该如何工作时遇到了问题。

场景如下:

我有一个用户列表和一个删除按钮。 每次我按下 remove 时，我都会调用 API:UsersApi.removeUser。可以同时删除多个用户。这意味着多个 UsersApi.removeUser 同时发生。

在每个 UsersApi.removeUser 之后，我需要调用一个 UsersApi.refreshUser

所以就伪代码而言，我在单击删除时所做的如下:

主持人:

public Observable<User> removeUser(int userId) {
        return UsersApi.removeUser(userId)
                .flatMap(user -> UsersApi.refreshUser(userId));
    }

fragment :

public void removeUser() {
   presenter.removeUser(userId)
     .subscribe(user -> {
        //remove user from ui
        // update number of total users
   })
}

这种方法的问题在于，由于删除的异步性质(允许多次删除)，我无法保证到达订阅的是最新的。订阅将达到两次，每次删除一个，用户信息可能不会更新或最新。这有意义吗？

我想要发生的事情:

1. 使用响应式(Reactive)方法并行/同时删除调用(由用户的多次删除点击触发)
2. 一个remove调用结束后，开始下一个remove调用

编辑:我想知道的是如何/如果可以使用 Rx 运算符来完成我所做的解决方案(请参阅 edit2)。

Edit2:我的解决方案是在 UsersApi.refreshUser(userId) 时使用 PublishSubject 将用户操作排队(在本例中为 remove)并发出通话结束。

基本上我所做的是(伪代码):

private final PublishSubject<UserOperation> userOperationObs;
private final ConcurrentLinkedQueue<UserOperation> pendingOperations;
private boolean executingOperation;

private void emitUserOperation(final UserOperation operation) {
      if (!executingOperation) {
            executingOperation = true;
            userOperationObs.onNext(operation);
        } else {
            executingOperation.add(operation);
        }
    }

public Observable<User> removeUser(UserOperation operation) {
        return UsersApi.removeUser(operation.getUserId)
                .switchMap(user -> UsersApi.refreshUser(operation.getUserId))
                .doOnNext(user -> {
                   executingOperation = false;
                   final UserOperation nextOperation = pendingOperations.poll();
                   if (nextOperation != null) {
                       userOperationObs.onNext(operation);
                   }
};
    }

Answer 1

您可以将您的 UI 点击变成 Observable(例如通过使用 RxBinding )。之后，您可以使用 concatMap 运算符执行 api 调用，以便在当前 api 调用完成后启动下一个网络调用。

// emit clicks as stream
Observable<?> clicks = RxView.clicks(removeView)

// listen clicks then perform network call in sequence
clicks.concatMap(ignored -> usersApi.refreshUser(userId))