[CHAT-5063] feat: add createWrapperSdkProxy implementation for kotlin adapter of ably-java

android/src/main/java/io/ably/lib/rest/AblyRest.java

@@ -1,6 +1,8 @@
 package io.ably.lib.rest;
 
 import android.content.Context;
+import io.ably.lib.http.Http;
+import io.ably.lib.http.HttpCore;
 import io.ably.lib.push.LocalDevice;
 import io.ably.lib.types.AblyException;
 import io.ably.lib.types.ClientOptions;
@@ -36,6 +38,24 @@ public AblyRest(ClientOptions options) throws AblyException {
         super(options, new AndroidPlatformAgentProvider());
     }
 
+    /**
+     * Constructor implementation to be able to have shallow copy of the client,
+     * allowing us to modify certain fields while implementing a proxy for the Realtime/Rest SDK wrapper
+     */
+    protected AblyRest(AblyRest underlyingClient, HttpCore httpCore, Http http) {
+        super(underlyingClient, httpCore, http);
+    }
+
+    /**
+     * [Internal Method]
+     * <p/>
+     * We use this method to create a shallow copy of the client, allowing us to modify certain fields
+     * while implementing a proxy for the Realtime/Rest SDK wrapper
+     */
+    public AblyRest createShallowCopy(HttpCore httpCore, Http http) {
+        return new AblyRest(this, httpCore, http);
+    }
+
     /**
      * Retrieves a {@link LocalDevice} object that represents the current state of the device as a target for push notifications.
      * <p>

gradle/libs.versions.toml

@@ -20,6 +20,8 @@ lombok = "8.10"
 okhttp = "4.12.0"
 test-retry = "1.6.0"
 kotlin = "2.1.10"
+coroutine = "1.9.0"
+turbine = "1.2.0"
 
 [libraries]
 gson = { group = "com.google.code.gson", name = "gson", version.ref = "gson" }
@@ -41,11 +43,14 @@ dexmaker = { group = "com.crittercism.dexmaker", name = "dexmaker", version.ref
 dexmaker-dx = { group = "com.crittercism.dexmaker", name = "dexmaker-dx", version.ref = "dexmaker" }
 dexmaker-mockito = { group = "com.crittercism.dexmaker", name = "dexmaker-mockito", version.ref = "dexmaker" }
 android-retrostreams = { group = "net.sourceforge.streamsupport", name = "android-retrostreams", version.ref = "android-retrostreams" }
-okhttp = { group ="com.squareup.okhttp3", name = "okhttp", version.ref = "okhttp" }
+okhttp = { group = "com.squareup.okhttp3", name = "okhttp", version.ref = "okhttp" }
+coroutine-core = { group = "org.jetbrains.kotlinx", name = "kotlinx-coroutines-core", version.ref = "coroutine" }
+coroutine-test = { group = "org.jetbrains.kotlinx", name = "kotlinx-coroutines-test", version.ref = "coroutine" }
+turbine = { group = "app.cash.turbine", name = "turbine", version.ref = "turbine" }
 
 [bundles]
 common = ["msgpack", "vcdiff-core"]
-tests = ["junit","hamcrest-all", "nanohttpd", "nanohttpd-nanolets", "nanohttpd-websocket", "mockito-core", "concurrentunit", "slf4j-simple"]
+tests = ["junit", "hamcrest-all", "nanohttpd", "nanohttpd-nanolets", "nanohttpd-websocket", "mockito-core", "concurrentunit", "slf4j-simple"]
 instrumental-android = ["android-test-runner", "android-test-rules", "dexmaker", "dexmaker-dx", "dexmaker-mockito", "android-retrostreams"]
 
 [plugins]

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

@@ -1,5 +1,7 @@
 package io.ably.lib.rest;
 
+import io.ably.lib.http.Http;
+import io.ably.lib.http.HttpCore;
 import io.ably.lib.types.AblyException;
 import io.ably.lib.types.ClientOptions;
 import io.ably.lib.util.JavaPlatformAgentProvider;
@@ -32,4 +34,22 @@ public AblyRest(String key) throws AblyException {
     public AblyRest(ClientOptions options) throws AblyException {
         super(options, new JavaPlatformAgentProvider());
     }
+
+    /**
+     * Constructor implementation to be able to have shallow copy of the client,
+     * allowing us to modify certain fields while implementing a proxy for the Realtime/Rest SDK wrapper
+     */
+    protected AblyRest(AblyRest underlyingClient, HttpCore httpCore, Http http) {
+        super(underlyingClient, httpCore, http);
+    }
+
+    /**
+     * [Internal Method]
+     * <p/>
+     * We use this method to create a shallow copy of the client, allowing us to modify certain fields
+     * while implementing a proxy for the Realtime/Rest SDK wrapper
+     */
+    public AblyRest createShallowCopy(HttpCore httpCore, Http http) {
+        return new AblyRest(this, httpCore, http);
+    }
 }

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

@@ -16,10 +16,23 @@ public AsyncHttpScheduler(HttpCore httpCore, ClientOptions options) {
         super(httpCore, new CloseableThreadPoolExecutor(options));
     }
 
+    private AsyncHttpScheduler(HttpCore httpCore, CloseableExecutor executor) {
+        super(httpCore, executor);
+    }
+
     private static final long KEEP_ALIVE_TIME = 2000L;
 
     protected static final String TAG = AsyncHttpScheduler.class.getName();
 
+    /**
+     * [Internal Method]
+     * <p>
+     * We use this method to implement proxy Realtime / Rest clients that add additional data to the underlying client.
+     */
+    public AsyncHttpScheduler exchangeHttpCore(HttpCore httpCore) {
+        return new AsyncHttpScheduler(httpCore, this.executor);
+    }
+
     private static class CloseableThreadPoolExecutor implements CloseableExecutor {
         private final ThreadPoolExecutor executor;
 

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

@@ -21,6 +21,15 @@ public void close() throws Exception {
         asyncHttp.close();
     }
 
+    /**
+     * [Internal Method]
+     * <p>
+     * We use this method to implement proxy Realtime / Rest clients that add additional data to the underlying client.
+     */
+    public Http exchangeHttpCore(HttpCore httpCore) {
+        return new Http(asyncHttp.exchangeHttpCore(httpCore), new SyncHttpScheduler(httpCore));
+    }
+
     public class Request<Result> {
         private final Execute<Result> execute;
 

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

@@ -68,6 +68,16 @@ public class HttpCore {
     private final HttpEngine engine;
     private HttpAuth proxyAuth;
 
+    /**
+     * This field is used for analytics purposes.
+     * <p>
+     * It holds additional agents that should be added after the Realtime/Rest client is initialized.
+     * - **Static agents** are set in `ClientOptions`.
+     * - **Dynamic agents** are added later by higher-level SDKs like Chat or Asset Tracking
+     *   and are provided in the `createWrapperSdkProxy` call.
+     */
+    private Map<String, String> dynamicAgents;
+
     /*************************
      *     Public API
      *************************/
@@ -103,6 +113,18 @@ public HttpCore(ClientOptions options, Auth auth, PlatformAgentProvider platform
         this.engine = engineFactory.create(new HttpEngineConfig(ClientOptionsUtils.convertToProxyConfig(options)));
     }
 
+    private HttpCore(HttpCore underlyingHttpCore, Map<String, String> dynamicAgents) {
+        this.options = underlyingHttpCore.options;
+        this.auth = underlyingHttpCore.auth;
+        this.platformAgentProvider = underlyingHttpCore.platformAgentProvider;
+        this.scheme = underlyingHttpCore.scheme;
+        this.port = underlyingHttpCore.port;
+        this.hosts = underlyingHttpCore.hosts;
+        this.proxyAuth = underlyingHttpCore.proxyAuth;
+        this.engine = underlyingHttpCore.engine;
+        this.dynamicAgents = dynamicAgents;
+    }
+
     /**
      * Make a synchronous HTTP request specified by URL and proxy, retrying if necessary on WWW-Authenticate
      *
@@ -307,7 +329,10 @@ private Map<String, String> collectRequestHeaders(URL url, String method, Param[
 
         /* pass required headers */
         requestHeaders.put(Defaults.ABLY_PROTOCOL_VERSION_HEADER, Defaults.ABLY_PROTOCOL_VERSION); // RSC7a
-        requestHeaders.put(Defaults.ABLY_AGENT_HEADER, AgentHeaderCreator.create(options.agents, platformAgentProvider));
+        Map<String, String> additionalAgents = new HashMap<>();
+        if (options.agents != null) additionalAgents.putAll(options.agents);
+        if (dynamicAgents != null) additionalAgents.putAll(dynamicAgents);
+        requestHeaders.put(Defaults.ABLY_AGENT_HEADER, AgentHeaderCreator.create(additionalAgents, platformAgentProvider));
         if (options.clientId != null)
             requestHeaders.put(Defaults.ABLY_CLIENT_ID_HEADER, Base64Coder.encodeString(options.clientId));
 
@@ -455,6 +480,15 @@ private Response executeRequest(HttpRequest request) {
         return response;
     }
 
+    /**
+     * [Internal Method]
+     * <p>
+     * We use this method to implement proxy Realtime / Rest clients that add additional agents to the underlying client.
+     */
+    public HttpCore injectDynamicAgents(Map<String, String> wrapperSDKAgents) {
+        return new HttpCore(this, wrapperSDKAgents);
+    }
+
     /**
      * Interface for an entity that supplies an httpCore request body
      */

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

@@ -440,7 +440,7 @@ public <T> Future<T> ablyHttpExecuteWithRetry(
         return request;
     }
 
-    private final CloseableExecutor executor;
+    protected final CloseableExecutor executor;
     private final HttpCore httpCore;
 
     protected static final String TAG = HttpScheduler.class.getName();

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

@@ -5,6 +5,8 @@
 import java.util.List;
 import java.util.Map;
 
+import io.ably.lib.http.Http;
+import io.ably.lib.http.HttpCore;
 import io.ably.lib.rest.AblyRest;
 import io.ably.lib.rest.Auth;
 import io.ably.lib.transport.ConnectionManager;
@@ -83,6 +85,16 @@ public void onConnectionStateChanged(ConnectionStateListener.ConnectionStateChan
         if(options.autoConnect) connection.connect();
     }
 
+    /**
+     * Constructor implementation to be able to have shallow copy of the client,
+     * allowing us to modify certain fields while implementing a proxy for the Realtime/Rest SDK wrapper
+     */
+    AblyRealtime(AblyRealtime underlyingClient, HttpCore httpCore, Http http) {
+        super(underlyingClient, httpCore, http);
+        this.channels = underlyingClient.channels;
+        this.connection = underlyingClient.connection;
+    }
+
     /**
      * Calls {@link Connection#connect} and causes the connection to open,
      * entering the connecting state. Explicitly calling connect() is unnecessary
@@ -118,6 +130,16 @@ public void close() {
         connection.close();
     }
 
+    /**
+     * [Internal Method]
+     * <p/>
+     * We use this method to create a shallow copy of the client, allowing us to modify certain fields
+     * while implementing a proxy for the Realtime/Rest SDK wrapper
+     */
+    public AblyRealtime createShallowCopy(HttpCore httpCore, Http http) {
+        return new AblyRealtime(this, httpCore, http);
+    }
+
     /**
      * Authentication token has changed.
      */

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

@@ -10,6 +10,7 @@
 import java.util.TimerTask;
 
 import io.ably.lib.http.BasePaginatedQuery;
+import io.ably.lib.http.Http;
 import io.ably.lib.http.HttpCore;
 import io.ably.lib.http.HttpUtils;
 import io.ably.lib.transport.ConnectionManager;
@@ -1150,7 +1151,11 @@ else if(!"false".equalsIgnoreCase(param.value)) {
      * @throws AblyException
      */
     public PaginatedResult<Message> history(Param[] params) throws AblyException {
-        return historyImpl(params).sync();
+        return historyImpl(ably.http, params).sync();
+    }
+
+    PaginatedResult<Message> history(Http http, Param[] params) throws AblyException {
+        return historyImpl(http, params).sync();
     }
 
     /**
@@ -1179,18 +1184,22 @@ public PaginatedResult<Message> history(Param[] params) throws AblyException {
      * @throws AblyException
      */
     public void historyAsync(Param[] params, Callback<AsyncPaginatedResult<Message>> callback) {
-        historyImpl(params).async(callback);
+        historyAsync(ably.http, params, callback);
+    }
+
+    void historyAsync(Http http, Param[] params, Callback<AsyncPaginatedResult<Message>> callback) {
+        historyImpl(http, params).async(callback);
     }
 
-    private BasePaginatedQuery.ResultRequest<Message> historyImpl(Param[] params) {
+    private BasePaginatedQuery.ResultRequest<Message> historyImpl(Http http, Param[] params) {
         try {
             params = replacePlaceholderParams((Channel) this, params);
         } catch (AblyException e) {
             return new BasePaginatedQuery.ResultRequest.Failed<Message>(e);
         }
 
         HttpCore.BodyHandler<Message> bodyHandler = MessageSerializer.getMessageResponseHandler(options);
-        return new BasePaginatedQuery<Message>(ably.http, basePath + "/history", HttpUtils.defaultAcceptHeaders(ably.options.useBinaryProtocol), params, bodyHandler).get();
+        return new BasePaginatedQuery<Message>(http, basePath + "/history", HttpUtils.defaultAcceptHeaders(ably.options.useBinaryProtocol), params, bodyHandler).get();
     }
 
     /************************************

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

@@ -1,6 +1,7 @@
 package io.ably.lib.realtime;
 
 import io.ably.lib.http.BasePaginatedQuery;
+import io.ably.lib.http.Http;
 import io.ably.lib.http.HttpCore;
 import io.ably.lib.http.HttpUtils;
 import io.ably.lib.transport.ConnectionManager;
@@ -794,7 +795,11 @@ public void updatePresence(PresenceMessage msg, CompletionListener listener) thr
      * @throws AblyException
      */
     public PaginatedResult<PresenceMessage> history(Param[] params) throws AblyException {
-        return historyImpl(params).sync();
+        return history(channel.ably.http, params);
+    }
+
+    PaginatedResult<PresenceMessage> history(Http http, Param[] params) throws AblyException {
+        return historyImpl(http, params).sync();
     }
 
     /**
@@ -821,10 +826,14 @@ public PaginatedResult<PresenceMessage> history(Param[] params) throws AblyExcep
      * @throws AblyException
      */
     public void historyAsync(Param[] params, Callback<AsyncPaginatedResult<PresenceMessage>> callback) {
-        historyImpl(params).async(callback);
+        historyImpl(channel.ably.http, params).async(callback);
+    }
+
+    void historyAsync(Http http, Param[] params, Callback<AsyncPaginatedResult<PresenceMessage>> callback) {
+        historyImpl(http, params).async(callback);
     }
 
-    private BasePaginatedQuery.ResultRequest<PresenceMessage> historyImpl(Param[] params) {
+    private BasePaginatedQuery.ResultRequest<PresenceMessage> historyImpl(Http http, Param[] params) {
         try {
             params = Channel.replacePlaceholderParams(channel, params);
         } catch (AblyException e) {
@@ -833,7 +842,7 @@ private BasePaginatedQuery.ResultRequest<PresenceMessage> historyImpl(Param[] pa
 
         AblyRealtime ably = channel.ably;
         HttpCore.BodyHandler<PresenceMessage> bodyHandler = PresenceSerializer.getPresenceResponseHandler(channel.options);
-        return new BasePaginatedQuery<PresenceMessage>(ably.http, channel.basePath + "/presence/history", HttpUtils.defaultAcceptHeaders(ably.options.useBinaryProtocol), params, bodyHandler).get();
+        return new BasePaginatedQuery<PresenceMessage>(http, channel.basePath + "/presence/history", HttpUtils.defaultAcceptHeaders(ably.options.useBinaryProtocol), params, bodyHandler).get();
     }
 
     /**

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

@@ -114,6 +114,20 @@ public AblyBase(ClientOptions options, PlatformAgentProvider platformAgentProvid
         push = new Push(this);
     }
 
+    /**
+     * We use empty constructor to be able to create proxy implementation of Realtime and Rest client
+     */
+    protected AblyBase(AblyBase underlyingClient, HttpCore httpCore, Http http) {
+        this.options = underlyingClient.options;
+        this.auth = underlyingClient.auth;
+        this.httpCore = httpCore;
+        this.http = http;
+        this.platform = underlyingClient.platform;
+        this.push = underlyingClient.push;
+        this.channels = underlyingClient.channels;
+        this.platformAgentProvider = underlyingClient.platformAgentProvider;
+    }
+
     /**
      * Causes the connection to close, entering the [{@link io.ably.lib.realtime.ConnectionState#closing} state.
      * Once closed, the library does not attempt to re-establish the connection without an explicit call to