chaining reactive functions rxjava

Question

I have this interface that then I create a List of implementation in order to execute sequentially, but this implementations are variable because in future releases we can add or delete several validations.

This is my interface:

import java.util.function.Function;

import io.reactivex.Single;

public interface Validator<T> {

  Single<T> validate(T data);

  default Function<T,Single<T>> reactiveFunction() {
    return (a) -> this.validate(a) ;
  }

  default Single<T> chainFunctions(T data, Function<T,Single<T>>... validators) {
    if (validators.length == 1) {
      return validators[0].apply(data);
    }
    else {
      Single<T> result = Single.just(data);
      for(Function<T,Single<T>> validator: validators) {
        result = result.flatMap(x -> validator.apply(x));
      }
      return result;
    }
  }
  
}

Can I do chainFunctions more reactive and functional like this answer?

@SafeVarargs
private static <T> Function<T, T> combineF(Function<T, T>... funcs) {
    return Arrays.stream(funcs).reduce(Function.identity(), Function::andThen);
}

[Edit with suggestions]

I changed my interface like this:

import java.util.function.Function;

import io.reactivex.Single;

public interface Validator<T> {

  Single<T> validate(final T data);

  default Function<Single<T>,Single<T>> reactiveFunction() {
    return x -> x.flatMap(this::stepValidator) ;
  }

  default Single<T> chainFunctions(final T data, Function<Single<T>,Single<T>>... validators) {
    return Flowable.fromArray(validators)
                   .reduce((s1,s2) -> s1.andThen(s2))
                   .switchIfEmpty(Single.fromCallable(() -> a -> a))
                   .flatMap(f -> f.apply(Single.just(data));

  }
  
}

score 0 · Answer 1 · answered Jul 01 '22 at 06:49

0

Not in RxJava proper.

Note though that result = result.flatMap may result in StackOverflow if you append too many operators.

In general, I'd suggest bringing the functions to the data instead:

Single.defer(() -> {
    AtomicReference<T> ref = new AtomicReference<>(data);

    return Flowable.fromArray(validators)
        .concatMapSingle(func -> {
            return func.apply(ref.get())
                   .doOnSuccess(ref::set);
        })
        .ignoreElements()
        .andThen(Single.fromCallable(ref::get));
});

answered Jul 01 '22 at 06:49

akarnokd

69,132
14
157
192

what about this way? modify reactiveFunction() to ```x -> x.flatMap(this::validate)``` and chainFunction with parameters like this ```T data, Function,Single>... validators)``` ```validators.stream().reduce(Function::andThen).orElseGet(Function::identity).apply(Single.just(data))``` and in another part of code, I invoke reactiveFunction in a list and invoke ```chainFunction``` – chavalife17 Jul 01 '22 at 15:32
That looks like the same as with the for-loop version so it has the same difficulty. – akarnokd Jul 04 '22 at 14:53
I add a refactor, maybe is better or not – chavalife17 Jul 04 '22 at 17:13

chaining reactive functions rxjava

[Edit with suggestions]

1 Answers1