QPID-5776 : Clean up sessions and links on AMQP 1.0 connection closing

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1612404 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 43 insertions, 9 deletions

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 c9212b1a1e..bc961cb86e 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
@@ -21,21 +21,38 @@
 
 package org.apache.qpid.amqp_1_0.transport;
 
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.TimeoutException;
+
 import org.apache.qpid.amqp_1_0.framing.OversizeFrameException;
-import org.apache.qpid.amqp_1_0.type.*;
+import org.apache.qpid.amqp_1_0.type.Binary;
+import org.apache.qpid.amqp_1_0.type.DeliveryState;
+import org.apache.qpid.amqp_1_0.type.DistributionMode;
+import org.apache.qpid.amqp_1_0.type.FrameBody;
+import org.apache.qpid.amqp_1_0.type.Outcome;
+import org.apache.qpid.amqp_1_0.type.UnsignedInteger;
 import org.apache.qpid.amqp_1_0.type.messaging.Source;
 import org.apache.qpid.amqp_1_0.type.messaging.Target;
 import org.apache.qpid.amqp_1_0.type.messaging.TerminusDurability;
 import org.apache.qpid.amqp_1_0.type.messaging.TerminusExpiryPolicy;
-import org.apache.qpid.amqp_1_0.type.transaction.*;
+import org.apache.qpid.amqp_1_0.type.transaction.Coordinator;
 import org.apache.qpid.amqp_1_0.type.transaction.TxnCapability;
-import org.apache.qpid.amqp_1_0.type.transport.*;
+import org.apache.qpid.amqp_1_0.type.transport.AmqpError;
+import org.apache.qpid.amqp_1_0.type.transport.Attach;
+import org.apache.qpid.amqp_1_0.type.transport.Begin;
+import org.apache.qpid.amqp_1_0.type.transport.Detach;
+import org.apache.qpid.amqp_1_0.type.transport.Disposition;
+import org.apache.qpid.amqp_1_0.type.transport.End;
 import org.apache.qpid.amqp_1_0.type.transport.Error;
-
-
-import java.nio.ByteBuffer;
-import java.util.*;
-import java.util.concurrent.TimeoutException;
+import org.apache.qpid.amqp_1_0.type.transport.Flow;
+import org.apache.qpid.amqp_1_0.type.transport.LinkError;
+import org.apache.qpid.amqp_1_0.type.transport.Role;
+import org.apache.qpid.amqp_1_0.type.transport.Transfer;
 
 public class SessionEndpoint
 {
@@ -605,7 +622,7 @@ public class SessionEndpoint
             ByteBuffer payload = xfr.getPayload();
             int payloadSent = send(xfr, payload);
 
-            if(payload != null && payloadSent < payload.remaining())
+            if(payload != null && payloadSent < payload.remaining() && payloadSent >= 0)
             {
                 payload = payload.duplicate();
                 payload.position(payload.position()+payloadSent);
@@ -849,6 +866,10 @@ public class SessionEndpoint
         return _linkMap;
     }
 
+    public Collection<LinkEndpoint> getLocalLinkEndpoints()
+    {
+        return new ArrayList<>(_localLinkEndpoints.keySet());
+    }
 
     public boolean isEnded()
     {
diff --git a/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java b/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
index 6d780461f4..31a962f71e 100644
--- a/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
+++ b/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
@@ -209,7 +209,15 @@ public class Connection_1_0 implements ConnectionEventListener, AMQConnectionMod
 
     public void closeReceived()
     {
+        Collection<Session_1_0> sessions = new ArrayList(_sessions);
+
+        for(Session_1_0 session : sessions)
+        {
+            session.remoteEnd(new End());
+        }
+
         List<Action<? super Connection_1_0>> taskCopy;
+
         synchronized (_closeTasks)
         {
             taskCopy = new ArrayList<Action<? super Connection_1_0>>(_closeTasks);
diff --git a/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java b/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
index 2c7884b3ce..b870eaf630 100644
--- a/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
+++ b/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
@@ -510,6 +510,11 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel<Sessio
             iter.remove();
         }
 
+        for(LinkEndpoint linkEndpoint : _endpoint.getLocalLinkEndpoints())
+        {
+            linkEndpoint.remoteDetached(new Detach());
+        }
+
         _connection.sessionEnded(this);
 
     }