Repository: ably/ably-java
Issue: waiter.close() is invoked early on onAuthUpdatedAsync method

After release of 1.2.15 
Me and KacperKluka realized that waiter is closed too early, just after execute call from background thread. That was a copy/paste mistake from the previous method.

We should move waiter.close method inside the background thread and close it after the while loop breaks.



┆Issue is synchronized with this [Jira Bug](https://ably.atlassian.net/browse/SDK-2468) by [Unito](https://www.unito.io)


Changed files:
lib/src/test/java/io/ably/lib/test/realtime/RealtimeAuthTest.java, lib/src/main/java/io/ably/lib/transport/ConnectionManager.java

diff --git a/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java b/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java
index fe29b824..2004cf8c 100644
--- a/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java
+++ b/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java
@@ -988,71 +988,69 @@ public class ConnectionManager implements ConnectListener {
      **/
     public void onAuthUpdatedAsync(final String token, final Auth.AuthUpdateResult authUpdateResult) {
         final ConnectionWaiter waiter = new ConnectionWaiter();
-        try {
-            switch (currentState.state) {
-                case connected:
-                    /* (RTC8a) If the connection is in the CONNECTED currentState and
-                     * auth.authorize is called or Ably requests a re-authentication
-                     * (see RTN22), the client must obtain a new token, then send an
-                     * AUTH ProtocolMessage to Ably with an auth attribute
-                     * containing an AuthDetails object with the token string. */
-                    try {
-                        ProtocolMessage msg = new ProtocolMessage(ProtocolMessage.Action.auth);
-                        msg.auth = new ProtocolMessage.AuthDetails(token);
-                        send(msg, false, null);
-                    } catch (AblyException e) {
-                        /* The send failed. Close the transport; if a subsequent
-                         * reconnect succeeds, it will be with the new token. */
-                        Log.v(TAG, "onAuthUpdated: closing transport after send failure");
-                        transport.close();
-                    }
-                    break;
+        switch (currentState.state) {
+            case connected:
+                /* (RTC8a) If the connection is in the CONNECTED currentState and
+                 * auth.authorize is called or Ably requests a re-authentication
+                 * (see RTN22), the client must obtain a new token, then send an
+                 * AUTH ProtocolMessage to Ably with an auth attribute
+                 * containing an AuthDetails object with the token string. */
+                try {
+                    ProtocolMessage msg = new ProtocolMessage(ProtocolMessage.Action.auth);
+                    msg.auth = new ProtocolMessage.AuthDetails(token);
+                    send(msg, false, null);
+                } catch (AblyException e) {
+                    /* The send failed. Close the transport; if a subsequent
+                     * reconnect succeeds, it will be with the new token. */
+                    Log.v(TAG, "onAuthUpdated: closing transport after send failure");
+                    transport.close();
+                }
+                break;
 
-                case connecting:
-                    /* Close the connecting transport. */
-                    Log.v(TAG, "onAuthUpdated: closing connecting transport");
-                    ErrorInfo disconnectError = new ErrorInfo("Aborting incomplete connection with superseded auth params", 503, 80003);
-                    requestState(new StateIndication(ConnectionState.disconnected, disconnectError, null, null));
-                    /* Start a new connection attempt. */
-                    connect();
-                    break;
+            case connecting:
+                /* Close the connecting transport. */
+                Log.v(TAG, "onAuthUpdated: closing connecting transport");
+                ErrorInfo disconnectError = new ErrorInfo("Aborting incomplete connection with superseded auth params", 503, 80003);
+                requestState(new StateIndication(ConnectionState.disconnected, disconnectError, null, null));
+                /* Start a new connection attempt. */
+                connect();
+                break;
 
-                default:
-                    /* Start a new connection attempt. */
-                    connect();
-                    break;
-            }
+            default:
+                /* Start a new connection attempt. */
+                connect();
+                break;
+        }
 
-            /* Wait for a currentState transition into anything other than connecting or
-             * disconnected in a background thread */
-            singleThreadExecutor.execute(() -> {
-                boolean waitingForConnected = true;
-                while (waitingForConnected) {
-                    final ErrorInfo reason = waiter.waitForChange();
-                    final ConnectionState connectionState = currentState.state;
-                    switch (connectionState) {
-                        case connected:
-                            authUpdateResult.onUpdate(true, null);
-                            Log.v(TAG, "onAuthUpdated: got connected");
-                            waitingForConnected = false;
-                            break;
+        /* Wait for a currentState transition into anything other than connecting or
+         * disconnected in a background thread */
+        singleThreadExecutor.execute(() -> {
+            boolean waitingForConnected = true;
+            while (waitingForConnected) {
+                final ErrorInfo reason = waiter.waitForChange();
+                final ConnectionState connectionState = currentState.state;
+                switch (connectionState) {
+                    case connected:
+                        authUpdateResult.onUpdate(true, null);
+                        Log.v(TAG, "onAuthUpdated: got connected");
+                        waitingForConnected = false;
+                        break;
 
-                        case connecting:
-                        case disconnected:
-                            Log.v(TAG, "onAuthUpdated: " + connectionState);
-                            break;
+                    case connecting:
+                    case disconnected:
+                        Log.v(TAG, "onAuthUpdated: " + connectionState);
+                        break;
 
-                        default:
-                            /* suspended/closed/error: throw the error. */
-                            Log.v(TAG, "onAuthUpdated: throwing exception");
-                            authUpdateResult.onUpdate(false, reason);
-                            waitingForConnected = false;
-                    }
+                    default:
+                        /* suspended/closed/error: throw the error. */
+                        Log.v(TAG, "onAuthUpdated: throwing exception");
+                        authUpdateResult.onUpdate(false, reason);
+                        waitingForConnected = false;
                 }
-            });
-        } finally {
+            }
             waiter.close();
-        }
+        });
+
     }
 
     /**
diff --git a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeAuthTest.java b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeAuthTest.java
index 4fa6acc3..f0cbf2b1 100644
--- a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeAuthTest.java
+++ b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeAuthTest.java
@@ -33,6 +33,10 @@ import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
 public class RealtimeAuthTest extends ParameterizedTest {
 
     @Rule
@@ -914,6 +918,7 @@ public class RealtimeAuthTest extends ParameterizedTest {
                 try {
                     opts.wait();
                 } catch(InterruptedException ie) {}
+
                 ably.auth.renewAuth((success, tokenDetails1, errorInfo) -> {
                     //Ignore completion handling
                 });
@@ -933,6 +938,84 @@ public class RealtimeAuthTest extends ParameterizedTest {
         }
     }
 
+    @Test
+    public void auth_renewAuth_callback_invoked() throws InterruptedException {
+        try {
+            /* get a TokenDetails */
+            final String testKey = testVars.keys[0].keyStr;
+            final ClientOptions clientOptions = createOptions(testKey);
+            final AblyRest ablyRest = new AblyRest(clientOptions);
+
+            final TokenDetails tokenDetails = ablyRest.auth.requestToken(new Auth.TokenParams() {{
+                ttl = 1000L;
+            }}, null);
+            assertNotNull("Expected token value", tokenDetails.token);
+
+            // create Ably realtime instance with token and authCallback
+            class ProtocolListener extends DebugOptions implements DebugOptions.RawProtocolListener {
+                ProtocolListener() {
+                    Setup.getTestVars().fillInOptions(this);
+                    protocolListener = this;
+                }
+
+                @Override
+                public void onRawConnectRequested(String url) {
+                    synchronized (this) {
+                        notify();
+                    }
+                }
+
+                @Override
+                public void onRawConnect(String url) {
+                }
+
+                @Override
+                public void onRawMessageSend(ProtocolMessage message) {
+                }
+
+                @Override
+                public void onRawMessageRecv(ProtocolMessage message) {
+                }
+            }
+
+            final ProtocolListener protocolListener = new ProtocolListener();
+            protocolListener.autoConnect = false;
+            protocolListener.tokenDetails = tokenDetails;
+            //   implement callback, using Ably instance with key
+            protocolListener.authCallback = params -> tokenDetails;
+
+            final AblyRealtime ably = new AblyRealtime(protocolListener);
+            synchronized (protocolListener) {
+                ably.connect();
+                try {
+                    protocolListener.wait();
+                } catch (InterruptedException ie) {
+                    fail("auth_expired_token_expire_renew protocolListener.wait(): interrupted -" + ie.getMessage());
+                }
+            }
+
+            final Helpers.ConnectionWaiter connectionWaiter = new Helpers.ConnectionWaiter(ably.connection);
+            boolean isConnected = connectionWaiter.waitFor(ConnectionState.connected, 1, 4000L);
+            if (isConnected) {
+                final CountDownLatch latch = new CountDownLatch(1);
+                final AtomicBoolean isCalled = new AtomicBoolean(false);
+                ably.auth.renewAuth((success, tokenDetails1, errorInfo) -> {
+                    latch.countDown();
+                    isCalled.set(true);
+                });
+                latch.await(30, TimeUnit.SECONDS);
+                assertTrue("Callback not invoked", isCalled.get());
+                ably.close();
+            } else {
+                fail("auth_renewAuth_callback_invoked: unable to connect; final state = " + ably.connection.state);
+            }
+        } catch (AblyException e) {
+            e.printStackTrace();
+            fail("auth_renewAuth_callback_invoked: Unexpected exception instantiating library: " + e.getMessage());
+        }
+    }
+
+
     /**
      * Verify that with queryTime=false, when instancing with an already-expired token and authCallback,
      * connection can succeed