android - 使用 RxJava 和 EventBus 将事件发送回 Activity/fragment

Question

Android Studio 3.2 Canary 8
com.squareup:otto:1.3.8
io.reactivex:rxjava:1.3.7
kotlin 1.2.31

我正在尝试使用 otto EventBus 将事件发送回我的 Activity。

但是，我正在使用 RxJava 执行一些后台工作，并且需要在第一个事件完成后发送事件。然而，在发布事件之后。 Activity 永远不会收到它。

此事件必须在主线程上执行此操作。 RxJava 在 IO 线程上。我不确定这样做的最佳方法是什么:

这是我的 RxJava 和 EventBus 发布的交互器代码

class Interactors(private val eventBus: Bus) {
    fun transmitMessage(): Completable {
        return insertTransmission()
                .andThen(onTransmissionChanged()) /* Send event to the activity */
                .andThen(requestTransmission())
    }

    private fun insertTransmission(): Completable {
        return Completable.fromCallable {
            Thread.sleep(4000)
            System.out.println("insertTransmission doing some long operation")
        }
    }

    private fun requestTransmission(): Completable {
        return Completable.fromCallable {
            Thread.sleep(2000)
            System.out.println("requestTransmission doing some long operation")
        }
    }

    /* Need to send this event back to the activity/fragment */
    private fun onTransmissionChanged(): Completable {
        return Completable.fromCallable {
            System.out.println("onTransmissionChanged send event to activity")
            eventBus.post(TransmissionChanged())
        }
    }
}

Activity :

public class HomeActivity extends AppCompatActivity {
    private Bus eventBus = new Bus();

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_home);

        eventBus.register(this);

        new Interactors(eventBus).transmitMessage()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe();
    }

    @Override
    protected void onDestroy() {
        eventBus.unregister(this);
        super.onDestroy();
    }

    @Subscribe
    public void onTransmissionChangedEvent(TransmissionChanged transmissionChanged) {
        System.out.println("onTransmissionChangedEvent");
    }
}

还有 EventBus 类:

class TransmissionChanged

这是我运行应用程序时的输出:

insertTransmission doing some long operation
onTransmissionChanged

我不确定 eventBus.post(..) 是否阻塞。实际上，这应该在主线程中完成，就像发回 Activity 以在 UI 中执行一些更新一样。

Answer 1

你真的需要混合使用 EventBus 和 RxJava 吗？对我来说，这引入了额外的复杂性，但没有太多好处。您的用例似乎是使用 Rx 流的完美示例，对每个发射做一些工作(在您的情况下，通过 onTransmissionChangedEvent() 更新 UI)。

我会将 transmitMessage() 方法更改为这样的:

fun transmitMessage(): Observable<TransmissionChanged> {
    return Observable.create(ObservableOnSubscribe<TransmissionChanged> { emitter ->
        insertTransmission()

        emitter.onNext(TransmissionChanged())  // perform this after the long running operation from insertTransmission() has finished

        requestTransmission()

        emitter.onComplete()   // after the long running operation from requestTransmission() has finished
    })
}

我猜你需要一些额外的数据来相应地更新你的 UI - 这被封装在 TransmissionChanged 类中 - 包括你需要的任何东西。需要注意的一件事 - 在 RxJava 1 中使用 Observable.create() 是危险的。我不记得这样做的安全方法是什么，并且没有使用 RxJava 1 的项目尝试...但是 Observable 类中有一个工厂方法可以安全地完成这项工作。

使用上述方法，您的 Activity 代码也会变得更简洁。不再需要 Otto，因为您的所有操作都通过单个 Rx 流处理。

@Override
protected void onCreate(Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);
    setContentView(R.layout.activity_home);

    new Interactors()
        .transmitMessage()
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(transmission -> onTransmissionChangedEvent(transmission),
                throwable -> handleError(throwable),
                () -> handleCompletion()
        );
}