Skip to content

call waiter.close() after breaking from while loop#825

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
QuintinWillison merged 5 commits into from
Jul 19, 2022
Merged

call waiter.close() after breaking from while loop #825

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
3bfa11f
Moved waiter.close inside background thread after while break
ikbalkaya Jul 15, 2022
ac9df42
Add simulation result for test CI
ikbalkaya Jul 18, 2022
fb5e5b8
Remove commented out code
ikbalkaya Jul 18, 2022
d548041
Reenact code
ikbalkaya Jul 18, 2022
370a444
Move latch to invoked test
ikbalkaya Jul 18, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
114 changes: 56 additions & 58 deletions lib/src/main/java/io/ably/lib/transport/ConnectionManager.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -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();
}
});

}

/**
Expand Down
83 changes: 83 additions & 0 deletions lib/src/test/java/io/ably/lib/test/realtime/RealtimeAuthTest.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -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
Expand DownExpand Up@@ -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
});
Expand All@@ -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) {
Comment thread
QuintinWillison marked this conversation as resolved.
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
Expand Down