From c626bccad37f6a56e71aa62db340db4780a6c2d3 Mon Sep 17 00:00:00 2001 From: Dave Syer Date: Mon, 15 May 2017 19:13:43 +0100 Subject: [PATCH] Add rest template to extract SSE events and assert them The message has to be sent after the endpoint starts up, which only happens after the first request is accepted, so we put that logic in the response extractor. --- spring-cloud-netflix-turbine-stream/pom.xml | 7 + .../stream/HystrixStreamAggregator.java | 123 ++++++------ .../TurbineStreamAutoConfiguration.java | 8 +- .../stream/TurbineStreamConfiguration.java | 184 +++++++++--------- .../stream/TurbineStreamProperties.java | 89 +++++---- .../turbine/stream/TurbineStreamTests.java | 155 ++++++++++++++- 6 files changed, 364 insertions(+), 202 deletions(-) diff --git a/spring-cloud-netflix-turbine-stream/pom.xml b/spring-cloud-netflix-turbine-stream/pom.xml index b17c352e..b7ca160e 100644 --- a/spring-cloud-netflix-turbine-stream/pom.xml +++ b/spring-cloud-netflix-turbine-stream/pom.xml @@ -15,6 +15,7 @@ ${basedir}/.. 2.0.0-DP.2 + 1.0.5.RELEASE @@ -104,5 +105,11 @@ spring-boot-starter-test test + + org.springframework.cloud + spring-cloud-starter-contract-stub-runner + ${spring-cloud-contract.version} + test + diff --git a/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/HystrixStreamAggregator.java b/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/HystrixStreamAggregator.java index d50de613..cf96e3d6 100644 --- a/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/HystrixStreamAggregator.java +++ b/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/HystrixStreamAggregator.java @@ -20,16 +20,17 @@ import java.io.IOException; import java.util.List; import java.util.Map; +import com.fasterxml.jackson.databind.ObjectMapper; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.messaging.handler.annotation.Payload; import org.springframework.stereotype.Component; import org.springframework.util.StringUtils; -import com.fasterxml.jackson.databind.ObjectMapper; - import rx.subjects.PublishSubject; /** @@ -38,70 +39,70 @@ import rx.subjects.PublishSubject; @Component // needed for ServiceActivator to be picked up public class HystrixStreamAggregator { - private static final Log log = LogFactory.getLog(HystrixStreamAggregator.class); + private static final Log log = LogFactory.getLog(HystrixStreamAggregator.class); - private ObjectMapper objectMapper; + private ObjectMapper objectMapper; - private PublishSubject> subject; + private PublishSubject> subject; - @Autowired - public HystrixStreamAggregator(ObjectMapper objectMapper, - PublishSubject> subject) { - this.objectMapper = objectMapper; - this.subject = subject; - } + @Autowired + public HystrixStreamAggregator(ObjectMapper objectMapper, + PublishSubject> subject) { + this.objectMapper = objectMapper; + this.subject = subject; + } - @ServiceActivator(inputChannel = TurbineStreamClient.INPUT) - public void sendToSubject(@Payload String payload) { - if (payload.startsWith("\"")) { - // Legacy payload from an Angel client - payload = payload.substring(1, payload.length() - 1); - payload = payload.replace("\\\"", "\""); - } - try { - if (payload.startsWith("[")) { - @SuppressWarnings("unchecked") - List> list = this.objectMapper.readValue(payload, - List.class); - for (Map map : list) { - sendMap(map); - } - } - else { - @SuppressWarnings("unchecked") - Map map = this.objectMapper.readValue(payload, Map.class); - sendMap(map); - } - } - catch (IOException ex) { - log.error("Error receiving hystrix stream payload: " + payload, ex); - } - } + @ServiceActivator(inputChannel = TurbineStreamClient.INPUT) + public void sendToSubject(@Payload String payload) { + if (payload.startsWith("\"")) { + // Legacy payload from an Angel client + payload = payload.substring(1, payload.length() - 1); + payload = payload.replace("\\\"", "\""); + } + try { + if (payload.startsWith("[")) { + @SuppressWarnings("unchecked") + List> list = this.objectMapper.readValue(payload, + List.class); + for (Map map : list) { + sendMap(map); + } + } + else { + @SuppressWarnings("unchecked") + Map map = this.objectMapper.readValue(payload, Map.class); + sendMap(map); + } + } + catch (IOException ex) { + log.error("Error receiving hystrix stream payload: " + payload, ex); + } + } - private void sendMap(Map map) { - Map data = getPayloadData(map); - if (log.isDebugEnabled()) { - log.debug("Received hystrix stream payload: " + data); - } - this.subject.onNext(data); - } + private void sendMap(Map map) { + Map data = getPayloadData(map); + if (log.isDebugEnabled()) { + log.debug("Received hystrix stream payload: " + data); + } + this.subject.onNext(data); + } - public static Map getPayloadData(Map jsonMap) { - @SuppressWarnings("unchecked") - Map origin = (Map) jsonMap.get("origin"); - String instanceId = null; - if (origin.containsKey("id")) { - instanceId = origin.get("id").toString(); - } - if (!StringUtils.hasText(instanceId)) { - // TODO: instanceid template - instanceId = origin.get("serviceId") + ":" + origin.get("host") + ":" - + origin.get("port"); - } - @SuppressWarnings("unchecked") - Map data = (Map) jsonMap.get("data"); - data.put("instanceId", instanceId); - return data; - } + public static Map getPayloadData(Map jsonMap) { + @SuppressWarnings("unchecked") + Map origin = (Map) jsonMap.get("origin"); + String instanceId = null; + if (origin.containsKey("id")) { + instanceId = origin.get("id").toString(); + } + if (!StringUtils.hasText(instanceId)) { + // TODO: instanceid template + instanceId = origin.get("serviceId") + ":" + origin.get("host") + ":" + + origin.get("port"); + } + @SuppressWarnings("unchecked") + Map data = (Map) jsonMap.get("data"); + data.put("instanceId", instanceId); + return data; + } } diff --git a/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamAutoConfiguration.java b/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamAutoConfiguration.java index ce6a1fa9..ecdb026c 100644 --- a/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamAutoConfiguration.java +++ b/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamAutoConfiguration.java @@ -20,17 +20,17 @@ import java.util.Map; import javax.annotation.PostConstruct; +import com.fasterxml.jackson.databind.ObjectMapper; + import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.config.BindingProperties; -import org.springframework.cloud.stream.config.ChannelBindingServiceProperties; +import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import com.fasterxml.jackson.databind.ObjectMapper; - import rx.subjects.PublishSubject; /** @@ -48,7 +48,7 @@ import rx.subjects.PublishSubject; public class TurbineStreamAutoConfiguration { @Autowired - private ChannelBindingServiceProperties bindings; + private BindingServiceProperties bindings; @Autowired private TurbineStreamProperties properties; diff --git a/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamConfiguration.java b/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamConfiguration.java index d67541c3..c83a7f12 100644 --- a/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamConfiguration.java +++ b/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamConfiguration.java @@ -16,13 +16,25 @@ package org.springframework.cloud.netflix.turbine.stream; +import java.nio.charset.StandardCharsets; import java.util.Collections; import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.reactivex.netty.RxNetty; +import io.reactivex.netty.protocol.http.server.HttpServer; +import io.reactivex.netty.protocol.http.sse.ServerSentEvent; + +import com.netflix.turbine.aggregator.InstanceKey; +import com.netflix.turbine.aggregator.StreamAggregator; +import com.netflix.turbine.internal.JsonUtility; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.client.actuator.HasFeatures; @@ -31,16 +43,8 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.util.SocketUtils; -import com.netflix.turbine.aggregator.InstanceKey; -import com.netflix.turbine.aggregator.StreamAggregator; -import com.netflix.turbine.internal.JsonUtility; +import static io.reactivex.netty.pipeline.PipelineConfigurators.serveSseConfigurator; -import static io.reactivex.netty.pipeline.PipelineConfigurators.sseServerConfigurator; - -import io.netty.buffer.ByteBuf; -import io.reactivex.netty.RxNetty; -import io.reactivex.netty.protocol.http.server.HttpServer; -import io.reactivex.netty.protocol.text.sse.ServerSentEvent; import rx.Observable; import rx.subjects.PublishSubject; @@ -51,102 +55,102 @@ import rx.subjects.PublishSubject; @EnableConfigurationProperties(TurbineStreamProperties.class) public class TurbineStreamConfiguration implements SmartLifecycle { - private static final Log log = LogFactory.getLog(TurbineStreamConfiguration.class); + private static final Log log = LogFactory.getLog(TurbineStreamConfiguration.class); - private AtomicBoolean running = new AtomicBoolean(false); + private AtomicBoolean running = new AtomicBoolean(false); - @Autowired - private TurbineStreamProperties properties; + @Autowired + private TurbineStreamProperties properties; - private int turbinePort; + private int turbinePort; - @Bean - public HasFeatures Feature() { - return HasFeatures.namedFeature("Turbine (Stream)", - TurbineStreamProperties.class); - } + @Bean + public HasFeatures Feature() { + return HasFeatures.namedFeature("Turbine (Stream)", + TurbineStreamProperties.class); + } - @Bean - public PublishSubject> hystrixSubject() { - return PublishSubject.create(); - } + @Bean + public PublishSubject> hystrixSubject() { + return PublishSubject.create(); + } - @Bean - @SuppressWarnings("deprecation") - public HttpServer aggregatorServer() { - // multicast so multiple concurrent subscribers get the same stream - Observable> publishedStreams = StreamAggregator - .aggregateGroupedStreams(hystrixSubject().groupBy( - data -> InstanceKey.create((String) data.get("instanceId")))) - .doOnUnsubscribe(() -> log.info("Unsubscribing aggregation.")) - .doOnSubscribe(() -> log.info("Starting aggregation")).flatMap(o -> o) - .publish().refCount(); - Observable> ping = Observable.timer(1, 10, TimeUnit.SECONDS) - .map(count -> Collections.singletonMap("type", (Object) "Ping")).publish() - .refCount(); - Observable> output = Observable.merge(publishedStreams, ping); + @Bean + @SuppressWarnings("deprecation") + public HttpServer aggregatorServer() { + // multicast so multiple concurrent subscribers get the same stream + Observable> publishedStreams = StreamAggregator + .aggregateGroupedStreams(hystrixSubject().groupBy( + data -> InstanceKey.create((String) data.get("instanceId")))) + .doOnUnsubscribe(() -> log.info("Unsubscribing aggregation.")) + .doOnSubscribe(() -> log.info("Starting aggregation")).flatMap(o -> o) + .publish().refCount(); + Observable> ping = Observable.timer(1, 10, TimeUnit.SECONDS) + .map(count -> Collections.singletonMap("type", (Object) "Ping")).publish() + .refCount(); + Observable> output = Observable.merge(publishedStreams, ping); - this.turbinePort = this.properties.getPort(); + this.turbinePort = this.properties.getPort(); - if (this.turbinePort <= 0) { - this.turbinePort = SocketUtils.findAvailableTcpPort(40000); - } + if (this.turbinePort <= 0) { + this.turbinePort = SocketUtils.findAvailableTcpPort(40000); + } - HttpServer httpServer = RxNetty - .createHttpServer(this.turbinePort, (request, response) -> { - log.info("SSE Request Received"); - response.getHeaders().setHeader("Content-Type", "text/event-stream"); - return output - .doOnUnsubscribe(() -> log - .info("Unsubscribing RxNetty server connection")) - .flatMap(data -> response.writeAndFlush(new ServerSentEvent( - null, null, JsonUtility.mapToJson(data)))); - }, sseServerConfigurator()); - return httpServer; - } + HttpServer httpServer = RxNetty + .createHttpServer(this.turbinePort, (request, response) -> { + log.info("SSE Request Received"); + response.getHeaders().setHeader("Content-Type", "text/event-stream"); + return output.doOnUnsubscribe( + () -> log.info("Unsubscribing RxNetty server connection")) + .flatMap(data -> response.writeAndFlush(new ServerSentEvent( + Unpooled.copiedBuffer(JsonUtility.mapToJson(data), + StandardCharsets.UTF_8)))); + }, serveSseConfigurator()); + return httpServer; + } - @Override - public boolean isAutoStartup() { - return true; - } + @Override + public boolean isAutoStartup() { + return true; + } - @Override - public void stop(Runnable callback) { - stop(); - callback.run(); - } + @Override + public void stop(Runnable callback) { + stop(); + callback.run(); + } - @Override - public void start() { - if (this.running.compareAndSet(false, true)) { - aggregatorServer().start(); - } - } + @Override + public void start() { + if (this.running.compareAndSet(false, true)) { + aggregatorServer().start(); + } + } - @Override - public void stop() { - if (this.running.compareAndSet(true, false)) { - try { - aggregatorServer().shutdown(); - } - catch (InterruptedException ex) { - log.error("Error shutting down", ex); - } - } - } + @Override + public void stop() { + if (this.running.compareAndSet(true, false)) { + try { + aggregatorServer().shutdown(); + } + catch (InterruptedException ex) { + log.error("Error shutting down", ex); + } + } + } - @Override - public boolean isRunning() { - return this.running.get(); - } + @Override + public boolean isRunning() { + return this.running.get(); + } - @Override - public int getPhase() { - return 0; - } + @Override + public int getPhase() { + return 0; + } - public int getTurbinePort() { - return this.turbinePort; - } + public int getTurbinePort() { + return this.turbinePort; + } } diff --git a/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamProperties.java b/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamProperties.java index 4ff99342..cab91b22 100644 --- a/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamProperties.java +++ b/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamProperties.java @@ -16,13 +16,13 @@ package org.springframework.cloud.netflix.turbine.stream; +import java.util.Objects; + import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.cloud.netflix.hystrix.HystrixConstants; import org.springframework.http.MediaType; -import java.util.Objects; - /** * @author Dave Syer * @author Gregor Zurowski @@ -30,59 +30,58 @@ import java.util.Objects; @ConfigurationProperties("turbine.stream") public class TurbineStreamProperties { - @Value("${server.port:8989}") - private int port = 8989; + @Value("${server.port:8989}") + private int port = 8989; - private String destination = HystrixConstants.HYSTRIX_STREAM_DESTINATION; + private String destination = HystrixConstants.HYSTRIX_STREAM_DESTINATION; - private String contentType = MediaType.APPLICATION_JSON_VALUE; + private String contentType = MediaType.APPLICATION_JSON_VALUE; - public int getPort() { - return port; - } + public int getPort() { + return port; + } - public void setPort(int port) { - this.port = port; - } + public void setPort(int port) { + this.port = port; + } - public String getDestination() { - return destination; - } + public String getDestination() { + return destination; + } - public void setDestination(String destination) { - this.destination = destination; - } + public void setDestination(String destination) { + this.destination = destination; + } - public String getContentType() { - return contentType; - } + public String getContentType() { + return contentType; + } - public void setContentType(String contentType) { - this.contentType = contentType; - } + public void setContentType(String contentType) { + this.contentType = contentType; + } - @Override - public boolean equals(Object o) { - if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; - TurbineStreamProperties that = (TurbineStreamProperties) o; - return port == that.port && - Objects.equals(destination, that.destination) && - Objects.equals(contentType, that.contentType); - } + @Override + public boolean equals(Object o) { + if (this == o) + return true; + if (o == null || getClass() != o.getClass()) + return false; + TurbineStreamProperties that = (TurbineStreamProperties) o; + return port == that.port && Objects.equals(destination, that.destination) + && Objects.equals(contentType, that.contentType); + } - @Override - public int hashCode() { - return Objects.hash(port, destination, contentType); - } + @Override + public int hashCode() { + return Objects.hash(port, destination, contentType); + } - @Override - public String toString() { - return new StringBuilder("TurbineStreamProperties{") - .append("port=").append(port).append(", ") - .append("destination='").append(destination).append("', ") - .append("contentType='").append(contentType).append("'}") - .toString(); - } + @Override + public String toString() { + return new StringBuilder("TurbineStreamProperties{").append("port=").append(port) + .append(", ").append("destination='").append(destination).append("', ") + .append("contentType='").append(contentType).append("'}").toString(); + } } diff --git a/spring-cloud-netflix-turbine-stream/src/test/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamTests.java b/spring-cloud-netflix-turbine-stream/src/test/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamTests.java index 5d2a7af7..8c2defbe 100644 --- a/spring-cloud-netflix-turbine-stream/src/test/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamTests.java +++ b/spring-cloud-netflix-turbine-stream/src/test/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamTests.java @@ -16,22 +16,78 @@ package org.springframework.cloud.netflix.turbine.stream; +import java.io.IOException; +import java.io.InputStream; +import java.net.URI; +import java.util.Map; +import java.util.concurrent.CountDownLatch; + +import com.fasterxml.jackson.databind.ObjectMapper; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.junit.Test; import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.boot.test.context.SpringBootTest.WebEnvironment; +import org.springframework.cloud.contract.stubrunner.StubTrigger; +import org.springframework.cloud.contract.stubrunner.spring.AutoConfigureStubRunner; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpMethod; +import org.springframework.http.HttpRequest; +import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; +import org.springframework.http.client.ClientHttpRequestExecution; +import org.springframework.http.client.ClientHttpRequestInterceptor; +import org.springframework.http.client.ClientHttpResponse; +import org.springframework.integration.support.management.MessageChannelMetrics; +import org.springframework.messaging.SubscribableChannel; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.StringUtils; +import org.springframework.web.client.RestTemplate; + +import static org.assertj.core.api.Assertions.assertThat; /** * @author Spencer Gibb */ @RunWith(SpringJUnit4ClassRunner.class) @SpringBootTest(classes = TurbineStreamTests.Application.class, webEnvironment = WebEnvironment.RANDOM_PORT, value = { - "turbine.stream.port=0", "spring.jmx.enabled=true" }) + "turbine.stream.port=0", + // TODO: we don't need this if we harmonize the turbine and hystrix destinations + // https://github.com/spring-cloud/spring-cloud-netflix/issues/1948 + "spring.cloud.stream.bindings.turbineStreamInput.destination=hystrixStreamOutput", + "logging.level.org.springframework.cloud.netflix.turbine=DEBUG", + "spring.jmx.enabled=true", "stubrunner.workOffline=true", + "stubrunner.ids=org.springframework.cloud:spring-cloud-netflix-hystrix-stream" }) +@AutoConfigureStubRunner public class TurbineStreamTests { + private static Log log = LogFactory.getLog(TurbineStreamTests.class); + + @Autowired + StubTrigger stubTrigger; + + @Autowired + ObjectMapper mapper; + + @Autowired + @Qualifier(TurbineStreamClient.INPUT) + SubscribableChannel input; + + RestTemplate rest = new RestTemplate(); + + @Autowired + TurbineStreamConfiguration turbine; + + private CountDownLatch latch = new CountDownLatch(1); + @EnableAutoConfiguration @EnableTurbineStream public static class Application { @@ -41,7 +97,102 @@ public class TurbineStreamTests { } @Test - public void contextLoads() { + public void contextLoads() throws Exception { + rest.getInterceptors().add(new NonClosingInterceptor()); + int count = ((MessageChannelMetrics) input).getSendCount(); + ResponseEntity response = rest.execute( + new URI("http://localhost:" + turbine.getTurbinePort() + "/"), + HttpMethod.GET, null, this::extract); + assertThat(response.getHeaders().getContentType()) + .isEqualTo(MediaType.TEXT_EVENT_STREAM); + assertThat(response.getStatusCode()).isEqualTo(HttpStatus.OK); + Map metrics = extractMetrics(response.getBody()); + assertThat(metrics).containsEntry("type", "HystrixCommand"); + assertThat(((MessageChannelMetrics) input).getSendCount()).isEqualTo(count + 1); } + @SuppressWarnings("unchecked") + private Map extractMetrics(String body) throws Exception { + String[] split = body.split("data:"); + for (String value : split) { + if (value.contains("Ping") || value.length() == 0) { + continue; + } + else { + return mapper.readValue(value, Map.class); + } + } + return null; + } + + private ResponseEntity extract(ClientHttpResponse response) + throws IOException { + // The message has to be sent after the endpoint is activated, so this is a + // convenient place to put it + stubTrigger.trigger("metrics"); + byte[] bytes = new byte[1024]; + StringBuilder builder = new StringBuilder(); + int read = 0; + while (read >= 0 + && StringUtils.countOccurrencesOf(builder.toString(), "\n") < 2) { + read = response.getBody().read(bytes, 0, bytes.length); + if (read > 0) { + latch.countDown(); + builder.append(new String(bytes, 0, read)); + } + log.info("Building: " + builder); + } + log.info("Done: " + builder); + return ResponseEntity.status(response.getStatusCode()) + .headers(response.getHeaders()).body(builder.toString()); + } + + private class NonClosingInterceptor implements ClientHttpRequestInterceptor { + + private class NonClosingResponse implements ClientHttpResponse { + + private ClientHttpResponse delegate; + + public NonClosingResponse(ClientHttpResponse delegate) { + this.delegate = delegate; + } + + @Override + public InputStream getBody() throws IOException { + return delegate.getBody(); + } + + @Override + public HttpHeaders getHeaders() { + return delegate.getHeaders(); + } + + @Override + public HttpStatus getStatusCode() throws IOException { + return delegate.getStatusCode(); + } + + @Override + public int getRawStatusCode() throws IOException { + return delegate.getRawStatusCode(); + } + + @Override + public String getStatusText() throws IOException { + return delegate.getStatusText(); + } + + @Override + public void close() { + } + + } + + @Override + public ClientHttpResponse intercept(HttpRequest request, byte[] body, + ClientHttpRequestExecution execution) throws IOException { + return new NonClosingResponse(execution.execute(request, body)); + } + + } }