You are viewing a plain text version of this content. The canonical link for it is here.
Posted to server-dev@james.apache.org by no...@apache.org on 2011/09/28 08:57:35 UTC

svn commit: r1176753 - /james/protocols/trunk/impl/src/main/java/org/apache/james/protocols/impl/NettyProtocolTransport.java

Author: norman
Date: Wed Sep 28 06:57:35 2011
New Revision: 1176753

URL: http://svn.apache.org/viewvc?rev=1176753&view=rev
Log:
Make sure the Response's are written in the right order even if we mix FutureResponse and Response implementations. See PROTOCOLS-36

Modified:
    james/protocols/trunk/impl/src/main/java/org/apache/james/protocols/impl/NettyProtocolTransport.java

Modified: james/protocols/trunk/impl/src/main/java/org/apache/james/protocols/impl/NettyProtocolTransport.java
URL: http://svn.apache.org/viewvc/james/protocols/trunk/impl/src/main/java/org/apache/james/protocols/impl/NettyProtocolTransport.java?rev=1176753&r1=1176752&r2=1176753&view=diff
==============================================================================
--- james/protocols/trunk/impl/src/main/java/org/apache/james/protocols/impl/NettyProtocolTransport.java (original)
+++ james/protocols/trunk/impl/src/main/java/org/apache/james/protocols/impl/NettyProtocolTransport.java Wed Sep 28 06:57:35 2011
@@ -20,6 +20,7 @@
 package org.apache.james.protocols.impl;
 
 import java.net.InetSocketAddress;
+import java.util.concurrent.ConcurrentLinkedQueue;
 
 import javax.net.ssl.SSLEngine;
 
@@ -40,10 +41,13 @@ import org.jboss.netty.handler.ssl.SslHa
  */
 public class NettyProtocolTransport implements ProtocolTransport {
     
-    private Channel channel;
-    private SSLEngine engine;
+    private final Channel channel;
+    private final SSLEngine engine;
     private int lineHandlerCount = 0;
-
+    
+    // TODO: Should we limit the size ?
+    private final ConcurrentLinkedQueue<Response> responses = new ConcurrentLinkedQueue<Response>();
+    
     public NettyProtocolTransport(Channel channel, SSLEngine engine) {
         this.channel = channel;
         this.engine = engine;
@@ -89,15 +93,52 @@ public class NettyProtocolTransport impl
      * @see org.apache.james.protocols.api.ProtocolTransport#writeResponse(org.apache.james.protocols.api.Response, org.apache.james.protocols.api.ProtocolSession)
      */
     public void writeResponse(Response response, final ProtocolSession session) {
-        if (response instanceof FutureResponse) {
-            ((FutureResponse) response).addListener(new ResponseListener() {
-                
-                public void onResponse(FutureResponse response) {
-                    writeResponseToChannel(response, session);                    
+        // just add the response to the queue. We will trigger the write operation later
+        responses.add(response);
+         
+        // trigger the write
+        writeQueuedResponses(session);
+    }
+    
+    /**
+     * Helper method which tries to write all queued {@link Response}'s to the remote client. This method is aware of {@link FutureResponse} and makes sure the {@link Response}'s are written
+     * in the correct order
+     * 
+     * This is related to PROTOCOLS-36
+     * 
+     * @param session
+     */
+    private void writeQueuedResponses(final ProtocolSession session) {
+        Response queuedResponse = null;
+        
+        // dequeue Responses until non is left
+        while ((queuedResponse = responses.poll()) != null) {
+            
+            // check if we need to take special care of FutureResponses
+            if (queuedResponse instanceof FutureResponse) {
+                FutureResponse futureResponse =(FutureResponse) queuedResponse;
+                if (futureResponse.isReady()) {
+                    // future is ready so we can write it without blocking the IO-Thread
+                    writeResponseToChannel(queuedResponse, session);
+                } else {
+                    
+                    // future is not ready so we need to write it via a ResponseListener otherwise we MAY block the IO-Thread
+                    futureResponse.addListener(new ResponseListener() {
+                        
+                        public void onResponse(FutureResponse response) {
+                            writeResponseToChannel(response, session);
+                            writeQueuedResponses(session);
+                        }
+                    });
+                    
+                    // just break here as we will trigger the dequeue later
+                    break;
                 }
-            });
-        } else {
-            writeResponseToChannel(response, session);
+                
+            } else {
+                // the Response is not a FutureResponse, so just write it back the the remote peer
+                writeResponseToChannel(queuedResponse, session);
+            }
         }
     }
     



---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org