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 2010/09/12 14:46:12 UTC
svn commit: r996303 - in /qpid/trunk/qpid/java:
broker/src/main/java/org/apache/qpid/server/transport/
common/src/main/java/org/apache/qpid/transport/
Author: rgodfrey
Date: Sun Sep 12 12:46:11 2010
New Revision: 996303
URL: http://svn.apache.org/viewvc?rev=996303&view=rev
Log:
QPID-2855 : Broker Transport should not block on awaiting session close confirmation
Modified:
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java?rev=996303&r1=996302&r2=996303&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java Sun Sep 12 12:46:11 2010
@@ -336,6 +336,12 @@ public class ServerSession extends Sessi
}
+ @Override
+ protected void awaitClose()
+ {
+ // Broker shouldn't block awaiting close - thus do override this method to do nothing
+ }
+
public void acknowledge(final Subscription_0_10 sub, final QueueEntry entry)
{
_transaction.dequeue(entry.getQueue(), entry.getMessage(),
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java?rev=996303&r1=996302&r2=996303&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java Sun Sep 12 12:46:11 2010
@@ -101,11 +101,14 @@ public class ServerSessionDelegate exten
public void command(Session session, Method method)
{
SecurityManager.setThreadPrincipal(session.getConnection().getAuthorizationID());
-
- super.command(session, method);
- if (method.isSync())
+
+ if(!session.isClosing())
{
- session.flushProcessed();
+ super.command(session, method);
+ if (method.isSync())
+ {
+ session.flushProcessed();
+ }
}
}
@@ -1189,6 +1192,12 @@ public class ServerSessionDelegate exten
((ServerSession)session).onClose();
}
+ @Override
+ public void detached(Session session)
+ {
+ closed(session);
+ }
+
public Collection<Subscription_0_10> getSubscriptions(Session session)
{
return ((ServerSession)session).getSubscriptions();
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java?rev=996303&r1=996302&r2=996303&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java Sun Sep 12 12:46:11 2010
@@ -543,7 +543,7 @@ public class Session extends SessionInvo
}
}
- if (state != OPEN && state != CLOSED)
+ if (state != OPEN && state != CLOSED && state != CLOSING)
{
Thread current = Thread.currentThread();
if (!current.equals(resumer))
@@ -568,6 +568,7 @@ public class Session extends SessionInvo
("timed out waiting for resume to finish");
}
break;
+ case CLOSING:
case CLOSED:
ExecutionException exc = getException();
if (exc != null)
@@ -906,18 +907,23 @@ public class Session extends SessionInvo
sessionRequestTimeout(0);
sessionDetach(name.getBytes());
- Waiter w = new Waiter(commands, timeout);
- while (w.hasTime() && state != CLOSED)
- {
- w.await();
- }
-
- if (state != CLOSED)
- {
- throw new SessionException("close() timed out");
- }
+ awaitClose();
- connection.removeSession(this);
+
+ }
+ }
+
+ protected void awaitClose()
+ {
+ Waiter w = new Waiter(commands, timeout);
+ while (w.hasTime() && state != CLOSED)
+ {
+ w.await();
+ }
+
+ if (state != CLOSED)
+ {
+ throw new SessionException("close() timed out");
}
}
@@ -960,6 +966,16 @@ public class Session extends SessionInvo
delegate.detached(this);
}
}
+
+ if(state == CLOSED)
+ {
+ connection.removeSession(this);
+ }
+ }
+
+ public boolean isClosing()
+ {
+ return state == CLOSED || state == CLOSING;
}
public String toString()
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org