diff --git a/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java b/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java index 4ae7b353a..616ac98f4 100644 --- a/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java +++ b/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java @@ -82,15 +82,13 @@ public abstract class ChannelBase extends EventEmitter modes; private String lastPayloadMessageId; + private String lastPayloadProtocolMessageChannelSerial; private boolean decodeFailureRecoveryInProgress; private final DecodingContext decodingContext; } diff --git a/lib/src/main/java/io/ably/lib/realtime/ChannelStateListener.java b/lib/src/main/java/io/ably/lib/realtime/ChannelStateListener.java index bf5582d4e..5dbb8ec48 100644 --- a/lib/src/main/java/io/ably/lib/realtime/ChannelStateListener.java +++ b/lib/src/main/java/io/ably/lib/realtime/ChannelStateListener.java @@ -77,7 +77,7 @@ public void onChannelStateChanged(ChannelStateChange stateChange) { for (final ChannelStateListener member : getMembers()) try { member.onChannelStateChanged(stateChange); - } catch(Throwable t) {} + } catch(Throwable ignored) {} } } diff --git a/lib/src/test/java/io/ably/lib/test/common/Helpers.java b/lib/src/test/java/io/ably/lib/test/common/Helpers.java index 8a9a7e38b..32033acc4 100644 --- a/lib/src/test/java/io/ably/lib/test/common/Helpers.java +++ b/lib/src/test/java/io/ably/lib/test/common/Helpers.java @@ -1,5 +1,8 @@ package io.ably.lib.test.common; +import java.lang.reflect.Field; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; import java.net.HttpURLConnection; import java.net.URL; import java.util.ArrayList; @@ -18,6 +21,7 @@ import java.util.Set; import java.util.concurrent.CopyOnWriteArraySet; import java.util.regex.Pattern; +import java.util.stream.Collectors; import com.google.gson.Gson; import com.google.gson.JsonArray; @@ -28,8 +32,10 @@ import io.ably.lib.debug.DebugOptions.RawProtocolListener; import io.ably.lib.http.HttpCore; import io.ably.lib.http.HttpUtils; +import io.ably.lib.realtime.AblyRealtime; import io.ably.lib.realtime.Channel; import io.ably.lib.realtime.Channel.MessageListener; +import io.ably.lib.realtime.ChannelEvent; import io.ably.lib.realtime.ChannelState; import io.ably.lib.realtime.ChannelStateListener; import io.ably.lib.realtime.CompletionListener; @@ -60,6 +66,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; +import static org.junit.Assert.fail; public class Helpers { @@ -263,7 +270,7 @@ public MessageWaiter(Channel channel, String event) { */ public synchronized void waitFor(int count) { while(receivedMessages.size() < count) - try { wait(); } catch(InterruptedException e) {} + try { wait(); } catch(InterruptedException ignored) {} } /** @@ -274,7 +281,7 @@ public synchronized void waitFor(int count, long time) { long targetTime = System.currentTimeMillis() + time; long remaining = time; while(receivedMessages.size() < count && remaining > 0) { - try { wait(remaining); } catch(InterruptedException e) {} + try { wait(remaining); } catch(InterruptedException ignored) {} remaining = targetTime - System.currentTimeMillis(); } } @@ -401,6 +408,48 @@ public PresenceMessage contains(String clientId, String connectionId, PresenceMe } } + public static class MutableConnectionManager { + ConnectionManager connectionManager; + + public MutableConnectionManager(AblyRealtime ablyRealtime) { + this.connectionManager = ablyRealtime.connection.connectionManager; + } + + public void setField(String fieldName, long value) { + try { + Field connectionStateField = ConnectionManager.class.getDeclaredField(fieldName); + connectionStateField.setAccessible(true); + connectionStateField.setLong(connectionManager, value); + } catch (NoSuchFieldException | IllegalAccessException e) { + fail("Failed updating " + fieldName + " with error " + e); + } + } + + public long getField(String fieldName) { + try { + Field connectionStateField = ConnectionManager.class.getDeclaredField(fieldName); + connectionStateField.setAccessible(true); + return connectionStateField.getLong(connectionManager); + } catch (NoSuchFieldException | IllegalAccessException e) { + fail("Failed accessing " + fieldName + " with error " + e); + } + return 0; + } + + /** + * Suppress automatic retries by the connection manager and disconnect + */ + public void disconnectAndSuppressRetries() { + try { + Method method = ConnectionManager.class.getDeclaredMethod("disconnectAndSuppressRetries"); + method.setAccessible(true); + method.invoke(connectionManager); + } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) { + fail("Unexpected exception in suppressing retries"); + } + } + } + /** * A class that listens for state change events on a connection. * @author paddy @@ -428,7 +477,7 @@ public synchronized ErrorInfo waitFor(ConnectionState state) { while (currentState() != state) { try { wait(); - } catch (InterruptedException e) { + } catch (InterruptedException ignored) { } } Log.d(TAG, "waitFor done: state=" + targetStateName + ")"); @@ -444,8 +493,8 @@ public synchronized void waitFor(ConnectionState state, int count) { Log.d(TAG, "waitFor(state=" + state.getConnectionEvent().name() + ", count=" + count + ")"); while(getStateCount(state) < count) - try { wait(); } catch(InterruptedException e) {} - Log.d(TAG, "waitFor done: state=" + latestChange.current.getConnectionEvent().name() + ", count=" + getStateCount(state) + ")"); + try { wait(); } catch(InterruptedException ignored) {} + Log.d(TAG, "waitFor done: state=" + lastStateChange().current.getConnectionEvent().name() + ", count=" + getStateCount(state) + ")"); } /** @@ -462,7 +511,7 @@ public synchronized boolean waitFor(ConnectionState state, int count, long time) long remaining = time; while(getStateCount(state) < count && remaining > 0) { Log.d(TAG, "waitFor(state=" + state.getConnectionEvent().name() + ", waiting for=" + remaining + ")"); - try { wait(remaining); } catch(InterruptedException e) {} + try { wait(remaining); } catch(InterruptedException ignored) {} remaining = targetTime - System.currentTimeMillis(); } int stateCount = getStateCount(state); @@ -503,7 +552,7 @@ public synchronized void reset() { @Override public void onConnectionStateChanged(ConnectionStateListener.ConnectionStateChange state) { synchronized(this) { - latestChange = state; + stateChanges.add(state); reason = state.reason; Counter counter = stateCounts.get(state.current); if(counter == null) stateCounts.put(state.current, (counter = new Counter())); counter.incr(); @@ -524,15 +573,23 @@ private synchronized int getStateCount(ConnectionState state) { } private synchronized ConnectionState currentState() { - return latestChange == null ? connection.state : latestChange.current; + ConnectionStateChange stateChange = lastStateChange(); + return stateChange == null ? connection.state : stateChange.current; + } + + public synchronized ConnectionStateChange lastStateChange() { + if (stateChanges.size() == 0) { + return null; + } + return stateChanges.get(stateChanges.size() -1); } /** * Internal */ - private Connection connection; + private final Connection connection; private ErrorInfo reason; - private ConnectionStateChange latestChange; + private final List stateChanges = new ArrayList<>(); private Map stateCounts; private static final String TAG = ConnectionWaiter.class.getName(); } @@ -557,14 +614,14 @@ public ConnectionManagerWaiter(ConnectionManager connectionManager) { */ public synchronized ErrorInfo waitFor(ConnectionState state) { while(connectionManager.getConnectionState().state != state) - try { wait(INTERVAL_POLLING); } catch(InterruptedException e) {} + try { wait(INTERVAL_POLLING); } catch(InterruptedException ignored) {} return connectionManager.getConnectionState().defaultErrorInfo; } /** * Internal */ - private ConnectionManager connectionManager; + private final ConnectionManager connectionManager; } /** @@ -577,7 +634,6 @@ public static class ChannelWaiter implements ChannelStateListener { /** * Public API - * @param channel */ public ChannelWaiter(Channel channel) { this.channel = channel; @@ -586,28 +642,80 @@ public ChannelWaiter(Channel channel) { /** * Wait for a given state to be reached. - * @param state */ - public synchronized ErrorInfo waitFor(ChannelState state) { - Log.d(TAG, "waitFor(" + state + ")"); - while(channel.state != state) - try { wait(); } catch(InterruptedException e) {} - Log.d(TAG, "waitFor done: " + channel.state + ", " + channel.reason + ")"); + public synchronized ErrorInfo waitFor(ChannelState ... states) { + for (ChannelState state : states) { + Log.d(TAG, "waitFor(" + state + ")"); + while(channel.state != state) + try { wait(); } catch(InterruptedException ignored) {} + Log.d(TAG, "waitFor done: " + channel.state + ", " + channel.reason + ")"); + } return channel.reason; } + /** + * Wait for a given ChannelEvent to be reached. + */ + public synchronized ChannelStateChange waitFor(ChannelEvent channelEvent) { + Log.d(TAG, "waitFor(" + channelEvent + ")"); + ChannelStateChange lastStateChange = getLastStateChange(); + while(lastStateChange.event != channelEvent) + try { wait(); } catch(InterruptedException ignored) {} + Log.d(TAG, "waitFor done: " + channel.state + ", " + channel.reason + ")"); + return lastStateChange; + } + /** * ChannelStateListener interface */ @Override - public void onChannelStateChanged(ChannelStateListener.ChannelStateChange stateChange) { - synchronized(this) { notify(); } + public void onChannelStateChanged(ChannelStateChange stateChange) { + synchronized(this) { + recordedStates.add(stateChange); + notify(); + } } + private final List recordedStates = Collections.synchronizedList(new ArrayList<>()); + + public List getRecordedStates() { + return recordedStates.stream().map(stateChange -> stateChange.current).collect(Collectors.toList()); + } + + public boolean hasFinalStates(ChannelState ... states) { + List rstates = getRecordedStates(); + List vettedList = rstates.subList(rstates.size() - states.length, rstates.size()); + return hasStates(vettedList, states); + } + + public boolean hasStates(ChannelState ... states) { + return hasStates(getRecordedStates(), states); + } + + private static boolean hasStates(List stateList, ChannelState ... states) { + boolean foundStates = false; + int statesCounter = 0; + for (ChannelState recordedState : stateList) { + if (states[statesCounter] != recordedState) { + statesCounter = 0; + } + if (states[statesCounter] == recordedState) { + statesCounter++; + } + if (statesCounter == states.length) { + foundStates = true; + } + } + return foundStates; + } + + public ChannelStateChange getLastStateChange() { + return recordedStates.get(recordedStates.size()-1); + } /** * Internal */ - private Channel channel; + private final Channel channel; } /** diff --git a/lib/src/test/java/io/ably/lib/test/realtime/ConnectionManagerTest.java b/lib/src/test/java/io/ably/lib/test/realtime/ConnectionManagerTest.java index a164120a0..292c96049 100644 --- a/lib/src/test/java/io/ably/lib/test/realtime/ConnectionManagerTest.java +++ b/lib/src/test/java/io/ably/lib/test/realtime/ConnectionManagerTest.java @@ -35,10 +35,7 @@ import java.lang.reflect.Constructor; import java.lang.reflect.Field; import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; import java.util.ArrayList; -import java.util.Arrays; -import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -46,8 +43,10 @@ import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.not; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -390,7 +389,7 @@ public void onConnectionStateChanged(ConnectionStateChange state) { /* wait for cm thread to exit */ try { Thread.sleep(2000L); - } catch(InterruptedException e) {} + } catch(InterruptedException ignored) {} assertEquals("Verify closed state is reached", ConnectionState.closed, ably.connection.state); Thread.State cmThreadState = threadContainer[0].getState(); @@ -464,7 +463,7 @@ public void run() { connectionWaiter.waitFor(ConnectionState.connected); assertEquals("Verify connected state is reached", ConnectionState.connected, ably.connection.state); - assertTrue("Not expecting token auth", ably.auth.getAuthMethod() == AuthMethod.basic); + assertSame("Not expecting token auth", ably.auth.getAuthMethod(), AuthMethod.basic); ably.close(); connectionWaiter.waitFor(ConnectionState.closed); @@ -473,7 +472,7 @@ public void run() { /* wait for cm thread to exit */ try { Thread.sleep(2000L); - } catch(InterruptedException e) {} + } catch(InterruptedException ignored) {} Thread.State cmThreadState = threadContainer[0].getState(); assertEquals("Verify cm thread has exited", cmThreadState, Thread.State.TERMINATED); @@ -512,7 +511,7 @@ public void run() { /* wait for cm thread to exit */ try { Thread.sleep(2000L); - } catch(InterruptedException e) {} + } catch(InterruptedException ignored) {} Thread.State cmThreadState = threadContainer[0].getState(); assertEquals("Verify cm thread has exited", cmThreadState, Thread.State.TERMINATED); @@ -529,29 +528,23 @@ public void run() { @Test public void connection_details_has_ttl() throws AblyException { ClientOptions opts = createOptions(testVars.keys[0].keyStr); + opts.autoConnect = false; try (AblyRealtime ably = new AblyRealtime(opts)) { - final boolean[] callbackWasRun = new boolean[1]; - ably.connection.on(ConnectionEvent.connected, new ConnectionStateListener() { - @Override - public void onConnectionStateChanged(ConnectionStateChange state) { - synchronized(callbackWasRun) { - callbackWasRun[0] = true; - try { - Field field = ably.connection.connectionManager.getClass().getDeclaredField("connectionStateTtl"); - field.setAccessible(true); - assertEquals("Verify connectionStateTtl has the default value", field.get(ably.connection.connectionManager), 120000L); - } catch (NoSuchFieldException|IllegalAccessException e) { - fail("Unexpected exception in checking connectionStateTtl"); - } - callbackWasRun.notify(); - } - } - }); + Helpers.MutableConnectionManager connectionManager = new Helpers.MutableConnectionManager(ably); - synchronized (callbackWasRun) { - try { callbackWasRun.wait(); } catch(InterruptedException ie) {} - assertTrue("Connected callback was not run", callbackWasRun[0]); - } + // connStateTtl set to default value + long connStateTtl = connectionManager.getField("connectionStateTtl"); + assertEquals(Defaults.connectionStateTtl, connStateTtl); + + connectionManager.setField("connectionStateTtl", 8000L); + long oldConnStateTtl = connectionManager.getField("connectionStateTtl"); + assertEquals(8000L, oldConnStateTtl); + + ably.connect(); + new ConnectionWaiter(ably.connection).waitFor(ConnectionState.connected); + long newConnStateTtl = connectionManager.getField("connectionStateTtl"); + // connStateTtl set by server to 120s + assertEquals(120000L, newConnStateTtl); } } @@ -563,23 +556,17 @@ public void connection_is_closed_after_max_idle_interval() throws AblyException ClientOptions opts = createOptions(testVars.keys[0].keyStr); opts.realtimeRequestTimeout = 2000; try(AblyRealtime ably = new AblyRealtime(opts)) { - final long newIdleInterval = 500L; - - // When we connect, we set the max idle interval to be very small - ably.connection.on(ConnectionEvent.connected, state -> { - try { - Field maxIdleField = ably.connection.connectionManager.getClass().getDeclaredField("maxIdleInterval"); - maxIdleField.setAccessible(true); - maxIdleField.setLong(ably.connection.connectionManager, newIdleInterval); - } catch (NoSuchFieldException | IllegalAccessException e) { - fail("Unexpected exception in checking connectionStateTtl"); - } - }); // The original max idle interval we receive from the server is 15s. // We should wait for this, plus a tiny bit extra (as we set the new idle interval to be very low // after connecting) to make sure that the connection is disconnected ConnectionWaiter connectionWaiter = new ConnectionWaiter(ably.connection); + connectionWaiter.waitFor(ConnectionState.connected); + + // When we connect, we set the max idle interval to be very small + Helpers.MutableConnectionManager connectionManager = new Helpers.MutableConnectionManager(ably); + connectionManager.setField("maxIdleInterval", 500L); + assertTrue(connectionWaiter.waitFor(ConnectionState.disconnected, 1, 25000)); } } @@ -593,46 +580,25 @@ public void connection_has_new_id_when_reconnecting_after_statettl_plus_idleinte ClientOptions opts = createOptions(testVars.keys[0].keyStr); opts.realtimeRequestTimeout = 2000L; try(AblyRealtime ably = new AblyRealtime(opts)) { - final long newTtl = 1000L; - final long newIdleInterval = 1000L; /* We want this greater than newTtl + newIdleInterval */ final long waitInDisconnectedState = 3000L; - ably.connection.on(ConnectionEvent.connected, new ConnectionStateListener() { - @Override - public void onConnectionStateChanged(ConnectionStateChange state) { - try { - Field connectionStateField = ably.connection.connectionManager.getClass().getDeclaredField("connectionStateTtl"); - connectionStateField.setAccessible(true); - connectionStateField.setLong(ably.connection.connectionManager, newTtl); - Field maxIdleField = ably.connection.connectionManager.getClass().getDeclaredField("maxIdleInterval"); - maxIdleField.setAccessible(true); - maxIdleField.setLong(ably.connection.connectionManager, newIdleInterval); - } catch (NoSuchFieldException | IllegalAccessException e) { - fail("Unexpected exception in checking connectionStateTtl"); - } - } - }); - ConnectionWaiter connectionWaiter = new ConnectionWaiter(ably.connection); connectionWaiter.waitFor(ConnectionState.connected); final String firstConnectionId = ably.connection.id; - /* suppress automatic retries by the connection manager and disconnect */ - try { - Method method = ably.connection.connectionManager.getClass().getDeclaredMethod("disconnectAndSuppressRetries"); - method.setAccessible(true); - method.invoke(ably.connection.connectionManager); - } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) { - fail("Unexpected exception in suppressing retries"); - } + Helpers.MutableConnectionManager connectionManager = new Helpers.MutableConnectionManager(ably); + connectionManager.setField("connectionStateTtl", 1000L); + connectionManager.setField("maxIdleInterval", 1000L); + + connectionManager.disconnectAndSuppressRetries(); connectionWaiter.waitFor(ConnectionState.disconnected); assertEquals("Disconnected state was not reached", ConnectionState.disconnected, ably.connection.state); /* Wait for the connection to go stale, then reconnect */ try { Thread.sleep(waitInDisconnectedState); - } catch (InterruptedException e) { + } catch (InterruptedException ignored) { } ably.connection.connect(); connectionWaiter.waitFor(ConnectionState.connected); @@ -675,72 +641,42 @@ public void connection_has_same_id_when_reconnecting_before_statettl_plus_idlein public void channels_are_reattached_after_reconnecting_when_statettl_plus_idleinterval_has_passed() throws AblyException { ClientOptions opts = createOptions(testVars.keys[0].keyStr); try(AblyRealtime ably = new AblyRealtime(opts)) { - final long newTtl = 1000L; - final long newIdleInterval = 1000L; /* We want this greater than newTtl + newIdleInterval */ final long waitInDisconnectedState = 3000L; - final List attachedChannelHistory = new ArrayList(); - final List expectedAttachedChannelHistory = Arrays.asList("attaching", "attached", "attaching", "attached"); - final List suspendedChannelHistory = new ArrayList(); - final List expectedSuspendedChannelHistory = Arrays.asList("attaching", "attached"); - ably.connection.on(ConnectionEvent.connected, new ConnectionStateListener() { - @Override - public void onConnectionStateChanged(ConnectionStateChange state) { - try { - Field connectionStateField = ably.connection.connectionManager.getClass().getDeclaredField("connectionStateTtl"); - connectionStateField.setAccessible(true); - connectionStateField.setLong(ably.connection.connectionManager, newTtl); - Field maxIdleField = ably.connection.connectionManager.getClass().getDeclaredField("maxIdleInterval"); - maxIdleField.setAccessible(true); - maxIdleField.setLong(ably.connection.connectionManager, newIdleInterval); - } catch (NoSuchFieldException | IllegalAccessException e) { - fail("Unexpected exception in checking connectionStateTtl"); - } - } - }); + final ChannelState[] expectedAttachedChannelHistory = new ChannelState[]{ + ChannelState.attaching, ChannelState.attached, ChannelState.attaching, ChannelState.attached}; - ConnectionWaiter connectionWaiter = new ConnectionWaiter(ably.connection); + final ChannelState[] expectedSuspendedChannelHistory = new ChannelState[]{ + ChannelState.attaching, ChannelState.attached}; + + ConnectionWaiter connectionWaiter = new ConnectionWaiter(ably.connection); connectionWaiter.waitFor(ConnectionState.connected); final String firstConnectionId = ably.connection.id; + Helpers.MutableConnectionManager connectionManager = new Helpers.MutableConnectionManager(ably); + connectionManager.setField("connectionStateTtl", 1000L); + connectionManager.setField("maxIdleInterval", 1000L); + /* Prepare channels */ final Channel attachedChannel = ably.channels.get("test-reattach-after-ttl" + testParams.name); ChannelWaiter attachedChannelWaiter = new Helpers.ChannelWaiter(attachedChannel); - attachedChannel.on(new ChannelStateListener() { - @Override - public void onChannelStateChanged(ChannelStateChange stateChange) { - attachedChannelHistory.add(stateChange.current.name()); - } - }); + final Channel suspendedChannel = ably.channels.get("test-reattach-suspended-after-ttl" + testParams.name); suspendedChannel.state = ChannelState.suspended; - suspendedChannel.on(new ChannelStateListener() { - @Override - public void onChannelStateChanged(ChannelStateChange stateChange) { - suspendedChannelHistory.add(stateChange.current.name()); - } - }); ChannelWaiter suspendedChannelWaiter = new Helpers.ChannelWaiter(suspendedChannel); /* attach first channel and wait for it to be attached */ attachedChannel.attach(); attachedChannelWaiter.waitFor(ChannelState.attached); - /* suppress automatic retries by the connection manager and disconnect */ - try { - Method method = ably.connection.connectionManager.getClass().getDeclaredMethod("disconnectAndSuppressRetries"); - method.setAccessible(true); - method.invoke(ably.connection.connectionManager); - } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) { - fail("Unexpected exception in suppressing retries"); - } + connectionManager.disconnectAndSuppressRetries(); connectionWaiter.waitFor(ConnectionState.disconnected); assertEquals("Disconnected state was not reached", ConnectionState.disconnected, ably.connection.state); /* Wait for the connection to go stale, then reconnect */ try { Thread.sleep(waitInDisconnectedState); - } catch (InterruptedException e) { + } catch (InterruptedException ignored) { } ably.connection.connect(); connectionWaiter.waitFor(ConnectionState.connected); @@ -754,15 +690,18 @@ public void onChannelStateChanged(ChannelStateChange stateChange) { attachedChannel.once(ChannelEvent.attached, new ChannelStateListener() { @Override public void onChannelStateChanged(ChannelStateChange stateChange) { - assertEquals("Resumed is true and should be false", stateChange.resumed, false); + assertFalse("Resumed is true and should be false", stateChange.resumed); } }); /* Wait for both channels to reattach and verify state histories match the expected ones */ attachedChannelWaiter.waitFor(ChannelState.attached); suspendedChannelWaiter.waitFor(ChannelState.attached); - assertEquals("Attached channel histories do not match", attachedChannelHistory, expectedAttachedChannelHistory); - assertEquals("Suspended channel histories do not match", suspendedChannelHistory, expectedSuspendedChannelHistory); + assertTrue("Attached channel histories do not match", + attachedChannelWaiter.hasFinalStates(expectedAttachedChannelHistory)); + + assertTrue("Suspended channel histories do not match", + suspendedChannelWaiter.hasFinalStates(expectedSuspendedChannelHistory)); } } 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 c8950eed9..d46014e58 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 @@ -4,7 +4,6 @@ import io.ably.lib.realtime.AblyRealtime; import io.ably.lib.realtime.Channel; import io.ably.lib.realtime.ChannelState; -import io.ably.lib.realtime.ConnectionEvent; import io.ably.lib.realtime.ConnectionState; import io.ably.lib.realtime.ConnectionStateListener; import io.ably.lib.rest.AblyRest; @@ -89,7 +88,7 @@ public void auth_client_match_tokendetails_null_clientId() { assertEquals("Verify connected state is reached", ConnectionState.connected, ablyRealtime.connection.state); /* check expected clientId */ - assertEquals("Auth#clientId is expected to be null", null, ablyRealtime.auth.clientId); + assertNull("Auth#clientId is expected to be null", ablyRealtime.auth.clientId); ablyRealtime.close(); } catch (AblyException e) { @@ -147,7 +146,8 @@ public void realtime_connection_with_auth_url_in_query_string_connects() { * Spec: RSA4d, RSA4d1 */ @Test - public void auth_client_fails_authorize_server_forbidden() { + public void auth_client_fails() { + AblyRealtime ablyRealtime = null; try { /* init ably for token */ ClientOptions optsForToken = createOptions(testVars.keys[0].keyStr); @@ -163,25 +163,13 @@ public void auth_client_fails_authorize_server_forbidden() { opts.authUrl = "https://echo.ably.io/respondwith"; opts.authParams = new Param[]{ new Param("status", 403)}; - final AblyRealtime ablyRealtime = new AblyRealtime(opts); + ablyRealtime = new AblyRealtime(opts); ablyRealtime.connection.connect(); /* wait for connected state */ Helpers.ConnectionWaiter connectionWaiter = new Helpers.ConnectionWaiter(ablyRealtime.connection); connectionWaiter.waitFor(ConnectionState.connected); - /* create listener for ConnectionEvent.failed */ - ablyRealtime.connection.once(ConnectionEvent.failed, new ConnectionStateListener() { - @Override - public void onConnectionStateChanged(ConnectionStateChange stateChange) { - /* assert that state changes correctly */ - assertEquals(ConnectionState.connected, stateChange.previous); - assertEquals(80019, stateChange.reason.code); - assertEquals(80019, ablyRealtime.connection.reason.code); - assertEquals(403, ablyRealtime.connection.reason.statusCode); - } - }); - try { opts.tokenDetails = null; /* try to authorize */ @@ -194,11 +182,21 @@ public void onConnectionStateChanged(ConnectionStateChange stateChange) { /* wait for failed state */ connectionWaiter.waitFor(ConnectionState.failed); + ConnectionStateListener.ConnectionStateChange lastStateChange = connectionWaiter.lastStateChange(); + assertEquals(ConnectionState.failed, lastStateChange.current); + assertEquals(80019, lastStateChange.reason.code); + assertEquals(403, lastStateChange.reason.statusCode); + assertEquals("Verify connected state has failed", ConnectionState.failed, ablyRealtime.connection.state); assertEquals("Check correct cause error code", 403, ablyRealtime.connection.reason.statusCode); + assertEquals(80019, ablyRealtime.connection.reason.code); + } catch (AblyException e) { e.printStackTrace(); fail(); + } finally { + assert ablyRealtime != null; + ablyRealtime.close(); } } @@ -350,7 +348,7 @@ public void auth_client_match_token_null_clientId() { assertEquals("Verify connected state is reached", ConnectionState.connected, ablyRealtime.connection.state); /* check expected clientId */ - assertEquals("Auth#clientId is expected to be null", null, ablyRealtime.auth.clientId); + assertNull("Auth#clientId is expected to be null", ablyRealtime.auth.clientId); ablyRealtime.close(); } catch (AblyException e) { @@ -383,7 +381,7 @@ public void auth_clientid_null_before_auth() { AblyRealtime ablyRealtime = new AblyRealtime(opts); /* check expected clientId */ - assertEquals("Auth#clientId is expected to be null", null, ablyRealtime.auth.clientId); + assertNull("Auth#clientId is expected to be null", ablyRealtime.auth.clientId); /* wait for connected state */ ablyRealtime.connection.connect(); @@ -688,7 +686,7 @@ public void auth_client_match_tokendetails_clientId_fail() { ClientOptions opts = createOptions(); opts.clientId = "options clientId"; opts.tokenDetails = tokenDetails; - AblyRealtime ablyRealtime = new AblyRealtime(opts); + new AblyRealtime(opts); } catch (AblyException e) { assertEquals("Verify error code indicates clientId mismatch", e.errorInfo.code, 40101); } @@ -773,7 +771,7 @@ public void auth_clientid_publish_implicit() { /* Get sent message */ Message messagePublished = protocolListener.sentMessages.get(0).messages[0]; - assertEquals("Sent message does not contain clientId", messagePublished.clientId, null); + assertNull("Sent message does not contain clientId", messagePublished.clientId); /* wait until message received on transport */ protocolListener.waitForRecv(1); @@ -819,7 +817,7 @@ public void auth_clientid_publish_implicit() { channel.publish(messageToPublish, pubComplete.add()); pubComplete.waitFor(); assertTrue("Verify publish callback called on completion", pubComplete.pending.isEmpty()); - assertTrue("Verify publish callback returns an error", pubComplete.errors.size() == 1); + assertEquals("Verify publish callback returns an error", 1, pubComplete.errors.size()); assertEquals("Verify publish callback error has expected error code", pubComplete.errors.iterator().next().code, 40012); /* verify no message sent or received on transport */ @@ -838,7 +836,7 @@ public void auth_clientid_publish_implicit() { /* Get sent message */ messagePublished = protocolListener.sentMessages.get(0).messages[0]; - assertEquals("Sent message does not contain clientId", messagePublished.clientId, null); + assertNull("Sent message does not contain clientId", messagePublished.clientId); /* wait until message received on transport */ protocolListener.waitForRecv(1); @@ -927,7 +925,7 @@ public void auth_clientid_publish_explicit_before_identified() { /* Get sent message */ messagePublished = protocolListener.sentMessages.get(0).messages[0]; - assertEquals("Sent message does not contain clientId", messagePublished.clientId, null); + assertNull("Sent message does not contain clientId", messagePublished.clientId); /* wait until message received on transport */ protocolListener.waitForRecv(1); @@ -996,7 +994,7 @@ public Object getTokenRequest(Auth.TokenParams params) { ably.connect(); try { opts.wait(); - } catch(InterruptedException ie) {} + } catch(InterruptedException ignored) {} ably.auth.renew(); } @@ -1066,7 +1064,7 @@ public Object getTokenRequest(Auth.TokenParams params) { ably.connect(); try { opts.wait(); - } catch(InterruptedException ie) {} + } catch(InterruptedException ignored) {} ably.auth.renewAuth((success, tokenDetails1, errorInfo) -> { //Ignore completion handling @@ -1183,7 +1181,7 @@ public void auth_expired_token_expire_before_connect_renew() { assertNotNull("Expected token value", tokenDetails.token); /* allow to expire */ - try { Thread.sleep(200L); } catch(InterruptedException ie) {} + try { Thread.sleep(200L); } catch(InterruptedException ignored) {} /* create Ably realtime instance with token and authCallback */ ClientOptions opts = createOptions(); diff --git a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelHistoryTest.java b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelHistoryTest.java index 17dbe8b24..b11439ff4 100644 --- a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelHistoryTest.java +++ b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelHistoryTest.java @@ -355,7 +355,7 @@ public void channelhistory_wait_b() { /* wait for the history to be persisted */ try { Thread.sleep(16000); - } catch(InterruptedException ie) {} + } catch(InterruptedException ignored) {} /* get the history for this channel */ PaginatedResult messages = channel.history(null); @@ -455,7 +455,7 @@ public void channelhistory_mixed_b() { /* wait for the history to be persisted */ try { Thread.sleep(16000); - } catch(InterruptedException ie) {} + } catch(InterruptedException ignored) {} /* publish to the channel */ msgComplete = new CompletionWaiter(); @@ -517,7 +517,7 @@ public void channelhistory_mixed_f() { /* wait for the history to be persisted */ try { Thread.sleep(16000); - } catch(InterruptedException ie) {} + } catch(InterruptedException ignored) {} /* publish to the channel */ msgComplete = new CompletionWaiter(); @@ -654,7 +654,6 @@ public void channelhistory_limit_b() { } catch (AblyException e) { e.printStackTrace(); fail("channelhistory_limit_b: Unexpected exception"); - return; } finally { if(ably != null) ably.close(); @@ -720,10 +719,7 @@ public void channelhistory_time_f() { for(int i = 20; i < 40; i++) expectedMessageHistory[i - 20] = messageContents.get("history" + i); Assert.assertArrayEquals("Expect messages in forward order", messages.items(), expectedMessageHistory); - } catch (AblyException e) { - e.printStackTrace(); - fail("channelhistory_time_f: Unexpected exception"); - } catch (InterruptedException e) { + } catch (AblyException | InterruptedException e) { e.printStackTrace(); fail("channelhistory_time_f: Unexpected exception"); } finally { @@ -791,10 +787,7 @@ public void channelhistory_time_b() { for(int i = 20; i < 40; i++) expectedMessageHistory[i - 20] = messageContents.get("history" + (59 - i)); Assert.assertArrayEquals("Expect messages in backwards order", messages.items(), expectedMessageHistory); - } catch (AblyException e) { - e.printStackTrace(); - fail("channelhistory_time_b: Unexpected exception"); - } catch (InterruptedException e) { + } catch (AblyException | InterruptedException e) { e.printStackTrace(); fail("channelhistory_time_b: Unexpected exception"); } finally { @@ -1205,7 +1198,7 @@ public void run() { /* wait 2 seconds */ try { Thread.sleep(2000L); - } catch(InterruptedException ie) {} + } catch(InterruptedException ignored) {} /* subscribe; this will trigger the attach */ MessageWaiter messageWaiter = new MessageWaiter(rxChannel); diff --git a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelTest.java b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelTest.java index 181dbc49c..a3a7b30cf 100644 --- a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelTest.java +++ b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelTest.java @@ -25,13 +25,10 @@ import io.ably.lib.types.Message; import io.ably.lib.types.ProtocolMessage; import io.ably.lib.util.Log; - import org.hamcrest.Matchers; import org.junit.Ignore; import org.junit.Test; -import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; import java.util.ArrayList; import java.util.Collections; import java.util.Comparator; @@ -45,6 +42,8 @@ import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -1453,13 +1452,13 @@ public void channel_state_on_connection_suspended() { } /* - * Establish connection, attach channel, simulate sending attached and detached messages + * Establish connection, attach channel, simulate sending attached message * from the server, test correct behaviour * - * Tests RTL12, RTL13a + * Tests RTL12 */ @Test - public void channel_server_initiated_attached_detached() throws AblyException { + public void channel_server_initiated_attached() throws AblyException { AblyRealtime ably = null; long oldRealtimeTimeout = Defaults.realtimeRequestTimeout; final String channelName = "channel_server_initiated_attach_detach"; @@ -1472,6 +1471,7 @@ public void channel_server_initiated_attached_detached() throws AblyException { opts.channelRetryTimeout = 1000; ably = new AblyRealtime(opts); + new ConnectionWaiter(ably.connection).waitFor(ConnectionState.connected); Channel channel = ably.channels.get(channelName); ChannelWaiter channelWaiter = new ChannelWaiter(channel); @@ -1479,24 +1479,55 @@ public void channel_server_initiated_attached_detached() throws AblyException { channel.attach(); channelWaiter.waitFor(ChannelState.attached); - final int[] updateEventsEmitted = new int[]{0}; - final boolean[] resumedFlag = new boolean[]{true}; - channel.on(ChannelEvent.update, new ChannelStateListener() { - @Override - public void onChannelStateChanged(ChannelStateChange stateChange) { - updateEventsEmitted[0]++; - resumedFlag[0] = stateChange.resumed; - } - }); - /* Inject attached message as if received from the server */ ProtocolMessage attachedMessage = new ProtocolMessage() {{ action = Action.attached; channel = channelName; - flags |= Flag.resumed.getMask(); }}; ably.connection.connectionManager.onMessage(null, attachedMessage); + ChannelStateListener.ChannelStateChange channelUpdateEvent = channelWaiter.waitFor(ChannelEvent.update); + assertEquals(ChannelEvent.update, channelUpdateEvent.event); + assertEquals(ChannelState.attached, channelUpdateEvent.previous); + assertEquals(ChannelState.attached, channelUpdateEvent.current); + assertFalse(channelUpdateEvent.resumed); + assertNull(channelUpdateEvent.reason); + + } finally { + if (ably != null) + ably.close(); + Defaults.realtimeRequestTimeout = oldRealtimeTimeout; + } + } + + /* + * Establish connection, attach channel, simulate sending detached messages + * from the server, test correct behaviour + * + * Tests RTL13a + */ + @Test + public void channel_server_initiated_detached() throws AblyException { + AblyRealtime ably = null; + long oldRealtimeTimeout = Defaults.realtimeRequestTimeout; + final String channelName = "channel_server_initiated_attach_detach"; + + try { + ClientOptions opts = createOptions(testVars.keys[0].keyStr); + + /* Make test faster */ + Defaults.realtimeRequestTimeout = 1000; + opts.channelRetryTimeout = 1000; + + ably = new AblyRealtime(opts); + new ConnectionWaiter(ably.connection).waitFor(ConnectionState.connected); + + Channel channel = ably.channels.get(channelName); + ChannelWaiter channelWaiter = new ChannelWaiter(channel); + + channel.attach(); + channelWaiter.waitFor(ChannelState.attached); + /* Inject detached message as if from the server */ ProtocolMessage detachedMessage = new ProtocolMessage() {{ action = Action.detached; @@ -1508,9 +1539,6 @@ public void onChannelStateChanged(ChannelStateChange stateChange) { channelWaiter.waitFor(ChannelState.attaching); channelWaiter.waitFor(ChannelState.attached); - /* Verify received UPDATE message on channel */ - assertEquals("Verify exactly one UPDATE event was emitted on the channel", updateEventsEmitted[0], 1); - assertTrue("Verify resumed flag set in UPDATE event", resumedFlag[0]); } finally { if (ably != null) ably.close(); @@ -1522,10 +1550,83 @@ public void onChannelStateChanged(ChannelStateChange stateChange) { * Establish connection, attach channel, disconnection and failed resume * verify that subsequent attaches are performed, and give rise to update events * - * Tests RTN15c3 + * Tests RTN15c6 + */ + @Test + public void channel_valid_resume_reattach_channels() throws AblyException { + AblyRealtime ably = null; + + try { + ClientOptions opts = createOptions(testVars.keys[0].keyStr); + ably = new AblyRealtime(opts); + ConnectionWaiter connectionWaiter = new ConnectionWaiter(ably.connection); + ably.connect(); + connectionWaiter.waitFor(ConnectionState.connected); + String originalConnectionId = ably.connection.id; + + /* prepare channels */ + Channel attachedChannel = ably.channels.get("attached_channel"); + ChannelWaiter attachedChannelWaiter = new ChannelWaiter(attachedChannel); + attachedChannel.attach(); + attachedChannelWaiter.waitFor(ChannelState.attached); + attachedChannel.publish("chat", "message"); + + Channel suspendedChannel = ably.channels.get("suspended_channel"); + ChannelWaiter suspendedChannelWaiter = new ChannelWaiter(suspendedChannel); + suspendedChannel.attach(); + suspendedChannelWaiter.waitFor(ChannelState.attached); + suspendedChannel.setSuspended(null, true); + suspendedChannelWaiter.waitFor(ChannelState.suspended); + + assertEquals(ably.connection.connectionManager.msgSerial, 1); + + new Helpers.MutableConnectionManager(ably).disconnectAndSuppressRetries(); + connectionWaiter.waitFor(ConnectionState.disconnected); + assertEquals("Verify disconnected state is reached", ConnectionState.disconnected, ably.connection.state); + + /* wait for connection to be reestablished */ + System.out.println("channel_resume_lost_continuity: initiating reconnection (resume)"); + ably.connection.connect(); + + ErrorInfo resumeError = connectionWaiter.waitFor(ConnectionState.connected); + assertNull(resumeError); + assertNull(ably.connection.connectionManager.getStateErrorInfo()); + assertEquals("Same connection is used", originalConnectionId, ably.connection.id); + assertEquals(ably.connection.connectionManager.msgSerial, 1); + + attachedChannelWaiter.waitFor(ChannelState.attaching, ChannelState.attached); + suspendedChannelWaiter.waitFor(ChannelState.attached); + + assertFalse("Verify channel was not suspended", + attachedChannelWaiter.hasStates(ChannelState.suspended)); + assertTrue("Verify channel was attaching and attached", + attachedChannelWaiter.hasFinalStates(ChannelState.attaching, ChannelState.attached)); + + ChannelStateListener.ChannelStateChange stateChange = attachedChannelWaiter.getLastStateChange(); + assertEquals(ChannelState.attached, stateChange.current); + assertEquals(ChannelState.attaching, stateChange.previous); + + assertTrue("Verify channel was attaching", + suspendedChannelWaiter.hasFinalStates(ChannelState.attaching, ChannelState.attached)); + + stateChange = suspendedChannelWaiter.getLastStateChange(); + assertEquals(ChannelState.attached, stateChange.current); + assertEquals(ChannelState.attaching, stateChange.previous); + + } finally { + if (ably != null) + ably.close(); + } + } + + /* + * Establish connection, attach channel, disconnection and failed resume + * verify that subsequent attaches are performed, and give rise to update events + * + * Tests RTN15c7 */ @Test - public void channel_resume_lost_continuity() throws AblyException { + public void channel_invalid_resume_reattach_channels() throws AblyException { AblyRealtime ably = null; final String attachedChannelName = "channel_resume_lost_continuity_attached"; final String suspendedChannelName = "channel_resume_lost_continuity_suspended"; @@ -1533,105 +1634,65 @@ public void channel_resume_lost_continuity() throws AblyException { try { ClientOptions opts = createOptions(testVars.keys[0].keyStr); ably = new AblyRealtime(opts); + ConnectionWaiter connectionWaiter = new ConnectionWaiter(ably.connection); + ably.connect(); + connectionWaiter.waitFor(ConnectionState.connected); + String originalConnectionId = ably.connection.id; /* prepare channels */ Channel attachedChannel = ably.channels.get(attachedChannelName); ChannelWaiter attachedChannelWaiter = new ChannelWaiter(attachedChannel); attachedChannel.attach(); attachedChannelWaiter.waitFor(ChannelState.attached); + attachedChannel.publish("chat", "message"); Channel suspendedChannel = ably.channels.get(suspendedChannelName); - suspendedChannel.state = ChannelState.suspended; ChannelWaiter suspendedChannelWaiter = new ChannelWaiter(suspendedChannel); + suspendedChannel.attach(); + suspendedChannelWaiter.waitFor(ChannelState.attached); + suspendedChannel.setSuspended(null, true); + suspendedChannelWaiter.waitFor(ChannelState.suspended); - final boolean[] suspendedStateReached = new boolean[2]; - final boolean[] attachingStateReached = new boolean[2]; - final boolean[] attachedStateReached = new boolean[2]; - final boolean[] resumedFlag = new boolean[]{true, true}; - attachedChannel.on(new ChannelStateListener() { - @Override - public void onChannelStateChanged(ChannelStateChange stateChange) { - switch(stateChange.current) { - case suspended: - suspendedStateReached[0] = true; - break; - case attaching: - attachingStateReached[0] = true; - break; - case attached: - attachedStateReached[0] = true; - resumedFlag[0] = stateChange.resumed; - break; - default: - break; - } - } - }); - suspendedChannel.on(new ChannelStateListener() { - @Override - public void onChannelStateChanged(ChannelStateChange stateChange) { - switch(stateChange.current) { - case attaching: - attachingStateReached[1] = true; - break; - case attached: - attachedStateReached[1] = true; - resumedFlag[1] = stateChange.resumed; - break; - default: - break; - } - } - }); - - /* disconnect, and sabotage the resume */ - String originalConnectionId = ably.connection.id; - ably.connection.key = "_____!ably___test_fake-key____"; - ably.connection.id = "ably___tes"; - ConnectionWaiter connectionWaiter = new ConnectionWaiter(ably.connection); - - /* suppress automatic retries by the connection manager */ - try { - Method method = ably.connection.connectionManager.getClass().getDeclaredMethod("disconnectAndSuppressRetries"); - method.setAccessible(true); - method.invoke(ably.connection.connectionManager); - } catch (NoSuchMethodException|IllegalAccessException|InvocationTargetException e) { - fail("Unexpected exception in suppressing retries"); - } + assertEquals(ably.connection.connectionManager.msgSerial, 1); + new Helpers.MutableConnectionManager(ably).disconnectAndSuppressRetries(); connectionWaiter.waitFor(ConnectionState.disconnected); assertEquals("Verify disconnected state is reached", ConnectionState.disconnected, ably.connection.state); - /* wait */ - try { Thread.sleep(2000L); } catch(InterruptedException e) {} + /* disconnect, and sabotage the resume */ + ably.connection.key = "_____!ably___test_fake-key____"; /* wait for connection to be reestablished */ System.out.println("channel_resume_lost_continuity: initiating reconnection (resume)"); ably.connection.connect(); - connectionWaiter.waitFor(ConnectionState.connected); - /* verify a new connection was assigned */ + + ErrorInfo resumeError = connectionWaiter.waitFor(ConnectionState.connected); + assertNotNull(resumeError); + assertTrue(resumeError.message.contains("Invalid connection key")); + assertSame(resumeError, ably.connection.connectionManager.getStateErrorInfo()); assertNotEquals("A new connection was created", originalConnectionId, ably.connection.id); + assertEquals(ably.connection.connectionManager.msgSerial, 0); - /* previously suspended channel should transition to attaching, then to attached */ + attachedChannelWaiter.waitFor(ChannelState.attaching, ChannelState.attached); suspendedChannelWaiter.waitFor(ChannelState.attached); - /* previously attached channel should remain attached */ - attachedChannelWaiter.waitFor(ChannelState.attached); + assertFalse("Verify channel was not suspended", + attachedChannelWaiter.hasStates(ChannelState.suspended)); + assertTrue("Verify channel was attaching and attached", + attachedChannelWaiter.hasFinalStates(ChannelState.attaching, ChannelState.attached)); + + ChannelStateListener.ChannelStateChange stateChange = attachedChannelWaiter.getLastStateChange(); + assertEquals(ChannelState.attached, stateChange.current); + assertEquals(ChannelState.attaching, stateChange.previous); + + assertTrue("Verify channel was attaching", + suspendedChannelWaiter.hasFinalStates(ChannelState.attaching, ChannelState.attached)); + + stateChange = suspendedChannelWaiter.getLastStateChange(); + assertEquals(ChannelState.attached, stateChange.current); + assertEquals(ChannelState.attaching, stateChange.previous); - /* - * Verify each channel undergoes relevant events: - * - previously attached channel does attaching, attached, without visiting suspended; - * - previously suspended channel does attaching, attached - */ - assertEquals("Verify channel was not suspended", suspendedStateReached[0], false); - assertEquals("Verify channel was attaching", attachingStateReached[0], true); - assertEquals("Verify channel was attached", attachedStateReached[0], true); - assertFalse("Verify resumed flag set false in ATTACHED event", resumedFlag[0]); - - assertEquals("Verify channel was attaching", attachingStateReached[1], true); - assertEquals("Verify channel was attached", attachedStateReached[1], true); - assertFalse("Verify resumed flag set false in ATTACHED event", resumedFlag[1]); } finally { if (ably != null) ably.close(); diff --git a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeConnectFailTest.java b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeConnectFailTest.java index 55f95d814..02a1d07d5 100644 --- a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeConnectFailTest.java +++ b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeConnectFailTest.java @@ -27,7 +27,6 @@ import org.junit.Test; import org.junit.rules.Timeout; -import java.lang.reflect.Field; import java.time.Duration; import java.time.Instant; import java.util.ArrayList; @@ -342,15 +341,16 @@ public void connect_unknown_recover_fail() { AblyRealtime ably = null; try { ClientOptions opts = createOptions(testVars.keys[0].keyStr); - String recoverConnectionId = "0123456789abcdef-99"; - opts.recover = recoverConnectionId + ":0"; + String recoveryKey = + "{\"connectionKey\":\"0123456789abcdef-99\",\"msgSerial\":5,\"channelSerials\":{\"channel1\":\"98\",\"channel2\":\"32\",\"channel3\":\"09\"}}"; + opts.recover = recoveryKey; ably = new AblyRealtime(opts); ConnectionWaiter connectionWaiter = new ConnectionWaiter(ably.connection); ErrorInfo connectedError = connectionWaiter.waitFor(ConnectionState.connected); assertEquals("Verify connected state is reached", ConnectionState.connected, ably.connection.state); assertNotNull("Verify error is returned", connectedError); - assertEquals("Verify correct error code is given", 80008, connectedError.code); - assertFalse("Verify new connection id is assigned", recoverConnectionId.equals(ably.connection.key)); + assertEquals("Verify correct error code is given", 80018, connectedError.code); + assertFalse("Verify new connection id is assigned", "0123456789abcdef-99".equals(ably.connection.key)); } catch (AblyException e) { e.printStackTrace(); fail("init0: Unexpected exception instantiating library"); @@ -417,13 +417,11 @@ public void onError(ErrorInfo reason) { */ @Test public void connect_reauth_failure_state_flow_test() { - try { - AblyRest ablyRest = null; ClientOptions opts = createOptions(testVars.keys[0].keyStr); - ablyRest = new AblyRest(opts); - final TokenDetails tokenDetails = ablyRest.auth.requestToken(new TokenParams() {{ ttl = 8000L; }}, null); + AblyRest ablyRest = new AblyRest(opts); + final TokenDetails tokenDetails = ablyRest.auth.requestToken(new TokenParams() {{ ttl = 2000L; }}, null); assertNotNull("Expected token value", tokenDetails.token); final ArrayList stateHistory = new ArrayList<>(); @@ -432,31 +430,14 @@ public void connect_reauth_failure_state_flow_test() { optsForRealtime.authCallback = new TokenCallback() { @Override public Object getTokenRequest(TokenParams params) throws AblyException { - // return already expired token + // always return same token return tokenDetails; } }; optsForRealtime.tokenDetails = tokenDetails; final AblyRealtime ablyRealtime = new AblyRealtime(optsForRealtime); - ablyRealtime.connection.on(ConnectionState.connected, new ConnectionStateListener() { - @Override - public void onConnectionStateChanged(ConnectionStateChange state) { - /* To go quicker into a disconnected state we use a - * smaller value for maxIdleInterval - */ - try { - Field field = ablyRealtime.connection.connectionManager.getClass().getDeclaredField("maxIdleInterval"); - field.setAccessible(true); - field.setLong(ablyRealtime.connection.connectionManager, 5000L); - } catch (NoSuchFieldException|IllegalAccessException e) { - fail("Unexpected exception in checking connectionStateTtl"); - } - } - }); - (new ConnectionWaiter(ablyRealtime.connection)).waitFor(ConnectionState.connected); - // TODO: improve by collecting and testing also auth attempts final List correctHistory = Arrays.asList( ConnectionState.disconnected, ConnectionState.connecting, @@ -474,7 +455,7 @@ public void onConnectionStateChanged(ConnectionStateChange state) { if (state.current == ConnectionState.disconnected) { disconnections++; if (disconnections == maxDisconnections) { - assertTrue("Verifying state change history", stateHistory.equals(correctHistory)); + assertEquals(correctHistory, stateHistory); ablyRealtime.close(); } } diff --git a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeDeltaDecoderTest.java b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeDeltaDecoderTest.java index d0fe03531..78d2ea026 100644 --- a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeDeltaDecoderTest.java +++ b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeDeltaDecoderTest.java @@ -62,7 +62,7 @@ public void simple_delta_codec() { Message message = messageWaiter.receivedMessages.get(i); int messageIndex = Integer.parseInt(message.name); assertEquals("Verify message order", i, messageIndex); - assertEquals("Verify message data", true, testData[messageIndex].equals(message.data)); + assertEquals("Verify message data", testData[messageIndex], message.data); } } catch(Exception e) { fail(testName + ": Unexpected exception " + e.getMessage()); diff --git a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeHttpHeaderTest.java b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeHttpHeaderTest.java index 07e945629..ac90c7ac3 100644 --- a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeHttpHeaderTest.java +++ b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeHttpHeaderTest.java @@ -81,7 +81,7 @@ public void realtime_websocket_param_test() { * Defaults.ABLY_VERSION_PARAM, as ultimately the request param has been derived from those values. */ assertEquals("Verify correct version", requestParameters.get("v"), - Collections.singletonList("1.0")); + Collections.singletonList("2")); /* Spec RSC7d3 * This test should not directly validate version against Defaults.ABLY_AGENT_VERSION, nor diff --git a/lib/src/test/java/io/ably/lib/test/realtime/RealtimePresenceTest.java b/lib/src/test/java/io/ably/lib/test/realtime/RealtimePresenceTest.java index 0bd2e7b57..a11a8c339 100644 --- a/lib/src/test/java/io/ably/lib/test/realtime/RealtimePresenceTest.java +++ b/lib/src/test/java/io/ably/lib/test/realtime/RealtimePresenceTest.java @@ -1655,7 +1655,6 @@ public void onPresenceMessage(PresenceMessage message) { * state will have all messages sent once the channel attaches, and all listeners will be called. *

* - * @throws AblyException */ @Test public void realtime_presence_update_multiple_queued_messages() throws AblyException { @@ -1717,7 +1716,7 @@ public void onPresenceMessage(PresenceMessage message) { !receivedMessageStack.get(receivedMessageStack.size()-1).data.equals("Dolor sit!")) receivedMessageStack.wait(); } - } catch(InterruptedException e) {} + } catch(InterruptedException ignored) {} /* Validate that, *- we received specific actions diff --git a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeResumeTest.java b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeResumeTest.java index 356dcf210..068054528 100644 --- a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeResumeTest.java +++ b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeResumeTest.java @@ -4,9 +4,8 @@ import io.ably.lib.realtime.AblyRealtime; import io.ably.lib.realtime.Channel; import io.ably.lib.realtime.ChannelState; -import io.ably.lib.realtime.ConnectionEvent; import io.ably.lib.realtime.ConnectionState; -import io.ably.lib.realtime.ConnectionStateListener; +import io.ably.lib.test.common.Helpers; import io.ably.lib.test.common.Helpers.ChannelWaiter; import io.ably.lib.test.common.Helpers.CompletionSet; import io.ably.lib.test.common.Helpers.ConnectionWaiter; @@ -26,9 +25,6 @@ import org.junit.Test; import org.junit.rules.Timeout; -import java.lang.reflect.Field; -import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -41,8 +37,6 @@ public class RealtimeResumeTest extends ParameterizedTest { - private static final String TAG = RealtimeResumeTest.class.getName(); - @Rule public Timeout testTimeout = Timeout.seconds(60); @@ -58,6 +52,8 @@ public void resume_none() { try { ClientOptions opts = createOptions(testVars.keys[0].keyStr); ably = new AblyRealtime(opts); + ConnectionWaiter connectionWaiter = new ConnectionWaiter(ably.connection); + connectionWaiter.waitFor(ConnectionState.connected); /* create and attach channel */ final Channel channel = ably.channels.get(channelName); @@ -66,27 +62,18 @@ public void resume_none() { (new ChannelWaiter(channel)).waitFor(ChannelState.attached); assertEquals("Verify attached state reached", channel.state, ChannelState.attached); - /* disconnect the connection, without closing, - /* suppressing automatic retries by the connection manager */ - System.out.println("Simulating dropped transport"); - try { - Method method = ably.connection.connectionManager.getClass().getDeclaredMethod("disconnectAndSuppressRetries"); - method.setAccessible(true); - method.invoke(ably.connection.connectionManager); - } catch (NoSuchMethodException|IllegalAccessException| InvocationTargetException e) { - fail("Unexpected exception in suppressing retries"); - } + new Helpers.MutableConnectionManager(ably).disconnectAndSuppressRetries(); + connectionWaiter.waitFor(ConnectionState.disconnected); /* reconnect the rx connection */ ably.connection.connect(); System.out.println("Waiting for reconnection"); - ConnectionWaiter connectionWaiter = new ConnectionWaiter(ably.connection); connectionWaiter.waitFor(ConnectionState.connected); assertEquals("Verify connected state is reached", ConnectionState.connected, ably.connection.state); /* wait */ System.out.println("Got reconnection; waiting 2s"); - try { Thread.sleep(2000L); } catch(InterruptedException e) {} + try { Thread.sleep(2000L); } catch(InterruptedException ignored) {} /* Check the channel is still attached. */ assertEquals("Verify channel still attached", channel.state, ChannelState.attached); @@ -140,12 +127,12 @@ public void resume_simple() { CompletionSet msgComplete1 = new CompletionSet(); for(int i = 0; i < messageCount; i++) { channelTx.publish("test_event", "Test message (resume_simple) " + i, msgComplete1.add()); - try { Thread.sleep(delay); } catch(InterruptedException e){} + try { Thread.sleep(delay); } catch(InterruptedException ignored){} } /* wait for the publish callback to be called */ ErrorInfo[] errors = msgComplete1.waitFor(); - assertTrue("Verify success from all message callbacks", errors.length == 0); + assertEquals("Verify success from all message callbacks", 0, errors.length); /* wait for the subscription callback to be called */ messageWaiter.waitFor(messageCount); @@ -159,7 +146,7 @@ public void resume_simple() { ablyRx.connection.connectionManager.requestState(ConnectionState.disconnected); /* wait */ - try { Thread.sleep(2000L); } catch(InterruptedException e) {} + try { Thread.sleep(2000L); } catch(InterruptedException ignored) {} /* reconnect the rx connection */ ablyRx.connection.connect(); @@ -168,12 +155,12 @@ public void resume_simple() { CompletionSet msgComplete2 = new CompletionSet(); for(int i = 0; i < messageCount; i++) { channelTx.publish("test_event", "Test message (resume_simple) " + i, msgComplete2.add()); - try { Thread.sleep(delay); } catch(InterruptedException e){} + try { Thread.sleep(delay); } catch(InterruptedException ignored){} } /* wait for the publish callback to be called */ errors = msgComplete2.waitFor(); - assertTrue("Verify success from all message callbacks", errors.length == 0); + assertEquals("Verify success from all message callbacks", 0, errors.length); /* wait for the subscription callback to be called */ messageWaiter.waitFor(messageCount); @@ -231,12 +218,12 @@ public void resume_disconnected() { CompletionSet msgComplete1 = new CompletionSet(); for(int i = 0; i < messageCount; i++) { channelTx.publish("test_event", "Test message (resume_disconnected) " + i, msgComplete1.add()); - try { Thread.sleep(delay); } catch(InterruptedException e){} + try { Thread.sleep(delay); } catch(InterruptedException ignored){} } /* wait for the publish callback to be called */ ErrorInfo[] errors = msgComplete1.waitFor(); - assertTrue("Verify success from all message callbacks", errors.length == 0); + assertEquals("Verify success from all message callbacks", 0, errors.length); /* wait for the subscription callback to be called */ messageWaiter.waitFor(messageCount); @@ -250,18 +237,18 @@ public void resume_disconnected() { ablyRx.connection.connectionManager.requestState(ConnectionState.disconnected); /* wait */ - try { Thread.sleep(2000L); } catch(InterruptedException e) {} + try { Thread.sleep(2000L); } catch(InterruptedException ignored) {} /* publish next messages to the channel */ CompletionSet msgComplete2 = new CompletionSet(); for(int i = 0; i < messageCount; i++) { channelTx.publish("test_event", "Test message (resume_disconnected) " + i, msgComplete2.add()); - try { Thread.sleep(delay); } catch(InterruptedException e){} + try { Thread.sleep(delay); } catch(InterruptedException ignored){} } /* wait for the publish callback to be called */ errors = msgComplete2.waitFor(); - assertTrue("Verify success from all message callbacks", errors.length == 0); + assertEquals("Verify success from all message callbacks", 0, errors.length); /* reconnect the rx connection, and expect the messages to be delivered */ ablyRx.connection.connect(); @@ -327,12 +314,12 @@ public void resume_multiple_channel() { for(int i = 0; i < messageCount; i++) { channelTx1.publish("test_event1", "Test message (resume_multiple_channel) " + i, msgComplete1.add()); channelTx2.publish("test_event2", "Test message (resume_multiple_channel) " + i, msgComplete1.add()); - try { Thread.sleep(delay); } catch(InterruptedException e){} + try { Thread.sleep(delay); } catch(InterruptedException ignored){} } /* wait for the publish callback to be called */ ErrorInfo[] errors = msgComplete1.waitFor(); - assertTrue("Verify success from all message callbacks", errors.length == 0); + assertEquals("Verify success from all message callbacks", 0, errors.length); /* wait for the subscription callback to be called */ messageWaiter1.waitFor(messageCount); @@ -349,19 +336,19 @@ public void resume_multiple_channel() { ablyRx.connection.connectionManager.requestState(ConnectionState.disconnected); /* wait */ - try { Thread.sleep(2000L); } catch(InterruptedException e) {} + try { Thread.sleep(2000L); } catch(InterruptedException ignored) {} /* publish next messages to the channel */ CompletionSet msgComplete2 = new CompletionSet(); for(int i = 0; i < messageCount; i++) { channelTx1.publish("test_event1", "Test message (resume_multiple_channel) " + i, msgComplete2.add()); channelTx2.publish("test_event2", "Test message (resume_multiple_channel) " + i, msgComplete2.add()); - try { Thread.sleep(delay); } catch(InterruptedException e){} + try { Thread.sleep(delay); } catch(InterruptedException ignored){} } /* wait for the publish callback to be called */ errors = msgComplete2.waitFor(); - assertTrue("Verify success from all message callbacks", errors.length == 0); + assertEquals("Verify success from all message callbacks", 0, errors.length); /* reconnect the rx connection, and expect the messages to be delivered */ ablyRx.connection.connect(); @@ -420,12 +407,12 @@ public void resume_multiple_interval() { CompletionSet msgComplete1 = new CompletionSet(); for(int i = 0; i < messageCount; i++) { channelTx.publish("test_event", "Test message (resume_multiple_interval) " + i, msgComplete1.add()); - try { Thread.sleep(delay); } catch(InterruptedException e){} + try { Thread.sleep(delay); } catch(InterruptedException ignored){} } /* wait for the publish callback to be called */ ErrorInfo[] errors = msgComplete1.waitFor(); - assertTrue("Verify success from all message callbacks", errors.length == 0); + assertEquals("Verify success from all message callbacks", 0, errors.length); /* wait for the subscription callback to be called */ messageWaiter.waitFor(messageCount); @@ -439,18 +426,18 @@ public void resume_multiple_interval() { ablyRx.connection.connectionManager.requestState(ConnectionState.disconnected); /* wait */ - try { Thread.sleep(20000L); } catch(InterruptedException e) {} + try { Thread.sleep(20000L); } catch(InterruptedException ignored) {} /* publish next messages to the channel */ CompletionSet msgComplete2 = new CompletionSet(); for(int i = 0; i < messageCount; i++) { channelTx.publish("test_event", "Test message (resume_multiple_interval) " + i, msgComplete2.add()); - try { Thread.sleep(delay); } catch(InterruptedException e){} + try { Thread.sleep(delay); } catch(InterruptedException ignored){} } /* wait for the publish callback to be called */ errors = msgComplete2.waitFor(); - assertTrue("Verify success from all message callbacks", errors.length == 0); + assertEquals("Verify success from all message callbacks", 0, errors.length); /* reconnect the rx connection, and expect the messages to be delivered */ ablyRx.connection.connect(); @@ -509,12 +496,12 @@ public void resume_verify_publish() { CompletionSet msgComplete1 = new CompletionSet(); for(int i = 0; i < messageCount; i++) { channelTx.publish("test_event", "Test message (resume_simple) " + i, msgComplete1.add()); - try { Thread.sleep(delay); } catch(InterruptedException e){} + try { Thread.sleep(delay); } catch(InterruptedException ignored){} } /* wait for the publish callback to be called */ ErrorInfo[] errors = msgComplete1.waitFor(); - assertTrue("Verify success from all message callbacks", errors.length == 0); + assertEquals("Verify success from all message callbacks", 0, errors.length); /* wait for the subscription callback to be called */ messageWaiter.waitFor(messageCount); @@ -526,17 +513,12 @@ public void resume_verify_publish() { * of the library, to simulate a dropped transport without * causing the connection itself to be disposed */ System.out.println("*** about to disconnect tx connection"); - /* suppress automatic retries by the connection manager */ - try { - Method method = ablyTx.connection.connectionManager.getClass().getDeclaredMethod("disconnectAndSuppressRetries"); - method.setAccessible(true); - method.invoke(ablyTx.connection.connectionManager); - } catch (NoSuchMethodException|IllegalAccessException|InvocationTargetException e) { - fail("Unexpected exception in suppressing retries"); - } + + new Helpers.MutableConnectionManager(ablyTx).disconnectAndSuppressRetries(); + (new ConnectionWaiter(ablyTx.connection)).waitFor(ConnectionState.disconnected); /* wait */ - try { Thread.sleep(2000L); } catch(InterruptedException e) {} + try { Thread.sleep(2000L); } catch(InterruptedException ignored) {} /* reconnect the tx connection */ System.out.println("*** about to reconnect tx connection"); @@ -547,7 +529,7 @@ public void resume_verify_publish() { CompletionSet msgComplete2 = new CompletionSet(); for(int i = 0; i < messageCount; i++) { channelTx.publish("test_event", "Test message (resume_simple) " + i, msgComplete2.add()); - try { Thread.sleep(delay); } catch(InterruptedException e){} + try { Thread.sleep(delay); } catch(InterruptedException ignored){} } /* wait for the publish callback to be called. This never finishes if @@ -556,7 +538,7 @@ public void resume_verify_publish() { System.out.println("*** published. About to wait for callbacks"); errors = msgComplete2.waitFor(); System.out.println("*** done"); - assertTrue("Verify success from all message callbacks", errors.length == 0); + assertEquals("Verify success from all message callbacks", 0, errors.length); /* wait for the subscription callback to be called */ messageWaiter.waitFor(messageCount); @@ -599,19 +581,16 @@ public void resume_publish_queue() { final Channel senderChannel = sender.channels.get(channelName); senderChannel.attach(); (new ChannelWaiter(senderChannel)).waitFor(ChannelState.attached); - assertEquals( - "The sender's channel should be attached", - senderChannel.state, ChannelState.attached - ); + assertEquals("The sender's channel should be attached", + senderChannel.state, ChannelState.attached); /* create and attach channel to recv on */ final Channel receiverChannel = receiver.channels.get(channelName); receiverChannel.attach(); (new ChannelWaiter(receiverChannel)).waitFor(ChannelState.attached); - assertEquals( - "The receiver's channel should be attached", - receiverChannel.state, ChannelState.attached - ); + assertEquals("The receiver's channel should be attached", + receiverChannel.state, ChannelState.attached); + /* subscribe */ MessageWaiter messageWaiter = new MessageWaiter(receiverChannel); @@ -619,21 +598,17 @@ public void resume_publish_queue() { CompletionSet msgComplete1 = new CompletionSet(); for(int i = 0; i < messageCount; i++) { senderChannel.publish("test_event", "Test message (resume_publish_queue) " + i, msgComplete1.add()); - try { Thread.sleep(delay); } catch(InterruptedException e){} + try { Thread.sleep(delay); } catch(InterruptedException ignored){} } /* wait for the publish callback to be called */ ErrorInfo[] errors = msgComplete1.waitFor(); - assertTrue( - "First round of messages has errors", errors.length == 0 - ); + assertEquals("First round of messages has errors", 0, errors.length); /* wait for the subscription callback to be called */ messageWaiter.waitFor(messageCount); - assertEquals( - "Did not receive the entire first round of messages", - messageWaiter.receivedMessages.size(), messageCount - ); + assertEquals("Did not receive the entire first round of messages", + messageWaiter.receivedMessages.size(), messageCount); messageWaiter.reset(); /* disconnect the sender, without closing; @@ -643,7 +618,7 @@ public void resume_publish_queue() { sender.connection.connectionManager.requestState(ConnectionState.disconnected); /* wait */ - try { Thread.sleep(2000L); } catch(InterruptedException e) {} + try { Thread.sleep(2000L); } catch(InterruptedException ignored) {} /* * publish further messages to the channel, which should be queued @@ -652,20 +627,16 @@ public void resume_publish_queue() { CompletionSet msgComplete2 = new CompletionSet(); for(int i = 0; i < messageCount; i++) { senderChannel.publish("queued_message_" + i, "Test queued message (resume_publish_queue) " + i, msgComplete2.add()); - try { Thread.sleep(delay); } catch(InterruptedException e){} + try { Thread.sleep(delay); } catch(InterruptedException ignored){} } /* reconnect the sender */ sender.connection.connect(); (new ConnectionWaiter(sender.connection)).waitFor(ConnectionState.connected); - /* wait for the publish callback to be called.*/ errors = msgComplete2.waitFor(); - assertTrue( - "Second round of messages (queued) has errors", - errors.length == 0 - ); + assertEquals("Second round of messages (queued) has errors", 0, errors.length); /* wait for the subscription callback to be called */ messageWaiter.waitFor(messageCount); @@ -676,10 +647,8 @@ public void resume_publish_queue() { received.size(), messageCount ); for(int i=0; i message.action == ProtocolMessage.Action.ack || message.action == ProtocolMessage.Action.nack); @@ -760,15 +725,7 @@ public void resume_publish_resend_pending_messages_when_resume_is_successful() { final String connectionId = sender.connection.id; - /* suppress automatic retries by the connection manager and disconnect */ - try { - Method method = sender.connection.connectionManager.getClass().getDeclaredMethod( - "disconnectAndSuppressRetries"); - method.setAccessible(true); - method.invoke(sender.connection.connectionManager); - } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) { - fail("Unexpected exception in suppressing retries"); - } + new Helpers.MutableConnectionManager(sender).disconnectAndSuppressRetries(); (new ConnectionWaiter(sender.connection)).waitFor(ConnectionState.disconnected); sender.connection.connectionManager.requestState(ConnectionState.disconnected); @@ -797,10 +754,7 @@ public void resume_publish_resend_pending_messages_when_resume_is_successful() { (new ChannelWaiter(senderChannel)).waitFor(ChannelState.attached); /* wait for the publish callback to be called.*/ ErrorInfo[] senderErrors = senderCompletion.waitFor(); - assertTrue( - "Second round of send has errors", - senderErrors.length == 0 - ); + assertEquals("Second round of send has errors", 0, senderErrors.length); assertEquals("Second round of messages has incorrect size", 6, transport.getPublishedMessages().size()); //make sure they were sent with correct serials @@ -837,28 +791,14 @@ public void resume_publish_resend_pending_messages_when_resume_failed() throws A try(AblyRealtime ably = new AblyRealtime(options)) { final long newTtl = 1000L; final long newIdleInterval = 1000L; - /* We want this greater than newTtl + newIdleInterval */ - final long waitInDisconnectedState = 3000L; - - ably.connection.on(ConnectionEvent.connected, new ConnectionStateListener() { - @Override - public void onConnectionStateChanged(ConnectionStateChange state) { - try { - Field connectionStateField = ably.connection.connectionManager.getClass().getDeclaredField("connectionStateTtl"); - connectionStateField.setAccessible(true); - connectionStateField.setLong(ably.connection.connectionManager, newTtl); - Field maxIdleField = ably.connection.connectionManager.getClass().getDeclaredField("maxIdleInterval"); - maxIdleField.setAccessible(true); - maxIdleField.setLong(ably.connection.connectionManager, newIdleInterval); - } catch (NoSuchFieldException | IllegalAccessException e) { - fail("Unexpected exception in checking connectionStateTtl"); - } - } - }); ConnectionWaiter connectionWaiter = new ConnectionWaiter(ably.connection); connectionWaiter.waitFor(ConnectionState.connected); + Helpers.MutableConnectionManager connectionManager = new Helpers.MutableConnectionManager(ably); + connectionManager.setField("connectionStateTtl", newTtl); + connectionManager.setField("maxIdleInterval", newIdleInterval); + final Channel senderChannel = ably.channels.get(channelName); senderChannel.attach(); (new ChannelWaiter(senderChannel)).waitFor(ChannelState.attached); @@ -896,14 +836,7 @@ public void onConnectionStateChanged(ConnectionStateChange state) { final String firstConnectionId = ably.connection.id; - /* suppress automatic retries by the connection manager and disconnect */ - try { - Method method = ably.connection.connectionManager.getClass().getDeclaredMethod("disconnectAndSuppressRetries"); - method.setAccessible(true); - method.invoke(ably.connection.connectionManager); - } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) { - fail("Unexpected exception in suppressing retries"); - } + connectionManager.disconnectAndSuppressRetries(); connectionWaiter.waitFor(ConnectionState.disconnected); assertEquals("Disconnected state was not reached", ConnectionState.disconnected, ably.connection.state); @@ -914,10 +847,13 @@ public void onConnectionStateChanged(ConnectionStateChange state) { } //now let's unblock the ack nacks and reconnect mockWebsocketFactory.blockReceiveProcessing(message -> false); + + /* We want this greater than newTtl + newIdleInterval */ + final long waitInDisconnectedState = 3000L; /* Wait for the connection to go stale, then reconnect */ try { Thread.sleep(waitInDisconnectedState); - } catch (InterruptedException e) { + } catch (InterruptedException ignored) { } ably.connection.connect(); connectionWaiter.waitFor(ConnectionState.connected); @@ -933,10 +869,7 @@ public void onConnectionStateChanged(ConnectionStateChange state) { assertEquals("Connection has the same id", ChannelState.attached, senderChannel.state); ErrorInfo[] resendErrors = senderCompletion.waitFor(); - assertTrue( - "Second round of messages (queued) has errors", - resendErrors.length == 0 - ); + assertEquals("Second round of messages (queued) has errors", 0, resendErrors.length); assertEquals("Second round of messages has incorrect size", 6, transport.getPublishedMessages().size()); //make sure they were sent with reset serials @@ -947,7 +880,6 @@ public void onConnectionStateChanged(ConnectionStateChange state) { } } - /** * In case of resume failure verify that presence messages are resent * */ @@ -956,44 +888,29 @@ public void resume_publish_reenter_when_resume_failed() throws AblyException { final String channelName = "sender_channel"; final MockWebsocketFactory mockWebsocketFactory = new MockWebsocketFactory(); final DebugOptions options = createOptions(testVars.keys[0].keyStr); - final String[] clients = new String[]{"client1","client2","client3", - "client4","client5","client6","client7","client8","client9"}; + final String[] clients = new String[]{"client1", "client2", "client3", "client4", "client5", + "client6", "client7", "client8", "client9"}; + options.logLevel = Log.VERBOSE; options.realtimeRequestTimeout = 2000L; - /* We want this greater than newTtl + newIdleInterval */ - final long waitInDisconnectedState = 5000L; options.transportFactory = mockWebsocketFactory; try(AblyRealtime ably = new AblyRealtime(options)) { - final long newTtl = 1000L; - final long newIdleInterval = 1000L; - /* We want this greater than newTtl + newIdleInterval */ - ably.connection.on(ConnectionEvent.connected, new ConnectionStateListener() { - @Override - public void onConnectionStateChanged(ConnectionStateChange state) { - try { - Field connectionStateField = ably.connection.connectionManager.getClass().getDeclaredField("connectionStateTtl"); - connectionStateField.setAccessible(true); - connectionStateField.setLong(ably.connection.connectionManager, newTtl); - Field maxIdleField = ably.connection.connectionManager.getClass().getDeclaredField("maxIdleInterval"); - maxIdleField.setAccessible(true); - maxIdleField.setLong(ably.connection.connectionManager, newIdleInterval); - } catch (NoSuchFieldException | IllegalAccessException e) { - fail("Unexpected exception in checking connectionStateTtl"); - } - } - }); ConnectionWaiter connectionWaiter = new ConnectionWaiter(ably.connection); connectionWaiter.waitFor(ConnectionState.connected); + final long newTtl = 1000L; + final long newIdleInterval = 1000L; + + Helpers.MutableConnectionManager connectionManager = new Helpers.MutableConnectionManager(ably); + connectionManager.setField("connectionStateTtl", newTtl); + connectionManager.setField("maxIdleInterval", newIdleInterval); + final Channel senderChannel = ably.channels.get(channelName); senderChannel.attach(); (new ChannelWaiter(senderChannel)).waitFor(ChannelState.attached); - assertEquals( - "The sender's channel should be attached", - senderChannel.state, ChannelState.attached - ); + assertEquals("The sender's channel should be attached", senderChannel.state, ChannelState.attached); MockWebsocketFactory.MockWebsocketTransport transport = mockWebsocketFactory.getCreatedTransport(); CompletionSet presenceCompletion = new CompletionSet(); @@ -1018,32 +935,27 @@ public void onConnectionStateChanged(ConnectionStateChange state) { message.action == ProtocolMessage.Action.nack); //enter next 3 clients - for (int i = 0; i < 3; i++) { - senderChannel.presence.enterClient(clients[i+3],null,presenceCompletion.add()); + for (int i = 3; i < 6; i++) { + senderChannel.presence.enterClient(clients[i],null,presenceCompletion.add()); } final String firstConnectionId = ably.connection.id; - /* suppress automatic retries by the connection manager and disconnect */ - try { - Method method = ably.connection.connectionManager.getClass().getDeclaredMethod("disconnectAndSuppressRetries"); - method.setAccessible(true); - method.invoke(ably.connection.connectionManager); - } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) { - fail("Unexpected exception in suppressing retries"); - } + connectionManager.disconnectAndSuppressRetries(); connectionWaiter.waitFor(ConnectionState.disconnected); assertEquals("Disconnected state was not reached", ConnectionState.disconnected, ably.connection.state); //enter last 3 clients while disconnected - for (int i = 0; i < 3; i++) { - senderChannel.presence.enterClient(clients[i+6],null,presenceCompletion.add()); + for (int i = 6; i < 9; i++) { + senderChannel.presence.enterClient(clients[i],null,presenceCompletion.add()); } + /* We want this greater than newTtl + newIdleInterval */ + final long waitInDisconnectedState = 5000L; /* Wait for the connection to go stale, then reconnect */ try { Thread.sleep(waitInDisconnectedState); - } catch (InterruptedException e) { + } catch (InterruptedException ignored) { } //now let's unblock the ack nacks and reconnect @@ -1070,24 +982,21 @@ public void onConnectionStateChanged(ConnectionStateChange state) { for (ErrorInfo resendError : resendErrors) { System.out.println("presence_resume_test: error "+resendError.message); } - assertTrue( - "Second round of messages (queued) has errors", - resendErrors.length == 0 - ); + assertEquals("Second round of messages (queued) has errors", 0, resendErrors.length); - for (PresenceMessage presenceMessage: - transport.getSentPresenceMessages()) { + for (PresenceMessage presenceMessage: transport.getSentPresenceMessages()) { System.out.println("presence_resume_test: sent message with client: "+presenceMessage.clientId +" " + " action:"+presenceMessage.action); } - assertEquals("Second round of messages has incorrect size", 9, transport.getSentPresenceMessages().size()); + assertEquals("Second round of messages has incorrect size", 6, transport.getSentPresenceMessages().size()); //make sure they were sent with correct client ids final Map sentPresenceMap = new HashMap<>(); for (PresenceMessage presenceMessage: transport.getSentPresenceMessages()){ sentPresenceMap.put(presenceMessage.clientId, presenceMessage); } - for (String client : clients) { - assertTrue("Client id isn't there:"+client, sentPresenceMap.containsKey(client)); + + for (int i = 3; i < 9; i++) { + assertTrue("Client id isn't there:" + clients[i], sentPresenceMap.containsKey(clients[i])); } } } @@ -1103,7 +1012,6 @@ public void resume_rewind_1 () String testName = "resume_rewind_1"; try { - ClientOptions common_opts = createOptions(testVars.keys[0].keyStr); sender = new AblyRealtime(common_opts); receiver1 = new AblyRealtime(common_opts); @@ -1125,21 +1033,22 @@ public void onRawMessageRecv(ProtocolMessage message) {} }; receiver2 = new AblyRealtime(receiver2_opts); - Channel recever1_channel = receiver1.channels.get("[?rewind=1]" + testName); - Channel recever2_channel = receiver2.channels.get("[?rewind=1]" + testName); - Channel sender_channel = sender.channels.get(testName); + Channel receiver1_channel = receiver1.channels.get("[?rewind=1]" + testName); + Channel receiver2_channel = receiver2.channels.get("[?rewind=1]" + testName); + + Channel sender_channel = sender.channels.get(testName); sender_channel.attach(); (new ChannelWaiter(sender_channel)).waitFor(ChannelState.attached); sender_channel.publish("0", testMessage); /* subscribe 1*/ - MessageWaiter messageWaiter_1 = new MessageWaiter(recever1_channel); + MessageWaiter messageWaiter_1 = new MessageWaiter(receiver1_channel); messageWaiter_1.waitFor(1); assertEquals("Verify rewound message", testMessage, messageWaiter_1.receivedMessages.get(0).data); /* subscribe 2*/ - MessageWaiter messageWaiter_2 = new MessageWaiter(recever2_channel); + MessageWaiter messageWaiter_2 = new MessageWaiter(receiver2_channel); messageWaiter_2.waitFor(1, 7000); assertEquals("Verify no message received on attach_rewind", 0, messageWaiter_2.receivedMessages.size());