You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2012/03/10 20:22:37 UTC
svn commit: r1299257 [10/26] - in
/qpid/branches/rg-amqp-1-0-sandbox/qpid/java: broker-plugins/
broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/
broker-plugins/access-control/src/main/java/org/apache/qpid/serve...
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java Sat Mar 10 19:22:10 2012
@@ -24,7 +24,6 @@ import org.apache.log4j.Logger;
import org.apache.qpid.AMQException;
import org.apache.qpid.common.AMQPFilterTypes;
-import org.apache.qpid.common.ClientProperties;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.server.AMQChannel;
@@ -320,9 +319,6 @@ public abstract class SubscriptionImpl i
private final Boolean _autoClose;
-
- private static final String CLIENT_PROPERTIES_INSTANCE = ClientProperties.instance.toString();
-
private AMQQueue _queue;
private final AtomicBoolean _deleted = new AtomicBoolean(false);
@@ -479,10 +475,6 @@ public abstract class SubscriptionImpl i
public boolean hasInterest(QueueEntry entry)
{
-
-
-
-
//check that the message hasn't been rejected
if (entry.isRejectedBy(getSubscriptionID()))
{
@@ -490,27 +482,21 @@ public abstract class SubscriptionImpl i
{
_logger.debug("Subscription:" + this + " rejected message:" + entry);
}
-// return false;
}
if (_noLocal)
{
-
AMQMessage message = (AMQMessage) entry.getMessage();
- //todo - client id should be recorded so we don't have to handle
- // the case where this is null.
- final Object publisher = message.getPublisherIdentifier();
+ final Object publisherReference = message.getConnectionIdentifier();
// We don't want local messages so check to see if message is one we sent
- Object localInstance = getProtocolSession();
+ Object localReference = getProtocolSession().getReference();
- if(publisher.equals(localInstance))
+ if(publisherReference != null && publisherReference.equals(localReference))
{
return false;
}
-
-
}
@@ -585,7 +571,7 @@ public abstract class SubscriptionImpl i
public boolean wouldSuspend(QueueEntry msg)
{
- return !_creditManager.useCreditForMessage(msg.getMessage().getSize());//_channel.wouldSuspend(msg.getMessage());
+ return !_creditManager.useCreditForMessage(msg.getMessage().getSize());
}
public boolean trySendLock()
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionList.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionList.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionList.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionList.java Sat Mar 10 19:22:10 2012
@@ -20,11 +20,9 @@
*/
package org.apache.qpid.server.subscription;
-import org.apache.qpid.server.subscription.Subscription;
-
-import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
public class SubscriptionList
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java Sat Mar 10 19:22:10 2012
@@ -20,46 +20,58 @@
*/
package org.apache.qpid.server.subscription;
-import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.SUBSCRIPTION_FORMAT;
-import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.QUEUE_FORMAT;
-
-import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.queue.BaseQueue;
-import org.apache.qpid.server.queue.InboundMessageAdapter;
-import org.apache.qpid.server.queue.QueueEntry;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.BasicContentHeaderProperties;
+import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.server.configuration.ConfigStore;
import org.apache.qpid.server.configuration.ConfiguredObject;
import org.apache.qpid.server.configuration.SessionConfig;
import org.apache.qpid.server.configuration.SubscriptionConfig;
import org.apache.qpid.server.configuration.SubscriptionConfigType;
import org.apache.qpid.server.exchange.Exchange;
-import org.apache.qpid.server.flow.FlowCreditManager;
+import org.apache.qpid.server.filter.FilterManager;
import org.apache.qpid.server.flow.CreditCreditManager;
-import org.apache.qpid.server.flow.WindowCreditManager;
+import org.apache.qpid.server.flow.FlowCreditManager;
import org.apache.qpid.server.flow.FlowCreditManager_0_10;
-import org.apache.qpid.server.filter.FilterManager;
+import org.apache.qpid.server.flow.WindowCreditManager;
+import org.apache.qpid.server.logging.LogActor;
+import org.apache.qpid.server.logging.LogSubject;
import org.apache.qpid.server.logging.actors.CurrentActor;
import org.apache.qpid.server.logging.actors.GenericActor;
import org.apache.qpid.server.logging.messages.ChannelMessages;
import org.apache.qpid.server.logging.messages.SubscriptionMessages;
-import org.apache.qpid.server.logging.LogActor;
-import org.apache.qpid.server.logging.LogSubject;
+import org.apache.qpid.server.message.AMQMessage;
import org.apache.qpid.server.message.InboundMessage;
-import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.message.MessageTransferMessage;
-import org.apache.qpid.server.message.AMQMessage;
+import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.BaseQueue;
+import org.apache.qpid.server.queue.InboundMessageAdapter;
+import org.apache.qpid.server.queue.QueueEntry;
import org.apache.qpid.server.transport.ServerSession;
-import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.txn.AutoCommitTransaction;
-import org.apache.qpid.transport.*;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.BasicContentHeaderProperties;
-import org.apache.qpid.framing.FieldTable;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.transport.util.Logger;
+import org.apache.qpid.server.txn.ServerTransaction;
+import org.apache.qpid.transport.DeliveryProperties;
+import org.apache.qpid.transport.Header;
+import org.apache.qpid.transport.MessageAcceptMode;
+import org.apache.qpid.transport.MessageAcquireMode;
+import org.apache.qpid.transport.MessageCreditUnit;
+import org.apache.qpid.transport.MessageDeliveryPriority;
+import org.apache.qpid.transport.MessageFlowMode;
+import org.apache.qpid.transport.MessageProperties;
+import org.apache.qpid.transport.MessageTransfer;
+import org.apache.qpid.transport.Method;
+import org.apache.qpid.transport.Option;
+import org.apache.qpid.transport.ReplyTo;
+import org.apache.qpid.transport.Struct;
import org.apache.qpid.url.AMQBindingURL;
+import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.QUEUE_FORMAT;
+import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.SUBSCRIPTION_FORMAT;
+
import java.net.URISyntaxException;
+import java.nio.ByteBuffer;
import java.text.MessageFormat;
import java.util.Arrays;
import java.util.Collections;
@@ -67,13 +79,12 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.ConcurrentHashMap;
-import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
public class Subscription_0_10 implements Subscription, FlowCreditManager.FlowCreditManagerListener, SubscriptionConfig, LogSubject
{
@@ -387,6 +398,10 @@ public class Subscription_0_10 implement
{
deliveryProps.setTimestamp(origDeliveryProps.getTimestamp());
}
+ if(origDeliveryProps.hasTtl())
+ {
+ deliveryProps.setTtl(origDeliveryProps.getTtl());
+ }
}
@@ -537,36 +552,8 @@ public class Subscription_0_10 implement
messageProps.setCorrelationId(serverMsg.getMessageHeader().getCorrelationId().getBytes());
}
-
// TODO - ReplyTo
-
- final Map<String, Object> appHeaders = new HashMap<String, Object>();
-
- /*properties.getHeaders().processOverElements(
- new FieldTable.FieldTableElementProcessor()
- {
-
- public boolean processElement(String propertyName, AMQTypedValue value)
- {
- Object val = value.getValue();
- if(val instanceof AMQShortString)
- {
- val = val.toString();
- }
- appHeaders.put(propertyName, val);
- return true;
- }
-
- public Object getResult()
- {
- return appHeaders;
- }
- });
-
-
- messageProps.setApplicationHeaders(appHeaders);
-*/
Header header = new Header(deliveryProps, messageProps, null);
xfr = batch ? new MessageTransfer(_destination,_acceptMode,_acquireMode,header, body, BATCHED)
: new MessageTransfer(_destination,_acceptMode,_acquireMode,header, body);
@@ -690,7 +677,10 @@ public class Subscription_0_10 implement
{
entry.setRedelivered();
entry.routeToAlternate();
-
+ if(entry.isAcquiredBy(this))
+ {
+ entry.discard();
+ }
}
void release(final QueueEntry entry, final boolean setRedelivered)
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/QpidAcceptor.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/QpidAcceptor.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/QpidAcceptor.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/QpidAcceptor.java Sat Mar 10 19:22:10 2012
@@ -24,8 +24,8 @@ import org.apache.qpid.transport.network
public class QpidAcceptor
{
- NetworkTransport _transport;
- String _protocol;
+ private NetworkTransport _transport;
+ private String _protocol;
public QpidAcceptor(NetworkTransport transport, String protocol)
{
_transport = transport;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java Sat Mar 10 19:22:10 2012
@@ -20,27 +20,6 @@
*/
package org.apache.qpid.server.transport;
-import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.CONNECTION_FORMAT;
-import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.SOCKET_FORMAT;
-import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.USER_FORMAT;
-
-import java.util.concurrent.atomic.AtomicLong;
-
-import javax.management.JMException;
-
-import org.apache.qpid.server.management.ManagedObject;
-
-import org.apache.qpid.server.management.Managable;
-
-import java.security.Principal;
-import java.text.MessageFormat;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.UUID;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import javax.security.auth.Subject;
-
import org.apache.qpid.AMQException;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.server.configuration.ConnectionConfig;
@@ -49,6 +28,8 @@ import org.apache.qpid.server.logging.Lo
import org.apache.qpid.server.logging.actors.CurrentActor;
import org.apache.qpid.server.logging.actors.GenericActor;
import org.apache.qpid.server.logging.messages.ConnectionMessages;
+import org.apache.qpid.server.management.Managable;
+import org.apache.qpid.server.management.ManagedObject;
import org.apache.qpid.server.protocol.AMQConnectionModel;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.security.AuthorizationHolder;
@@ -62,8 +43,20 @@ import org.apache.qpid.transport.Executi
import org.apache.qpid.transport.Method;
import org.apache.qpid.transport.ProtocolEvent;
import org.apache.qpid.transport.Session;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+
+import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.CONNECTION_FORMAT;
+import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.SOCKET_FORMAT;
+import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.USER_FORMAT;
+
+import javax.management.JMException;
+import javax.security.auth.Subject;
+import java.security.Principal;
+import java.text.MessageFormat;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
public class ServerConnection extends Connection implements Managable, AMQConnectionModel, LogSubject, AuthorizationHolder
{
@@ -114,7 +107,7 @@ public class ServerConnection extends Co
{
_onOpenTask.run();
}
- _actor.message(ConnectionMessages.OPEN(getClientId(), "0-10", true, true));
+ _actor.message(ConnectionMessages.OPEN(getClientId(), "0-10", getClientVersion(), true, true, true));
getVirtualHost().getConnectionRegistry().registerConnection(this);
}
@@ -495,4 +488,14 @@ public class ServerConnection extends Co
_mBean = null;
}
}
+
+ public String getClientId()
+ {
+ return getConnectionDelegate().getClientId();
+ }
+
+ public String getClientVersion()
+ {
+ return getConnectionDelegate().getClientVersion();
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java Sat Mar 10 19:22:10 2012
@@ -20,17 +20,10 @@
*/
package org.apache.qpid.server.transport;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.StringTokenizer;
-import javax.security.sasl.SaslException;
-import javax.security.sasl.SaslServer;
+import static org.apache.qpid.transport.Connection.State.CLOSE_RCVD;
+
import org.apache.qpid.common.ServerPropertyNames;
+import org.apache.qpid.properties.ConnectionStartProperties;
import org.apache.qpid.protocol.ProtocolEngine;
import org.apache.qpid.server.configuration.BrokerConfig;
import org.apache.qpid.server.protocol.AMQConnectionModel;
@@ -41,24 +34,25 @@ import org.apache.qpid.server.security.a
import org.apache.qpid.server.security.auth.AuthenticationResult.AuthenticationStatus;
import org.apache.qpid.server.subscription.Subscription_0_10;
import org.apache.qpid.server.virtualhost.VirtualHost;
-import org.apache.qpid.transport.Binary;
-import org.apache.qpid.transport.Connection;
-import org.apache.qpid.transport.ConnectionClose;
-import org.apache.qpid.transport.ConnectionCloseCode;
-import org.apache.qpid.transport.ConnectionOpen;
-import org.apache.qpid.transport.ConnectionOpenOk;
-import org.apache.qpid.transport.ConnectionStartOk;
-import org.apache.qpid.transport.ConnectionTuneOk;
-import org.apache.qpid.transport.ServerDelegate;
-import org.apache.qpid.transport.Session;
-import org.apache.qpid.transport.SessionAttach;
-import org.apache.qpid.transport.SessionDelegate;
-import org.apache.qpid.transport.SessionDetach;
-import org.apache.qpid.transport.SessionDetachCode;
-import org.apache.qpid.transport.SessionDetached;
+import org.apache.qpid.transport.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.StringTokenizer;
public class ServerConnectionDelegate extends ServerDelegate
{
+ private static final Logger LOGGER = LoggerFactory.getLogger(ServerConnectionDelegate.class);
+
private final String _localFQDN;
private final IApplicationRegistry _appRegistry;
private int _maxNoOfChannels;
@@ -140,17 +134,20 @@ public class ServerConnectionDelegate ex
}
}
+ @Override
public void connectionClose(Connection conn, ConnectionClose close)
{
+ final ServerConnection sconn = (ServerConnection) conn;
try
{
- ((ServerConnection) conn).logClosed();
+ sconn.logClosed();
}
finally
{
- super.connectionClose(conn, close);
+ sconn.closeCode(close);
+ sconn.setState(CLOSE_RCVD);
+ sendConnectionCloseOkAndCloseSender(conn);
}
-
}
public void connectionOpen(Connection conn, ConnectionOpen open)
@@ -177,19 +174,19 @@ public class ServerConnectionDelegate ex
if (!vhost.getSecurityManager().accessVirtualhost(vhostName, ((ProtocolEngine) sconn.getConfig()).getRemoteAddress()))
{
- sconn.invoke(new ConnectionClose(ConnectionCloseCode.CONNECTION_FORCED, "Permission denied '"+vhostName+"'"));
sconn.setState(Connection.State.CLOSING);
+ sconn.invoke(new ConnectionClose(ConnectionCloseCode.CONNECTION_FORCED, "Permission denied '"+vhostName+"'"));
}
else
{
- sconn.invoke(new ConnectionOpenOk(Collections.emptyList()));
- sconn.setState(Connection.State.OPEN);
+ sconn.setState(Connection.State.OPEN);
+ sconn.invoke(new ConnectionOpenOk(Collections.emptyList()));
}
}
else
{
- sconn.invoke(new ConnectionClose(ConnectionCloseCode.INVALID_PATH, "Unknown virtualhost '"+vhostName+"'"));
sconn.setState(Connection.State.CLOSING);
+ sconn.invoke(new ConnectionClose(ConnectionCloseCode.INVALID_PATH, "Unknown virtualhost '"+vhostName+"'"));
}
}
@@ -202,9 +199,9 @@ public class ServerConnectionDelegate ex
if (okChannelMax > getChannelMax())
{
- _logger.error("Connection '" + sconn.getConnectionId() + "' being severed, " +
+ LOGGER.error("Connection '" + sconn.getConnectionId() + "' being severed, " +
"client connectionTuneOk returned a channelMax (" + okChannelMax +
- ") above the servers offered limit (" + getChannelMax() +")");
+ ") above the server's offered limit (" + getChannelMax() +")");
//Due to the error we must forcefully close the connection without negotiation
sconn.getSender().close();
@@ -234,23 +231,26 @@ public class ServerConnectionDelegate ex
@Override public void sessionDetach(Connection conn, SessionDetach dtc)
{
- // To ensure a clean detach, we unregister any remaining subscriptions. Unregister ensures
- // that any in-progress delivery (SubFlushRunner/QueueRunner) is completed before the unregister
+ // To ensure a clean detach, we stop any remaining subscriptions. Stop ensures
+ // that any in-progress delivery (SubFlushRunner/QueueRunner) is completed before the stop
// completes.
- unregisterAllSubscriptions(conn, dtc);
+ stopAllSubscriptions(conn, dtc);
+ Session ssn = conn.getSession(dtc.getChannel());
+ ((ServerSession)ssn).setClose(true);
super.sessionDetach(conn, dtc);
}
- private void unregisterAllSubscriptions(Connection conn, SessionDetach dtc)
+ private void stopAllSubscriptions(Connection conn, SessionDetach dtc)
{
final ServerSession ssn = (ServerSession) conn.getSession(dtc.getChannel());
final Collection<Subscription_0_10> subs = ssn.getSubscriptions();
for (Subscription_0_10 subscription_0_10 : subs)
{
- ssn.unregister(subscription_0_10);
+ subscription_0_10.stop();
}
}
+
@Override
public void sessionAttach(final Connection conn, final SessionAttach atc)
{
@@ -258,8 +258,7 @@ public class ServerConnectionDelegate ex
if(isSessionNameUnique(atc.getName(), conn))
{
- ssn = sessionAttachImpl(conn, atc);
- conn.registerSession(ssn);
+ super.sessionAttach(conn, atc);
((ServerConnection)conn).checkForNotification();
}
else
@@ -299,4 +298,14 @@ public class ServerConnectionDelegate ex
{
return _clientProperties;
}
+
+ public String getClientId()
+ {
+ return _clientProperties == null ? null : (String) _clientProperties.get(ConnectionStartProperties.CLIENT_ID_0_10);
+ }
+
+ public String getClientVersion()
+ {
+ return _clientProperties == null ? null : (String) _clientProperties.get(ConnectionStartProperties.VERSION_0_10);
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionMBean.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionMBean.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionMBean.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionMBean.java Sat Mar 10 19:22:10 2012
@@ -20,17 +20,6 @@
*/
package org.apache.qpid.server.transport;
-import java.io.IOException;
-import java.util.Date;
-import java.util.List;
-import javax.management.JMException;
-import javax.management.NotCompliantMBeanException;
-import javax.management.openmbean.CompositeData;
-import javax.management.openmbean.CompositeDataSupport;
-import javax.management.openmbean.TabularData;
-import javax.management.openmbean.TabularDataSupport;
-
-import org.apache.qpid.common.ClientProperties;
import org.apache.qpid.management.common.mbeans.annotations.MBeanConstructor;
import org.apache.qpid.management.common.mbeans.annotations.MBeanDescription;
import org.apache.qpid.server.logging.actors.CurrentActor;
@@ -39,6 +28,16 @@ import org.apache.qpid.server.management
import org.apache.qpid.server.management.ManagedObject;
import org.apache.qpid.server.protocol.AMQSessionModel;
+import javax.management.JMException;
+import javax.management.NotCompliantMBeanException;
+import javax.management.openmbean.CompositeData;
+import javax.management.openmbean.CompositeDataSupport;
+import javax.management.openmbean.TabularData;
+import javax.management.openmbean.TabularDataSupport;
+import java.io.IOException;
+import java.util.Date;
+import java.util.List;
+
/**
* This MBean class implements the management interface. In order to make more attributes, operations and notifications
* available over JMX simply augment the ManagedConnection interface and add the appropriate implementation here.
@@ -76,7 +75,7 @@ public class ServerConnectionMBean exten
@Override
public String getVersion()
{
- return String.valueOf(_serverConnection.getConnectionDelegate().getClientProperties().get(ClientProperties.version.toString()));
+ return String.valueOf(_serverConnection.getClientVersion());
}
@Override
@@ -132,7 +131,7 @@ public class ServerConnectionMBean exten
}
else if (session.isTransactional())
{
- CurrentActor.set(new ManagementActor(_logActor.getRootMessageLogger()));
+ CurrentActor.set(new ManagementActor(getLogActor().getRootMessageLogger()));
try
{
session.commit();
@@ -154,7 +153,7 @@ public class ServerConnectionMBean exten
}
else if (session.isTransactional())
{
- CurrentActor.set(new ManagementActor(_logActor.getRootMessageLogger()));
+ CurrentActor.set(new ManagementActor(getLogActor().getRootMessageLogger()));
try
{
session.rollback();
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java Sat Mar 10 19:22:10 2012
@@ -21,6 +21,11 @@
package org.apache.qpid.server.transport;
import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.CHANNEL_FORMAT;
+import org.apache.qpid.server.message.InboundMessage;
+import org.apache.qpid.server.message.MessageMetaData_0_10;
+import org.apache.qpid.server.message.MessageTransferMessage;
+import org.apache.qpid.server.txn.RollbackOnlyDtxException;
+import org.apache.qpid.server.txn.TimeoutDtxException;
import static org.apache.qpid.util.Serial.gt;
import java.security.Principal;
@@ -30,17 +35,21 @@ import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.SortedMap;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
+
import javax.security.auth.Subject;
+
import org.apache.qpid.AMQException;
+import org.apache.qpid.AMQStoreException;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.protocol.ProtocolEngine;
import org.apache.qpid.server.configuration.ConfigStore;
@@ -64,24 +73,19 @@ import org.apache.qpid.server.queue.Queu
import org.apache.qpid.server.security.AuthorizationHolder;
import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.subscription.Subscription_0_10;
+import org.apache.qpid.server.txn.AlreadyKnownDtxException;
import org.apache.qpid.server.txn.AsyncAutoCommitTransaction;
+import org.apache.qpid.server.txn.DistributedTransaction;
+import org.apache.qpid.server.txn.DtxNotSelectedException;
+import org.apache.qpid.server.txn.IncorrectDtxStateException;
+import org.apache.qpid.server.txn.JoinAndResumeDtxException;
import org.apache.qpid.server.txn.LocalTransaction;
+import org.apache.qpid.server.txn.NotAssociatedDtxException;
import org.apache.qpid.server.txn.ServerTransaction;
+import org.apache.qpid.server.txn.SuspendAndFailDtxException;
+import org.apache.qpid.server.txn.UnknownDtxBranchException;
import org.apache.qpid.server.virtualhost.VirtualHost;
-import org.apache.qpid.transport.Binary;
-import org.apache.qpid.transport.Connection;
-import org.apache.qpid.transport.MessageCreditUnit;
-import org.apache.qpid.transport.MessageFlow;
-import org.apache.qpid.transport.MessageFlowMode;
-import org.apache.qpid.transport.MessageSetFlowMode;
-import org.apache.qpid.transport.MessageStop;
-import org.apache.qpid.transport.MessageTransfer;
-import org.apache.qpid.transport.Method;
-import org.apache.qpid.transport.Range;
-import org.apache.qpid.transport.RangeSet;
-import org.apache.qpid.transport.RangeSetFactory;
-import org.apache.qpid.transport.Session;
-import org.apache.qpid.transport.SessionDelegate;
+import org.apache.qpid.transport.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -100,13 +104,12 @@ public class ServerSession extends Sessi
private long _createTime = System.currentTimeMillis();
private LogActor _actor = GenericActor.getInstance(this);
- private final ConcurrentMap<AMQQueue, Boolean> _blockingQueues = new ConcurrentHashMap<AMQQueue, Boolean>();
+ private final Set<AMQQueue> _blockingQueues = new ConcurrentSkipListSet<AMQQueue>();
private final AtomicBoolean _blocking = new AtomicBoolean(false);
private ChannelLogSubject _logSubject;
private final AtomicInteger _outstandingCredit = new AtomicInteger(UNLIMITED_CREDIT);
-
public static interface MessageDispositionChangeListener
{
public void onAccept();
@@ -356,7 +359,15 @@ public class ServerSession extends Sessi
public void onClose()
{
- _transaction.rollback();
+ if(_transaction instanceof LocalTransaction)
+ {
+ _transaction.rollback();
+ }
+ else if(_transaction instanceof DistributedTransaction)
+ {
+ getVirtualHost().getDtxRegistry().endAssociations(this);
+ }
+
for(MessageDispositionChangeListener listener : _messageDispositionListenerMap.values())
{
listener.onRelease(true);
@@ -392,6 +403,9 @@ public class ServerSession extends Sessi
public void onRollback()
{
+ // The client has acknowledge the message and therefore have seen it.
+ // In the event of rollback, the message must be marked as redelivered.
+ entry.setRedelivered();
entry.release();
}
});
@@ -452,6 +466,95 @@ public class ServerSession extends Sessi
_txnStarts.incrementAndGet();
}
+ public void selectDtx()
+ {
+ _transaction = new DistributedTransaction(this, getMessageStore(), getVirtualHost());
+
+ }
+
+
+ public void startDtx(Xid xid, boolean join, boolean resume)
+ throws JoinAndResumeDtxException,
+ UnknownDtxBranchException,
+ AlreadyKnownDtxException,
+ DtxNotSelectedException
+ {
+ DistributedTransaction distributedTransaction = assertDtxTransaction();
+ distributedTransaction.start(xid, join, resume);
+ }
+
+
+ public void endDtx(Xid xid, boolean fail, boolean suspend)
+ throws NotAssociatedDtxException,
+ UnknownDtxBranchException,
+ DtxNotSelectedException,
+ SuspendAndFailDtxException, TimeoutDtxException
+ {
+ DistributedTransaction distributedTransaction = assertDtxTransaction();
+ distributedTransaction.end(xid, fail, suspend);
+ }
+
+
+ public long getTimeoutDtx(Xid xid)
+ throws UnknownDtxBranchException
+ {
+ return getVirtualHost().getDtxRegistry().getTimeout(xid);
+ }
+
+
+ public void setTimeoutDtx(Xid xid, long timeout)
+ throws UnknownDtxBranchException
+ {
+ getVirtualHost().getDtxRegistry().setTimeout(xid, timeout);
+ }
+
+
+ public void prepareDtx(Xid xid)
+ throws UnknownDtxBranchException,
+ IncorrectDtxStateException, AMQStoreException, RollbackOnlyDtxException, TimeoutDtxException
+ {
+ getVirtualHost().getDtxRegistry().prepare(xid);
+ }
+
+ public void commitDtx(Xid xid, boolean onePhase)
+ throws UnknownDtxBranchException,
+ IncorrectDtxStateException, AMQStoreException, RollbackOnlyDtxException, TimeoutDtxException
+ {
+ getVirtualHost().getDtxRegistry().commit(xid, onePhase);
+ }
+
+
+ public void rollbackDtx(Xid xid)
+ throws UnknownDtxBranchException,
+ IncorrectDtxStateException, AMQStoreException, TimeoutDtxException
+ {
+ getVirtualHost().getDtxRegistry().rollback(xid);
+ }
+
+
+ public void forgetDtx(Xid xid) throws UnknownDtxBranchException, IncorrectDtxStateException
+ {
+ getVirtualHost().getDtxRegistry().forget(xid);
+ }
+
+ public List<Xid> recoverDtx()
+ {
+ return getVirtualHost().getDtxRegistry().recover();
+ }
+
+ private DistributedTransaction assertDtxTransaction() throws DtxNotSelectedException
+ {
+ if(_transaction instanceof DistributedTransaction)
+ {
+ return (DistributedTransaction) _transaction;
+ }
+ else
+ {
+ throw new DtxNotSelectedException();
+ }
+ }
+
+
public void commit()
{
_transaction.commit();
@@ -524,12 +627,12 @@ public class ServerSession extends Sessi
public Principal getAuthorizedPrincipal()
{
- return ((ServerConnection) getConnection()).getAuthorizedPrincipal();
+ return getConnection().getAuthorizedPrincipal();
}
public Subject getAuthorizedSubject()
{
- return ((ServerConnection) getConnection()).getAuthorizedSubject();
+ return getConnection().getAuthorizedSubject();
}
public void addSessionCloseTask(Task task)
@@ -544,7 +647,7 @@ public class ServerSession extends Sessi
public Object getReference()
{
- return ((ServerConnection) getConnection()).getReference();
+ return getConnection().getReference();
}
public MessageStore getMessageStore()
@@ -624,7 +727,7 @@ public class ServerSession extends Sessi
public AMQConnectionModel getConnectionModel()
{
- return (ServerConnection) getConnection();
+ return getConnection();
}
public String getClientID()
@@ -632,6 +735,12 @@ public class ServerSession extends Sessi
return getConnection().getClientId();
}
+ @Override
+ public ServerConnection getConnection()
+ {
+ return (ServerConnection) super.getConnection();
+ }
+
public LogActor getLogActor()
{
return _actor;
@@ -676,7 +785,8 @@ public class ServerSession extends Sessi
public void block(AMQQueue queue)
{
- if(_blockingQueues.putIfAbsent(queue, Boolean.TRUE) == null)
+
+ if(_blockingQueues.add(queue))
{
if(_blocking.compareAndSet(false,true))
@@ -694,7 +804,7 @@ public class ServerSession extends Sessi
{
if(_blockingQueues.remove(queue) && _blockingQueues.isEmpty())
{
- if(_blocking.compareAndSet(true,false))
+ if(_blocking.compareAndSet(true,false) && !isClosing())
{
_actor.message(_logSubject, ChannelMessages.FLOW_REMOVED());
@@ -710,11 +820,19 @@ public class ServerSession extends Sessi
}
}
+ public boolean onSameConnection(InboundMessage inbound)
+ {
+ return ((inbound instanceof MessageTransferMessage)
+ && ((MessageTransferMessage)inbound).getConnectionReference() == getConnection().getReference())
+ || ((inbound instanceof MessageMetaData_0_10)
+ && (((MessageMetaData_0_10)inbound).getConnectionReference())== getConnection().getReference());
+ }
+
public String toLogString()
{
- long connectionId = getConnection() instanceof ServerConnection
- ? ((ServerConnection) getConnection()).getConnectionId()
+ long connectionId = super.getConnection() instanceof ServerConnection
+ ? getConnection().getConnectionId()
: -1;
String remoteAddress = _connectionConfig instanceof ProtocolEngine
@@ -749,6 +867,16 @@ public class ServerSession extends Sessi
}
}
+ void stopSubscriptions()
+ {
+ final Collection<Subscription_0_10> subscriptions = getSubscriptions();
+ for (Subscription_0_10 subscription_0_10 : subscriptions)
+ {
+ subscription_0_10.stop();
+ }
+ }
+
+
public void receivedComplete()
{
final Collection<Subscription_0_10> subscriptions = getSubscriptions();
@@ -889,4 +1017,14 @@ public class ServerSession extends Sessi
return _future.isComplete();
}
}
+
+ protected void setClose(boolean close)
+ {
+ super.setClose(close);
+ }
+
+ public int compareTo(AMQSessionModel session)
+ {
+ return getId().toString().compareTo(session.getID().toString());
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java Sat Mar 10 19:22:10 2012
@@ -20,13 +20,10 @@
*/
package org.apache.qpid.server.transport;
-import java.nio.ByteBuffer;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-
import org.apache.log4j.Logger;
+
import org.apache.qpid.AMQException;
+import org.apache.qpid.AMQStoreException;
import org.apache.qpid.AMQUnknownExchangeType;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.FieldTable;
@@ -54,10 +51,24 @@ import org.apache.qpid.server.store.Mess
import org.apache.qpid.server.store.StoredMessage;
import org.apache.qpid.server.subscription.SubscriptionFactoryImpl;
import org.apache.qpid.server.subscription.Subscription_0_10;
+import org.apache.qpid.server.txn.AlreadyKnownDtxException;
+import org.apache.qpid.server.txn.DtxNotSelectedException;
+import org.apache.qpid.server.txn.IncorrectDtxStateException;
+import org.apache.qpid.server.txn.JoinAndResumeDtxException;
+import org.apache.qpid.server.txn.NotAssociatedDtxException;
+import org.apache.qpid.server.txn.RollbackOnlyDtxException;
import org.apache.qpid.server.txn.ServerTransaction;
+import org.apache.qpid.server.txn.SuspendAndFailDtxException;
+import org.apache.qpid.server.txn.TimeoutDtxException;
+import org.apache.qpid.server.txn.UnknownDtxBranchException;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.transport.*;
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
public class ServerSessionDelegate extends SessionDelegate
{
private static final Logger LOGGER = Logger.getLogger(ServerSessionDelegate.class);
@@ -154,7 +165,12 @@ public class ServerSessionDelegate exten
@Override
public void messageSubscribe(Session session, MessageSubscribe method)
{
- //TODO - work around broken Python tests
+ /*
+ TODO - work around broken Python tests
+ Correct code should read like
+ if not hasAcceptMode() exception ILLEGAL_ARGUMENT "Accept-mode not supplied"
+ else if not method.hasAcquireMode() exception ExecutionErrorCode.ILLEGAL_ARGUMENT, "Acquire-mode not supplied"
+ */
if(!method.hasAcceptMode())
{
method.setAcceptMode(MessageAcceptMode.EXPLICIT);
@@ -165,15 +181,7 @@ public class ServerSessionDelegate exten
}
- /* if(!method.hasAcceptMode())
- {
- exception(session,method,ExecutionErrorCode.ILLEGAL_ARGUMENT, "Accept-mode not supplied");
- }
- else if(!method.hasAcquireMode())
- {
- exception(session,method,ExecutionErrorCode.ILLEGAL_ARGUMENT, "Acquire-mode not supplied");
- }
- else */if(!method.hasQueue())
+ if(!method.hasQueue())
{
exception(session,method,ExecutionErrorCode.ILLEGAL_ARGUMENT, "queue not supplied");
}
@@ -201,6 +209,10 @@ public class ServerSessionDelegate exten
{
exception(session,method,ExecutionErrorCode.RESOURCE_LOCKED, "Exclusive Queue: " + queueName + " owned exclusively by another session");
}
+ else if(queue.isExclusive() && queue.getExclusiveOwningSession() != null && queue.getExclusiveOwningSession() != session)
+ {
+ exception(session,method,ExecutionErrorCode.RESOURCE_LOCKED, "Exclusive Queue: " + queueName + " owned exclusively by another session");
+ }
else
{
if(queue.isExclusive())
@@ -223,7 +235,6 @@ public class ServerSessionDelegate exten
}
});
}
-
}
FlowCreditManager_0_10 creditManager = new WindowCreditManager(0L,0L);
@@ -283,6 +294,7 @@ public class ServerSessionDelegate exten
}
final MessageMetaData_0_10 messageMetaData = new MessageMetaData_0_10(xfr);
+ messageMetaData.setConnectionReference(((ServerSession)ssn).getReference());
if (!getVirtualHost(ssn).getSecurityManager().authorisePublish(messageMetaData.isImmediate(), messageMetaData.getRoutingKey(), exchange.getName()))
{
@@ -428,6 +440,235 @@ public class ServerSessionDelegate exten
((ServerSession)session).rollback();
}
+ @Override
+ public void dtxSelect(Session session, DtxSelect method)
+ {
+ // TODO - check current tx mode
+ ((ServerSession)session).selectDtx();
+ }
+
+ @Override
+ public void dtxStart(Session session, DtxStart method)
+ {
+ XaResult result = new XaResult();
+ result.setStatus(DtxXaStatus.XA_OK);
+ try
+ {
+ ((ServerSession)session).startDtx(method.getXid(), method.getJoin(), method.getResume());
+ session.executionResult(method.getId(), result);
+ }
+ catch(JoinAndResumeDtxException e)
+ {
+ exception(session, method, ExecutionErrorCode.COMMAND_INVALID, e.getMessage());
+ }
+ catch(UnknownDtxBranchException e)
+ {
+ exception(session, method, ExecutionErrorCode.NOT_ALLOWED, "Unknown xid " + method.getXid());
+ }
+ catch(AlreadyKnownDtxException e)
+ {
+ exception(session, method, ExecutionErrorCode.NOT_ALLOWED, "Xid already started an neither join nor " +
+ "resume set" + method.getXid());
+ }
+ catch(DtxNotSelectedException e)
+ {
+ exception(session, method, ExecutionErrorCode.COMMAND_INVALID, e.getMessage());
+ }
+
+ }
+
+ @Override
+ public void dtxEnd(Session session, DtxEnd method)
+ {
+ XaResult result = new XaResult();
+ result.setStatus(DtxXaStatus.XA_OK);
+ try
+ {
+ try
+ {
+ ((ServerSession)session).endDtx(method.getXid(), method.getFail(), method.getSuspend());
+ }
+ catch (TimeoutDtxException e)
+ {
+ result.setStatus(DtxXaStatus.XA_RBTIMEOUT);
+ }
+ session.executionResult(method.getId(), result);
+ }
+ catch(UnknownDtxBranchException e)
+ {
+ exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
+ }
+ catch(NotAssociatedDtxException e)
+ {
+ exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
+ }
+ catch(DtxNotSelectedException e)
+ {
+ exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
+ }
+ catch(SuspendAndFailDtxException e)
+ {
+ exception(session, method, ExecutionErrorCode.COMMAND_INVALID, e.getMessage());
+ }
+
+ }
+
+ @Override
+ public void dtxCommit(Session session, DtxCommit method)
+ {
+ XaResult result = new XaResult();
+ result.setStatus(DtxXaStatus.XA_OK);
+ try
+ {
+ try
+ {
+ ((ServerSession)session).commitDtx(method.getXid(), method.getOnePhase());
+ }
+ catch (RollbackOnlyDtxException e)
+ {
+ result.setStatus(DtxXaStatus.XA_RBROLLBACK);
+ }
+ catch (TimeoutDtxException e)
+ {
+ result.setStatus(DtxXaStatus.XA_RBTIMEOUT);
+ }
+ session.executionResult(method.getId(), result);
+ }
+ catch(UnknownDtxBranchException e)
+ {
+ exception(session, method, ExecutionErrorCode.NOT_FOUND, e.getMessage());
+ }
+ catch(IncorrectDtxStateException e)
+ {
+ exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
+ }
+ catch(AMQStoreException e)
+ {
+ exception(session, method, ExecutionErrorCode.INTERNAL_ERROR, e.getMessage());
+ }
+ }
+
+ @Override
+ public void dtxForget(Session session, DtxForget method)
+ {
+ try
+ {
+ ((ServerSession)session).forgetDtx(method.getXid());
+ }
+ catch(UnknownDtxBranchException e)
+ {
+ exception(session, method, ExecutionErrorCode.NOT_FOUND, e.getMessage());
+ }
+ catch(IncorrectDtxStateException e)
+ {
+ exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
+ }
+
+ }
+
+ @Override
+ public void dtxGetTimeout(Session session, DtxGetTimeout method)
+ {
+ GetTimeoutResult result = new GetTimeoutResult();
+ try
+ {
+ result.setTimeout(((ServerSession) session).getTimeoutDtx(method.getXid()));
+ session.executionResult(method.getId(), result);
+ }
+ catch(UnknownDtxBranchException e)
+ {
+ exception(session, method, ExecutionErrorCode.NOT_FOUND, e.getMessage());
+ }
+ }
+
+ @Override
+ public void dtxPrepare(Session session, DtxPrepare method)
+ {
+ XaResult result = new XaResult();
+ result.setStatus(DtxXaStatus.XA_OK);
+ try
+ {
+ try
+ {
+ ((ServerSession)session).prepareDtx(method.getXid());
+ }
+ catch (RollbackOnlyDtxException e)
+ {
+ result.setStatus(DtxXaStatus.XA_RBROLLBACK);
+ }
+ catch (TimeoutDtxException e)
+ {
+ result.setStatus(DtxXaStatus.XA_RBTIMEOUT);
+ }
+ session.executionResult((int) method.getId(), result);
+ }
+ catch(UnknownDtxBranchException e)
+ {
+ exception(session, method, ExecutionErrorCode.NOT_FOUND, e.getMessage());
+ }
+ catch(IncorrectDtxStateException e)
+ {
+ exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
+ }
+ catch(AMQStoreException e)
+ {
+ exception(session, method, ExecutionErrorCode.INTERNAL_ERROR, e.getMessage());
+ }
+ }
+
+ @Override
+ public void dtxRecover(Session session, DtxRecover method)
+ {
+ RecoverResult result = new RecoverResult();
+ List inDoubt = ((ServerSession)session).recoverDtx();
+ result.setInDoubt(inDoubt);
+ session.executionResult(method.getId(), result);
+ }
+
+ @Override
+ public void dtxRollback(Session session, DtxRollback method)
+ {
+
+ XaResult result = new XaResult();
+ result.setStatus(DtxXaStatus.XA_OK);
+ try
+ {
+ try
+ {
+ ((ServerSession)session).rollbackDtx(method.getXid());
+ }
+ catch (TimeoutDtxException e)
+ {
+ result.setStatus(DtxXaStatus.XA_RBTIMEOUT);
+ }
+ session.executionResult(method.getId(), result);
+ }
+ catch(UnknownDtxBranchException e)
+ {
+ exception(session, method, ExecutionErrorCode.NOT_FOUND, e.getMessage());
+ }
+ catch(IncorrectDtxStateException e)
+ {
+ exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
+ }
+ catch(AMQStoreException e)
+ {
+ exception(session, method, ExecutionErrorCode.INTERNAL_ERROR, e.getMessage());
+ }
+ }
+
+ @Override
+ public void dtxSetTimeout(Session session, DtxSetTimeout method)
+ {
+ try
+ {
+ ((ServerSession)session).setTimeoutDtx(method.getXid(), method.getTimeout());
+ }
+ catch(UnknownDtxBranchException e)
+ {
+ exception(session, method, ExecutionErrorCode.NOT_FOUND, e.getMessage());
+ }
+ }
@Override
public void executionSync(final Session ssn, final ExecutionSync sync)
@@ -465,9 +706,9 @@ public class ServerSessionDelegate exten
}
else
{
- if(!exchange.getTypeShortString().toString().equals(method.getType()))
+ if(!exchange.getTypeShortString().toString().equals(method.getType()) && (method.getType() != null && method.getType().length() > 0))
{
- exception(session, method, ExecutionErrorCode.NOT_ALLOWED, "Cannot redeclare with a different exchange type");
+ exception(session, method, ExecutionErrorCode.NOT_ALLOWED, "Attempt to redeclare exchange: " + exchangeName + " of type " + exchange.getTypeShortString() + " to " + method.getType() +".");
}
}
@@ -476,48 +717,96 @@ public class ServerSessionDelegate exten
{
if (exchange == null)
{
- ExchangeRegistry exchangeRegistry = getExchangeRegistry(session);
- ExchangeFactory exchangeFactory = virtualHost.getExchangeFactory();
+ if(exchangeName.startsWith("amq."))
+ {
+ exception(session, method, ExecutionErrorCode.NOT_ALLOWED,
+ "Attempt to declare exchange: " + exchangeName +
+ " which begins with reserved prefix 'amq.'.");
+ }
+ else if(exchangeName.startsWith("qpid."))
+ {
+ exception(session, method, ExecutionErrorCode.NOT_ALLOWED,
+ "Attempt to declare exchange: " + exchangeName +
+ " which begins with reserved prefix 'qpid.'.");
+ }
+ else
+ {
+ ExchangeRegistry exchangeRegistry = getExchangeRegistry(session);
+ ExchangeFactory exchangeFactory = virtualHost.getExchangeFactory();
- try
- {
+ try
+ {
- exchange = exchangeFactory.createExchange(method.getExchange(),
- method.getType(),
- method.getDurable(),
- method.getAutoDelete());
+ exchange = exchangeFactory.createExchange(method.getExchange(),
+ method.getType(),
+ method.getDurable(),
+ method.getAutoDelete());
- String alternateExchangeName = method.getAlternateExchange();
- if(alternateExchangeName != null && alternateExchangeName.length() != 0)
+ String alternateExchangeName = method.getAlternateExchange();
+ boolean validAlternate;
+ if(alternateExchangeName != null && alternateExchangeName.length() != 0)
+ {
+ Exchange alternate = getExchange(session, alternateExchangeName);
+ if(alternate == null)
+ {
+ validAlternate = false;
+ }
+ else
+ {
+ exchange.setAlternateExchange(alternate);
+ validAlternate = true;
+ }
+ }
+ else
+ {
+ validAlternate = true;
+ }
+
+ if(validAlternate)
+ {
+ if (exchange.isDurable())
+ {
+ DurableConfigurationStore store = virtualHost.getDurableConfigurationStore();
+ store.createExchange(exchange);
+ }
+
+ exchangeRegistry.registerExchange(exchange);
+ }
+ else
+ {
+ exception(session, method, ExecutionErrorCode.NOT_FOUND,
+ "Unknown alternate exchange " + alternateExchangeName);
+ }
+ }
+ catch(AMQUnknownExchangeType e)
{
- Exchange alternate = getExchange(session, alternateExchangeName);
- exchange.setAlternateExchange(alternate);
+ exception(session, method, ExecutionErrorCode.NOT_FOUND, "Unknown Exchange Type: " + method.getType());
}
-
- if (exchange.isDurable())
+ catch (AMQException e)
{
- DurableConfigurationStore store = virtualHost.getDurableConfigurationStore();
- store.createExchange(exchange);
+ exception(session, method, e, "Cannot declare exchange '" + exchangeName);
}
-
- exchangeRegistry.registerExchange(exchange);
- }
- catch(AMQUnknownExchangeType e)
- {
- exception(session, method, ExecutionErrorCode.NOT_FOUND, "Unknown Exchange Type: " + method.getType());
- }
- catch (AMQException e)
- {
- exception(session, method, e, "Cannot declare exchange '" + exchangeName);
}
}
else
{
if(!exchange.getTypeShortString().toString().equals(method.getType()))
{
- exception(session, method, ExecutionErrorCode.NOT_ALLOWED, "Cannot redeclare with a different exchange type");
+ exception(session, method, ExecutionErrorCode.NOT_ALLOWED,
+ "Attempt to redeclare exchange: " + exchangeName
+ + " of type " + exchange.getTypeShortString()
+ + " to " + method.getType() +".");
+ }
+ else if(method.hasAlternateExchange()
+ && (exchange.getAlternateExchange() == null ||
+ !method.getAlternateExchange().equals(exchange.getAlternateExchange().getName())))
+ {
+ exception(session, method, ExecutionErrorCode.NOT_ALLOWED,
+ "Attempt to change alternate exchange of: " + exchangeName
+ + " from " + exchange.getAlternateExchange()
+ + " to " + method.getAlternateExchange() +".");
}
}
@@ -710,15 +999,10 @@ public class ServerSessionDelegate exten
{
exception(session, method, ExecutionErrorCode.INVALID_ARGUMENT, "Bind not allowed for default exchange");
}
-/*
- else if (!method.hasBindingKey())
- {
- exception(session, method, ExecutionErrorCode.ILLEGAL_ARGUMENT, "binding-key not set");
- }
-*/
else
{
//TODO - here because of non-compiant python tests
+ // should raise exception ILLEGAL_ARGUMENT "binding-key not set"
if (!method.hasBindingKey())
{
method.setBindingKey(method.getQueue());
@@ -739,10 +1023,7 @@ public class ServerSessionDelegate exten
}
else
{
- AMQShortString routingKey = new AMQShortString(method.getBindingKey());
- FieldTable fieldTable = FieldTable.convertToFieldTable(method.getArguments());
-
- if (!exchange.isBound(routingKey, fieldTable, queue))
+ if (!exchange.isBound(method.getBindingKey(), method.getArguments(), queue))
{
try
{
@@ -854,12 +1135,6 @@ public class ServerSessionDelegate exten
if(method.hasBindingKey())
{
- if(method.hasArguments())
- {
- FieldTable args = FieldTable.convertToFieldTable(method.getArguments());
-
- result.setArgsNotMatched(!exchange.isBound(new AMQShortString(method.getBindingKey()), args, queue));
- }
if(queueMatched)
{
result.setKeyNotMatched(!exchange.isBound(method.getBindingKey(), queue));
@@ -868,23 +1143,28 @@ public class ServerSessionDelegate exten
{
result.setKeyNotMatched(!exchange.isBound(method.getBindingKey()));
}
+
+ if(method.hasArguments())
+ {
+ result.setArgsNotMatched(!exchange.isBound(result.getKeyNotMatched() ? null : method.getBindingKey(), method.getArguments(), queueMatched ? queue : null));
+ }
+
}
else if (method.hasArguments())
{
- // TODO
-
+ result.setArgsNotMatched(!exchange.isBound(null, method.getArguments(), queueMatched ? queue : null));
}
- result.setQueueNotMatched(!exchange.isBound(queue));
-
}
else if(exchange != null && method.hasBindingKey())
{
+ result.setKeyNotMatched(!exchange.isBound(method.getBindingKey()));
+
if(method.hasArguments())
{
- // TODO
+ result.setArgsNotMatched(!exchange.isBound(result.getKeyNotMatched() ? null : method.getBindingKey(), method.getArguments(), queue));
}
- result.setKeyNotMatched(!exchange.isBound(method.getBindingKey()));
+
}
@@ -893,11 +1173,15 @@ public class ServerSessionDelegate exten
{
if(method.hasArguments())
{
- // TODO
+ result.setArgsNotMatched(!exchange.isBound(method.getBindingKey(), method.getArguments(), null));
}
result.setKeyNotMatched(!exchange.isBound(method.getBindingKey()));
}
+ else if(exchange != null && method.hasArguments())
+ {
+ result.setArgsNotMatched(!exchange.isBound(null, method.getArguments(), null));
+ }
session.executionResult((int) method.getId(), result);
@@ -1141,6 +1425,10 @@ public class ServerSessionDelegate exten
{
exception(session,method,ExecutionErrorCode.RESOURCE_LOCKED, "Exclusive Queue: " + queueName + " owned exclusively by another session");
}
+ else if(queue.isExclusive() && queue.getExclusiveOwningSession() != null && queue.getExclusiveOwningSession() != session)
+ {
+ exception(session,method,ExecutionErrorCode.RESOURCE_LOCKED, "Exclusive Queue: " + queueName + " owned exclusively by another session");
+ }
else if (method.getIfEmpty() && !queue.isEmpty())
{
exception(session, method, ExecutionErrorCode.PRECONDITION_FAILED, "Queue " + queueName + " not empty");
@@ -1287,8 +1575,9 @@ public class ServerSessionDelegate exten
ServerSession serverSession = (ServerSession)session;
- serverSession.unregisterSubscriptions();
+ serverSession.stopSubscriptions();
serverSession.onClose();
+ serverSession.unregisterSubscriptions();
}
@Override
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/AsyncAutoCommitTransaction.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/AsyncAutoCommitTransaction.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/AsyncAutoCommitTransaction.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/AsyncAutoCommitTransaction.java Sat Mar 10 19:22:10 2012
@@ -21,6 +21,7 @@
package org.apache.qpid.server.txn;
import org.apache.log4j.Logger;
+
import org.apache.qpid.AMQException;
import org.apache.qpid.AMQStoreException;
import org.apache.qpid.server.message.EnqueableMessage;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/AutoCommitTransaction.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/AutoCommitTransaction.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/AutoCommitTransaction.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/AutoCommitTransaction.java Sat Mar 10 19:22:10 2012
@@ -20,11 +20,8 @@
*/
package org.apache.qpid.server.txn;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-
import org.apache.log4j.Logger;
+
import org.apache.qpid.AMQException;
import org.apache.qpid.AMQStoreException;
import org.apache.qpid.server.message.EnqueableMessage;
@@ -33,6 +30,9 @@ import org.apache.qpid.server.queue.Base
import org.apache.qpid.server.queue.QueueEntry;
import org.apache.qpid.server.store.MessageStore;
+import java.util.Collection;
+import java.util.List;
+
/**
* An implementation of ServerTransaction where each enqueue/dequeue
* operation takes place within it own transaction.
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransaction.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransaction.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransaction.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransaction.java Sat Mar 10 19:22:10 2012
@@ -20,9 +20,8 @@ package org.apache.qpid.server.txn;
*
*/
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.apache.qpid.AMQException;
import org.apache.qpid.server.message.EnqueableMessage;
@@ -31,8 +30,9 @@ import org.apache.qpid.server.queue.Base
import org.apache.qpid.server.queue.QueueEntry;
import org.apache.qpid.server.store.MessageStore;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
/**
* A concrete implementation of ServerTransaction where enqueue/dequeue
Propchange: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Sat Mar 10 19:22:10 2012
@@ -1,3 +1,3 @@
/qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost:930288
/qpid/branches/qpid-2935/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost:1061302-1072333
-/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost:1073294-1157765,1160415-1162726,1162729-1166086,1166089-1235000
+/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost:1073294-1299226
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/HouseKeepingTask.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/HouseKeepingTask.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/HouseKeepingTask.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/HouseKeepingTask.java Sat Mar 10 19:22:10 2012
@@ -21,13 +21,14 @@
package org.apache.qpid.server.virtualhost;
import org.apache.log4j.Logger;
+
import org.apache.qpid.server.logging.RootMessageLogger;
import org.apache.qpid.server.logging.actors.AbstractActor;
import org.apache.qpid.server.logging.actors.CurrentActor;
public abstract class HouseKeepingTask implements Runnable
{
- Logger _logger = Logger.getLogger(this.getClass());
+ private Logger _logger = Logger.getLogger(this.getClass());
private VirtualHost _virtualHost;
@@ -59,7 +60,7 @@ public abstract class HouseKeepingTask i
{
execute();
}
- catch (Throwable e)
+ catch (Exception e)
{
_logger.warn(this.getClass().getSimpleName() + " throw exception: " + e, e);
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/ManagedVirtualHost.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/ManagedVirtualHost.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/ManagedVirtualHost.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/ManagedVirtualHost.java Sat Mar 10 19:22:10 2012
@@ -20,10 +20,10 @@
*/
package org.apache.qpid.server.virtualhost;
-import java.io.IOException;
-
import org.apache.qpid.management.common.mbeans.annotations.MBeanAttribute;
+import java.io.IOException;
+
/**
* The management interface exposed to allow management of a virtualHost
*/
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java Sat Mar 10 19:22:10 2012
@@ -22,7 +22,7 @@ package org.apache.qpid.server.virtualho
import java.util.Map;
import java.util.UUID;
-
+import java.util.concurrent.ScheduledFuture;
import org.apache.qpid.common.Closeable;
import org.apache.qpid.server.binding.BindingFactory;
import org.apache.qpid.server.configuration.ConfigStore;
@@ -37,10 +37,10 @@ import org.apache.qpid.server.protocol.v
import org.apache.qpid.server.queue.QueueRegistry;
import org.apache.qpid.server.registry.IApplicationRegistry;
import org.apache.qpid.server.security.SecurityManager;
-import org.apache.qpid.server.security.auth.manager.AuthenticationManager;
import org.apache.qpid.server.stats.StatisticsGatherer;
import org.apache.qpid.server.store.DurableConfigurationStore;
import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.txn.DtxRegistry;
public interface VirtualHost extends DurableConfigurationStore.Source, VirtualHostConfig, Closeable, StatisticsGatherer
{
@@ -60,8 +60,6 @@ public interface VirtualHost extends Dur
DurableConfigurationStore getDurableConfigurationStore();
- AuthenticationManager getAuthenticationManager();
-
SecurityManager getSecurityManager();
void close();
@@ -97,7 +95,11 @@ public interface VirtualHost extends Dur
ConfigStore getConfigStore();
+ DtxRegistry getDtxRegistry();
+
void removeBrokerConnection(BrokerLink brokerLink);
LinkRegistry getLinkRegistry(String remoteContainerId);
+
+ ScheduledFuture<?> scheduleTask(long delay, Runnable timeoutTask);
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostConfigRecoveryHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostConfigRecoveryHandler.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostConfigRecoveryHandler.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostConfigRecoveryHandler.java Sat Mar 10 19:22:10 2012
@@ -20,43 +20,47 @@
*/
package org.apache.qpid.server.virtualhost;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.UUID;
+import org.apache.log4j.Logger;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.AMQStoreException;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.server.binding.BindingFactory;
+import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.federation.BrokerLink;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.logging.messages.TransactionLogMessages;
+import org.apache.qpid.server.logging.subjects.MessageStoreLogSubject;
+import org.apache.qpid.server.message.AMQMessage;
+import org.apache.qpid.server.message.AbstractServerMessageImpl;
import org.apache.qpid.server.message.EnqueableMessage;
+import org.apache.qpid.server.message.MessageTransferMessage;
+import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.AMQQueueFactory;
+import org.apache.qpid.server.queue.QueueEntry;
import org.apache.qpid.server.store.ConfigurationRecoveryHandler;
import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.store.MessageStoreRecoveryHandler;
import org.apache.qpid.server.store.StoredMessage;
import org.apache.qpid.server.store.TransactionLogRecoveryHandler;
import org.apache.qpid.server.store.TransactionLogResource;
-import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.queue.AMQQueueFactory;
-import org.apache.qpid.server.queue.QueueRegistry;
-import org.apache.qpid.server.exchange.Exchange;
-import org.apache.qpid.server.logging.subjects.MessageStoreLogSubject;
-import org.apache.qpid.server.logging.actors.CurrentActor;
-import org.apache.qpid.server.logging.messages.TransactionLogMessages;
-import org.apache.qpid.server.message.AMQMessage;
-import org.apache.qpid.server.message.ServerMessage;
-import org.apache.qpid.server.message.MessageTransferMessage;
-import org.apache.qpid.server.binding.BindingFactory;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.FieldTable;
-import org.apache.qpid.AMQException;
-
-import org.apache.log4j.Logger;
+import org.apache.qpid.server.txn.DtxBranch;
+import org.apache.qpid.server.txn.DtxRegistry;
+import org.apache.qpid.server.txn.ServerTransaction;
+import org.apache.qpid.transport.Xid;
+import org.apache.qpid.transport.util.Functions;
import org.apache.qpid.util.ByteBufferInputStream;
-import java.io.DataInputStream;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-
-import java.util.List;
-import java.util.ArrayList;
-import java.util.Map;
-import java.util.HashMap;
-import java.util.TreeMap;
-import java.util.UUID;
-
public class VirtualHostConfigRecoveryHandler implements ConfigurationRecoveryHandler,
ConfigurationRecoveryHandler.QueueRecoveryHandler,
ConfigurationRecoveryHandler.ExchangeRecoveryHandler,
@@ -65,7 +69,8 @@ public class VirtualHostConfigRecoveryHa
MessageStoreRecoveryHandler,
MessageStoreRecoveryHandler.StoredMessageRecoveryHandler,
TransactionLogRecoveryHandler,
- TransactionLogRecoveryHandler.QueueEntryRecoveryHandler
+ TransactionLogRecoveryHandler.QueueEntryRecoveryHandler,
+ TransactionLogRecoveryHandler.DtxRecordRecoveryHandler
{
private static final Logger _logger = Logger.getLogger(VirtualHostConfigRecoveryHandler.class);
@@ -78,7 +83,7 @@ public class VirtualHostConfigRecoveryHa
private MessageStore _store;
private final Map<String, Integer> _queueRecoveries = new TreeMap<String, Integer>();
- private Map<Long, ServerMessage> _recoveredMessages = new HashMap<Long, ServerMessage>();
+ private Map<Long, AbstractServerMessageImpl> _recoveredMessages = new HashMap<Long, AbstractServerMessageImpl>();
private Map<Long, StoredMessage> _unusedMessages = new HashMap<Long, StoredMessage>();
@@ -160,7 +165,7 @@ public class VirtualHostConfigRecoveryHa
public void message(StoredMessage message)
{
- ServerMessage serverMessage;
+ AbstractServerMessageImpl serverMessage;
switch(message.getMetaData().getType())
{
case META_DATA_0_8:
@@ -173,9 +178,6 @@ public class VirtualHostConfigRecoveryHa
throw new RuntimeException("Unknown message type retrieved from store " + message.getMetaData().getClass());
}
- //_logger.debug("Recovered message with id " + serverMessage);
-
-
_recoveredMessages.put(message.getMessageNumber(), serverMessage);
_unusedMessages.put(message.getMessageNumber(), message);
}
@@ -198,6 +200,164 @@ public class VirtualHostConfigRecoveryHa
{
}
+ public void dtxRecord(long format, byte[] globalId, byte[] branchId,
+ MessageStore.Transaction.Record[] enqueues,
+ MessageStore.Transaction.Record[] dequeues)
+ {
+ Xid id = new Xid(format, globalId, branchId);
+ DtxRegistry dtxRegistry = _virtualHost.getDtxRegistry();
+ DtxBranch branch = dtxRegistry.getBranch(id);
+ if(branch == null)
+ {
+ branch = new DtxBranch(id, _store, _virtualHost);
+ dtxRegistry.registerBranch(branch);
+ }
+ for(MessageStore.Transaction.Record record : enqueues)
+ {
+ final AMQQueue queue = _virtualHost.getQueueRegistry().getQueue(record.getQueue().getResourceName());
+ if(queue != null)
+ {
+ final long messageId = record.getMessage().getMessageNumber();
+ final AbstractServerMessageImpl message = _recoveredMessages.get(messageId);
+ _unusedMessages.remove(messageId);
+
+ if(message != null)
+ {
+ message.incrementReference();
+
+ branch.enqueue(queue,message);
+
+ branch.addPostTransactionAcion(new ServerTransaction.Action()
+ {
+
+ public void postCommit()
+ {
+ try
+ {
+
+ queue.enqueue(message, true, null);
+ message.decrementReference();
+ }
+ catch (AMQException e)
+ {
+ _logger.error("Unable to enqueue message " + message.getMessageNumber() + " into " +
+ "queue " + queue.getName() + " (from XA transaction)", e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ public void onRollback()
+ {
+ message.decrementReference();
+ }
+ });
+ }
+ else
+ {
+ StringBuilder xidString = xidAsString(id);
+ String messageNumberString = String.valueOf(message.getMessageNumber());
+ CurrentActor.get().message(_logSubject,
+ TransactionLogMessages.XA_INCOMPLETE_MESSAGE(xidString.toString(),
+ messageNumberString));
+
+ }
+
+ }
+ else
+ {
+ StringBuilder xidString = xidAsString(id);
+ CurrentActor.get().message(_logSubject,
+ TransactionLogMessages.XA_INCOMPLETE_QUEUE(xidString.toString(),
+ record.getQueue().getResourceName()));
+
+ }
+ }
+ for(MessageStore.Transaction.Record record : dequeues)
+ {
+ final AMQQueue queue = _virtualHost.getQueueRegistry().getQueue(record.getQueue().getResourceName());
+ if(queue != null)
+ {
+ final long messageId = record.getMessage().getMessageNumber();
+ final AbstractServerMessageImpl message = _recoveredMessages.get(messageId);
+ _unusedMessages.remove(messageId);
+
+ if(message != null)
+ {
+ final QueueEntry entry = queue.getMessageOnTheQueue(messageId);
+
+ entry.acquire();
+
+ branch.dequeue(queue, message);
+
+ branch.addPostTransactionAcion(new ServerTransaction.Action()
+ {
+
+ public void postCommit()
+ {
+ entry.discard();
+ }
+
+ public void onRollback()
+ {
+ entry.release();
+ }
+ });
+ }
+ else
+ {
+ StringBuilder xidString = xidAsString(id);
+ String messageNumberString = String.valueOf(message.getMessageNumber());
+ CurrentActor.get().message(_logSubject,
+ TransactionLogMessages.XA_INCOMPLETE_MESSAGE(xidString.toString(),
+ messageNumberString));
+
+ }
+
+ }
+ else
+ {
+ StringBuilder xidString = xidAsString(id);
+ CurrentActor.get().message(_logSubject,
+ TransactionLogMessages.XA_INCOMPLETE_QUEUE(xidString.toString(),
+ queue.getName()));
+ }
+
+ }
+
+ try
+ {
+ branch.setState(DtxBranch.State.PREPARED);
+ branch.prePrepareTransaction();
+ }
+ catch (AMQStoreException e)
+ {
+ _logger.error("Unexpected database exception when attempting to prepare a recovered XA transaction " +
+ xidAsString(id), e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static StringBuilder xidAsString(Xid id)
+ {
+ return new StringBuilder("(")
+ .append(id.getFormat())
+ .append(',')
+ .append(Functions.str(id.getGlobalId()))
+ .append(',')
+ .append(Functions.str(id.getBranchId()))
+ .append(')');
+ }
+
+ public void completeDtxRecordRecovery()
+ {
+ for(StoredMessage m : _unusedMessages.values())
+ {
+ _logger.warn("Message id " + m.getMessageNumber() + " in store, but not in any queue - removing....");
+ m.remove();
+ }
+ CurrentActor.get().message(_logSubject, TransactionLogMessages.RECOVERY_COMPLETE(null, false));
+ }
+
private static final class ProcessAction
{
private final AMQQueue _queue;
@@ -354,15 +514,9 @@ public class VirtualHostConfigRecoveryHa
}
- public void completeQueueEntryRecovery()
+ public DtxRecordRecoveryHandler completeQueueEntryRecovery()
{
- for(StoredMessage m : _unusedMessages.values())
- {
- _logger.warn("Message id " + m.getMessageNumber() + " in store, but not in any queue - removing....");
- m.remove();
- }
-
for(Map.Entry<String,Integer> entry : _queueRecoveries.entrySet())
{
CurrentActor.get().message(_logSubject, TransactionLogMessages.RECOVERED(entry.getValue(), entry.getKey()));
@@ -370,7 +524,9 @@ public class VirtualHostConfigRecoveryHa
CurrentActor.get().message(_logSubject, TransactionLogMessages.RECOVERY_COMPLETE(entry.getKey(), true));
}
- CurrentActor.get().message(_logSubject, TransactionLogMessages.RECOVERY_COMPLETE(null, false));
+
+
+ return this;
}
private static class DummyMessage implements EnqueableMessage
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org