1

我正在开发一个应用程序，显示它从互联网上获取的项目列表。我有 2 个按钮 loadMore 和刷新，loadMore - 加载下一批项目，刷新 - 从头开始​​加载项目。

我正在使用 MVI（模型视图意图）模式。为了简单起见，我使用数字列表创建了一个示例，每个数字将代表一批项目：

    val loadSubject = BehaviorSubject.create<Unit>()
    val refreshSubject = PublishSubject.create<Unit>()

    val list = loadSubject.scanWith(
            { Observable.just(emptyList<Int>()) },
            { listObservable, _ ->
                listObservable
                        .map { it + ++count }
                        .replay().autoConnect()
            }
    )
            .flatMap { it }
            .filter { it.isNotEmpty() }


    val listSubscription = {
        list.subscribe {
            //do whatever with the list
        }
    }

    refreshSubject.scanWith(
            listSubscription,
            { disposable, _ ->
                disposable.dispose()
                listSubscription()
            }
    ).subscribe()

所以现在它可以完美地工作，但订阅在我的 Intent 中，我需要一个带有 Rx 的方法，它可以做完全相同的事情，但让我的 View 订阅。

我想要得到的是：

假设我的清单是[1,2,3]

加载更多新闻我会得到[1,2,3,4]

在刷新新闻生病得到[5]

4

2 回答 2

0 
val loadSubject = PublishSubject.create<Unit>()
val refreshSubject = PublishSubject.create<Unit>()

val loadMoreSequence = defer {
  loadSubject.scanWith(
    { Observable.just(listOf(0)) },
    { listObservable, _ ->
      listObservable
        .map { it + it.size }
        .replay().autoConnect()
    }
  )
    .flatMap { it }
}

val list = concat(Unit.asObservable(), refreshSubject)
  .switchMap { loadMoreSequence }

// testing

list.subscribe {
  println(it)
}

loadSubject.onNext(Unit)
loadSubject.onNext(Unit)
refreshSubject.onNext(Unit)
loadSubject.onNext(Unit)

结果：

[0]
[0, 1]
[0, 1, 2]
[0]
[0, 1]
于 2018-01-24T20:07:17.517 回答
0

我设法让一些工作：

    val loadSubject = PublishSubject.create<Unit>()
    val refreshSubject = PublishSubject.create<Unit>()

    val loadList: (ItemsProvider) -> Observable<SingleEvent<List<Int>>> = { provider ->
        val markedGet = Observable.fromCallable(provider::fetchItem)
                .markStartAndEnd()

        loadSubject.concatWith(Unit)
                .flatMapWithDrop(markedGet)
                .publish().autoConnect()
    }

    val listEvents = refreshSubject.concatWith(Unit)
            .switchMap { loadList(itemsProvider) }

    val list = refreshSubject.concatWith(Unit)
            .switchMap {
                listEvents.filter { it is SingleEvent.Result }
                        .map { (it as SingleEvent.Result).data }
                        .scan { list: List<Int>, newList: List<Int> ->
                            list + newList
                        }
            }

    val isListEmpty = list.map { it.isEmpty() }

    val isDownloading = listEvents.map { it.isRunning() }

    val isRefreshing =
            Observable.merge(
                    refreshSubject.map { true },
                    isDownloading.filter { !it }.skip(1)
            )

扩展是：

fun <T> Observable<T>.concatWith(item: T): Observable<T> =
        Observable.concat(Observable.just(item), this)

sealed class SingleEvent<T> {
    class Start<T> : SingleEvent<T>()
    data class Result<T>(val data: T) : SingleEvent<T>()
}

fun <T> Observable<T>.markStartAndEnd(): Observable<SingleEvent<T>> {
    return this.map { SingleEvent.Result(it) as SingleEvent<T> }
            .startWith(SingleEvent.Start())
}

/**
* Flatmaps upstream items into [source] items.
* Ignores upstream items if there is any [source] instance is currently running.
*
* upstream ----u-----u---u-------u---------------|-->
*              ↓                 ↓               ↓
* source       ---s-------|->    ---s-------|->  ↓
*                 ↓                 ↓            ↓
* result   -------s-----------------s------------|-->
*/
fun <T, R> Observable<T>.flatMapWithDrop(source: Observable<R>): Observable<R> =
        this.toFlowable(BackpressureStrategy.DROP)
                .flatMap({ source.toFlowable(BackpressureStrategy.MISSING) }, 1)
                .toObservable()

基于 Dmitry Ryadnenko博客文章。

但是我不能让 flatMapWithDrop 工作，因为我在做网络调用我需要丢弃任何传入的刷新请求，直到当前的刷新请求完成。

更新：有没有其他方法可以缓存列表（不再使用refreshSubject.concatWith(Unit).switchMap {}）？

于 2018-01-25T12:05:06.590 回答