Feature Request: Add unfold construct to Flowable/Observable

Ouverte
  • Quelque chose ne va pas ?
#7,422 17 commentaires 0 réactions 0 personnes assignées Voir sur GitHub

Les mainteneurs répondent en général sous 1 jour

Personne n'a encore pris cette issue.

Évaluation

Difficulté
5/5
Temps estimé
Plus d'une semaine
Accessibilité débutants
35/100
Type d'issue
Fonctionnalité
Clarté
Plutôt claire
Activité
À l'abandon
Stack technique
java
Domaine
backend-api-design

Piste de recherche

Commencez par lire les points d’entrée existants Flowable.create, Flowable.generate et Flowable.fromFuture, puis comparez leur comportement en matière de backpressure et leur comportement asynchrone. Examinez les propositions Flowable.unfold, Flowable.unfoldMaybe et leurs équivalents Observable, puis utilisez les exemples de range fournis pour définir la complétion, la demande et le comportement asynchrone avant d’envisager l’implémentation et les tests.

Rédigé par le modèle d'indexation à partir du texte de l'issue.

Description

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)
  }
}
Langage dominant
Java
Étoiles
48.2k
Forks
7.6k
Merge moyen
1 h 46 min
PR mergées (30 j)
25

Préparer son environnement

Par où commencer

  1. Lisez l'issue en entier, puis le guide de contribution du projet.
  2. Signalez en commentaire que vous la prenez — cela évite que deux personnes fassent le même travail.
  3. Forkez le dépôt et travaillez sur une branche.
  4. Ouvrez une pull request qui référence le numéro de l'issue.

Autres issues de ReactiveX/RxJava

Toutes les issues de ReactiveX/RxJava

Issues similaires

Plus d'issues Java

Recevez les nouvelles issues par e-mail

Un résumé court des issues GitHub adaptées aux débutants.