gistfile1.txt

@ContextConfiguration
@SpringBootTest(webEnvironment=SpringBootTest.WebEnvironment.RANDOM_PORT)
@ExtendWith(SpringExtension.class)
public class WebClientTest {
    private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(WebClient.class);

    private static final String BODY = "this is the body";
    private static final String ENDPOINT = "endpoint";

    @LocalServerPort
    int port;

    private StringDecoder stringDecoder = StringDecoder.textPlainOnly(false);

    @Test
    void testTwoStreamsFromBody() {
        WebClient webclient = WebClient.create("http://localhost:" + port);

        Flux<DataBuffer> body = webclient
        .get()
        .uri(ENDPOINT)
        .exchange()
        .doOnNext(this::validateIsSuccess)
        .map(response -> response.bodyToFlux(DataBuffer.class))
        .block()
        .log()
        .cache() // comment out to get duplicate subscribe error, leave in to get ref cnt error
        ;

        StepVerifier.create(
            Flux.from(decodeToString(body))
            .concatWith(decodeToString(body))
        )
        .expectNext(BODY)
        .expectNext(BODY)
        .verifyComplete();
    }

    private void validateIsSuccess(ClientResponse r) {
        if (!r.statusCode().is2xxSuccessful()) throw new RuntimeException("Received " + r.statusCode());
    }

    private Mono<String> decodeToString(Flux<DataBuffer> stream) {
        return stringDecoder.decode(stream, ResolvableType.forClass(String.class), null, null)
        .reduce(new StringBuilder(), (builder, e) -> { builder.append(e); return builder; })
        .map(StringBuilder::toString);
    }

    @RestController
    static class Service {
        @GetMapping(ENDPOINT)
        public String endpoint() {
            logger.info("endpoint called");
            return BODY;
        }
    }

    @Configuration
    @SpringBootApplication
    static class TestConfiguration {
        @Bean
        Service service() { return new Service(); }
    }
}