[cxf] branch 3.3.x-fixes updated: CXF-8215: SSE breaks Pipeline Processing. Adding proper headers propagation.

[reta]

This is an automated email from the ASF dual-hosted git repository.

reta pushed a commit to branch 3.3.x-fixes
in repository https://gitbox.apache.org/repos/asf/cxf.git


The following commit(s) were added to refs/heads/3.3.x-fixes by this push:
     new 3a3ea18  CXF-8215: SSE breaks Pipeline Processing. Adding proper 
headers propagation.
3a3ea18 is described below

commit 3a3ea182277cd25c5cdd9c779970e1d1127e5688
Author: reta <drr...@gmail.com>
AuthorDate: Sun Mar 1 10:50:44 2020 -0500

    CXF-8215: SSE breaks Pipeline Processing. Adding proper headers propagation.
    
    (cherry picked from commit 7ed31b3a6c6b1b602b5e60e9379ec94503abb1b6)
---
 .../cxf/jaxrs/sse/interceptor/SseInterceptor.java  | 61 ++++++++++++++++++++--
 1 file changed, 56 insertions(+), 5 deletions(-)

diff --git 
a/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/interceptor/SseInterceptor.java
 
b/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/interceptor/SseInterceptor.java
index 8cbdaa1..560e10c 100644
--- 
a/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/interceptor/SseInterceptor.java
+++ 
b/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/interceptor/SseInterceptor.java
@@ -19,16 +19,21 @@
 
 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;
@@ -49,28 +54,51 @@ public class SseInterceptor extends 
AbstractPhaseInterceptor<Message> {
     public void handleMessage(Message message) {
         if (!isRequestor(message) && message.get(SseInterceptor.class) == 
null) {
             message.put(SseInterceptor.class, this);
-            
+
             final Exchange exchange = message.getExchange();
             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()));
+        }
+    }
 }