SSE breaks Pipeline Processing. Adding proper headers propagation.
package org.apache.cxf.jaxrs.sse.interceptor;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.servlet.http.HttpServletResponse;
+import javax.ws.rs.core.MultivaluedMap;
import javax.ws.rs.core.Response;
import org.apache.cxf.common.logging.LogUtils;
import org.apache.cxf.jaxrs.impl.ResponseImpl;
import org.apache.cxf.jaxrs.model.OperationResourceInfo;
import org.apache.cxf.jaxrs.provider.ServerProviderFactory;
+import org.apache.cxf.jaxrs.utils.HttpUtils;
import org.apache.cxf.jaxrs.utils.JAXRSUtils;
import org.apache.cxf.message.Exchange;
import org.apache.cxf.message.Message;
@@ -54,23 +59,46 @@ public class SseInterceptor extends AbstractPhaseInterceptor<Message> {
OperationResourceInfo ori = (OperationResourceInfo)exchange.get(OperationResourceInfo.class.getName());
if (ori != null) {
Response.ResponseBuilder builder = Response.ok();
+ HttpServletResponse servletResponse = null;
final ServerProviderFactory providerFactory = ServerProviderFactory.getInstance(message);
final Object response = message.get(AbstractHTTPDestination.HTTP_RESPONSE);
if (response instanceof HttpServletResponse) {
- final HttpServletResponse servletResponse = (HttpServletResponse)response;
-
+ servletResponse = (HttpServletResponse)response;
builder = Response.status(servletResponse.getStatus());
+
+ @SuppressWarnings("unchecked")
+ final Map<String, List<Object>> userHeaders = (Map<String, List<Object>>)message
+ .get(Message.PROTOCOL_HEADERS);
+
+ if (userHeaders != null) {
+ for (Map.Entry<String, List<Object>> entry: userHeaders.entrySet()) {
+ addHeader(builder, entry);
+ }
+ }
+
for (final String header: servletResponse.getHeaderNames()) {
- builder.header(header, servletResponse.getHeaders(header));
+ final Collection<String> headers = servletResponse.getHeaders(header);
+ addHeader(builder, header, headers);
}
}
// Run the filters
try {
final ResponseImpl responseImpl = (ResponseImpl)builder.build();
+
JAXRSUtils.runContainerResponseFilters(providerFactory, responseImpl,
message, ori, ori.getAnnotatedMethod());
+
+ if (servletResponse != null) {
+ final MultivaluedMap<String, String> headers = responseImpl.getStringHeaders();
+ servletResponse.setStatus(responseImpl.getStatus());
+ if (headers != null) {
+ for (Map.Entry<String, List<String>> entry: headers.entrySet()) {
+ setHeader(servletResponse, entry);
+ }
+ }
+ }
} catch (Throwable ex) {
if (LOG.isLoggable(Level.FINE)) {
LOG.log(Level.FINE, ex.getMessage(), ex);
@@ -79,4 +107,27 @@ public class SseInterceptor extends AbstractPhaseInterceptor<Message> {
}
}
}
+
+ private void addHeader(Response.ResponseBuilder builder, Map.Entry<String, List<Object>> entry) {
+ if (entry.getValue() != null) {
+ for (Object value: entry.getValue()) {
+ builder.header(entry.getKey(), value);
+ }
+ }
+ }
+
+ private void addHeader(Response.ResponseBuilder builder, final String header, final Collection<String> headers) {
+ if (headers != null) {
+ for (Object value: headers) {
+ builder.header(header, value);
+ }
+ }
+ }
+
+ private void setHeader(HttpServletResponse servletResponse, Map.Entry<String, List<String>> entry) {
+ if (entry.getValue() != null) {
+ servletResponse.setHeader(entry.getKey(),
+ HttpUtils.getHeaderString(entry.getValue()));
+ }
+ }
}
