From 3bfa11faa98798bd7caa51503decb1f7fc299105 Mon Sep 17 00:00:00 2001 From: ikbalkaya Date: Fri, 15 Jul 2022 14:52:17 +0100 Subject: [PATCH 1/5] Moved waiter.close inside background thread after while break --- .../ably/lib/transport/ConnectionManager.java | 116 +++++++++--------- 1 file changed, 57 insertions(+), 59 deletions(-) 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..c68fb2a46 100644 --- a/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java +++ b/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java @@ -987,72 +987,70 @@ public void onAuthUpdated(final String token, final boolean waitForResponse) thr * 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; + 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(() -> { + final ConnectionWaiter waiter = new ConnectionWaiter(); + 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(); - } + }); + } /** From ac9df421b69f78c1b4a68ff5b3b1baec2ba9d28f Mon Sep 17 00:00:00 2001 From: ikbalkaya Date: Mon, 18 Jul 2022 19:30:45 +0100 Subject: [PATCH 2/5] Add simulation result for test CI --- .../ably/lib/transport/ConnectionManager.java | 15 +++- .../lib/test/realtime/RealtimeAuthTest.java | 79 +++++++++++++++++++ 2 files changed, 90 insertions(+), 4 deletions(-) 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 c68fb2a46..eff6c9121 100644 --- a/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java +++ b/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java @@ -987,6 +987,7 @@ public void onAuthUpdated(final String token, final boolean waitForResponse) thr * 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(); switch (currentState.state) { case connected: /* (RTC8a) If the connection is in the CONNECTED currentState and @@ -1024,8 +1025,14 @@ public void onAuthUpdatedAsync(final String token, final Auth.AuthUpdateResult a /* Wait for a currentState transition into anything other than connecting or * disconnected in a background thread */ singleThreadExecutor.execute(() -> { - final ConnectionWaiter waiter = new ConnectionWaiter(); - boolean waitingForConnected = true; + //simulate result + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + e.printStackTrace(); + } + authUpdateResult.onUpdate(true, null); + /* boolean waitingForConnected = true; while (waitingForConnected) { final ErrorInfo reason = waiter.waitForChange(); final ConnectionState connectionState = currentState.state; @@ -1042,12 +1049,12 @@ public void onAuthUpdatedAsync(final String token, final Auth.AuthUpdateResult a break; default: - /* suspended/closed/error: throw the error. */ + *//* suspended/closed/error: throw the error. *//* Log.v(TAG, "onAuthUpdated: throwing exception"); authUpdateResult.onUpdate(false, reason); waitingForConnected = false; } - } + }*/ 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..8013f91b0 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,9 @@ import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicBoolean; + public class RealtimeAuthTest extends ParameterizedTest { @Rule @@ -914,9 +917,16 @@ public Object getTokenRequest(Auth.TokenParams params) { try { opts.wait(); } catch(InterruptedException ie) {} + final CountDownLatch latch = new CountDownLatch(1); ably.auth.renewAuth((success, tokenDetails1, errorInfo) -> { //Ignore completion handling + latch.countDown(); }); + try { + latch.await(); + } catch (InterruptedException e) { + fail("auth_expired_token_expire_renew: interrupted"); + } } Helpers.ConnectionWaiter connectionWaiter = new Helpers.ConnectionWaiter(ably.connection); @@ -933,6 +943,75 @@ 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) { + AtomicBoolean isCalled = new AtomicBoolean(false); + ably.auth.renewAuth((success, tokenDetails1, errorInfo) -> { + isCalled.set(true); + }); + Thread.sleep(1200); + assertTrue("Callback not invoked", isCalled.get()); + assertTrue(isConnected); + 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 From fb5e5b83ae8f31069f2941617472a8bb224a97db Mon Sep 17 00:00:00 2001 From: ikbalkaya Date: Mon, 18 Jul 2022 19:38:36 +0100 Subject: [PATCH 3/5] Remove commented out code --- .../ably/lib/transport/ConnectionManager.java | 23 ------------------- 1 file changed, 23 deletions(-) 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 eff6c9121..57dd3e5c1 100644 --- a/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java +++ b/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java @@ -1032,29 +1032,6 @@ public void onAuthUpdatedAsync(final String token, final Auth.AuthUpdateResult a e.printStackTrace(); } authUpdateResult.onUpdate(true, null); - /* 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; - } - }*/ waiter.close(); }); From d548041f8039a820c52ff4ae92908d2ab8f96faf Mon Sep 17 00:00:00 2001 From: ikbalkaya Date: Mon, 18 Jul 2022 21:07:05 +0100 Subject: [PATCH 4/5] Reenact code --- .../ably/lib/transport/ConnectionManager.java | 28 +++++++++++++++---- 1 file changed, 22 insertions(+), 6 deletions(-) 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 57dd3e5c1..2004cf8cf 100644 --- a/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java +++ b/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java @@ -1025,13 +1025,29 @@ public void onAuthUpdatedAsync(final String token, final Auth.AuthUpdateResult a /* Wait for a currentState transition into anything other than connecting or * disconnected in a background thread */ singleThreadExecutor.execute(() -> { - //simulate result - try { - Thread.sleep(1000); - } catch (InterruptedException e) { - e.printStackTrace(); + 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; + } } - authUpdateResult.onUpdate(true, null); waiter.close(); }); From 370a444d305bc47d29e5b0ed654cf25e5097cb33 Mon Sep 17 00:00:00 2001 From: ikbalkaya Date: Mon, 18 Jul 2022 21:52:08 +0100 Subject: [PATCH 5/5] Move latch to invoked test --- .../lib/test/realtime/RealtimeAuthTest.java | 42 ++++++++++--------- 1 file changed, 23 insertions(+), 19 deletions(-) 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 8013f91b0..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 @@ -34,6 +34,7 @@ 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 { @@ -917,16 +918,10 @@ public Object getTokenRequest(Auth.TokenParams params) { try { opts.wait(); } catch(InterruptedException ie) {} - final CountDownLatch latch = new CountDownLatch(1); + ably.auth.renewAuth((success, tokenDetails1, errorInfo) -> { //Ignore completion handling - latch.countDown(); }); - try { - latch.await(); - } catch (InterruptedException e) { - fail("auth_expired_token_expire_renew: interrupted"); - } } Helpers.ConnectionWaiter connectionWaiter = new Helpers.ConnectionWaiter(ably.connection); @@ -951,7 +946,9 @@ public void auth_renewAuth_callback_invoked() throws InterruptedException { final ClientOptions clientOptions = createOptions(testKey); final AblyRest ablyRest = new AblyRest(clientOptions); - final TokenDetails tokenDetails = ablyRest.auth.requestToken(new Auth.TokenParams(){{ ttl = 1000L; }}, null); + 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 @@ -960,19 +957,25 @@ class ProtocolListener extends DebugOptions implements DebugOptions.RawProtocolL Setup.getTestVars().fillInOptions(this); protocolListener = this; } + @Override public void onRawConnectRequested(String url) { - synchronized(this) { + synchronized (this) { notify(); } } @Override - public void onRawConnect(String url) {} + public void onRawConnect(String url) { + } + @Override - public void onRawMessageSend(ProtocolMessage message) {} + public void onRawMessageSend(ProtocolMessage message) { + } + @Override - public void onRawMessageRecv(ProtocolMessage message) {} + public void onRawMessageRecv(ProtocolMessage message) { + } } final ProtocolListener protocolListener = new ProtocolListener(); @@ -986,21 +989,22 @@ public void onRawMessageRecv(ProtocolMessage message) {} ably.connect(); try { protocolListener.wait(); - } catch(InterruptedException ie) { - fail( "auth_expired_token_expire_renew protocolListener.wait(): interrupted -"+ie.getMessage()); + } 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) { - AtomicBoolean isCalled = new AtomicBoolean(false); - ably.auth.renewAuth((success, tokenDetails1, errorInfo) -> { + 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); }); - Thread.sleep(1200); + latch.await(30, TimeUnit.SECONDS); assertTrue("Callback not invoked", isCalled.get()); - assertTrue(isConnected); ably.close(); } else { fail("auth_renewAuth_callback_invoked: unable to connect; final state = " + ably.connection.state);