SPR-12538 AsyncRestTemplate interceptors

spring-web/src/main/java/org/springframework/http/client/AsyncClientHttpRequestExecution.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2002-2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.http.client;
+
+import org.springframework.http.HttpRequest;
+import org.springframework.util.concurrent.ListenableFuture;
+
+import java.io.IOException;
+
+/**
+ * The execution context of asynchronous client http request.
+ *
+ * @author Jakub Narloch
+ * @see AsyncClientHttpRequestInterceptor
+ */
+public interface AsyncClientHttpRequestExecution {
+
+    /**
+     * Resumes the request execution by invoking next interceptor in the chain or executing the
+     * request to the remote service.
+     *
+     * @param request the http request, containing the http method and headers
+     * @param body    the body of the request
+     * @return the future
+     * @throws IOException in case of I/O errors
+     */
+    ListenableFuture<ClientHttpResponse> executeAsync(HttpRequest request, byte[] body) throws IOException;
+}

spring-web/src/main/java/org/springframework/http/client/AsyncClientHttpRequestInterceptor.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2002-2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.http.client;
+
+import org.springframework.http.HttpRequest;
+import org.springframework.http.client.support.InterceptingAsyncHttpAccessor;
+import org.springframework.util.concurrent.ListenableFuture;
+
+import java.io.IOException;
+
+/**
+ * The asynchronous HTTP request interceptor.
+ *
+ * @author Jakub Narloch
+ * @see org.springframework.web.client.AsyncRestTemplate
+ * @see InterceptingAsyncHttpAccessor
+ */
+public interface AsyncClientHttpRequestInterceptor {
+
+    /**
+     * Intercepts the outgoing client HTTP request.
+     *
+     * @param request   the request
+     * @param body      the request's body
+     * @param execution the request execution context
+     * @return the future
+     * @throws IOException in case of I/O errors
+     */
+    ListenableFuture<ClientHttpResponse> interceptRequest(
+            HttpRequest request, byte[] body, AsyncClientHttpRequestExecution execution) throws IOException;
+}

spring-web/src/main/java/org/springframework/http/client/InterceptingAsyncClientHttpRequest.java

@@ -0,0 +1,115 @@
+/*
+ * Copyright 2002-2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.http.client;
+
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.HttpRequest;
+import org.springframework.util.StreamUtils;
+import org.springframework.util.concurrent.ListenableFuture;
+import org.springframework.util.concurrent.ListenableFutureAdapter;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+
+/**
+ * A {@link AsyncClientHttpRequest} wrapper that enriches it proceeds the actual request execution with calling
+ * the registered interceptors.
+ *
+ * @author Jakub Narloch
+ * @see InterceptingAsyncClientHttpRequestFactory
+ */
+class InterceptingAsyncClientHttpRequest extends AbstractBufferingAsyncClientHttpRequest {
+
+    private AsyncClientHttpRequestFactory requestFactory;
+
+    private List<AsyncClientHttpRequestInterceptor> interceptors;
+
+    private URI uri;
+
+    private HttpMethod httpMethod;
+
+    /**
+     * Creates new instance of {@link InterceptingAsyncClientHttpRequest}.
+     *
+     * @param requestFactory the async request factory
+     * @param interceptors   the list of interceptors
+     * @param uri            the request URI
+     * @param httpMethod     the HTTP method
+     */
+    public InterceptingAsyncClientHttpRequest(AsyncClientHttpRequestFactory requestFactory,
+                                              List<AsyncClientHttpRequestInterceptor> interceptors, URI uri,
+                                              HttpMethod httpMethod) {
+
+        this.requestFactory = requestFactory;
+        this.interceptors = interceptors;
+        this.uri = uri;
+        this.httpMethod = httpMethod;
+    }
+
+    @Override
+    protected ListenableFuture<ClientHttpResponse> executeInternal(HttpHeaders headers, byte[] body) throws IOException {
+        return new AsyncRequestExecution().executeAsync(this, body);
+    }
+
+    @Override
+    public HttpMethod getMethod() {
+        return httpMethod;
+    }
+
+    @Override
+    public URI getURI() {
+        return uri;
+    }
+
+    private class AsyncRequestExecution implements AsyncClientHttpRequestExecution {
+
+        private Iterator<AsyncClientHttpRequestInterceptor> nextInterceptor = interceptors.iterator();
+
+        @Override
+        public ListenableFuture<ClientHttpResponse> executeAsync(HttpRequest request, byte[] body) throws IOException {
+            if (nextInterceptor.hasNext()) {
+                AsyncClientHttpRequestInterceptor interceptor = nextInterceptor.next();
+                ListenableFuture<ClientHttpResponse> future = interceptor.interceptRequest(request, body, this);
+                return new IdentityListenableFutureAdapter<ClientHttpResponse>(future);
+            }
+            else {
+                AsyncClientHttpRequest req = requestFactory.createAsyncRequest(uri, httpMethod);
+                req.getHeaders().putAll(getHeaders());
+                if (body.length > 0) {
+                    StreamUtils.copy(body, req.getBody());
+                }
+                return req.executeAsync();
+            }
+        }
+    }
+
+    private static class IdentityListenableFutureAdapter<T> extends ListenableFutureAdapter<T, T> {
+
+        protected IdentityListenableFutureAdapter(ListenableFuture<T> adaptee) {
+            super(adaptee);
+        }
+
+        @Override
+        protected T adapt(T adapteeResult) throws ExecutionException {
+            return adapteeResult;
+        }
+    }
+}

spring-web/src/main/java/org/springframework/http/client/InterceptingAsyncClientHttpRequestFactory.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2002-2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.http.client;
+
+import org.springframework.http.HttpMethod;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * The intercepting request factory.
+ *
+ * @author Jakub Narloch
+ * @see InterceptingAsyncClientHttpRequest
+ */
+public class InterceptingAsyncClientHttpRequestFactory implements AsyncClientHttpRequestFactory {
+
+    private AsyncClientHttpRequestFactory delegate;
+
+    private List<AsyncClientHttpRequestInterceptor> interceptors;
+
+    /**
+     * Creates new instance of {@link InterceptingAsyncClientHttpRequestFactory} with delegated request factory and
+     * list of interceptors.
+     *
+     * @param delegate     the delegated request factory
+     * @param interceptors the list of interceptors.
+     */
+    public InterceptingAsyncClientHttpRequestFactory(AsyncClientHttpRequestFactory delegate, List<AsyncClientHttpRequestInterceptor> interceptors) {
+
+        this.delegate = delegate;
+        this.interceptors = interceptors != null ? interceptors : Collections.<AsyncClientHttpRequestInterceptor>emptyList();
+    }
+
+    @Override
+    public AsyncClientHttpRequest createAsyncRequest(URI uri, HttpMethod httpMethod) throws IOException {
+
+        return new InterceptingAsyncClientHttpRequest(delegate, interceptors, uri, httpMethod);
+    }
+}

spring-web/src/main/java/org/springframework/http/client/support/InterceptingAsyncHttpAccessor.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2002-2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.http.client.support;
+
+import org.springframework.http.client.AsyncClientHttpRequestFactory;
+import org.springframework.http.client.AsyncClientHttpRequestInterceptor;
+import org.springframework.http.client.InterceptingAsyncClientHttpRequestFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * The HTTP accessor that extends the base {@link AsyncHttpAccessor} with request intercepting functionality.
+ *
+ * @author Jakub Narloch
+ */
+public abstract class InterceptingAsyncHttpAccessor extends AsyncHttpAccessor {
+
+    private List<AsyncClientHttpRequestInterceptor> interceptors = new ArrayList<AsyncClientHttpRequestInterceptor>();
+
+    /**
+     * Retrieves the list of interceptors.
+     *
+     * @return the list of interceptors
+     */
+    public List<AsyncClientHttpRequestInterceptor> getInterceptors() {
+        return interceptors;
+    }
+
+    /**
+     * Sets the list of interceptors.
+     *
+     * @param interceptors the list of interceptors
+     */
+    public void setInterceptors(List<AsyncClientHttpRequestInterceptor> interceptors) {
+        this.interceptors = interceptors;
+    }
+
+    @Override
+    public AsyncClientHttpRequestFactory getAsyncRequestFactory() {
+        AsyncClientHttpRequestFactory asyncRequestFactory = super.getAsyncRequestFactory();
+        if(interceptors.isEmpty()) {
+            return asyncRequestFactory;
+        }
+        return new InterceptingAsyncClientHttpRequestFactory(asyncRequestFactory, getInterceptors());
+    }
+}

spring-web/src/main/java/org/springframework/web/client/AsyncRestTemplate.java

@@ -41,7 +41,7 @@ import org.springframework.http.client.ClientHttpRequest;
 import org.springframework.http.client.ClientHttpRequestFactory;
 import org.springframework.http.client.ClientHttpResponse;
 import org.springframework.http.client.SimpleClientHttpRequestFactory;
-import org.springframework.http.client.support.AsyncHttpAccessor;
+import org.springframework.http.client.support.InterceptingAsyncHttpAccessor;
 import org.springframework.http.converter.HttpMessageConverter;
 import org.springframework.util.Assert;
 import org.springframework.util.concurrent.FailureCallback;
@@ -74,7 +74,7 @@ import org.springframework.web.util.UriTemplateHandler;
  * @since 4.0
  * @see RestTemplate
  */
-public class AsyncRestTemplate extends AsyncHttpAccessor implements AsyncRestOperations {
+public class AsyncRestTemplate extends InterceptingAsyncHttpAccessor implements AsyncRestOperations {
 
 	private final RestTemplate syncTemplate;
 

spring-web/src/test/java/org/springframework/web/client/AsyncRestTemplateIntegrationTests.java

@@ -16,8 +16,11 @@
 
 package org.springframework.web.client;
 
+import java.io.IOException;
 import java.net.URI;
 import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
 import java.util.EnumSet;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
@@ -32,9 +35,13 @@ import org.springframework.core.io.Resource;
 import org.springframework.http.HttpEntity;
 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.AsyncClientHttpRequestExecution;
+import org.springframework.http.client.AsyncClientHttpRequestInterceptor;
+import org.springframework.http.client.ClientHttpResponse;
 import org.springframework.http.client.HttpComponentsAsyncClientHttpRequestFactory;
 import org.springframework.util.LinkedMultiValueMap;
 import org.springframework.util.MultiValueMap;
@@ -600,4 +607,47 @@ public class AsyncRestTemplateIntegrationTests extends AbstractJettyServerTestCa
 		future.get();
 	}
 
+	@Test
+	public void getAndInterceptResponse() throws Exception {
+		RequestInterceptor interceptor = new RequestInterceptor();
+		template.setInterceptors(Arrays.<AsyncClientHttpRequestInterceptor>asList(interceptor));
+		ListenableFuture<ResponseEntity<String>> future = template.getForEntity(baseUrl + "/get", String.class);
+
+		ResponseEntity<String> response = future.get();
+		assertNotNull(interceptor.response);
+		assertEquals(HttpStatus.OK, interceptor.response.getStatusCode());
+		assertNull(interceptor.exception);
+		assertEquals(helloWorld, response.getBody());
+	}
+
+	@Test
+	public void getAndInterceptError() throws Exception {
+		RequestInterceptor interceptor = new RequestInterceptor();
+		template.setInterceptors(Arrays.<AsyncClientHttpRequestInterceptor>asList(interceptor));
+		ListenableFuture<ResponseEntity<String>> future = template.getForEntity(baseUrl + "/status/notfound", String.class);
+
+		try {
+			future.get();
+			fail("No exception thrown");
+		} catch (ExecutionException ex) {
+		}
+		assertNotNull(interceptor.response);
+		assertEquals(HttpStatus.NOT_FOUND, interceptor.response.getStatusCode());
+		assertNull(interceptor.exception);
+	}
+
+	public static class RequestInterceptor implements AsyncClientHttpRequestInterceptor {
+
+		private ClientHttpResponse response;
+
+		private Throwable exception;
+
+		@Override
+		public ListenableFuture<ClientHttpResponse> interceptRequest(HttpRequest request, byte[] body,
+																	 AsyncClientHttpRequestExecution execution) throws IOException {
+			ListenableFuture<ClientHttpResponse> future = execution.executeAsync(request, body);
+			future.addCallback(resp -> response = resp, ex -> exception = ex);
+			return future;
+		}
+	}
 }