diff options
| author | Robert Godfrey <rgodfrey@apache.org> | 2014-02-17 20:19:36 +0000 |
|---|---|---|
| committer | Robert Godfrey <rgodfrey@apache.org> | 2014-02-17 20:19:36 +0000 |
| commit | add5c695d1138bc25bb89cd0e1b1724bf542f676 (patch) | |
| tree | f222dd1710831dc383ada64bc4479bcfd4c8bbc4 /java/amqp-1-0-common | |
| parent | e31aa33452bd54ce118078ad38d8291cc3a3d1db (diff) | |
| download | qpid-python-java-broker-amqp-1-0-management.tar.gz | |
Update Queue implementation to better define lifetime and exclusivity policiesjava-broker-amqp-1-0-management
git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/java-broker-amqp-1-0-management@1569102 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'java/amqp-1-0-common')
2 files changed, 33 insertions, 5 deletions
diff --git a/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java b/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java index 270abef88b..1556876681 100644 --- a/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java +++ b/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java @@ -563,7 +563,7 @@ public class ConnectionEndpoint implements DescribedTypeConstructorRegistry.Sour { _receivingSessions[channel] = null; - endpoint.end(end); + endpoint.receiveEnd(end); } else { diff --git a/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/SessionEndpoint.java b/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/SessionEndpoint.java index c37c52c6ea..c9212b1a1e 100644 --- a/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/SessionEndpoint.java +++ b/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/SessionEndpoint.java @@ -153,19 +153,47 @@ public class SessionEndpoint public void end() { - end(null); + end(new End()); } - public void end(final End end) + public void end(End end) { synchronized(getLock()) { switch(_state) { case BEGIN_SENT: - _connection.sendEnd(getSendingChannel(), new End(), false); + _connection.sendEnd(getSendingChannel(), end, false); _state = SessionState.END_PIPE; break; + case ACTIVE: + detachLinks(); + short sendChannel = getSendingChannel(); + _connection.sendEnd(sendChannel, end, true); + _state = SessionState.END_SENT; + break; + default: + sendChannel = getSendingChannel(); + End reply = new End(); + Error error = new Error(); + error.setCondition(AmqpError.ILLEGAL_STATE); + error.setDescription("END called on Session which has not been opened"); + reply.setError(error); + _connection.sendEnd(sendChannel, reply, true); + break; + + + } + getLock().notifyAll(); + } + } + + public void receiveEnd(final End end) + { + synchronized(getLock()) + { + switch(_state) + { case END_SENT: _state = SessionState.ENDED; break; @@ -174,7 +202,7 @@ public class SessionEndpoint _sessionEventListener.remoteEnd(end); short sendChannel = getSendingChannel(); _connection.sendEnd(sendChannel, new End(), true); - _state = end == null ? SessionState.END_SENT : SessionState.ENDED; + _state = SessionState.ENDED; break; default: sendChannel = getSendingChannel(); |