From 5701b4ba67b8e475326acfc9f28735aead8d9dfc Mon Sep 17 00:00:00 2001 From: Robert Godfrey Date: Wed, 11 Feb 2015 16:54:29 +0000 Subject: more refactoring git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/QPID-6262-JavaBrokerNIO@1659004 13f79535-47bb-0310-9956-ffa450edef68 --- .../server/protocol/v0_10/ProtocolEngine_0_10.java | 16 ++++++++++++++++ .../qpid/server/protocol/v0_10/ServerConnection.java | 3 --- .../qpid/server/protocol/v0_8/AMQProtocolEngine.java | 20 +++++++++++++++++--- .../qpid/server/protocol/v1_0/Connection_1_0.java | 3 --- .../protocol/v1_0/ProtocolEngine_1_0_0_SASL.java | 17 +++++++++++++++++ 5 files changed, 50 insertions(+), 9 deletions(-) (limited to 'qpid/java/broker-plugins') 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 e521754edf..5f99ddf56c 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 @@ -25,6 +25,7 @@ import java.nio.ByteBuffer; import java.security.AccessController; import java.security.PrivilegedAction; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import javax.security.auth.Subject; @@ -36,6 +37,7 @@ import org.apache.qpid.server.logging.messages.ConnectionMessages; import org.apache.qpid.server.model.Consumer; import org.apache.qpid.server.model.Port; import org.apache.qpid.server.protocol.AMQSessionModel; +import org.apache.qpid.server.util.Action; import org.apache.qpid.transport.ByteBufferSender; import org.apache.qpid.transport.Constant; import org.apache.qpid.transport.network.Assembler; @@ -61,6 +63,8 @@ public class ProtocolEngine_0_10 extends InputHandler implements ServerProtocol private volatile boolean _messageAssignmentSuspended; private final AtomicBoolean _stateChanged = new AtomicBoolean(); + private final AtomicReference> _workListener = new AtomicReference<>(); + public ProtocolEngine_0_10(ServerConnection conn, NetworkConnection network) @@ -302,6 +306,12 @@ public class ProtocolEngine_0_10 extends InputHandler implements ServerProtocol public void notifyWork() { _stateChanged.set(true); + + final Action listener = _workListener.get(); + if(listener != null) + { + listener.performAction(this); + } } @Override @@ -309,4 +319,10 @@ public class ProtocolEngine_0_10 extends InputHandler implements ServerProtocol { _stateChanged.set(false); } + + @Override + public void setWorkListener(final Action listener) + { + _workListener.set(listener); + } } 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 dce656e7fd..caa8b90485 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 @@ -690,9 +690,6 @@ public class ServerConnection extends Connection implements AMQConnectionModel { + + enum ConnectionState { INIT, @@ -120,6 +123,7 @@ public class AMQProtocolEngine implements ServerProtocolEngine, private final AmqpPort _port; private final long _creationTime; private final AtomicBoolean _stateChanged = new AtomicBoolean(); + private final AtomicReference> _workListener = new AtomicReference<>(); private AMQShortString _contextKey; @@ -218,7 +222,6 @@ public class AMQProtocolEngine implements ServerProtocolEngine, public void setMessageAssignmentSuspended(final boolean messageAssignmentSuspended) { _messageAssignmentSuspended = messageAssignmentSuspended; - if(!messageAssignmentSuspended) { for(AMQSessionModel session : getSessionModels()) @@ -2095,8 +2098,13 @@ public class AMQProtocolEngine implements ServerProtocolEngine, { _stateChanged.set(true); - // TODO - _sender.flush(); + final Action listener = _workListener.get(); + _logger.info("Work lister is null? " + (listener == null)); + if(listener != null) + { + + listener.performAction(this); + } } @Override @@ -2104,4 +2112,10 @@ public class AMQProtocolEngine implements ServerProtocolEngine, { _stateChanged.set(false); } + + @Override + public void setWorkListener(final Action listener) + { + _workListener.set(listener); + } } 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 09250ea6ac..097abe9d8b 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 @@ -503,9 +503,6 @@ public class Connection_1_0 implements ConnectionEventListener, AMQConnectionMod public void notifyWork() { _protocolEngine.notifyWork(); - - // TODO - _protocolEngine.flushBatched(); } @Override 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 856aa14947..3d7712d1ad 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 @@ -30,6 +30,7 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import javax.security.auth.Subject; import javax.security.sasl.SaslException; @@ -62,6 +63,7 @@ import org.apache.qpid.server.model.port.AmqpPort; import org.apache.qpid.server.protocol.AMQSessionModel; import org.apache.qpid.server.security.SubjectCreator; import org.apache.qpid.server.security.auth.UsernamePrincipal; +import org.apache.qpid.server.util.Action; import org.apache.qpid.server.util.ServerScopedRuntimeException; import org.apache.qpid.transport.ByteBufferSender; import org.apache.qpid.transport.TransportException; @@ -84,6 +86,8 @@ public class ProtocolEngine_1_0_0_SASL implements ServerProtocolEngine, FrameOut private ConnectionEndpoint _endpoint; private long _connectionId; private final AtomicBoolean _stateChanged = new AtomicBoolean(); + private final AtomicReference> _workListener = new AtomicReference<>(); + private static final ByteBuffer HEADER = ByteBuffer.wrap(new byte[] @@ -633,6 +637,12 @@ public class ProtocolEngine_1_0_0_SASL implements ServerProtocolEngine, FrameOut public void notifyWork() { _stateChanged.set(true); + + final Action listener = _workListener.get(); + if(listener != null) + { + listener.performAction(this); + } } @Override @@ -640,4 +650,11 @@ public class ProtocolEngine_1_0_0_SASL implements ServerProtocolEngine, FrameOut { _stateChanged.set(false); } + + @Override + public void setWorkListener(final Action listener) + { + _workListener.set(listener); + } + } -- cgit v1.2.1