Flux en série, où le deuxième flux est créé à la volée avec la dernière valeur du premier flux ?

L'éditeur PHP Shinichi peut expliquer ce concept de manière succincte et claire lorsqu'il explique "Les flux concaténés, où le deuxième flux est créé instantanément avec la dernière valeur du premier flux". Dans les flux en série, la valeur du premier flux est transmise au deuxième flux, et la valeur du deuxième flux est générée à la volée en fonction de la dernière valeur du premier flux. Ce mécanisme peut être utilisé pour mettre en œuvre un transfert et un traitement dynamiques des données, rendant le déroulement du programme plus flexible et plus efficace. En utilisant rationnellement le flux série, les performances et la maintenabilité du programme peuvent être améliorées, et une meilleure expérience utilisateur peut être fournie.

Contenu de la question

Je soupçonnais qu'il devait s'agir d'un doublon, mais j'ai simplement recherché le mauvais terme sur Google.

J'ai deux flux a et b, mais b ne peut être créé qu'en utilisant la dernière valeur de a.

Je veux créer un flux qui est essentiellement la concaténation de a et b , mais la création de b est différée jusqu'à ce que nous obtenions la dernière valeur de a .

Peut-être que ça ressemble à ceci :

fluxC = fluxA.concatWith(lastA -> createFluxB(lastA))

Workaround

Je ne sais pas s'il existe une fonction dans la bibliothèque qui fait exactement cela.

Cependant, vous pouvez créer un tel opérateur en :

1. Cache la dernière valeur du trafic d'entrée
2. Créez la séquence suivante à l'aide de l'opération de concatérisation standard pour obtenir la dernière valeur du flux de cache.

Remarque : Cela ne devrait pas entraîner trop de surcharge puisque l'opération ne met en cache qu'une valeur à la fois et que la deuxième partie de l'algorithme doit récupérer directement la valeur mise en cache sans déclencher le flux source inverse.

Voici un exemple d'implémentation et de test :

import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;

public class TestConcatLast {

    /**
     * Create a stream that emit all elements from input flux,
     * followed by all items from a flux created by provided function.
     * Input function is triggered using last element of source flux as input.
     *
     * @param source The flux providing elements of the first part of the concatenation.
     * @param createFromLastElement A function that provides the tail of the concatenation from a given element.
     *                              It will be triggered <em>at most once</em> using the last element of input flux.
     * @param errorIfSourceEmpty If true and input stream is empty, the returned flow will trigger an error.
     *                           If false, an empty flux is produced if input is empty.
     */
    public <T> Flux<T> concatLast(Flux<T> source, boolean errorIfSourceEmpty, Function<T, Flux<T>> createFromLastElement) {
        var sourceWithLatestCached = source.cache(1);
        final Mono<T> deferLast = Mono.defer(errorIfSourceEmpty ? sourceWithLatestCached::last : sourceWithLatestCached::next);
        return sourceWithLatestCached.concatWith(
                deferLast.flatMapMany(createFromLastElement)
        );
    }

    @Test
    public void testConcat() {
        var nextExpectedElement = new AtomicInteger(1);
        var elts = Flux.just(1, 2, 3, 4)
                // Check cache works and no element has been fetched back from source
                .doOnNext(i -> {
                    assert nextExpectedElement.compareAndSet(i, i+1);
                });

        var concatenated = concatLast(elts, true, i -> Flux.just(i + 1, i + 2, i + 3));
        StepVerifier.create(concatenated)
                .expectNext(1, 2, 3, 4, 5, 6, 7)
                .verifyComplete();
    }
}

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!