时间戳

首页 > 解决方案 > 在发射器响应后按顺序运行多个任务

kotlin - 在发射器响应后按顺序运行多个任务

问题描述

我正在尝试为总是延迟响应的硬件设备创建一个通信控制器。如果我只请求一个值，我可以创建一个Single<ByteArray>并在.subscribe{ ...}.

但是当我请求多个值时，我需要确保第二个请求发生在第一个请求完全关闭之后。

这是我可以用 RxJava 做的事情吗，例如defer？还是我应该自己创建一个队列并使用我的队列手动处理事件序列？

无论如何，我们都在使用 RxJava（我显然对它很陌生），当然也可以将它用于此目的。但这是一个很好的用例吗？

编辑：

我可以使用的代码，但不够通用：

hardware.write(byteArray)
    .subscribe(
        {
            hardware.receiveResult().take(1)
                .doFinally { /* dispose code */ }
                .subscribe(
                    { /* onSuccess */ }
                    { /* onError */ }
                .let { disposable = it }
        },
        { /* onError */ }
    )

队列中下一个请求的所有代码都可以放在 inner 中onSuccess，然后下一个放在 that 中onSuccess。这将按顺序执行，但不够通用。任何其他发出请求的类最终都会破坏我的序列。

我正在寻找一种在硬件通信控制器类中自动建立队列的解决方案。

标签： kotlinrx-java

解决方案

很长一段时间过去了，项目开发了，我们很久以前就有了解决方案。现在我想在这里分享它：

fun writeSequential(data1: ByteArray, data2: ByteArray) {
    disposable = hardwareWrite(data1)
        .zipWith(hardwareWrite(data2))
        .subscribe(
            {
                /* handle results.
                  it.first will be the first response,
                  it.second the second. */
            },
            { /* handle error */ }
        )

    compositeDisposable.add(disposable)
}

fun hardwareWrite(data: ByteArray): Disposable {
    var emitter: SingleEmitter<ByteArray>? = null
    var single = Single.create<ByteArray> { emitter = it }

    return hardware.write(data)
        .subscribe(
            { hardwareRead(emitter) },
            { /* onError */ }
        ))
}

fun hardwareRead(emitter: SingleEmitter<ByteArray>): Disposable {
    return hardware.receiveResult()
        .take(1)
        .timeout( /* your timeout */ )
        .single( /* default value */ )
        .doFinally( /* cleanup queue */ )
        .subscribe(
            { emitter.onSuccess(it) }
            { emitter.onError(it) }
        )
}

解决方案并不完美，现在我看到中间部分对一次性结果没有任何作用。

同样在示例中，它有点复杂，因为 hardwareWrite 不会立即触发而是排队。通过这种方式，我们可以确保硬件是按顺序访问的，并且结果不会混淆。

我仍然希望这可能对正在寻找解决方案的人有所帮助，并且可能是 kotlin 和/或 RxJava 的新手（就像我在项目开始时一样）。

推荐阅读