Add support for Server-Sent Events

This commit adds ResponseBodyEmitter and SseEmitter (and also
ResponseEntity<ResponseBodyEmitter> and ResponseEntity<SseEmitter>) as
new return value types supported on @RequestMapping controller methods.

See Javadoc on respective types for more details.

Issue: SPR-12212

spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/RequestMappingHandlerAdapter.java

@@ -632,6 +632,7 @@ public class RequestMappingHandlerAdapter extends AbstractHandlerMethodAdapter
 		handlers.add(new ModelAndViewMethodReturnValueHandler());
 		handlers.add(new ModelMethodProcessor());
 		handlers.add(new ViewMethodReturnValueHandler());
+		handlers.add(new ResponseBodyEmitterReturnValueHandler(getMessageConverters()));
 		handlers.add(new HttpEntityMethodProcessor(
 				getMessageConverters(), this.contentNegotiationManager, this.responseBodyAdvice));
 		handlers.add(new HttpHeadersReturnValueHandler());

spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/ResponseBodyEmitter.java

@@ -0,0 +1,198 @@
+/*
+ * 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.web.servlet.mvc.method.annotation;
+
+import java.io.IOException;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.springframework.http.MediaType;
+import org.springframework.http.server.ServerHttpResponse;
+import org.springframework.util.Assert;
+
+
+/**
+ * A controller method return value type for asynchronous request processing
+ * where one or more objects are written to the response. While
+ * {@link org.springframework.web.context.request.async.DeferredResult DeferredResult}
+ * is used to produce a single result, a {@code ResponseBodyEmitter} can be used
+ * to send multiple objects where each object is written with a compatible
+ * {@link org.springframework.http.converter.HttpMessageConverter HttpMessageConverter}.
+ *
+ * <p>Supported as a return type on its own as well as within a
+ * {@link org.springframework.http.ResponseEntity ResponseEntity}.
+ *
+ * <pre>
+ * &#064;RequestMapping(value="/stream", method=RequestMethod.GET)
+ * public ResponseBodyEmitter handle() {
+ * 	ResponseBodyEmitter emitter = new ResponseBodyEmitter();
+ * 	// Pass the emitter to another component...
+ * 	return emitter;
+ * }
+ *
+ * // in another thread
+ * emitter.send(foo1);
+ *
+ * // and again
+ * emitter.send(foo2);
+ *
+ * // and done
+ * emitter.complete();
+ * </pre>
+ *
+ * <p><strong>Note:</strong> this class is not thread-safe. Callers must ensure
+ * that use from multiple threads is synchronized.
+ *
+ * @author Rossen Stoyanchev
+ * @since 4.2
+ */
+public class ResponseBodyEmitter {
+
+	private Handler handler;
+
+	/* Cache for objects sent before handler is set. */
+	private final Map<Object, MediaType> initHandlerCache = new LinkedHashMap<Object, MediaType>(10);
+
+	private volatile boolean complete;
+
+	private Throwable failure;
+
+
+	/**
+	 * Invoked after the response is updated with the status code and headers,
+	 * if the ResponseBodyEmitter is wrapped in a ResponseEntity, but before the
+	 * response is committed, i.e. before the response body has been written to.
+	 * <p>The default implementation is empty.
+	 */
+	protected void extendResponse(ServerHttpResponse outputMessage) {
+	}
+
+	void initialize(Handler handler) throws IOException {
+		synchronized (this) {
+			this.handler = handler;
+			for (Map.Entry<Object, MediaType> entry : this.initHandlerCache.entrySet()) {
+				try {
+					sendInternal(entry.getKey(), entry.getValue());
+				}
+				catch (Throwable ex) {
+					return;
+				}
+			}
+			if (this.complete) {
+				if (this.failure != null) {
+					this.handler.completeWithError(this.failure);
+				}
+				else {
+					this.handler.complete();
+				}
+			}
+		}
+	}
+
+	/**
+	 * Write the given object to the response.
+	 * <p>If any exception occurs a dispatch is made back to the app server where
+	 * Spring MVC will pass the exception through its exception handling mechanism.
+	 * @param object the object to write
+	 * @throws IOException raised when an I/O error occurs
+	 * @throws java.lang.IllegalStateException wraps any other errors
+	 */
+	public void send(Object object) throws IOException {
+		send(object, null);
+	}
+
+	/**
+	 * Write the given object to the response also using a MediaType hint.
+	 * <p>If any exception occurs a dispatch is made back to the app server where
+	 * Spring MVC will pass the exception through its exception handling mechanism.
+	 * @param object the object to write
+	 * @param mediaType a MediaType hint for selecting an HttpMessageConverter
+	 * @throws IOException raised when an I/O error occurs
+	 * @throws java.lang.IllegalStateException wraps any other errors
+	 */
+	public void send(Object object, MediaType mediaType) throws IOException {
+		Assert.state(!this.complete, "ResponseBodyEmitter is already set complete.");
+		sendInternal(object, mediaType);
+	}
+
+	private void sendInternal(Object object, MediaType mediaType) throws IOException {
+		if (object == null) {
+			return;
+		}
+		if (handler == null) {
+			synchronized (this) {
+				if (handler == null) {
+					this.initHandlerCache.put(object, mediaType);
+					return;
+				}
+			}
+		}
+		try {
+			this.handler.send(object, mediaType);
+		}
+		catch(IOException ex){
+			this.handler.completeWithError(ex);
+			throw ex;
+		}
+		catch(Throwable ex){
+			this.handler.completeWithError(ex);
+			throw new IllegalStateException("Failed to send " + object, ex);
+		}
+	}
+
+	/**
+	 * Complete request processing.
+	 * <p>A dispatch is made into the app server where Spring MVC completes
+	 * asynchronous request processing.
+	 */
+	public void complete() {
+		synchronized (this) {
+			this.complete = true;
+			if (handler != null) {
+				this.handler.complete();
+			}
+		}
+	}
+
+	/**
+	 * Complete request processing with an error.
+	 * <p>A dispatch is made into the app server where Spring MVC will pass the
+	 * exception through its exception handling mechanism.
+	 */
+	public void completeWithError(Throwable ex) {
+		synchronized (this) {
+			this.complete = true;
+			this.failure = ex;
+			if (handler != null) {
+				this.handler.completeWithError(ex);
+			}
+		}
+	}
+
+
+	/**
+	 * Handle sent objects and complete request processing.
+	 */
+	interface Handler {
+
+		void send(Object data, MediaType mediaType) throws IOException;
+
+		void complete();
+
+		void completeWithError(Throwable failure);
+	}
+
+}

spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/ResponseBodyEmitterReturnValueHandler.java

@@ -0,0 +1,211 @@
+/*
+ * 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.web.servlet.mvc.method.annotation;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.util.List;
+
+import javax.servlet.http.HttpServletResponse;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.core.MethodParameter;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.http.converter.HttpMessageConverter;
+import org.springframework.http.server.ServerHttpResponse;
+import org.springframework.http.server.ServletServerHttpResponse;
+import org.springframework.util.Assert;
+import org.springframework.web.context.request.NativeWebRequest;
+import org.springframework.web.context.request.async.DeferredResult;
+import org.springframework.web.context.request.async.WebAsyncUtils;
+import org.springframework.web.method.support.HandlerMethodReturnValueHandler;
+import org.springframework.web.method.support.ModelAndViewContainer;
+
+
+/**
+ * Supports return values of type {@link ResponseBodyEmitter} and also
+ * {@code ResponseEntity<ResponseBodyEmitter>}.
+ *
+ * @author Rossen Stoyanchev
+ * @since 4.2
+ */
+public class ResponseBodyEmitterReturnValueHandler implements HandlerMethodReturnValueHandler {
+
+	private static final Log logger = LogFactory.getLog(ResponseBodyEmitterReturnValueHandler.class);
+
+	private final List<HttpMessageConverter<?>> messageConverters;
+
+
+	public ResponseBodyEmitterReturnValueHandler(List<HttpMessageConverter<?>> messageConverters) {
+		Assert.notEmpty(messageConverters, "'messageConverters' must not be empty");
+		this.messageConverters = messageConverters;
+	}
+
+	@Override
+	public boolean supportsReturnType(MethodParameter returnType) {
+		if (ResponseBodyEmitter.class.isAssignableFrom(returnType.getParameterType())) {
+			return true;
+		}
+		else if (ResponseEntity.class.isAssignableFrom(returnType.getParameterType())) {
+			Type paramType = returnType.getGenericParameterType();
+			if (paramType instanceof ParameterizedType) {
+				ParameterizedType type = (ParameterizedType) paramType;
+				Type[] typeArguments = type.getActualTypeArguments();
+				if (typeArguments.length == 1) {
+					return ResponseBodyEmitter.class.isAssignableFrom((Class<?>) typeArguments[0]);
+				}
+			}
+		}
+		return false;
+	}
+
+	@Override
+	public void handleReturnValue(Object returnValue, MethodParameter returnType,
+			ModelAndViewContainer mavContainer, NativeWebRequest webRequest) throws Exception {
+
+		if (returnValue == null) {
+			mavContainer.setRequestHandled(true);
+			return;
+		}
+
+		HttpServletResponse response = webRequest.getNativeResponse(HttpServletResponse.class);
+		ServerHttpResponse outputMessage = new ServletServerHttpResponse(response);
+
+		if (ResponseEntity.class.isAssignableFrom(returnValue.getClass())) {
+			ResponseEntity<?> responseEntity = (ResponseEntity<?>) returnValue;
+			outputMessage.setStatusCode(responseEntity.getStatusCode());
+			outputMessage.getHeaders().putAll(responseEntity.getHeaders());
+			returnValue = responseEntity.getBody();
+			if (returnValue == null) {
+				mavContainer.setRequestHandled(true);
+				return;
+			}
+		}
+
+		Assert.isInstanceOf(ResponseBodyEmitter.class, returnValue);
+		ResponseBodyEmitter emitter = (ResponseBodyEmitter) returnValue;
+		emitter.extendResponse(outputMessage);
+
+		// Commit the response and wrap to ignore further header changes
+		outputMessage.getBody();
+		outputMessage = new StreamingServletServerHttpResponse(outputMessage);
+
+		DeferredResult<?> deferredResult = new DeferredResult<Object>();
+		WebAsyncUtils.getAsyncManager(webRequest).startDeferredResultProcessing(deferredResult, mavContainer);
+
+		HttpMessageConvertingHandler handler = new HttpMessageConvertingHandler(outputMessage, deferredResult);
+		emitter.initialize(handler);
+	}
+
+
+	/**
+	 * ResponseBodyEmitter.Handler that writes with HttpMessageConverter's.
+	 */
+	private class HttpMessageConvertingHandler implements ResponseBodyEmitter.Handler {
+
+		private final ServerHttpResponse outputMessage;
+
+		private final DeferredResult<?> deferredResult;
+
+
+		public HttpMessageConvertingHandler(ServerHttpResponse outputMessage, DeferredResult<?> deferredResult) {
+			this.outputMessage = outputMessage;
+			this.deferredResult = deferredResult;
+		}
+
+
+		@Override
+		public void send(Object data, MediaType mediaType) throws IOException {
+			sendInternal(data, mediaType);
+		}
+
+		@SuppressWarnings("unchecked")
+		private <T> void sendInternal(T data, MediaType mediaType) throws IOException {
+			for (HttpMessageConverter<?> converter : ResponseBodyEmitterReturnValueHandler.this.messageConverters) {
+				if (converter.canWrite(data.getClass(), mediaType)) {
+					((HttpMessageConverter<T>) converter).write(data, mediaType, this.outputMessage);
+					this.outputMessage.flush();
+					if (logger.isDebugEnabled()) {
+						logger.debug("Written [" + data + "] using [" + converter + "]");
+					}
+					return;
+				}
+			}
+			throw new IllegalArgumentException("No suitable converter for " + data);
+		}
+
+		@Override
+		public void complete() {
+			this.deferredResult.setResult(null);
+		}
+
+		@Override
+		public void completeWithError(Throwable failure) {
+			this.deferredResult.setErrorResult(failure);
+		}
+	}
+
+
+	/**
+	 * Wrap to silently ignore header changes HttpMessageConverter's that would
+	 * otherwise cause HttpHeaders to raise exceptions.
+	 */
+	private static class StreamingServletServerHttpResponse implements ServerHttpResponse {
+
+		private final ServerHttpResponse delegate;
+
+		private final HttpHeaders mutableHeaders = new HttpHeaders();
+
+
+		public StreamingServletServerHttpResponse(ServerHttpResponse delegate) {
+			this.delegate = delegate;
+			this.mutableHeaders.putAll(delegate.getHeaders());
+		}
+
+
+		@Override
+		public void setStatusCode(HttpStatus status) {
+			this.delegate.setStatusCode(status);
+		}
+
+		@Override
+		public HttpHeaders getHeaders() {
+			return this.mutableHeaders;
+		}
+
+		@Override
+		public OutputStream getBody() throws IOException {
+			return this.delegate.getBody();
+		}
+
+		@Override
+		public void flush() throws IOException {
+			this.delegate.flush();
+		}
+
+		@Override
+		public void close() {
+			this.delegate.close();
+		}
+	}
+
+}

spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/ServletInvocableHandlerMethod.java

@@ -25,6 +25,7 @@ import java.util.concurrent.Callable;
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ResolvableType;
 import org.springframework.http.HttpStatus;
+import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.StringUtils;
 import org.springframework.web.bind.annotation.ResponseStatus;
@@ -260,7 +261,16 @@ public class ServletInvocableHandlerMethod extends InvocableHandlerMethod {
 
 		@Override
 		public Class<?> getParameterType() {
-			return (this.returnValue != null ? this.returnValue.getClass() : this.returnType.getRawClass());
+			if (this.returnValue != null) {
+				return this.returnValue.getClass();
+			}
+			Class<?> parameterType = super.getParameterType();
+			if (ResponseBodyEmitter.class.isAssignableFrom(parameterType)) {
+				return parameterType;
+			}
+			Assert.isTrue(!ResolvableType.NONE.equals(this.returnType), "Expected one of" +
+					"Callable, DeferredResult, or ListenableFuture: " + super.getParameterType());
+			return this.returnType.getRawClass();
 		}
 
 		@Override

spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/SseEmitter.java

@@ -0,0 +1,233 @@
+/*
+ * 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.web.servlet.mvc.method.annotation;
+
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.http.server.ServerHttpResponse;
+
+/**
+ * A specialization of
+ * {@link org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter
+ * ResponseBodyEmitter} for sending
+ * <a href="http://www.w3.org/TR/eventsource/">Server-Sent Events</a>.
+ *
+ * @author Rossen Stoyanchev
+ * @since 4.2
+ */
+public class SseEmitter extends ResponseBodyEmitter {
+
+	public static final MediaType TEXT_PLAIN = new MediaType("text", "plain", Charset.forName("UTF-8"));
+
+
+	@Override
+	protected void extendResponse(ServerHttpResponse outputMessage) {
+		super.extendResponse(outputMessage);
+		HttpHeaders headers = outputMessage.getHeaders();
+		if (headers.getContentType() == null) {
+			headers.setContentType(new MediaType("text", "event-stream"));
+		}
+	}
+
+	/**
+	 * Send the object formatted as a single SSE "data" line. It's equivalent to:
+	 * <pre>
+	 *
+	 * // static import of SseEmitter.*
+	 *
+	 * SseEmitter emitter = new SseEmitter();
+	 * emitter.send(event().data(myObject));
+	 * </pre>
+	 * @param object the object to write
+	 * @throws IOException raised when an I/O error occurs
+	 * @throws java.lang.IllegalStateException wraps any other errors
+	 */
+	@Override
+	public void send(Object object) throws IOException {
+		send(object, null);
+	}
+
+	/**
+	 * Send the object formatted as a single SSE "data" line. It's equivalent to:
+	 * <pre>
+	 *
+	 * // static import of SseEmitter.*
+	 *
+	 * SseEmitter emitter = new SseEmitter();
+	 * emitter.send(event().data(myObject, MediaType.APPLICATION_JSON));
+	 * </pre>
+	 * @param object the object to write
+	 * @param mediaType a MediaType hint for selecting an HttpMessageConverter
+	 * @throws IOException raised when an I/O error occurs
+	 * @throws java.lang.IllegalStateException wraps any other errors
+	 */
+	@Override
+	public void send(Object object, MediaType mediaType) throws IOException {
+		if (object == null) {
+			return;
+		}
+		send(event().data(object, mediaType));
+	}
+
+	/**
+	 * Send an SSE event prepared with the given builder. For example:
+	 * <pre>
+	 *
+	 * // static import of SseEmitter
+	 *
+	 * SseEmitter emitter = new SseEmitter();
+	 * emitter.send(event().name("update").id("1").data(myObject));
+	 * </pre>
+	 * @param builder a builder for an SSE formatted event.
+	 * @throws IOException raised when an I/O error occurs
+	 * @throws java.lang.IllegalStateException wraps any other errors
+	 */
+	public void send(SseEventBuilder builder) throws IOException {
+		Map<Object, MediaType> map = builder.build();
+		for (Map.Entry<Object, MediaType> entry : map.entrySet()) {
+			super.send(entry.getKey(), entry.getValue());
+		}
+	}
+
+	public static SseEventBuilder event() {
+		return new DefaultSseEventBuilder();
+	}
+
+
+	/**
+	 * A builder for an SSE event.
+	 */
+	public interface SseEventBuilder {
+
+		/**
+		 * Add an SSE "comment" line.
+		 */
+		SseEventBuilder comment(String comment);
+
+		/**
+		 * Add an SSE "event" line.
+		 */
+		SseEventBuilder name(String eventName);
+
+		/**
+		 * Add an SSE "id" line.
+		 */
+		SseEventBuilder id(String id);
+
+		/**
+		 * Add an SSE "event" line.
+		 */
+		SseEventBuilder reconnectTime(long reconnectTimeMillis);
+
+		/**
+		 * Add an SSE "data" line.
+		 */
+		SseEventBuilder data(Object object);
+
+		/**
+		 * Add an SSE "data" line.
+		 */
+		SseEventBuilder data(Object object, MediaType mediaType);
+
+		/**
+		 * Return a map with objects that represent the data to be written to
+		 * the response as well as the required SSE text formatting that
+		 * surrounds it.
+		 */
+		Map<Object, MediaType> build();
+	}
+
+	/**
+	 * Default implementation of SseEventBuilder.
+	 */
+	private static class DefaultSseEventBuilder implements SseEventBuilder {
+
+		private final Map<Object, MediaType> map = new LinkedHashMap<Object, MediaType>(4);
+
+		private StringBuilder sb;
+
+
+		@Override
+		public SseEventBuilder comment(String comment) {
+			append(":").append(comment != null ? comment : "").append("\n");
+			return this;
+		}
+
+		@Override
+		public SseEventBuilder name(String name) {
+			append("name:").append(name != null ? name : "").append("\n");
+			return this;
+		}
+
+		@Override
+		public SseEventBuilder id(String id) {
+			append("id:").append(id != null ? id : "").append("\n");
+			return this;
+		}
+
+		@Override
+		public SseEventBuilder reconnectTime(long reconnectTimeMillis) {
+			append("retry:").append(String.valueOf(reconnectTimeMillis)).append("\n");
+			return this;
+		}
+
+		@Override
+		public SseEventBuilder data(Object object) {
+			return data(object, null);
+		}
+
+		@Override
+		public SseEventBuilder data(Object object, MediaType mediaType) {
+			append("data:");
+			saveAppendedText();
+			this.map.put(object, mediaType);
+			append("\n");
+			return this;
+		}
+
+		DefaultSseEventBuilder append(String text) {
+			if (this.sb == null) {
+				this.sb = new StringBuilder();
+			}
+			this.sb.append(text);
+			return this;
+		}
+
+		private void saveAppendedText() {
+			if (this.sb != null) {
+				this.map.put(this.sb.toString(), TEXT_PLAIN);
+				this.sb = null;
+			}
+		}
+
+		@Override
+		public Map<Object, MediaType> build() {
+			if (this.sb == null || this.sb.length() == 0 && this.map.isEmpty()) {
+				return Collections.<Object, MediaType>emptyMap();
+			}
+			append("\n");
+			saveAppendedText();
+			return this.map;
+		}
+	}
+
+}

spring-webmvc/src/main/java/org/springframework/web/servlet/support/ServletUriComponentsBuilder.java

@@ -79,7 +79,7 @@ public class ServletUriComponentsBuilder extends UriComponentsBuilder {
 	 * {@code "/"} or {@code "*.do"}, the result will be the same as
 	 * if calling {@link #fromContextPath(HttpServletRequest)}.
 	 */
-	public static ServletUriComponentsBuilder fromServletMapping(HttpServletRequest request) {
+		public static ServletUriComponentsBuilder fromServletMapping(HttpServletRequest request) {
 		ServletUriComponentsBuilder builder = fromContextPath(request);
 		if (StringUtils.hasText(new UrlPathHelper().getPathWithinServletMapping(request))) {
 			builder.path(request.getServletPath());