[AIT-767] uts: live objects ground work

.github/workflows/check.yml

@@ -23,4 +23,4 @@ jobs:
  distribution: 'temurin'
  - name: Set up Gradle
  uses: gradle/actions/setup-gradle@d9c87d481d55275bb5441eef3fe0e46805f9ef70 # v3
- - run: ./gradlew checkWithCodenarc checkstyleMain checkstyleTest runUnitTests runLiveObjectUnitTests
+ - run: ./gradlew checkWithCodenarc checkstyleMain checkstyleTest runUnitTests runLiveObjectUnitTests :uts:test

lib/src/main/java/io/ably/lib/debug/DebugOptions.java

@@ -4,11 +4,14 @@
 import java.util.Map;
 
 import io.ably.lib.http.HttpCore;
+import io.ably.lib.network.HttpEngine;
 import io.ably.lib.network.HttpRequest;
+import io.ably.lib.network.WebSocketEngineFactory;
 import io.ably.lib.transport.ITransport;
 import io.ably.lib.types.AblyException;
 import io.ably.lib.types.ClientOptions;
 import io.ably.lib.types.ProtocolMessage;
+import io.ably.lib.util.Clock;
 
 public class DebugOptions extends ClientOptions {
  public interface RawProtocolListener {
@@ -31,12 +34,18 @@ public interface RawHttpListener {
  public RawProtocolListener protocolListener;
  public RawHttpListener httpListener;
  public ITransport.Factory transportFactory;
+ public HttpEngine httpEngine;
+ public WebSocketEngineFactory webSocketEngineFactory;
+ public Clock clock;
 
  public DebugOptions copy() {
  DebugOptions copied = new DebugOptions();
  copied.protocolListener = protocolListener;
  copied.httpListener = httpListener;
  copied.transportFactory = transportFactory;
+ copied.httpEngine = httpEngine;
+ copied.webSocketEngineFactory = webSocketEngineFactory;
+ copied.clock = clock;
  copied.clientId = clientId;
  copied.logLevel = logLevel;
  copied.logHandler = logHandler;

lib/src/main/java/io/ably/lib/http/HttpCore.java

@@ -108,9 +108,13 @@ public HttpCore(ClientOptions options, Auth auth, PlatformAgentProvider platform
  proxyAuth = new HttpAuth(proxyUser, proxyPassword, proxyOptions.prefAuthType);
  }
  }
- HttpEngineFactory engineFactory = HttpEngineFactory.getFirstAvailable();
- Log.v(TAG, String.format("Using %s HTTP Engine", engineFactory.getEngineType().name()));
- this.engine = engineFactory.create(new HttpEngineConfig(ClientOptionsUtils.convertToProxyConfig(options)));
+ if (options instanceof DebugOptions && ((DebugOptions) options).httpEngine != null) {
+ this.engine = ((DebugOptions) options).httpEngine;
+ } else {
+ HttpEngineFactory engineFactory = HttpEngineFactory.getFirstAvailable();
+ Log.v(TAG, String.format("Using %s HTTP Engine", engineFactory.getEngineType().name()));
+ this.engine = engineFactory.create(new HttpEngineConfig(ClientOptionsUtils.convertToProxyConfig(options)));
+ }
  }
 
  private HttpCore(HttpCore underlyingHttpCore, Map<String, String> dynamicAgents) {

lib/src/main/java/io/ably/lib/http/HttpScheduler.java

@@ -13,7 +13,9 @@
 import io.ably.lib.types.Callback;
 import io.ably.lib.types.ErrorInfo;
 import io.ably.lib.types.Param;
+import io.ably.lib.util.Clock;
 import io.ably.lib.util.Log;
+import io.ably.lib.util.SystemClock;
 
 /**
  * HttpScheduler schedules HttpCore operations to an Executor, exposing a generic async API.
@@ -286,12 +288,12 @@ public T get() throws InterruptedException, ExecutionException {
  }
  @Override
  public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
- long remaining = unit.toMillis(timeout), deadline = System.currentTimeMillis() + remaining;
+ long remaining = unit.toMillis(timeout), deadline = clock.currentTimeMillis() + remaining;
  synchronized(this) {
  while(remaining > 0) {
-wait(remaining);
+clock.waitOn(this, remaining);
  if(isDone) { break; }
- remaining = deadline - System.currentTimeMillis();
+ remaining = deadline - clock.currentTimeMillis();
  }
  if(!isDone) {
  throw new TimeoutException();
@@ -360,6 +362,7 @@ protected synchronized boolean disposeConnection() {
  protected HttpScheduler(HttpCore httpCore, CloseableExecutor executor) {
  this.httpCore = httpCore;
  this.executor = executor;
+ this.clock = SystemClock.clockFrom(httpCore.options);
  }
 
  @Override
@@ -446,6 +449,7 @@ public <T> Future<T> ablyHttpExecuteWithRetry(
 
  protected final CloseableExecutor executor;
  private final HttpCore httpCore;
+ private final Clock clock;
 
  protected static final String TAG = HttpScheduler.class.getName();
 

lib/src/main/java/io/ably/lib/realtime/ChannelBase.java

@@ -6,7 +6,6 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
-import java.util.Timer;
 import java.util.TimerTask;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -46,12 +45,15 @@
 import io.ably.lib.types.PublishResult;
 import io.ably.lib.types.Summary;
 import io.ably.lib.types.UpdateDeleteResult;
+import io.ably.lib.util.Clock;
 import io.ably.lib.util.CollectionUtils;
 import io.ably.lib.util.EventEmitter;
 import io.ably.lib.util.Listeners;
 import io.ably.lib.util.Log;
+import io.ably.lib.util.AblyTimer;
 import io.ably.lib.util.ReconnectionStrategy;
 import io.ably.lib.util.StringUtils;
+import io.ably.lib.util.SystemClock;
 import org.jetbrains.annotations.Blocking;
 import org.jetbrains.annotations.NonBlocking;
 import org.jetbrains.annotations.Nullable;
@@ -508,21 +510,20 @@ private void setFailed(ErrorInfo reason) {
  }
 
  /* Timer for attach operation */
- private Timer attachTimer;
+ private AblyTimer attachTimer;
 
  /* Timer for reattaching if attach failed */
- private Timer reattachTimer;
+ private AblyTimer reattachTimer;
 
  /**
  * Cancel attach/reattach timers
  */
  synchronized private void clearAttachTimers() {
-Timer[] timers = new Timer[]{attachTimer, reattachTimer};
+AblyTimer[] timers = new AblyTimer[]{attachTimer, reattachTimer};
  attachTimer = reattachTimer = null;
- for (Timer t: timers) {
+ for (AblyTimer t: timers) {
  if (t != null) {
  t.cancel();
- t.purge();
  }
  }
  }
@@ -537,9 +538,9 @@ private void attachWithTimeout(final CompletionListener listener) throws AblyExc
  */
  synchronized private void attachWithTimeout(final boolean forceReattach, final CompletionListener listener, ErrorInfo reattachmentReason) {
  checkChannelIsNotReleased();
-Timer currentAttachTimer;
+AblyTimer currentAttachTimer;
  try {
- currentAttachTimer = new Timer();
+ currentAttachTimer = clock.newTimer("attach-timer");
  } catch(Throwable t) {
  /* an exception instancing the timer can arise because the runtime is exiting */
  callCompletionListenerError(listener, ErrorInfo.fromThrowable(t));
@@ -571,7 +572,7 @@ public void onError(ErrorInfo reason) {
  return;
  }
 
- final Timer inProgressTimer = currentAttachTimer;
+ final AblyTimer inProgressTimer = currentAttachTimer;
  attachTimer.schedule(
  new TimerTask() {
  @Override
@@ -601,9 +602,9 @@ private void checkChannelIsNotReleased() {
  * try to attach the channel
  */
  synchronized private void reattachAfterTimeout() {
-Timer currentReattachTimer;
+AblyTimer currentReattachTimer;
  try {
- currentReattachTimer = new Timer();
+ currentReattachTimer = clock.newTimer("reattach-timer");
  } catch(Throwable t) {
  /* an exception instancing the timer can arise because the runtime is exiting */
  return;
@@ -613,7 +614,7 @@ synchronized private void reattachAfterTimeout() {
  this.retryAttempt++;
  int retryDelay = ReconnectionStrategy.getRetryTime(ably.options.channelRetryTimeout, retryAttempt);
 
- final Timer inProgressTimer = currentReattachTimer;
+ final AblyTimer inProgressTimer = currentReattachTimer;
  reattachTimer.schedule(new TimerTask() {
  @Override
  public void run() {
@@ -640,9 +641,9 @@ public void run() {
  */
  synchronized private void detachWithTimeout(final CompletionListener listener) {
  final ChannelState originalState = state;
-Timer currentDetachTimer;
+AblyTimer currentDetachTimer;
  try {
- currentDetachTimer = released.get() ? null : new Timer();
+ currentDetachTimer = released.get() ? null : clock.newTimer("detach-timer");
  } catch(Throwable t) {
  /* an exception instancing the timer can arise because the runtime is exiting */
  callCompletionListenerError(listener, ErrorInfo.fromThrowable(t));
@@ -676,7 +677,7 @@ public void onError(ErrorInfo reason) {
  return;
  }
 
- final Timer inProgressTimer = currentDetachTimer;
+ final AblyTimer inProgressTimer = currentDetachTimer;
  attachTimer.schedule(new TimerTask() {
  @Override
  public void run() {
@@ -1684,6 +1685,7 @@ else if(stateChange.current.equals(failureState)) {
  ChannelBase(AblyRealtime ably, String name, ChannelOptions options, @Nullable LiveObjectsPlugin liveObjectsPlugin) throws AblyException {
  Log.v(TAG, "RealtimeChannel(); channel = " + name);
  this.ably = ably;
+ this.clock = SystemClock.clockFrom(ably.options);
  this.name = name;
  this.basePath = "/channels/" + HttpUtils.encodeURIComponent(name);
  this.setOptions(options);
@@ -1808,6 +1810,7 @@ public void sendProtocolMessage(ProtocolMessage protocolMessage, CompletionListe
 
  private static final String TAG = Channel.class.getName();
  final AblyRealtime ably;
+ final Clock clock;
  final String basePath;
  ChannelOptions options;
  /**

lib/src/main/java/io/ably/lib/realtime/Presence.java

@@ -331,7 +331,7 @@ private void endSync() {
  for (PresenceMessage member: residualMembers) { // RTP19
  member.action = PresenceMessage.Action.leave;
  member.id = null;
- member.timestamp = System.currentTimeMillis();
+ member.timestamp = channel.clock.currentTimeMillis();
  }
  broadcastPresence(residualMembers);
  }

lib/src/main/java/io/ably/lib/rest/Auth.java

@@ -27,8 +27,10 @@
 import io.ably.lib.types.NonRetriableTokenException;
 import io.ably.lib.types.Param;
 import io.ably.lib.util.Base64Coder;
+import io.ably.lib.util.Clock;
 import io.ably.lib.util.Log;
 import io.ably.lib.util.Serialisation;
+import io.ably.lib.util.SystemClock;
 
 /**
  * Token-generation and authentication operations for the Ably API.
@@ -921,7 +923,7 @@ else if(!request.keyName.equals(keyName))
  if(request.timestamp == 0) {
  if(options.queryTime) {
  long oldNanoTimeDelta = nanoTimeDelta;
- long currentNanoTimeDelta = System.currentTimeMillis() - System.nanoTime()/(1000*1000);
+ long currentNanoTimeDelta = clock.currentTimeMillis() - clock.nanoTime()/(1000*1000);
 
  if (timeDelta != Long.MAX_VALUE) {
  /* system time changed by more than 500ms since last time? */
@@ -1036,7 +1038,7 @@ public void onAuthError(ErrorInfo err) {
  clearTokenDetails();
  }
 
- public static long timestamp() { return System.currentTimeMillis(); }
+ public long timestamp() { return clock.currentTimeMillis(); }
 
  /********************
  * internal
@@ -1050,6 +1052,8 @@ public void onAuthError(ErrorInfo err) {
  */
  Auth(AblyBase ably, ClientOptions options) throws AblyException {
  this.ably = ably;
+ this.clock = SystemClock.clockFrom(options);
+ this.nanoTimeDelta = clock.currentTimeMillis() - clock.nanoTime()/(1000*1000);
  authOptions = options;
  tokenParams = options.defaultTokenParams != null ?
  options.defaultTokenParams : new TokenParams();
@@ -1304,6 +1308,7 @@ public long serverTimestamp() {
 
  private static final String TAG = Auth.class.getName();
  private final AblyBase ably;
+ private final Clock clock;
  private final AuthMethod method;
  private AuthOptions authOptions;
  private TokenParams tokenParams;
@@ -1320,7 +1325,7 @@ public long serverTimestamp() {
  * Time delta between System.nanoTime() and System.currentTimeMillis. If it changes significantly it
  * suggests device time/date has changed
  */
- private long nanoTimeDelta = System.currentTimeMillis() - System.nanoTime()/(1000*1000);
+ private long nanoTimeDelta;
 
  public static final String WILDCARD_CLIENTID = "*";
  /**

lib/src/main/java/io/ably/lib/transport/ConnectionManager.java

@@ -36,9 +36,11 @@
 import io.ably.lib.types.ProtocolMessage;
 import io.ably.lib.types.ProtocolSerializer;
 import io.ably.lib.types.PublishResult;
+import io.ably.lib.util.Clock;
 import io.ably.lib.util.Log;
 import io.ably.lib.util.PlatformAgentProvider;
 import io.ably.lib.util.ReconnectionStrategy;
+import io.ably.lib.util.SystemClock;
 import org.jetbrains.annotations.Nullable;
 
 public class ConnectionManager implements ConnectListener {
@@ -782,6 +784,7 @@ public void run() {
 
  public ConnectionManager(final AblyRealtime ably, final Connection connection, final Channels channels, final PlatformAgentProvider platformAgentProvider, LiveObjectsPlugin liveObjectsPlugin) throws AblyException {
  this.ably = ably;
+ this.clock = SystemClock.clockFrom(ably.options);
  this.connection = connection;
  this.channels = channels;
  this.platformAgentProvider = platformAgentProvider;
@@ -992,7 +995,7 @@ public void run() {
  boolean pending;
  synchronized(heartbeatWaiters) {
  try {
-heartbeatWaiters.wait(HEARTBEAT_TIMEOUT);
+clock.waitOn(heartbeatWaiters, HEARTBEAT_TIMEOUT);
  } catch (InterruptedException ie) {
  }
  pending = clear();
@@ -1447,7 +1450,7 @@ private boolean checkConnectionStale() {
  if(lastActivity == 0) {
  return false;
  }
- long now = System.currentTimeMillis();
+ long now = clock.currentTimeMillis();
  long intervalSinceLastActivity = now - lastActivity;
  if(intervalSinceLastActivity > (maxIdleInterval + connectionStateTtl)) {
  /* RTN15g1, RTN15g2 Force a new connection if the previous one is stale;
@@ -1465,7 +1468,7 @@ private boolean checkConnectionStale() {
  }
 
  private synchronized void setSuspendTime() {
- suspendTime = (System.currentTimeMillis() + connectionStateTtl);
+ suspendTime = (clock.currentTimeMillis() + connectionStateTtl);
  }
 
  /**
@@ -1490,7 +1493,7 @@ private StateIndication checkFallback(ErrorInfo reason) {
  }
 
  private synchronized StateIndication checkSuspended(ErrorInfo reason) {
- long currentTime = System.currentTimeMillis();
+ long currentTime = clock.currentTimeMillis();
  long timeToSuspend = suspendTime - currentTime;
  boolean suspendMode = timeToSuspend <= 0;
  Log.v(TAG, "checkSuspended: timeToSuspend = " + timeToSuspend + "ms; suspendMode = " + suspendMode);
@@ -1503,7 +1506,7 @@ private void tryWait(long timeout) {
  if(timeout == 0) {
  wait();
  } else {
-wait(timeout);
+clock.waitOn(this, timeout);
  }
  } catch (InterruptedException e) {}
  }
@@ -2015,6 +2018,7 @@ private boolean isFatalError(ErrorInfo err) {
  ******************/
 
  final AblyRealtime ably;
+ private final Clock clock;
  private final Channels channels;
  private final Connection connection;
  private final ITransport.Factory transportFactory;