Reactor ‐ 在流上操作 - CCH0124/spring-sandbox GitHub Wiki

@Test
    public void map_test() {
        List<Integer> eles = new ArrayList<>();
        Flux.just(1, 2, 3, 4)
                .log()
                .map(i -> i * 10)
                .log()
                .subscribe(eles::add);

        Assertions.assertIterableEquals(eles, List.of(10, 20, 30, 40));
    }

    @Test
    public void zip_test() {
        List<String> eles = new ArrayList<>();
        Flux.just(1, 2, 3, 4)
            .log()
            .map(i -> i * 10)
            .zipWith(Flux.range(0, Integer.MAX_VALUE), 
                (one, two) -> String.format("First Flux: %d, Second Flux: %d", 
                one, two))
            .subscribe(eles::add);

        Assertions.assertIterableEquals(eles, List.of("First Flux: 10, Second Flux: 0",
        "First Flux: 20, Second Flux: 1",
        "First Flux: 30, Second Flux: 2",
        "First Flux: 40, Second Flux: 3"));
    }

    @Test
    public void infinite_stream() {
        ConnectableFlux<Object> publish = Flux.create(sink -> {
            int i = 100_000;
            while (i > 0) {
                sink.next(i--);
            }
        }).publish();

        publish.subscribe(System.out::println); 
        publish.subscribe(System.out::println);
        publish.subscribe(System.out::println);

        publish.connect();
    }

@Test
    public void test() {
        List<Integer> eles = new ArrayList<>();
        Flux.just(1, 2, 3, 4)
        .log()
        .map(x -> x * x)
        .subscribeOn(Schedulers.parallel())
        .subscribe(eles::add);
    }