JacksonResponseConverterFunction implements ResponseConverterFunction { @Override public HttpResponse convertResponse(ServiceRequestContext ctx, ResponseHeaders headers, @Nullable Object result, HttpHeaders trailers) throws Exception { final MediaType mediaType = headers.contentType(); if (mediaType != null) { if (mediaType.is(MediaType.JSON) || mediaType.subtype().endsWith("+json")) { final Charset charset = mediaType.charset(StandardCharsets.UTF_8); if (charset.contains(StandardCharsets.UTF_8)) { if (result instanceof Publisher) { return aggregateFrom((Publisher<?>) result, headers, trailers, this::toJsonHttpData); } if (result instanceof Stream) { return aggregateFrom((Stream<?>) result, headers, trailers, this::toJsonHttpData, ctx.blockingTaskExecutor()); } return HttpResponse.of(headers, toJsonHttpData(result), trailers); } } if (mediaType.is(MediaType.JSON_SEQ)) { if (result instanceof Publisher) { return JsonTextSequences.fromPublisher(headers, (Publisher<?>) result, trailers, mapper); } if (result instanceof Stream) { return JsonTextSequences.fromStream(headers, (Stream<?>) result, trailers, ctx.blockingTaskExecutor(), mapper); } return JsonTextSequences.fromObject(headers, result, trailers, mapper); } } else if (result instanceof JsonNode) { return HttpResponse.of(headers.toBuilder().contentType(MediaType.JSON_UTF_8).build(), toJsonHttpData(result), trailers); } return ResponseConverterFunction.fallthrough(); } JacksonResponseConverterFunction();  JacksonResponseConverterFunction(ObjectMapper mapper); @Override HttpResponse convertResponse(ServiceRequestContext ctx,
                                        ResponseHeaders headers,
                                        @Nullable Object result,
                                        HttpHeaders trailers);  }
@Test void aggregatedJson_streamError() throws Exception { final Stream<String> stream = Stream.of(TEST_STRINGS).map(s -> { throw new AnticipatedException(); }); final HttpResponse response = function.convertResponse(ctx, JSON_HEADERS, stream, DEFAULT_TRAILERS); StepVerifier.create(response) .expectError(AnticipatedException.class) .verify(); }
@Test void aggregatedJson_otherTypes() throws Exception { StepVerifier.create(function.convertResponse( ctx, JSON_HEADERS, "abc", DEFAULT_TRAILERS)) .expectNext(JSON_HEADERS.toBuilder() .setInt(HttpHeaderNames.CONTENT_LENGTH, 5) .build()) .expectNext(HttpData.ofUtf8("\"abc\"")) .expectComplete() .verify(); StepVerifier.create(function.convertResponse( ctx, JSON_HEADERS, 123, DEFAULT_TRAILERS)) .expectNext(JSON_HEADERS.toBuilder() .setInt(HttpHeaderNames.CONTENT_LENGTH, 3) .build()) .expectNext(HttpData.ofUtf8("123")) .expectComplete() .verify(); }
@Test void jsonTextSequences_streamError() throws Exception { final Stream<String> stream = Stream.of(TEST_STRINGS).map(s -> { throw new AnticipatedException(); }); final HttpResponse response = function.convertResponse(ctx, JSON_SEQ_HEADERS, stream, DEFAULT_TRAILERS); StepVerifier.create(response) .expectError(AnticipatedException.class) .verify(); }
@Test void jsonTextSequences_otherTypes() throws Exception { StepVerifier.create(function.convertResponse( ctx, JSON_SEQ_HEADERS, "abc", DEFAULT_TRAILERS)) .expectNext(JSON_SEQ_HEADERS.toBuilder() .setInt(HttpHeaderNames.CONTENT_LENGTH, 7) .build()) .expectNext(HttpData.wrap( new byte[] { RECORD_SEPARATOR, '\"', 'a', 'b', 'c', '\"', LINE_FEED })) .expectComplete() .verify(); StepVerifier.create(function.convertResponse( ctx, JSON_SEQ_HEADERS, 123, DEFAULT_TRAILERS)) .expectNext(JSON_SEQ_HEADERS.toBuilder() .setInt(HttpHeaderNames.CONTENT_LENGTH, 5) .build()) .expectNext(HttpData.wrap(new byte[] { RECORD_SEPARATOR, '1', '2', '3', LINE_FEED })) .expectComplete() .verify(); }