Gracefully disconnect XMPPTCPConnection

That is, wait with a timeout for a closing stream element if we have
sent a closing stream element to the server. See RFC 6120 § 4.4.

Fixes SMACK-633.

smack-core/src/main/java/org/jivesoftware/smack/SynchronizationPoint.java

@@ -65,8 +65,9 @@ public class SynchronizationPoint<E extends Exception> {
      * @param request the plain stream element to send.
      * @throws NoResponseException if no response was received.
      * @throws NotConnectedException if the connection is not connected.
+     * @return <code>true</code> if synchronization point was successful, <code>false</code> on failure.
      */
-    public void sendAndWaitForResponse(TopLevelStreamElement request) throws NoResponseException,
+    public boolean sendAndWaitForResponse(TopLevelStreamElement request) throws NoResponseException,
                     NotConnectedException, InterruptedException {
         assert (state == State.Initial);
         connectionLock.lock();
@@ -87,7 +88,7 @@ public class SynchronizationPoint<E extends Exception> {
         finally {
             connectionLock.unlock();
         }
-        checkForResponse();
+        return checkForResponse();
     }
 
     /**
@@ -129,19 +130,26 @@ public class SynchronizationPoint<E extends Exception> {
      * Check if this synchronization point is successful or wait the connections reply timeout.
      * @throws NoResponseException if there was no response marking the synchronization point as success or failed.
      * @throws InterruptedException
+     * @return <code>true</code> if synchronization point was successful, <code>false</code> on failure.
      */
-    public void checkIfSuccessOrWait() throws NoResponseException, InterruptedException {
+    public boolean checkIfSuccessOrWait() throws NoResponseException, InterruptedException {
         connectionLock.lock();
         try {
-            if (state == State.Success) {
+            switch (state) {
-                // Return immediately
+            // Return immediately on success or failure
-                return;
+            case Success:
+                return true;
+            case Failure:
+                return false;
+            default:
+                // Do nothing
+                break;
             }
             waitForConditionOrTimeout();
         } finally {
             connectionLock.unlock();
         }
-        checkForResponse();
+        return checkForResponse();
     }
 
     /**
@@ -240,15 +248,18 @@ public class SynchronizationPoint<E extends Exception> {
      * </p>
      * @throws NoResponseException
      */
-    private void checkForResponse() throws NoResponseException {
+    private boolean checkForResponse() throws NoResponseException {
         switch (state) {
         case Initial:
         case NoResponse:
         case RequestSent:
             throw NoResponseException.newWith(connection);
+        case Success:
+            return true;
+        case Failure:
+            return false;
         default:
-            // Do nothing
+            throw new AssertionError("Unknown state " + state);
-            break;
         }
     }
 

smack-integration-test/src/main/java/org/igniterealtime/smack/inttest/IntTestUtil.java

@@ -75,6 +75,17 @@ public class IntTestUtil {
 
     public static void disconnectAndMaybeDelete(XMPPTCPConnection connection, boolean delete)
                     throws InterruptedException {
+        // If the connection is disconnected, then re-reconnect and login. This could happen when
+        // (low-level) integration tests disconnect the connection, e.g. to test disconnection
+        // mechanisms
+        if (!connection.isConnected()) {
+            try {
+                connection.connect().login();
+            }
+            catch (XMPPException | SmackException | IOException e) {
+                LOGGER.log(Level.WARNING, "Exception reconnection account for deletion", e);
+            }
+        }
         try {
             if (delete) {
                 final int maxAttempts = 3;

smack-integration-test/src/main/java/org/jivesoftware/smack/WaitForClosingStreamElementTest.java

@@ -0,0 +1,45 @@
+/**
+ *
+ * Copyright 2015 Florian Schmaus
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.jivesoftware.smack;
+
+import static org.junit.Assert.assertTrue;
+
+import java.lang.reflect.Field;
+
+import org.igniterealtime.smack.inttest.AbstractSmackLowLevelIntegrationTest;
+import org.igniterealtime.smack.inttest.Configuration;
+import org.igniterealtime.smack.inttest.SmackIntegrationTest;
+import org.jivesoftware.smack.tcp.XMPPTCPConnection;
+
+public class WaitForClosingStreamElementTest extends AbstractSmackLowLevelIntegrationTest {
+
+    public WaitForClosingStreamElementTest(Configuration configuration, String testRunId) {
+        super(configuration, testRunId);
+    }
+
+    @SmackIntegrationTest
+    public void waitForClosingStreamElementTest(XMPPTCPConnection connection)
+                    throws NoSuchFieldException, SecurityException, IllegalArgumentException,
+                    IllegalAccessException {
+        connection.disconnect();
+
+        Field closingStreamReceivedField = connection.getClass().getDeclaredField("closingStreamReceived");
+        closingStreamReceivedField.setAccessible(true);
+        SynchronizationPoint<?> closingStreamReceived = (SynchronizationPoint<?>) closingStreamReceivedField.get(connection);
+        assertTrue(closingStreamReceived.wasSuccessful());
+    }
+}

smack-tcp/src/main/java/org/jivesoftware/smack/tcp/XMPPTCPConnection.java

@@ -188,6 +188,12 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
     private final SynchronizationPoint<XMPPException> compressSyncPoint = new SynchronizationPoint<XMPPException>(
                     this);
 
+    /**
+     * A synchronization point which is successful if this connection has received the closing
+     * stream element from the remote end-point, i.e. the server.
+     */
+    private final SynchronizationPoint<Exception> closingStreamReceived = new SynchronizationPoint<Exception>(this);
+
     /**
      * The default bundle and defer callback, used for new connections.
      * @see bundleAndDeferCallback
@@ -458,17 +464,29 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
         shutdown(true);
     }
 
+    /**
+     * <code>true</code> if this connection is currently disconnecting the session by performing a
+     * shut down.
+     */
+    private volatile boolean shutdownInProgress;
+
     private void shutdown(boolean instant) {
         if (disconnectedButResumeable) {
             return;
         }
-        if (packetReader != null) {
+
-                packetReader.shutdown();
+        shutdownInProgress = true;
-        }
+
+        // First shutdown the writer, this will result in a closing stream element getting send to
+        // the server
         if (packetWriter != null) {
             packetWriter.shutdown(instant);
         }
 
+        if (packetReader != null) {
+                packetReader.shutdown();
+        }
+
         // Set socketClosed to true. This will cause the PacketReader
         // and PacketWriter to ignore any Exceptions that are thrown
         // because of a read/write from/to a closed stream.
@@ -492,6 +510,7 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
             // stream tag, there is no longer a stream to resume.
             smSessionId = null;
         }
+        shutdownInProgress = false;
         authenticated = false;
         connected = false;
         usingTLS = false;
@@ -789,6 +808,7 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
      */
     @Override
     protected void connectInternal() throws SmackException, IOException, XMPPException, InterruptedException {
+        closingStreamReceived.init();
         // Establishes the TCP connection to the server and does setup the reader and writer. Throws an exception if
         // there is an error establishing the connection
         connectUsingConfiguration();
@@ -1102,9 +1122,20 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
                         break;
                     case XmlPullParser.END_TAG:
                         if (parser.getName().equals("stream")) {
-                            // Disconnect the connection
+                            closingStreamReceived.reportSuccess();
+                            if (shutdownInProgress) {
+                                // We received a closing stream element *after* we initiated the
+                                // termination of the session by sending a closing stream element to
+                                // the server first
+                                return;
+                            } else {
+                                // We received a closing stream element from the server without us
+                                // sending a closing stream element first. This means that the
+                                // server wants to terminate the session, therefore disconnect
+                                // the connection
                                 disconnect();
                             }
+                        }
                         break;
                     case XmlPullParser.END_DOCUMENT:
                         // END_DOCUMENT only happens in an error case, as otherwise we would see a
@@ -1116,6 +1147,7 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
                 }
             }
             catch (Exception e) {
+                closingStreamReceived.reportFailure(e);
                 // The exception can be ignored if the the connection is 'done'
                 // or if the it was caused because the socket got closed
                 if (!(done || isSocketClosed())) {
@@ -1362,6 +1394,18 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
                     catch (Exception e) {
                         LOGGER.log(Level.WARNING, "Exception writing closing stream element", e);
                     }
+
+                    try {
+                        // After we send the closing stream element, check if there was already a
+                        // closing stream element sent by the server or wait with a timeout for a
+                        // closing stream element to be received from the server.
+                        closingStreamReceived.checkIfSuccessOrWait();
+                    } catch (NoResponseException e) {
+                        LOGGER.log(Level.INFO, "No response while waiting for closing stream element from the server (" + getConnectionCounter() + ")", e);
+                    } catch (InterruptedException e) {
+                        LOGGER.log(Level.INFO, "Waiting for closing stream element from the server was interrupted (" + getConnectionCounter() + ")", e);
+                    }
+
                     // Delete the queue contents (hopefully nothing is left).
                     queue.clear();
                 } else if (instantShutdown && isSmEnabled()) {