4

我知道这违反了很多 Rx 规则，但我真的很喜欢RxJava-JDBC，我的队友也很喜欢。关系数据库是我们工作的核心，Rx 也是如此。

但是，在某些情况下，我们不想作为 an 发出，Observable<ResultSet>而是希望只使用基于拉取的 Java 8Stream<ResultSet>或 Kotlin Sequence<ResultSet>。但是我们非常习惯 RxJava-JDBC 库，它只返回一个Observable<ResultSet>.

因此，我想知道是否有一种方法可以使用扩展功能将 anObservable<ResultSet>转换为 a Sequence<ResultSet>，而不进行任何中间收集或toBlocking()调用。以下是我目前所拥有的一切，但我现在正在尝试连接基于推拉的系统，我无法缓冲，因为ResultSet每次onNext()调用都是有状态的。这是不可能完成的任务吗？

import rx.Observable
import rx.Subscriber
import java.sql.ResultSet

fun Observable<ResultSet>.asSequence() = object: Iterator<ResultSet>, Subscriber<ResultSet>() {

    private var isComplete = false

    override fun onCompleted() {
        isComplete = true
    }

    override fun onError(e: Throwable?) {
        throw UnsupportedOperationException()
    }

    override fun onNext(rs: ResultSet?) {
        throw UnsupportedOperationException()
    }


    override fun hasNext(): Boolean {
        throw UnsupportedOperationException()
    }

    override fun next(): ResultSet {
        throw UnsupportedOperationException()
    }

}.asSequence()
4

2 回答 2

4

我不确定这是实现您想要的最简单的方法，但您可以尝试此代码。它通过创建一个阻塞队列并将所有事件从该队列发布到该队列来将一个转换Observable为一个。如果没有，则从队列中拉出事件并阻塞。然后它根据接收到的当前事件修改自己的状态。IteratorObservableIterable

class ObservableIterator<T>(
    observable: Observable<T>,
    scheduler: Scheduler
) : Iterator<T>, Closeable {

  private val queue = LinkedBlockingQueue<Notification<T>>()
  private var cached: Notification<T>? = null
  private var completed: Boolean = false

  private val subscription =
      observable
          .materialize()
          .subscribeOn(scheduler)
          .subscribe({ queue.put(it) })

  override fun hasNext(): Boolean {
    cacheNext()
    return !completed
  }

  override fun next(): T {
    cacheNext()
    val notification = cached ?: throw NoSuchElementException()
    check(notification.isOnNext)
    cached = null
    return notification.value
  }

  private fun cacheNext() {
    if (completed) {
      return
    }

    if (cached == null) {
      queue.take().let { notification ->
        if (notification.isOnError) {
          completed = true
          throw RuntimeException(notification.throwable)
        } else if (notification.isOnCompleted) {
          completed = true
        } else {
          cached = notification
        }
      }
    }
  }

  override fun close() {
    subscription.unsubscribe()
    completed = true
    cached = null
  }
}
于 2016-05-24T20:35:18.550 回答
3

您可以使用以下辅助函数：

fun <T> Observable<T>.asSequence() = Sequence { toBlocking().getIterator() }

当为迭代器调用返回的序列时，将订阅 observable。

如果一个 observable 在它订阅的同一线程上发出元素（Observable.just例如），它将在有机会返回之前填充迭代器的缓冲区。在这种情况下，您可能需要通过调用来直接订阅不同的线程subscribeOn：

observable.subscribeOn(scheduler).asSequence()

但是，虽然toBlocking().getIterator()不会缓冲所有结果，但如果迭代器没有及时使用它们，它可以缓冲其中一些结果。ResultSet如果下一个到达时 a 以某种方式过期，那可能是一个问题ResultSet。

于 2016-05-25T03:21:51.057 回答