0

基本的に、私は2つFlowables Fあり、それらでG使用したいのですが、完了した場合（まだ実行中であっても）combineLatest、結合Flowableされたものがすでに完了していることを望みます。FG

これは、醜いソリューションで何を達成するかの例です。

fun combineFandGbutTerminateIfFTerminates(F: Flowable<Int>, G: Flowable<Int>) : Flowable<Pair<Int, Int>> {
  val _F = F.share()
  val _G = G.takeUntil(_F.ignoreElements().toFlowable<Nothing>())
  val FandG = Flowables.combineLatest(_F, _G)
  return FandG
}

それを拡張関数に抽出できます。

fun<T> Flowable<T>.completeWith(other: Flowable<*>) : Flowable<T> {
    return takeUntil(other.ignoreElements().toFlowable<Nothing>())
}

それを表現するより良い方法はありますか？

4

1 に答える 1

0

次の解決策にたどり着きました。1 つのマスターを多くのスレーブ ソースと組み合わせることができます。マスターが完了すると、組み合わせFlowableが完了します。ただし、スレーブがマスターより先に完了すると、エラーSlaveCompletedPrematurelyErrorが伝搬されます。

class SlaveCompletedPrematurelyError(message: String) : Throwable(message)

/**
 * Combine this Flowable with one slave source.
 */
@CheckReturnValue
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
fun <T, T1, R> Flowable<T>.combineLatestSlaves(
  slaveSource: Flowable<T1>,
  combineFunction: (T, T1) -> R
): Flowable<R> = combineLatestSlaves(Functions.toFunction(combineFunction), slaveSource)

/**
 * Combine this Flowable with two slave sources.
 */
@CheckReturnValue
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
fun <T, T1, T2, R> Flowable<T>.combineLatestSlaves(
  slaveSource1: Flowable<T1>,
  slaveSource2: Flowable<T2>,
  combineFunction: (T, T1, T2) -> R
) =
  combineLatestSlaves(Functions.toFunction(combineFunction), slaveSource1, slaveSource2)

/**
 * Combine this Flowable with three slave sources.
 */
@CheckReturnValue
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
fun <T, T1, T2, T3, R> Flowable<T>.combineLatestSlaves(
  slaveSource1: Flowable<T1>,
  slaveSource2: Flowable<T2>,
  slaveSource3: Flowable<T3>,
  combineFunction: (T, T1, T2, T3) -> R
) =
  combineLatestSlaves(Functions.toFunction(combineFunction), slaveSource1, slaveSource2, slaveSource3)

/**
 * Combine this Flowable with many slave sources.
 */
@SchedulerSupport(SchedulerSupport.NONE)
@CheckReturnValue
@BackpressureSupport(BackpressureKind.FULL)
fun <T : U, U, R> Flowable<T>.combineLatestSlaves(
  combiner: Function<in Array<Any>, out R>,
  vararg slaveSources: Publisher<out U>
): Flowable<R> =
  combineLatestSlaves(slaveSources, combiner, bufferSize())

/**
 * Combine this Flowable with many slave sources.
 *
 * This function is identical of using combineLatest with this and the slave sources except with the following changes:
 * - If this Flowable completes, the resulting Flowable completes even if the slave sources are still running.
 * - If a slave source completes before this Flowable, a SlaveCompletedPrematurelyError error is triggered.
 */
@SchedulerSupport(SchedulerSupport.NONE)
@CheckReturnValue
@BackpressureSupport(BackpressureKind.FULL)
fun <T : U, U, R> Flowable<T>.combineLatestSlaves(
  slaveSources: Array<out Publisher<out U>>,
  combiner: Function<in Array<Any>, out R>,
  bufferSize: Int
): Flowable<R> {
  val masterCompleted = Throwable()

  val sources = Array<Publisher<out U>>(slaveSources.size + 1) {
    when (it) {
      0 -> Flowable.error<U>(masterCompleted).startWith(this)
      else -> Flowable.error<U> { SlaveCompletedPrematurelyError(slaveSources[it - 1].toString()) }.startWith(
        slaveSources[it - 1]
      )
    }
  }

  return combineLatest(sources, combiner, bufferSize).onErrorComplete { it == masterCompleted }
}

/**
 * Errors encountered in the stream for which the provided `predicate` returns true will be silently turned into graceful completion.
 */
@CheckReturnValue
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
inline fun <T> Flowable<T>.onErrorComplete(crossinline predicate: (Throwable) -> Boolean): Flowable<T> =
  onErrorResumeNext { error: Throwable ->
    if (predicate(error)) Flowable.empty<T>() else Flowable.error<T>(
      error
    )
  }
于 2019-04-08T11:31:02.970 に答える