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 fe29b8241..2004cf8cf 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 void onAuthUpdated(final String token, final boolean waitForResponse) thr **/ 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 4fa6acc3b..f0cbf2b14 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.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 Object getTokenRequest(Auth.TokenParams params) { try { opts.wait(); } catch(InterruptedException ie) {} + ably.auth.renewAuth((success, tokenDetails1, errorInfo) -> { //Ignore completion handling }); @@ -933,6 +938,84 @@ public Object getTokenRequest(Auth.TokenParams params) { } } + @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