More refactoring

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/QPID-6262-JavaBrokerNIO@1658990 13f79535-47bb-0310-9956-ffa450edef68

6 files changed, 72 insertions, 9 deletions

diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java
index 5f227e5f18..e521754edf 100755
--- a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java
@@ -24,6 +24,7 @@ import java.net.SocketAddress;
 import java.nio.ByteBuffer;
 import java.security.AccessController;
 import java.security.PrivilegedAction;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import javax.security.auth.Subject;
 
@@ -59,6 +60,7 @@ public class ProtocolEngine_0_10  extends InputHandler implements ServerProtocol
     private volatile boolean _transportBlockedForWriting;
 
     private volatile boolean _messageAssignmentSuspended;
+    private final AtomicBoolean _stateChanged = new AtomicBoolean();
 
     public ProtocolEngine_0_10(ServerConnection conn,
                                NetworkConnection network)
@@ -289,4 +291,22 @@ public class ProtocolEngine_0_10  extends InputHandler implements ServerProtocol
             session.processPendingMessages();
         }
     }
+
+    @Override
+    public boolean hasWork()
+    {
+        return _stateChanged.get();
+    }
+
+    @Override
+    public void notifyWork()
+    {
+        _stateChanged.set(true);
+    }
+
+    @Override
+    public void clearWork()
+    {
+        _stateChanged.set(false);
+    }
 }
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
index a2f1f1a4ba..dce656e7fd 100644
--- a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
@@ -687,8 +687,11 @@ public class ServerConnection extends Connection implements AMQConnectionModel<S
     }
 
     @Override
-    public void flushBatched()
+    public void notifyWork()
     {
+        _serverProtocolEngine.notifyWork();
+
+        // TODO
         getSender().flush();
     }
 
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
index ed075038e6..b1547f13e2 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
@@ -97,6 +97,7 @@ public class AMQProtocolEngine implements ServerProtocolEngine,
                                           AMQConnectionModel<AMQProtocolEngine, AMQChannel>,
                                           ServerMethodProcessor<ServerChannelMethodProcessor>
 {
+
     enum ConnectionState
     {
         INIT,
@@ -118,6 +119,7 @@ public class AMQProtocolEngine implements ServerProtocolEngine,
     private static final long AWAIT_CLOSED_TIMEOUT = 60000;
     private final AmqpPort<?> _port;
     private final long _creationTime;
+    private final AtomicBoolean _stateChanged = new AtomicBoolean();
 
     private AMQShortString _contextKey;
 
@@ -339,10 +341,6 @@ public class AMQProtocolEngine implements ServerProtocolEngine,
         return _closing.get();
     }
 
-    public synchronized void flushBatched()
-    {
-        _sender.flush();
-    }
 
 
     public ClientDeliveryMethod createDeliveryMethod(int channelId)
@@ -2085,4 +2083,25 @@ public class AMQProtocolEngine implements ServerProtocolEngine,
             session.processPendingMessages();
         }
     }
+
+    @Override
+    public boolean hasWork()
+    {
+        return _stateChanged.get();
+    }
+
+    @Override
+    public void notifyWork()
+    {
+        _stateChanged.set(true);
+
+        // TODO
+        _sender.flush();
+    }
+
+    @Override
+    public void clearWork()
+    {
+        _stateChanged.set(false);
+    }
 }
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java
index d33a4aafd8..f9560fa0d2 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java
@@ -21,7 +21,6 @@
 package org.apache.qpid.server.protocol.v0_8;
 
 import java.util.List;
-import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
@@ -34,7 +33,6 @@ import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.FieldTable;
 import org.apache.qpid.server.consumer.AbstractConsumerTarget;
 import org.apache.qpid.server.consumer.ConsumerImpl;
-import org.apache.qpid.server.consumer.ConsumerMessageInstancePair;
 import org.apache.qpid.server.flow.FlowCreditManager;
 import org.apache.qpid.server.message.InstanceProperties;
 import org.apache.qpid.server.message.MessageInstance;
@@ -524,7 +522,7 @@ public abstract class ConsumerTarget_0_8 extends AbstractConsumerTarget implemen
     {
         _channel.getConnection().setDeferFlush(false);
 
-        _channel.getConnection().flushBatched();
+        _channel.getConnection().notifyWork();
     }
 
     protected void addUnacknowledgedMessage(MessageInstance entry)
diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
index a442b5c437..09250ea6ac 100644
--- a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
+++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
@@ -500,8 +500,11 @@ public class Connection_1_0 implements ConnectionEventListener, AMQConnectionMod
     }
 
     @Override
-    public void flushBatched()
+    public void notifyWork()
     {
+        _protocolEngine.notifyWork();
+
+        // TODO
         _protocolEngine.flushBatched();
     }
 
diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngine_1_0_0_SASL.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngine_1_0_0_SASL.java
index d361dce682..856aa14947 100644
--- a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngine_1_0_0_SASL.java
+++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngine_1_0_0_SASL.java
@@ -29,6 +29,7 @@ import java.security.PrivilegedAction;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import javax.security.auth.Subject;
 import javax.security.sasl.SaslException;
@@ -82,6 +83,7 @@ public class ProtocolEngine_1_0_0_SASL implements ServerProtocolEngine, FrameOut
     private long _createTime = System.currentTimeMillis();
     private ConnectionEndpoint _endpoint;
     private long _connectionId;
+    private final AtomicBoolean _stateChanged = new AtomicBoolean();
 
     private static final ByteBuffer HEADER =
            ByteBuffer.wrap(new byte[]
@@ -620,4 +622,22 @@ public class ProtocolEngine_1_0_0_SASL implements ServerProtocolEngine, FrameOut
             session.processPendingMessages();
         }
     }
+
+    @Override
+    public boolean hasWork()
+    {
+        return _stateChanged.get();
+    }
+
+    @Override
+    public void notifyWork()
+    {
+        _stateChanged.set(true);
+    }
+
+    @Override
+    public void clearWork()
+    {
+        _stateChanged.set(false);
+    }
 }