diff options
Diffstat (limited to 'java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0')
4 files changed, 106 insertions, 31 deletions
diff --git a/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java b/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java index f429d8ba9f..cf4164c244 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java +++ b/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java @@ -31,7 +31,6 @@ import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.server.logging.LogSubject; import org.apache.qpid.server.protocol.AMQConnectionModel; import org.apache.qpid.server.protocol.AMQSessionModel; -import org.apache.qpid.server.registry.IApplicationRegistry; import org.apache.qpid.server.stats.StatisticsCounter; import org.apache.qpid.server.virtualhost.VirtualHost; @@ -44,7 +43,6 @@ import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.CONNECTIO public class Connection_1_0 implements ConnectionEventListener { - private IApplicationRegistry _appRegistry; private VirtualHost _vhost; private final ConnectionEndpoint _conn; private final long _connectionId; @@ -62,10 +60,9 @@ public class Connection_1_0 implements ConnectionEventListener - public Connection_1_0(IApplicationRegistry appRegistry, ConnectionEndpoint conn, long connectionId) + public Connection_1_0(VirtualHost virtualHost, ConnectionEndpoint conn, long connectionId) { - _appRegistry = appRegistry; - _vhost = _appRegistry.getVirtualHostRegistry().getDefaultVirtualHost(); + _vhost = virtualHost; _conn = conn; _connectionId = connectionId; _vhost.getConnectionRegistry().registerConnection(_model); @@ -74,7 +71,7 @@ public class Connection_1_0 implements ConnectionEventListener public void remoteSessionCreation(SessionEndpoint endpoint) { - Session_1_0 session = new Session_1_0(_vhost, _appRegistry, this); + Session_1_0 session = new Session_1_0(_vhost, this); _sessions.add(session); endpoint.setSessionEventListener(session); } diff --git a/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/ExchangeDestination.java b/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/ExchangeDestination.java index ba1a1ca45c..2cef27267b 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/ExchangeDestination.java +++ b/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/ExchangeDestination.java @@ -80,7 +80,7 @@ public class ExchangeDestination implements ReceivingDestination, SendingDestina { // NO-OP } - }, System.currentTimeMillis()); + }); return ACCEPTED; } diff --git a/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Message_1_0.java b/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Message_1_0.java index 140a815f57..fbce1666b7 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Message_1_0.java +++ b/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Message_1_0.java @@ -23,9 +23,10 @@ package org.apache.qpid.server.protocol.v1_0; import java.lang.ref.WeakReference; import java.nio.ByteBuffer; +import java.util.ArrayList; import java.util.List; +import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.server.configuration.SessionConfig; import org.apache.qpid.server.message.InboundMessage; import org.apache.qpid.server.message.MessageMetaData_1_0; import org.apache.qpid.server.message.MessageReference; @@ -34,11 +35,45 @@ import org.apache.qpid.server.store.StoredMessage; public class Message_1_0 implements ServerMessage, InboundMessage { + + + private static final AtomicIntegerFieldUpdater<Message_1_0> _refCountUpdater = + AtomicIntegerFieldUpdater.newUpdater(Message_1_0.class, "_referenceCount"); + + private volatile int _referenceCount = 0; + private final StoredMessage<MessageMetaData_1_0> _storedMessage; private List<ByteBuffer> _fragments; private WeakReference<Session_1_0> _session; + public Message_1_0(final StoredMessage<MessageMetaData_1_0> storedMessage) + { + _storedMessage = storedMessage; + _session = null; + _fragments = restoreFragments(storedMessage); + } + + private static List<ByteBuffer> restoreFragments(StoredMessage<MessageMetaData_1_0> storedMessage) + { + ArrayList<ByteBuffer> fragments = new ArrayList<ByteBuffer>(); + final int FRAGMENT_SIZE = 2048; + int offset = 0; + ByteBuffer b; + do + { + + b = storedMessage.getContent(offset,FRAGMENT_SIZE); + if(b.hasRemaining()) + { + fragments.add(b); + offset+= b.remaining(); + } + } + while(b.hasRemaining()); + return fragments; + } + public Message_1_0(final StoredMessage<MessageMetaData_1_0> storedMessage, final List<ByteBuffer> fragments, final Session_1_0 session) @@ -136,11 +171,6 @@ public class Message_1_0 implements ServerMessage, InboundMessage return buf; } - public SessionConfig getSessionConfig() - { - return null; //TODO - } - public List<ByteBuffer> getFragments() { return _fragments; @@ -148,7 +178,61 @@ public class Message_1_0 implements ServerMessage, InboundMessage public Session_1_0 getSession() { - return _session.get(); + return _session == null ? null : _session.get(); + } + + + public boolean incrementReference() + { + if(_refCountUpdater.incrementAndGet(this) <= 0) + { + _refCountUpdater.decrementAndGet(this); + return false; + } + else + { + return true; + } + } + + /** + * Threadsafe. This will decrement the reference count and when it reaches zero will remove the message from the + * message store. + * + * + * @throws org.apache.qpid.server.queue.MessageCleanupException when an attempt was made to remove the message from the message store and that + * failed + */ + public void decrementReference() + { + int count = _refCountUpdater.decrementAndGet(this); + + // note that the operation of decrementing the reference count and then removing the message does not + // have to be atomic since the ref count starts at 1 and the exchange itself decrements that after + // the message has been passed to all queues. i.e. we are + // not relying on the all the increments having taken place before the delivery manager decrements. + if (count == 0) + { + // set the reference count way below 0 so that we can detect that the message has been deleted + // this is to guard against the message being spontaneously recreated (from the mgmt console) + // by copying from other queues at the same time as it is being removed. + _refCountUpdater.set(this,Integer.MIN_VALUE/2); + + // must check if the handle is null since there may be cases where we decide to throw away a message + // and the handle has not yet been constructed + if (_storedMessage != null) + { + _storedMessage.remove(); + } + } + else + { + if (count < 0) + { + throw new RuntimeException("Reference count for message id " + getMessageNumber() + + " has gone below 0."); + } + } } public static class Reference extends MessageReference<Message_1_0> @@ -160,13 +244,13 @@ public class Message_1_0 implements ServerMessage, InboundMessage protected void onReference(Message_1_0 message) { - + message.incrementReference(); } protected void onRelease(Message_1_0 message) { - + message.decrementReference(); } -} + } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java b/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java index 999ffc55e5..a0ed824c58 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java +++ b/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java @@ -36,7 +36,6 @@ import org.apache.qpid.amqp_1_0.type.transport.*; import org.apache.qpid.amqp_1_0.type.transport.Error; import org.apache.qpid.AMQException; import org.apache.qpid.AMQSecurityException; -import org.apache.qpid.protocol.ProtocolEngine; import org.apache.qpid.server.exchange.Exchange; import org.apache.qpid.server.logging.LogSubject; import org.apache.qpid.server.message.InboundMessage; @@ -45,8 +44,6 @@ import org.apache.qpid.server.protocol.AMQConnectionModel; import org.apache.qpid.server.protocol.AMQSessionModel; import org.apache.qpid.server.queue.AMQQueue; import org.apache.qpid.server.queue.AMQQueueFactory; -import org.apache.qpid.server.registry.IApplicationRegistry; -import org.apache.qpid.server.transport.ServerConnection; import org.apache.qpid.server.txn.AutoCommitTransaction; import org.apache.qpid.server.txn.ServerTransaction; import org.apache.qpid.server.virtualhost.VirtualHost; @@ -58,7 +55,6 @@ import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.CHANNEL_F public class Session_1_0 implements SessionEventListener, AMQSessionModel, LogSubject { private static final Symbol LIFETIME_POLICY = Symbol.valueOf("lifetime-policy"); - private IApplicationRegistry _appRegistry; private VirtualHost _vhost; private AutoCommitTransaction _transaction; @@ -68,9 +64,8 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel, LogSu private UUID _id = UUID.randomUUID(); - public Session_1_0(VirtualHost vhost, IApplicationRegistry appRegistry, final Connection_1_0 connection) + public Session_1_0(VirtualHost vhost, final Connection_1_0 connection) { - _appRegistry = appRegistry; _vhost = vhost; _transaction = new AutoCommitTransaction(vhost.getMessageStore()); _connection = connection; @@ -456,8 +451,9 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel, LogSu { } + @Override - public UUID getQMFId() + public UUID getId() { return _id; } @@ -580,13 +576,6 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel, LogSu return 0; } - @Override - public int compareTo(AMQSessionModel o) - { - return getQMFId().compareTo(o.getQMFId()); - } - - public String toLogString() { @@ -604,4 +593,9 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel, LogSu + "] "; } + @Override + public int compareTo(AMQSessionModel o) + { + return getId().compareTo(o.getId()); + } } |