lib/src/main/java/edu/utexas/tacc/tapis/files/lib/services/ChildTaskTransferService.java
public Flux<TransferTaskChild> runPipeline() {
        // Deserialize the message so that we can pass that into the reactor chain.
        return transfersService.streamChildMessages()
            .groupBy((m)-> {
                try {
                    return this.childTaskGrouper(m);
                } catch (IOException ex) {
                    return Mono.empty();
                }
            })
            .flatMap(group -> {
                Scheduler scheduler = Schedulers.newBoundedElastic(5, 100, "ChildPool:" + group.key());
                return group
                    .flatMap(
                        //We need the message in scope so we can ack/nack it later
                        m -> deserializeChildMessage(m)
                            .publishOn(scheduler)
                            .flatMap(t1 -> Mono.fromCallable(() -> chevronOne(t1))
                                .retryWhen(
                                    Retry.backoff(MAX_RETRIES, Duration.ofMillis(10))
                                        .maxBackoff(Duration.ofSeconds(10))
                                        .scheduler(scheduler)
                                ).onErrorResume(e -> doErrorChevronOne(m, e, t1))
                            )
                            .flatMap(t2 -> Mono.fromCallable(() -> doTransfer(t2))
                                .retryWhen(
                                    Retry.backoff( 10, Duration.ofMillis(100))
                                        .maxBackoff(Duration.ofSeconds(120))
                                        .scheduler(scheduler)
                                        .doBeforeRetry(signal-> log.error("RETRY", signal.failure()))
                                        .filter(e -> e.getClass().equals(IOException.class))
                                )
                                .onErrorResume(e -> doErrorChevronOne(m, e, t2))
                            )
                            .flatMap(t3 -> Mono.fromCallable(() -> chevronThree(t3))
                                .retryWhen(
                                    Retry.backoff(MAX_RETRIES, Duration.ofMillis(10))
                                        .maxBackoff(Duration.ofSeconds(10))
                                        .scheduler(scheduler)
                                ).onErrorResume(e -> doErrorChevronOne(m, e, t3))
                            )
                            .flatMap(t4 -> Mono.fromCallable(() -> chevronFour(t4))
                                .retryWhen(
                                    Retry.backoff(MAX_RETRIES, Duration.ofMillis(10))
                                        .maxBackoff(Duration.ofSeconds(10))
                                        .scheduler(scheduler)
                                ).onErrorResume(e -> doErrorChevronOne(m, e, t4))
                            )
                            .flatMap(t5 -> {
                                m.ack();
                                return Mono.just(t5);
                            })
                            .flatMap(t6 -> Mono.fromCallable(() -> chevronFive(t6)))
                    );
            });
    }
/**
     * This is the main processing workflow. Starting with the raw message stream created by streamChildMessages(),
     * we walk through the transfer process. A Flux<TransferTaskChild> is returned and can be subscribed to later
     * for further processing / notifications / logging
     *
     * @return a Flux of TransferTaskChild
     */
This is the main processing workflow. Starting with the raw message stream created by streamChildMessages(),
we walk through the transfer process. A Flux is returned and can be subscribed to later
for further processing / notifications / logging

@return a Flux of TransferTaskChild
This is the main processing workflow . Starting with the raw message stream created by streamChildMessages () we walk through the transfer process . A Flux is returned and can be subscribed to later for further processing / notifications / logging @return a Flux of TransferTaskChild
public Flux<TransferTaskChild> runPipeline() {
        return transfersService.streamChildMessages()
            .groupBy((m)-> {
                try {
                    return this.childTaskGrouper(m);
                } catch (IOException ex) {
                    return Mono.empty();
                }
            })
            .flatMap(group -> {
                Scheduler scheduler = Schedulers.newBoundedElastic(5, 100, "ChildPool:" + group.key());
                return group
                    .flatMap(
                        m -> deserializeChildMessage(m)
                            .publishOn(scheduler)
                            .flatMap(t1 -> Mono.fromCallable(() -> chevronOne(t1))
                                .retryWhen(
                                    Retry.backoff(MAX_RETRIES, Duration.ofMillis(10))
                                        .maxBackoff(Duration.ofSeconds(10))
                                        .scheduler(scheduler)
                                ).onErrorResume(e -> doErrorChevronOne(m, e, t1))
                            )
                            .flatMap(t2 -> Mono.fromCallable(() -> doTransfer(t2))
                                .retryWhen(
                                    Retry.backoff( 10, Duration.ofMillis(100))
                                        .maxBackoff(Duration.ofSeconds(120))
                                        .scheduler(scheduler)
                                        .doBeforeRetry(signal-> log.error("RETRY", signal.failure()))
                                        .filter(e -> e.getClass().equals(IOException.class))
                                )
                                .onErrorResume(e -> doErrorChevronOne(m, e, t2))
                            )
                            .flatMap(t3 -> Mono.fromCallable(() -> chevronThree(t3))
                                .retryWhen(
                                    Retry.backoff(MAX_RETRIES, Duration.ofMillis(10))
                                        .maxBackoff(Duration.ofSeconds(10))
                                        .scheduler(scheduler)
                                ).onErrorResume(e -> doErrorChevronOne(m, e, t3))
                            )
                            .flatMap(t4 -> Mono.fromCallable(() -> chevronFour(t4))
                                .retryWhen(
                                    Retry.backoff(MAX_RETRIES, Duration.ofMillis(10))
                                        .maxBackoff(Duration.ofSeconds(10))
                                        .scheduler(scheduler)
                                ).onErrorResume(e -> doErrorChevronOne(m, e, t4))
                            )
                            .flatMap(t5 -> {
                                m.ack();
                                return Mono.just(t5);
                            })
                            .flatMap(t6 -> Mono.fromCallable(() -> chevronFive(t6)))
                    );
            });
    }
public Flux < TransferTaskChild > runPipeline ( ) { return transfersService . streamChildMessages ( ) . groupBy ( ( m ) -> { try { return this . childTaskGrouper ( m ) ; } catch ( IOException ex ) { return Mono . empty ( ) ; } } ) . flatMap ( group -> { Scheduler scheduler = Schedulers . newBoundedElastic ( 5 , 100 , "ChildPool:" + group . key ( ) ) ; return group . flatMap ( m -> deserializeChildMessage ( m ) . publishOn ( scheduler ) . flatMap ( t1 -> Mono . fromCallable ( ( ) -> chevronOne ( t1 ) ) . retryWhen ( Retry . backoff ( MAX_RETRIES , Duration . ofMillis ( 10 ) ) . maxBackoff ( Duration . ofSeconds ( 10 ) ) . scheduler ( scheduler ) ) . onErrorResume ( e -> doErrorChevronOne ( m , e , t1 ) ) ) . flatMap ( t2 -> Mono . fromCallable ( ( ) -> doTransfer ( t2 ) ) . retryWhen ( Retry . backoff ( 10 , Duration . ofMillis ( 100 ) ) . maxBackoff ( Duration . ofSeconds ( 120 ) ) . scheduler ( scheduler ) . doBeforeRetry ( signal -> log . error ( "RETRY" , signal . failure ( ) ) ) . filter ( e -> e . getClass ( ) . equals ( IOException . class ) ) ) . onErrorResume ( e -> doErrorChevronOne ( m , e , t2 ) ) ) . flatMap ( t3 -> Mono . fromCallable ( ( ) -> chevronThree ( t3 ) ) . retryWhen ( Retry . backoff ( MAX_RETRIES , Duration . ofMillis ( 10 ) ) . maxBackoff ( Duration . ofSeconds ( 10 ) ) . scheduler ( scheduler ) ) . onErrorResume ( e -> doErrorChevronOne ( m , e , t3 ) ) ) . flatMap ( t4 -> Mono . fromCallable ( ( ) -> chevronFour ( t4 ) ) . retryWhen ( Retry . backoff ( MAX_RETRIES , Duration . ofMillis ( 10 ) ) . maxBackoff ( Duration . ofSeconds ( 10 ) ) . scheduler ( scheduler ) ) . onErrorResume ( e -> doErrorChevronOne ( m , e , t4 ) ) ) . flatMap ( t5 -> { m . ack ( ) ; return Mono . just ( t5 ) ; } ) . flatMap ( t6 -> Mono . fromCallable ( ( ) -> chevronFive ( t6 ) ) ) ) ; } ) ; }
This is the main processing workflow.
This is the main processing workflow .
// Deserialize the message so that we can pass that into the reactor chain. //We need the message in scope so we can ack/nack it later