Repository: ably/ably-java
Issue: Early return from  onAuthUpdated creates issues

`AblyBase` and `ConnectionManager` have a method called `onAuthUpdated` that contains a `waitForResponse` and is used to decide whether to wait for async response before returning from method. 
This seems to be set to `false`  only when `Auth.renew()` is called. 

This feature doesn't seem to be on spec. Apart from this, this method is implicitly asynchronous. When the `waitForResponse` is set to `false` there is no opportunity for caller to know about the result. For our specific case we want to know when the token renewal is completed. As a result we either have to


* Remove this parameter and wait for response for all cases, so that it is safe to call it synchronously
or
* Provide a callback so that  callers are notified of completion. 

Please see [this thread](https://github.com/ably/ably-asset-tracking-android/pull/703/files#r907540279) to see where this discussion has started.



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


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

diff --git a/lib/src/main/java/io/ably/lib/realtime/AblyRealtime.java b/lib/src/main/java/io/ably/lib/realtime/AblyRealtime.java
index 2a27a50f..f91e3dbd 100644
--- a/lib/src/main/java/io/ably/lib/realtime/AblyRealtime.java
+++ b/lib/src/main/java/io/ably/lib/realtime/AblyRealtime.java
@@ -4,6 +4,7 @@ import java.util.Iterator;
 import java.util.Map;
 
 import io.ably.lib.rest.AblyRest;
+import io.ably.lib.rest.Auth;
 import io.ably.lib.transport.ConnectionManager;
 import io.ably.lib.types.AblyException;
 import io.ably.lib.types.ChannelOptions;
@@ -100,6 +101,14 @@ public class AblyRealtime extends AblyRest {
         connection.connectionManager.onAuthUpdated(token, waitForResponse);
     }
 
+    /**
+     * Authentication token has changed. Async version
+     */
+    @Override
+    protected void onAuthUpdatedAsync(String token, Auth.AuthUpdateResult authUpdateResult)  {
+        connection.connectionManager.onAuthUpdatedAsync(token,authUpdateResult);
+    }
+
     /**
      * Authentication error occurred
      */
diff --git a/lib/src/main/java/io/ably/lib/rest/AblyBase.java b/lib/src/main/java/io/ably/lib/rest/AblyBase.java
index 57b14b54..deee49b8 100644
--- a/lib/src/main/java/io/ably/lib/rest/AblyBase.java
+++ b/lib/src/main/java/io/ably/lib/rest/AblyBase.java
@@ -316,6 +316,15 @@ public abstract class AblyBase implements AutoCloseable {
         /* Default is to do nothing. Overridden by subclass. */
     }
 
+    /**
+     * Override this method in AblyRealtime and pass updated token to ConnectionManager
+     * @param token new token
+     * @param authUpdateResult Callback result
+     */
+    protected void onAuthUpdatedAsync(String token, Auth.AuthUpdateResult authUpdateResult)  {
+        //this must be overriden by subclass
+    }
+
     /**
      * Authentication error occurred
      */
diff --git a/lib/src/main/java/io/ably/lib/rest/Auth.java b/lib/src/main/java/io/ably/lib/rest/Auth.java
index c9d7eec2..9ad0abc2 100644
--- a/lib/src/main/java/io/ably/lib/rest/Auth.java
+++ b/lib/src/main/java/io/ably/lib/rest/Auth.java
@@ -35,7 +35,6 @@ import io.ably.lib.util.Serialisation;
  *
  */
 public class Auth {
-
     /**
      * Authentication methods
      */
@@ -467,6 +466,33 @@ public class Auth {
         }
     }
 
+    /**
+     * An interface providing update result for onAuthUpdated
+     */
+    public interface AuthUpdateResult{
+        /**
+         * Signals an update from {@link io.ably.lib.transport.ConnectionManager#onAuthUpdatedAsync(String, AuthUpdateResult)}
+         * @param success If Update was successful
+         * @param errorInfo optional errorInfo if update wasn't successful
+         */
+        void onUpdate(boolean success, ErrorInfo errorInfo);
+    }
+
+    /**
+     * An interface providing completion callbackk for renewAuth
+     */
+    public interface RenewAuthResult {
+        /**
+         * Signals completion of {@link Auth#renewAuth(RenewAuthResult)}
+         * @param success if token renewal was successful. Please note that success for this operation means that
+         *                other operations relating to this also succeeded.
+         * @param tokenDetails New token details. Please note that this value can exist regardless of value of
+         *                     success state.
+         * @param errorInfo Error details if operation is completed with error.
+         */
+        void onCompletion(boolean success,TokenDetails tokenDetails, ErrorInfo errorInfo);
+    }
+
     /**
      * An interface implemented by a callback that provides either tokens,
      * or signed token requests, in response to a request with given token params.
@@ -827,13 +853,31 @@ public class Auth {
      * Renew auth credentials.
      * Will obtain a new token, even if we already have an apparently valid one.
      * Authorization will use the parameters supplied on construction.
+     * @deprecated Because the method returns early before renew() completes and does not provide a completion
+     * handler for callers.
+     * Please use {@link Auth#renewAuth} instead
      */
+    @Deprecated
     public TokenDetails renew() throws AblyException {
         TokenDetails tokenDetails = assertValidToken(this.tokenParams, this.authOptions, true);
         ably.onAuthUpdated(tokenDetails.token, false);
         return tokenDetails;
     }
 
+    /**
+     * Renew auth credentials.
+     * Will obtain a new token, even if we already have an apparently valid one.
+     * Authorization will use the parameters supplied on construction.
+     * @param result Asynchronous result the completion
+     * Please note that completion callback  {@link RenewAuthResult#onCompletion(boolean, TokenDetails, ErrorInfo)}
+     *              is called on a background thread.
+     */
+    public void renewAuth(RenewAuthResult result) throws AblyException {
+        final TokenDetails tokenDetails = assertValidToken(this.tokenParams, this.authOptions, true);
+
+        ably.onAuthUpdatedAsync(tokenDetails.token, (success, errorInfo) -> result.onCompletion(success,tokenDetails,errorInfo));
+    }
+
     public void onAuthError(ErrorInfo err) {
         /* we're only interested in token expiry errors */
         if(err.code >= 40140 && err.code < 40150)
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 7cc5ddaa..fe29b824 100644
--- a/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java
+++ b/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java
@@ -1,5 +1,16 @@
 package io.ably.lib.transport;
 
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
 import io.ably.lib.debug.DebugOptions;
 import io.ably.lib.debug.DebugOptions.RawProtocolListener;
 import io.ably.lib.http.HttpHelpers;
@@ -10,8 +21,10 @@ import io.ably.lib.realtime.Connection;
 import io.ably.lib.realtime.ConnectionState;
 import io.ably.lib.realtime.ConnectionStateListener;
 import io.ably.lib.realtime.ConnectionStateListener.ConnectionStateChange;
+import io.ably.lib.rest.Auth;
 import io.ably.lib.transport.ITransport.ConnectListener;
 import io.ably.lib.transport.ITransport.TransportParams;
+import io.ably.lib.transport.NetworkConnectivity.NetworkConnectivityListener;
 import io.ably.lib.types.AblyException;
 import io.ably.lib.types.ClientOptions;
 import io.ably.lib.types.ConnectionDetails;
@@ -19,18 +32,12 @@ import io.ably.lib.types.ErrorInfo;
 import io.ably.lib.types.ProtocolMessage;
 import io.ably.lib.types.ProtocolSerializer;
 import io.ably.lib.util.Log;
-import io.ably.lib.transport.NetworkConnectivity.NetworkConnectivityListener;
 import io.ably.lib.util.PlatformAgentProvider;
 
-import java.util.ArrayDeque;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Locale;
-import java.util.Map;
-
 public class ConnectionManager implements ConnectListener {
+    final ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
+    final ExecutorCompletionService<Void> executorCompletionService =
+        new ExecutorCompletionService<>(singleThreadExecutor);
 
     /**************************************************************
      * ConnectionManager
@@ -975,6 +982,79 @@ public class ConnectionManager implements ConnectListener {
         }
     }
 
+
+    /**
+     * Async version of onAuthUpdated that returns a Future that includes an option Ably exception
+     **/
+    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;
+
+                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;
+            }
+
+            /* 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;
+
+                        default:
+                            /* suspended/closed/error: throw the error. */
+                            Log.v(TAG, "onAuthUpdated: throwing exception");
+                            authUpdateResult.onUpdate(false, reason);
+                            waitingForConnected = false;
+                    }
+                }
+            });
+        } finally {
+            waiter.close();
+        }
+    }
+
     /**
      * Called when where was an error during authentication attempt
      *
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 44dedb2b..4fa6acc3 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
@@ -861,6 +861,78 @@ public class RealtimeAuthTest extends ParameterizedTest {
         }
     }
 
+
+    /**
+     * Call renewAuth() whilst connecting; verify there's no crash (see https://github.com/ably/ably-java/issues/503)
+     */
+    @Test
+    public void auth_renewAuth_whilst_connecting() {
+        try {
+            /* get a TokenDetails */
+            final String testKey = testVars.keys[0].keyStr;
+            ClientOptions optsForToken = createOptions(testKey);
+            final AblyRest ablyForToken = new AblyRest(optsForToken);
+
+            final TokenDetails tokenDetails = ablyForToken.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) {}
+            }
+
+            ProtocolListener opts = new ProtocolListener();
+            opts.autoConnect = false;
+            opts.tokenDetails = tokenDetails;
+            opts.authCallback = new Auth.TokenCallback() {
+                /* implement callback, using Ably instance with key */
+                @Override
+                public Object getTokenRequest(Auth.TokenParams params) {
+                    return tokenDetails;
+                }
+            };
+
+            final AblyRealtime ably = new AblyRealtime(opts);
+            synchronized (opts) {
+                ably.connect();
+                try {
+                    opts.wait();
+                } catch(InterruptedException ie) {}
+                ably.auth.renewAuth((success, tokenDetails1, errorInfo) -> {
+                    //Ignore completion handling
+                });
+            }
+
+            Helpers.ConnectionWaiter connectionWaiter = new Helpers.ConnectionWaiter(ably.connection);
+            boolean isConnected = connectionWaiter.waitFor(ConnectionState.connected, 1, 4000L);
+            if(isConnected) {
+                /* done */
+                ably.close();
+            } else {
+                fail("auth_expired_token_expire_renew: unable to connect; final state = " + ably.connection.state);
+            }
+        } catch (AblyException e) {
+            e.printStackTrace();
+            fail("auth_expired_token_expire_renew: Unexpected exception instantiating library");
+        }
+    }
+
     /**
      * Verify that with queryTime=false, when instancing with an already-expired token and authCallback,
      * connection can succeed