Skip to content

Commit dac477f

Browse files
sacOO7sacOO7
authored
Merge pull request ably#1146 from ably/fix/getroot-implicit-attach
[ECO-5517] Implement getRoot implicit attach
2 parents 631a5dc + a035ee7 commit dac477f

12 files changed

Lines changed: 570 additions & 147 deletions

File tree

‎lib/src/main/java/io/ably/lib/objects/Adapter.java‎

Lines changed: 13 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,11 @@
11
package io.ably.lib.objects;
22

33
import io.ably.lib.realtime.AblyRealtime;
4-
import io.ably.lib.realtime.ChannelState;
5-
import io.ably.lib.realtime.CompletionListener;
4+
import io.ably.lib.realtime.ChannelBase;
65
import io.ably.lib.transport.ConnectionManager;
76
import io.ably.lib.types.AblyException;
8-
import io.ably.lib.types.ChannelMode;
9-
import io.ably.lib.types.ChannelOptions;
107
import io.ably.lib.types.ClientOptions;
11-
import io.ably.lib.types.ProtocolMessage;
8+
import io.ably.lib.types.ErrorInfo;
129
import io.ably.lib.util.Log;
1310
import org.jetbrains.annotations.NotNull;
1411

@@ -20,54 +17,6 @@ public Adapter(@NotNull AblyRealtime ably) {
2017
this.ably = ably;
2118
}
2219

23-
@Override
24-
public void setChannelSerial(@NotNull String channelName, @NotNull String channelSerial) {
25-
if (ably.channels.containsKey(channelName)) {
26-
ably.channels.get(channelName).properties.channelSerial = channelSerial;
27-
} else {
28-
Log.e(TAG, "setChannelSerial(): channel not found: " + channelName);
29-
}
30-
}
31-
32-
@Override
33-
public void send(@NotNull ProtocolMessage msg, @NotNull CompletionListener listener) throws AblyException {
34-
// Always queue LiveObjects messages to ensure reliable state synchronization and proper acknowledgment
35-
ably.connection.connectionManager.send(msg, true, listener);
36-
}
37-
38-
@Override
39-
public int maxMessageSizeLimit() {
40-
return ably.connection.connectionManager.maxMessageSize;
41-
}
42-
43-
@Override
44-
public ChannelMode[] getChannelModes(@NotNull String channelName) {
45-
if (ably.channels.containsKey(channelName)) {
46-
// RTO2a - channel.modes is only populated on channel attachment, so use it only if it is set
47-
ChannelMode[] modes = ably.channels.get(channelName).getModes();
48-
if (modes != null) {
49-
return modes;
50-
}
51-
// RTO2b - otherwise as a best effort use user provided channel options
52-
ChannelOptions options = ably.channels.get(channelName).getOptions();
53-
if (options != null && options.hasModes()) {
54-
return options.modes;
55-
}
56-
return null;
57-
}
58-
Log.e(TAG, "getChannelMode(): channel not found: " + channelName);
59-
return null;
60-
}
61-
62-
@Override
63-
public ChannelState getChannelState(@NotNull String channelName) {
64-
if (ably.channels.containsKey(channelName)) {
65-
return ably.channels.get(channelName).state;
66-
}
67-
Log.e(TAG, "getChannelState(): channel not found: " + channelName);
68-
return null;
69-
}
70-
7120
@Override
7221
public @NotNull ClientOptions getClientOptions() {
7322
return ably.options;
@@ -82,4 +31,15 @@ public ChannelState getChannelState(@NotNull String channelName) {
8231
public long getTime() throws AblyException {
8332
return ably.time();
8433
}
34+
35+
@Override
36+
public @NotNull ChannelBase getChannel(@NotNull String channelName) throws AblyException {
37+
if (ably.channels.containsKey(channelName)) {
38+
return ably.channels.get(channelName);
39+
} else {
40+
Log.e(TAG, "attachChannel(): channel not found: " + channelName);
41+
ErrorInfo errorInfo = new ErrorInfo("Channel not found: " + channelName, 404);
42+
throw AblyException.fromErrorInfo(errorInfo);
43+
}
44+
}
8545
}

‎lib/src/main/java/io/ably/lib/objects/LiveObjectsAdapter.java‎

Lines changed: 11 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -1,62 +1,13 @@
11
package io.ably.lib.objects;
22

3-
import io.ably.lib.realtime.ChannelState;
4-
import io.ably.lib.realtime.CompletionListener;
3+
import io.ably.lib.realtime.ChannelBase;
54
import io.ably.lib.transport.ConnectionManager;
65
import io.ably.lib.types.AblyException;
7-
import io.ably.lib.types.ChannelMode;
86
import io.ably.lib.types.ClientOptions;
9-
import io.ably.lib.types.ProtocolMessage;
107
import org.jetbrains.annotations.Blocking;
118
import org.jetbrains.annotations.NotNull;
12-
import org.jetbrains.annotations.Nullable;
139

1410
public interface LiveObjectsAdapter {
15-
/**
16-
* Sends a protocol message to its intended recipient.
17-
* This method transmits a protocol message, allowing for queuing events if necessary,
18-
* and notifies the provided listener upon the success or failure of the send operation.
19-
*
20-
* @param msg the protocol message to send.
21-
* @param listener a listener to be notified of the success or failure of the send operation.
22-
* @throws AblyException if an error occurs during the send operation.
23-
*/
24-
void send(@NotNull ProtocolMessage msg, @NotNull CompletionListener listener) throws AblyException;
25-
26-
/**
27-
* Sets the channel serial for a specific channel.
28-
* @param channelName the name of the channel for which to set the serial
29-
* @param channelSerial the serial to set for the channel
30-
*/
31-
void setChannelSerial(@NotNull String channelName, @NotNull String channelSerial);
32-
33-
/**
34-
* Retrieves the maximum message size allowed for the messages.
35-
* This method returns the maximum size in bytes that a message can have.
36-
*
37-
* @return the maximum message size limit in bytes.
38-
*/
39-
int maxMessageSizeLimit();
40-
41-
/**
42-
* Retrieves the channel modes for a specific channel.
43-
* This method returns the modes that are set for the specified channel.
44-
*
45-
* @param channelName the name of the channel for which to retrieve the modes
46-
* @return the array of channel modes for the specified channel, or null if the channel is not found
47-
* Spec: RTO2a, RTO2b
48-
*/
49-
@Nullable ChannelMode[] getChannelModes(@NotNull String channelName);
50-
51-
/**
52-
* Retrieves the current state of a specific channel.
53-
* This method returns the state of the specified channel, which indicates its connection status.
54-
*
55-
* @param channelName the name of the channel for which to retrieve the state
56-
* @return the current state of the specified channel, or null if the channel is not found
57-
*/
58-
@Nullable ChannelState getChannelState(@NotNull String channelName);
59-
6011
/**
6112
* Retrieves the client options configured for the Ably client.
6213
* Used to access client configuration parameters such as echoMessages setting
@@ -81,5 +32,15 @@ public interface LiveObjectsAdapter {
8132
*/
8233
@Blocking
8334
long getTime() throws AblyException;
35+
36+
/**
37+
* Retrieves the channel instance for the specified channel name.
38+
* If the channel does not exist, an AblyException is thrown.
39+
*
40+
* @param channelName the name of the channel to retrieve
41+
* @return the ChannelBase instance for the specified channel
42+
* @throws AblyException if the channel is not found or cannot be retrieved
43+
*/
44+
@NotNull ChannelBase getChannel(@NotNull String channelName) throws AblyException;
8445
}
8546

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1286,6 +1286,9 @@ public Map<String, String> getParams() {
12861286
}
12871287

12881288
public ChannelMode[] getModes() {
1289+
if (modes == null) {
1290+
return new ChannelMode[0];
1291+
}
12891292
return modes.toArray(new ChannelMode[modes.size()]);
12901293
}
12911294

‎live-objects/src/main/kotlin/io/ably/lib/objects/DefaultLiveObjects.kt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,7 @@ internal class DefaultLiveObjects(internal val channelName: String, internal val
9393

9494
private suspend fun getRootAsync(): LiveMap = withContext(sequentialScope.coroutineContext) {
9595
adapter.throwIfInvalidAccessApiConfiguration(channelName)
96+
adapter.ensureAttached(channelName)
9697
objectsManager.ensureSynced(state)
9798
objectsPool.get(ROOT_OBJECT_ID) as LiveMap
9899
}

‎live-objects/src/main/kotlin/io/ably/lib/objects/Helpers.kt‎

Lines changed: 79 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import io.ably.lib.realtime.CompletionListener
55
import io.ably.lib.types.ChannelMode
66
import io.ably.lib.types.ErrorInfo
77
import io.ably.lib.types.ProtocolMessage
8+
import kotlinx.coroutines.CompletableDeferred
89
import kotlinx.coroutines.suspendCancellableCoroutine
910
import kotlin.coroutines.resume
1011
import kotlin.coroutines.resumeWithException
@@ -14,7 +15,7 @@ import kotlin.coroutines.resumeWithException
1415
*/
1516
internal suspend fun LiveObjectsAdapter.sendAsync(message: ProtocolMessage) = suspendCancellableCoroutine { continuation ->
1617
try {
17-
this.send(message, object : CompletionListener {
18+
connectionManager.send(message, clientOptions.queueMessages, object : CompletionListener {
1819
override fun onSuccess() {
1920
continuation.resume(Unit)
2021
}
@@ -28,11 +29,54 @@ internal suspend fun LiveObjectsAdapter.sendAsync(message: ProtocolMessage) = su
2829
}
2930
}
3031

32+
internal suspend fun LiveObjectsAdapter.attachAsync(channelName: String) = suspendCancellableCoroutine { continuation ->
33+
try {
34+
getChannel(channelName).attach(object : CompletionListener {
35+
override fun onSuccess() {
36+
continuation.resume(Unit)
37+
}
38+
39+
override fun onError(reason: ErrorInfo) {
40+
continuation.resumeWithException(ablyException(reason))
41+
}
42+
})
43+
} catch (e: Exception) {
44+
continuation.resumeWithException(e)
45+
}
46+
}
47+
48+
/**
49+
* Retrieves the channel modes for a specific channel.
50+
* This method returns the modes that are set for the specified channel.
51+
*
52+
* @param channelName the name of the channel for which to retrieve the modes
53+
* @return the array of channel modes for the specified channel, or null if the channel is not found
54+
* Spec: RTO2a, RTO2b
55+
*/
56+
internal fun LiveObjectsAdapter.getChannelModes(channelName: String): Array<ChannelMode>? {
57+
val channel = getChannel(channelName)
58+
59+
// RTO2a - channel.modes is only populated on channel attachment, so use it only if it is set
60+
channel.modes?.let { modes ->
61+
if (modes.isNotEmpty()) {
62+
return modes
63+
}
64+
}
65+
66+
// RTO2b - otherwise as a best effort use user provided channel options
67+
channel.options?.let { options ->
68+
if (options.hasModes()) {
69+
return options.modes
70+
}
71+
}
72+
return null
73+
}
74+
3175
/**
3276
* Spec: RTO15d
3377
*/
3478
internal fun LiveObjectsAdapter.ensureMessageSizeWithinLimit(objectMessages: Array<ObjectMessage>) {
35-
val maximumAllowedSize = maxMessageSizeLimit()
79+
val maximumAllowedSize = connectionManager.maxMessageSize
3680
val objectsTotalMessageSize = objectMessages.sumOf { it.size() }
3781
if (objectsTotalMessageSize > maximumAllowedSize) {
3882
throw ablyException("ObjectMessages size $objectsTotalMessageSize exceeds maximum allowed size of $maximumAllowedSize bytes",
@@ -44,19 +88,46 @@ internal fun LiveObjectsAdapter.setChannelSerial(channelName: String, protocolMe
4488
if (protocolMessage.action != ProtocolMessage.Action.`object`) return
4589
val channelSerial = protocolMessage.channelSerial
4690
if (channelSerial.isNullOrEmpty()) return
47-
setChannelSerial(channelName, channelSerial)
91+
getChannel(channelName).properties.channelSerial = channelSerial
92+
}
93+
94+
internal suspend fun LiveObjectsAdapter.ensureAttached(channelName: String) {
95+
val channel = getChannel(channelName)
96+
when (val currentChannelStatus = channel.state) {
97+
ChannelState.initialized -> attachAsync(channelName)
98+
ChannelState.attached -> return
99+
ChannelState.attaching -> {
100+
val attachDeferred = CompletableDeferred<Unit>()
101+
getChannel(channelName).once {
102+
when(it.current) {
103+
ChannelState.attached -> attachDeferred.complete(Unit)
104+
else -> {
105+
val exception = ablyException("Channel $channelName is in invalid state: ${it.current}, " +
106+
"error: ${it.reason}", ErrorCode.ChannelStateError)
107+
attachDeferred.completeExceptionally(exception)
108+
}
109+
}
110+
}
111+
if (channel.state == ChannelState.attached) {
112+
attachDeferred.complete(Unit)
113+
}
114+
attachDeferred.await()
115+
}
116+
else ->
117+
throw ablyException("Channel $channelName is in invalid state: $currentChannelStatus", ErrorCode.ChannelStateError)
118+
}
48119
}
49120

50121
// Spec: RTLO4b1, RTLO4b2
51122
internal fun LiveObjectsAdapter.throwIfInvalidAccessApiConfiguration(channelName: String) {
52-
throwIfMissingChannelMode(channelName, ChannelMode.object_subscribe)
53123
throwIfInChannelState(channelName, arrayOf(ChannelState.detached, ChannelState.failed))
124+
throwIfMissingChannelMode(channelName, ChannelMode.object_subscribe)
54125
}
55126

56127
internal fun LiveObjectsAdapter.throwIfInvalidWriteApiConfiguration(channelName: String) {
57128
throwIfEchoMessagesDisabled()
58-
throwIfMissingChannelMode(channelName, ChannelMode.object_publish)
59129
throwIfInChannelState(channelName, arrayOf(ChannelState.detached, ChannelState.failed, ChannelState.suspended))
130+
throwIfMissingChannelMode(channelName, ChannelMode.object_publish)
60131
}
61132

62133
internal fun LiveObjectsAdapter.throwIfUnpublishableState(channelName: String) {
@@ -67,16 +138,16 @@ internal fun LiveObjectsAdapter.throwIfUnpublishableState(channelName: String) {
67138
}
68139

69140
// Spec: RTO2
70-
internal fun LiveObjectsAdapter.throwIfMissingChannelMode(channelName: String, channelMode: ChannelMode) {
141+
private fun LiveObjectsAdapter.throwIfMissingChannelMode(channelName: String, channelMode: ChannelMode) {
71142
val channelModes = getChannelModes(channelName)
72143
if (channelModes == null || !channelModes.contains(channelMode)) {
73144
// Spec: RTO2a2, RTO2b2
74145
throw ablyException("\"${channelMode.name}\" channel mode must be set for this operation", ErrorCode.ChannelModeRequired)
75146
}
76147
}
77148

78-
internal fun LiveObjectsAdapter.throwIfInChannelState(channelName: String, channelStates: Array<ChannelState>) {
79-
val currentState = getChannelState(channelName)
149+
private fun LiveObjectsAdapter.throwIfInChannelState(channelName: String, channelStates: Array<ChannelState>) {
150+
val currentState = getChannel(channelName).state
80151
if (currentState == null || channelStates.contains(currentState)) {
81152
throw ablyException("Channel is in invalid state: $currentState", ErrorCode.ChannelStateError)
82153
}

‎live-objects/src/test/kotlin/io/ably/lib/objects/integration/DefaultLiveObjectsTest.kt‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ class DefaultLiveObjectsTest : IntegrationTest() {
3333
// Initialize the root map on the channel with initial data
3434
restObjects.initializeRootMap(channelName)
3535

36-
val channel = getRealtimeChannel(channelName, autoAttach = false)
36+
val channel = getRealtimeChannel(channelName)
3737
val objects = channel.objects
3838
assertNotNull(objects)
3939

‎live-objects/src/test/kotlin/io/ably/lib/objects/integration/setup/IntegrationTest.kt‎

Lines changed: 4 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -29,14 +29,13 @@ abstract class IntegrationTest {
2929
/**
3030
* Retrieves a realtime channel for the specified channel name and client ID
3131
* If a client with the given clientID does not exist, a new client is created using the provided options.
32-
* The channel is attached and ensured to be in the attached state before returning.
3332
*
3433
* @param channelName Name of the channel
3534
* @param clientId The ID of the client to use or create. Defaults to "client1".
36-
* @return The attached realtime channel.
37-
* @throws Exception If the channel fails to attach or the client fails to connect.
35+
* @return The realtime channel in the INITIALIZED state.
36+
* @throws Exception If the client fails to connect.
3837
*/
39-
internal suspend fun getRealtimeChannel(channelName: String, clientId: String = "client1", autoAttach: Boolean = true): Channel {
38+
internal suspend fun getRealtimeChannel(channelName: String, clientId: String = "client1"): Channel {
4039
val client = realtimeClients.getOrPut(clientId) {
4140
sandbox.createRealtimeClient {
4241
this.clientId = clientId
@@ -46,12 +45,7 @@ abstract class IntegrationTest {
4645
val channelOpts = ChannelOptions().apply {
4746
modes = arrayOf(ChannelMode.object_publish, ChannelMode.object_subscribe)
4847
}
49-
return client.channels.get(channelName, channelOpts).apply {
50-
if (autoAttach) {
51-
attach()
52-
ensureAttached()
53-
}
54-
}
48+
return client.channels.get(channelName, channelOpts)
5549
}
5650

5751
/**

‎live-objects/src/test/kotlin/io/ably/lib/objects/integration/setup/Sandbox.kt‎

Lines changed: 0 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -92,20 +92,3 @@ internal suspend fun AblyRealtime.ensureConnected() {
9292
}
9393
connectedDeferred.await()
9494
}
95-
96-
internal suspend fun Channel.ensureAttached() {
97-
if (this.state == ChannelState.attached) {
98-
return
99-
}
100-
val attachedDeferred = CompletableDeferred<Unit>()
101-
this.on {
102-
if (it.event == ChannelEvent.attached) {
103-
attachedDeferred.complete(Unit)
104-
this.off()
105-
} else if (it.event != ChannelEvent.attaching) {
106-
attachedDeferred.completeExceptionally(ablyException(it.reason))
107-
this.off()
108-
}
109-
}
110-
attachedDeferred.await()
111-
}

0 commit comments

Comments
 (0)