You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kw...@apache.org on 2011/08/25 12:46:40 UTC
svn commit: r1161492 - in /qpid/trunk/qpid/java:
broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
common/src/main/java/org/apache/qpid/transport/Connection.java
Author: kwall
Date: Thu Aug 25 10:46:39 2011
New Revision: 1161492
URL: http://svn.apache.org/viewvc?rev=1161492&view=rev
Log:
QPID-3452: Broker now unregisters any remaining subscriptions on receipt of SessionDetach to prevent SubFlushRunner and QueueRunner sending erroneous frames causing a ProtocolViolationException on the client.
Modified:
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java?rev=1161492&r1=1161491&r2=1161492&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java Thu Aug 25 10:46:39 2011
@@ -21,6 +21,7 @@
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.List;
@@ -36,6 +37,7 @@ import org.apache.qpid.server.registry.I
import org.apache.qpid.server.security.SecurityManager;
import org.apache.qpid.server.security.auth.AuthenticationResult;
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.*;
@@ -194,4 +196,23 @@ public class ServerConnectionDelegate ex
{
return ApplicationRegistry.getInstance().getConfiguration().getMaxChannelCount();
}
+
+ @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
+ // completes.
+ unregisterAllSubscriptions(conn, dtc);
+ super.sessionDetach(conn, dtc);
+ }
+
+ private void unregisterAllSubscriptions(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);
+ }
+ }
}
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java?rev=1161492&r1=1161491&r2=1161492&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java Thu Aug 25 10:46:39 2011
@@ -35,7 +35,6 @@ import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
import javax.security.sasl.SaslClient;
import javax.security.sasl.SaslServer;
@@ -406,7 +405,7 @@ public class Connection extends Connecti
else
{
throw new ProtocolViolationException(
- "Received frames for an already dettached session", null);
+ "Received frames for an already detached session", null);
}
}
@@ -455,7 +454,7 @@ public class Connection extends Connecti
}
}
- protected Session getSession(int channel)
+ public Session getSession(int channel)
{
synchronized (lock)
{
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org