Hacktoberfest 2026：维护者为十月标记出来的 issue，仍然开放、适合新手。 浏览 Hacktoberfest issue

Feature Request: Add unfold construct to Flowable/Observable

未关闭
  • 哪里不对？
#7,422 17 条评论 0 个 reaction 已指派 0 人 在 GitHub 查看

维护者通常 1 天内回复

还没有人认领这个 Issue。

评估

难度
5/5
预计耗时
一周以上
新手友好度
35/100
Issue 类型
功能
描述清晰度
基本清楚
活跃度
停滞
技术栈
java
领域
backend-api-design

调研方向

首先阅读现有的 Flowable.create、Flowable.generate 和 Flowable.fromFuture 入口点，并比较它们的背压和异步行为。查看提议的 Flowable.unfold、Flowable.unfoldMaybe 及其 Observable 对应项，然后使用提供的 range 示例定义完成、需求和异步行为，再考虑实现和测试。

由索引模型根据 Issue 内容生成。

描述

4.x Feature-Request

Version 3.1.4

Scala as of 2.13 and Akka (another reactive framework) have added various unfold patterns, and I'm finding a need for this currently. Here's the use case - pagination of responses to a "cursor"-like REST api.

Essentially, I'd like to make a Flowable that makes requests to get another page - where it starts without a "continuation" parameter, and keeps making requests (honoring backpressure) until the response comes without a "continuation".

Flowable.create doesn't seem to deal with the backpressure well, and Flowable.generate is blocking, which seems disappointing as Flowable.fromFuture is great at dealing with async http clients.

This unfold pattern seems to offer what would be very helpful to the cause - where a new request isn't made until the subscriber asks for more, and the termination state is known from some inspection of the previous element in the feed (the http response, or the json parsing of it).

Here's a stab of making these using RxJava constructs; it would be nice if Flowable.unfold and Flowable.unfoldMaybe (and Observable too) could be added to RxJava itself.

import java.util.Optional;
import java.util.function.Function;

import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.core.Maybe;
import io.reactivex.rxjava3.core.MaybeSource;

public class Unfold {
    public static class Pair<L, R> {
        private final L left;
        private final R right;

        public Pair(L left, R right) {
            this.left = left;
            this.right = right;
        }

        public L getLeft() {
            return left;
        }

        public R getRight() {
            return right;
        }
    }

    public static <S, E> Flowable<E> flowableUnfold(S state, Function<S, Optional<Pair<S, E>>> generator) {
        return Flowable.defer(() -> Flowable.fromOptional(generator.apply(state)))
                .flatMap(pair -> Flowable.just(pair.getRight())
                        .concatWith(flowableUnfold(pair.getLeft(), generator)));
    }

    public static <S, E> Flowable<E> flowableUnfoldMaybe(S state, Function<S, ? extends MaybeSource<Pair<S, E>>> generator) {
        return Flowable.defer(() -> Flowable.fromMaybe(generator.apply(state)))
                .flatMap(pair -> Flowable.just(pair.getRight())
                        .concatWith(flowableUnfoldMaybe(pair.getLeft(), generator)));
    }
}

And some scala to test it

@RunWith(classOf[JUnitRunner])
class UnfoldTest extends AnyFlatSpec with Matchers {
  "flowableUnfold" should "work with range of 0 to 10" in {
    val flowable = Unfold.flowableUnfold[(Int, Int), Int]((0, 10), {
      case (start: Int, end: Int) if start < end =>
        Option(new Pair(start + 1 -> end, start)).asJava
      case _ => None.asJava
    })
    flowable.toList.blockingGet().asScala shouldBe (0 until 10)
  }

  "flowableUnfoldMaybe" should "work with range of 0 to 10" in {
    val flowable = Unfold.flowableUnfoldMaybe[(Int, Int), Int]((0, 10), {
      case (start: Int, end: Int) if start < end =>
        Maybe.fromCallable(() => new Pair(start + 1 -> end, start))
      case _ => Maybe.empty()
    })
    flowable.toList.blockingGet().asScala shouldBe (0 until 10)
  }
}
主要语言
Java
星标
48.2k
派生
7.6k
平均合并
3 小时 2 分钟
30 天内合并 PR
27

环境准备

从这里开始

  1. 先读完整个 Issue，再读项目的贡献指南。
  2. 在 Issue 下留言说明你要接手 —— 这能避免两个人做同样的事。
  3. Fork 仓库，在一个分支上完成修改。
  4. 提交 Pull Request，并在描述里引用这个 Issue 编号。

ReactiveX/RxJava 的其他 Issue

查看 ReactiveX/RxJava 的全部 Issue

相似的 Issue

更多 Java Issue

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。