Spring 5 Web Reactive Programming - WebClient ClassCastException при демаршалинге JSON из Spring Reactive Controller, который передает данные

Этот вопрос связан с этим one, в котором я спросил, как передавать данные из Reactive Spring Controller.

Как указал Россен, мы должны использовать text/event-stream для отправки результатов потоковой передачи в качестве событий, отправленных сервером, пока все хорошо.

У меня есть такая услуга:

@GetMapping(value="/accounts/alertsStreaming", headers="accept=text/event-stream")
public Flux<Alert> getAccountAlertsStreaming() {
    return Flux.fromArray(new Alert[]{new Alert((long)1, "Alert message"), 
                                      new Alert((long)2, "Alert message2"),
                                      new Alert((long)3, "Alert message3")})
               .delayMillis(1000)
               .log();
}

Вызывая его из браузеров, 3 результата начинают поступать с задержкой в ​​1 секунду.

Я хотел вызвать эту службу из WebClient и реализовал ее следующим образом:

@Component
public class AccountsServiceClient {

    @Autowired
    private WebClient webClient;

    public Flux<Alert> getAccountAlertsStreaming(String serviceBaseUrl){
        Flux<Alert> response = webClient
                .perform(get(serviceBaseUrl+"/accounts/alertsStreaming").header("Accept", "text/event-stream"))
                .extract(bodyStream(Alert.class));
        return response;
    }       
}

А это тестовый код:

@Test
@ContextConfiguration(classes={WebClientConfig.class, AccountsServiceClient.class})
public class AccountsServiceClientTest extends AbstractTestNGSpringContextTests{

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private AccountsServiceClient client;

    public void testNumbersServiceClientStreamingTest() throws InterruptedException{

        CountDownLatch latch = new CountDownLatch(1);

        Flux<Alert> alerts = client.getAccountAlertsStreaming("http://localhost:8080");
        alerts.doOnComplete( () -> {
            latch.countDown();
        }).subscribe( (n) -> {
            logger.info("------------> GOT ALERT {}", n);
        });

        latch.await();
    }
}

Проблема в том, что когда клиент пытается извлечь результаты по мере их получения, ни один из HttpMessageReader' не может прочитать text/event-stream + Alert.class.

public class ResponseExtractors {

    protected static HttpMessageReader<?> resolveMessageReader(List<HttpMessageReader<?>> messageReaders,
                ResolvableType responseType, MediaType contentType) {

            return messageReaders.stream()
                    .filter(e -> e.canRead(responseType, contentType))
                    .findFirst()
                    .orElseThrow(() ->
                            new WebClientException(
                                    "Could not decode response body of type '" + contentType
                                            + "' with target type '" + respons

eType.toString() + "'"));
    }

Исключение:

reactor.core.Exceptions$BubblingException: org.springframework.web.client.reactive.WebClientException: Could not decode response body of type 'text/event-stream' with target type 'com.codependent.spring5.playground.reactive.dto.Alert'
    at reactor.core.Exceptions.bubble(Exceptions.java:97)
    at reactor.core.Exceptions.onErrorDropped(Exceptions.java:263)
    at reactor.core.publisher.LambdaSubscriber.onError(LambdaSubscriber.java:126)
    at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:183)
    at reactor.core.publisher.MonoFlatMap$FlattenSubscriber.onNext(MonoFlatMap.java:128)
    at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:169)
    at reactor.core.publisher.FluxLog$LoggerSubscriber.doNext(FluxLog.java:161)
    at reactor.core.publisher.OperatorAdapter.onNext(OperatorAdapter.java:88)
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:123)
    at reactor.core.publisher.FluxResume$ResumeSubscriber.onNext(FluxResume.java:75)
    at reactor.core.publisher.FluxJust$WeakScalarSubscription.request(FluxJust.java:103)
    at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:1010)
    at reactor.core.publisher.FluxResume$ResumeSubscriber.onSubscribe(FluxResume.java:70)
    at reactor.core.publisher.FluxJust.subscribe(FluxJust.java:71)
    at reactor.ipc.netty.http.NettyHttpClientHandler.channelRead(NettyHttpClientHandler.java:120)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:372)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:358)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:350)
    at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:435)
    at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:293)
    at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:280)
    at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:396)
    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:248)
    at io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:250)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:372)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:358)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:350)
    at io.netty.handler.logging.LoggingHandler.channelRead(LoggingHandler.java:233)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:372)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:358)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:350)
    at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1334)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:372)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:358)
    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:926)
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:123)
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:571)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:512)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:426)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:398)
    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:877)
    at java.lang.Thread.run(Thread.java:745)
Caused by: org.springframework.web.client.reactive.WebClientException: Could not decode response body of type 'text/event-stream' with target type 'com.codependent.spring5.playground.reactive.dto.Alert'
    at org.springframework.web.client.reactive.ResponseExtractors.lambda$resolveMessageReader$23(ResponseExtractors.java:203)
    at org.springframework.web.client.reactive.ResponseExtractors$$Lambda$61/1950155746.get(Unknown Source)
    at java.util.Optional.orElseThrow(Optional.java:290)
    at org.springframework.web.client.reactive.ResponseExtractors.resolveMessageReader(ResponseExtractors.java:200)
    at org.springframework.web.client.reactive.ResponseExtractors.decodeResponseBody(ResponseExtractors.java:181)
    at org.springframework.web.client.reactive.ResponseExtractors.lambda$null$12(ResponseExtractors.java:89)
    at org.springframework.web.client.reactive.ResponseExtractors$$Lambda$36/70386506.apply(Unknown Source)
    at reactor.core.publisher.MonoFlatMap$FlattenSubscriber.onNext(MonoFlatMap.java:126)
    ... 37 common frames omitted

spring reactive-programming project-reactor rx-java
person codependent    schedule 05.08.2016    source источник
comment
В чем разница между WebClient и веб-браузером с точки зрения сервера?   -  person Tassos Bassoukos    schedule 06.08.2016
comment
Я понимаю вашу точку зрения, но, помимо новой семантики, если мы не собираемся передавать результаты, а все они одновременно, какой смысл использовать новый WebClient вместо старого RestTemplate?   -  person codependent    schedule 06.08.2016

Ответы (2)


arrow_upward
0
arrow_downward

Для этого уже есть проблема. Прокомментируйте / проголосуйте за SPR-14539.

person Brian Clozel    schedule 29.08.2016

arrow_upward
0
arrow_downward

Возможно, это должно обрабатываться автоматически фреймворком. В любом случае, я решил это, разобрав для себя данные потока JSON:

WebConfigClient:

@Configuration
public class WebClientConfig {

    @Bean
    public ObjectMapper jacksonObjectMapper(){
        return new ObjectMapper();
    }

    @Bean
    public WebClient webClient(){
        WebClient webClient = new WebClient(new ReactorClientHttpConnector());
        return webClient;
    }

}

Сервисный клиент:

@Component
public class AccountsServiceClient {

    @Autowired
    private WebClient webClient;

    @Autowired
    private ObjectMapper jacksonObjectMapper;

    public Flux<Alert> getAccountAlertsStreaming(String serviceBaseUrl){
        Flux<Alert> response = webClient
                .perform(get(serviceBaseUrl+"/accounts/alertsStreaming").header("Accept", "text/event-stream"))
                .extract(bodyStream(String.class))
                .map((e -> {
                    try {
                        e = e.substring(e.indexOf(":")+1);
                        Alert a = jacksonObjectMapper.readValue(e, Alert.class);
                        return a;
                    } catch (Exception e1) {
                        e1.printStackTrace();
                        return null;
                    }

                }));
        return response;
    }

}

ОБНОВЛЕНИЕ. Начиная с Spring 5 M4, это выполняется фреймворком. Вы можете проверить решение с новым API здесь: Spring 5 Web Reactive - Как мы можем использовать WebClient для получения потоковых данных в Flux?

person codependent    schedule 09.08.2016