如何将flux转换为mono

Question

private Mono<List<Integer>> filterAndMap(Mono<List<Integer>> listMono) {
  return listMono.filter(integers -> integers.size() == 3)
      .map(integers -> integers.stream()
          .map(integer -> integer * 10)
          .collect(Collectors.toList()));
}

void method1() {
  Flux.just(Arrays.asList(1, 2, 3), Arrays.asList(4, 5, 6), Arrays.asList(7, 8, 9, 10))
      .flatMap(integers -> this.filterAndMap(Mono.just(integers)))
      .doOnNext(System.out::println)
      .blockLast();
}

void method2() {
  Flux.just(Arrays.asList(1, 2, 3), Arrays.asList(4, 5, 6), Arrays.asList(7, 8, 9, 10))
      .convertFluxToMono()?
      .transform(this::filterAndMap)
      .doOnNext(System.out::println)
      .block();
}

更新：

一点我想要实现的内容。我有两个服务--一个通过Http返回memono > ，另一个通过Redis返回flus 。对于这两种情况，我有完全相同的功能-10-15个操作符链，我想要实现的是避免重复代码。

例如：

  void f1() {
    Mono<List<Integer>> mono = getFromHttp();
    
    mono
        .map(integers -> integers.stream().collect(Collectors.groupingBy(Function.identity())))
        .filter(entry -> entry.size() > 5)
        //...
        //many other operators
        //...
        //.flatMap()
        //.switchIfEmpty()
        //.doOnNext()
        .retryWhen();
  }

  void f2() {
    Flux<List<Integer>> flux = getFromRedis();
    
    flux.
        //...
        //same functionality here
        //...
  }

private Flux<List<Integer>> filterAndMap(Flux<List<Integer>> listMono) {
  return listMono.filter(integers -> integers.size() == 3)
      .map(integers -> integers.stream()
          .map(integer -> integer * 10)
          .collect(Collectors.toList()));
}

Answer 1

根据注释，这里的问题是“如何在不重复代码的情况下对mono和flux应用相同的转换？”

您可以使用一个简单的实用函数，如下所示：

static <T> Function<Mono<T>, Mono<T>> toMonoTransformer(
    Function<Flux<T>, Flux<T>> transformer) {
  return listMono -> transformer.apply(listMono.flux()).next();
}

此功能可用于任何类型的任何变压器。然后，可以对流量按原样使用转换函数：

void method2() {
  Flux.just(Arrays.asList(1, 2, 3), Arrays.asList(4, 5, 6), Arrays.asList(7, 8, 9, 10))
      .transform(this::filterAndMap)
      .doOnNext(System.out::println)
      .blockLast();
}

void method1() {
  Mono.just(Arrays.asList(1, 2, 3))
      .transform(toMonoTransformer(this::filterAndMap))
      .doOnNext(System.out::println)
      .block();
}