diff --git a/lib/src/main/java/io/ably/lib/realtime/AblyRealtime.java b/lib/src/main/java/io/ably/lib/realtime/AblyRealtime.java index efd8bcad6..e07465b9a 100644 --- a/lib/src/main/java/io/ably/lib/realtime/AblyRealtime.java +++ b/lib/src/main/java/io/ably/lib/realtime/AblyRealtime.java @@ -188,12 +188,14 @@ public interface Channels extends ReadOnlyMap { /** * Releases a {@link Channel} object, deleting it, and enabling it to be garbage collected. * It also removes any listeners associated with the channel. - * To release a channel, the {@link ChannelState} must be INITIALIZED, DETACHED, or FAILED. + * A realtime channel can only be released when it is in the INITIALIZED, DETACHED, or FAILED state. *

* Spec: RSN4, RTS4 * @param channelName The channel name. + * @throws AblyException If the channel is in any other state. Call {@link Channel#detach(CompletionListener)} + * and wait for it to complete before calling release(). */ - void release(String channelName); + void release(String channelName) throws AblyException; } private class InternalChannels extends InternalMap implements Channels, ConnectionManager.Channels { @@ -232,15 +234,25 @@ public Channel get(final String channelName, final ChannelOptions channelOptions } @Override - public void release(String channelName) { - Channel channel = map.remove(channelName); - if(channel != null) { - channel.markAsReleased(); - try { - channel.detach(); - } catch (AblyException e) { - Log.e(TAG, "Unexpected exception detaching channel; channelName = " + channelName, e); + public void release(String channelName) throws AblyException { + Channel channel = map.get(channelName); + if (channel == null) { + return; + } + // The channel's state transitions hold its monitor, so holding it here stops the channel + // from starting to attach between the state check and it being marked as released. + synchronized (channel) { + ChannelState state = channel.state; + if (state != ChannelState.initialized && state != ChannelState.detached && state != ChannelState.failed) { + // RTS4e + throw AblyException.fromErrorInfo(new ErrorInfo( + "Can only release a channel in a state where there is no possibility of further updates from the server being received (initialized, detached, or failed). The current state is " + state, + 400, 90011)); } + if (!map.remove(channelName, channel)) { + return; + } + channel.markAsReleased(); } if (liveObjectsPlugin != null) { liveObjectsPlugin.dispose(channelName); 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 fe68a481d..ae17649c4 100644 --- a/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java +++ b/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java @@ -400,11 +400,7 @@ private void sendDetachMessage(CompletionListener listener) throws AblyException } this.attachResume = false; - if (released.get()) { - setDetached(null); - } else { - setState(ChannelState.detaching, null); - } + setState(ChannelState.detaching, null); ably.connection.connectionManager.send(detachMessage, true, null); } catch(AblyException e) { throw e; @@ -635,7 +631,7 @@ synchronized private void detachWithTimeout(final CompletionListener listener) { final ChannelState originalState = state; AblyTimer currentDetachTimer; try { - currentDetachTimer = released.get() ? null : clock.newTimer("detach-timer"); + currentDetachTimer = clock.newTimer("detach-timer"); } catch(Throwable t) { /* an exception instancing the timer can arise because the runtime is exiting */ callCompletionListenerError(listener, ErrorInfo.fromThrowable(t)); @@ -644,8 +640,7 @@ synchronized private void detachWithTimeout(final CompletionListener listener) { attachTimer = currentDetachTimer; try { - // If channel has been released, completionListener won't be invoked anyway - CompletionListener completionListener = released.get() ? null : new CompletionListener() { + CompletionListener completionListener = new CompletionListener() { @Override public void onSuccess() { clearAttachTimers(); 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 8fa2a3bbc..e7db4deed 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 @@ -2442,41 +2442,6 @@ public void no_messages_when_channel_state_not_attached() { } } - /* - * Checks that the DETACHED message sent by the server when a channel is released is dropped. - */ - @Test - public void detach_message_to_released_channel_is_dropped() throws AblyException { - AblyRealtime ably = null; - long oldRealtimeTimeout = Defaults.realtimeRequestTimeout; - final String channelName = "detach_message_to_released_channel_is_dropped"; - - try { - DebugOptions opts = createOptions(testVars.keys[0].keyStr); - Helpers.RawProtocolMonitor monitor = Helpers.RawProtocolMonitor.createReceiver(ProtocolMessage.Action.detached); - opts.protocolListener = monitor; - - /* Make test faster */ - Defaults.realtimeRequestTimeout = 1000; - opts.channelRetryTimeout = 1000; - - ably = new AblyRealtime(opts); - Channel channel = ably.channels.get(channelName); - channel.attach(); - (new ChannelWaiter(channel)).waitFor(ChannelState.attached); - - // Listen for detach messages and release the channel - ably.channels.release(channelName); - monitor.waitForRecv(1, 10000); - - assertFalse(ably.channels.containsKey("messages_to_non_existent_channels_are_dropped")); - } finally { - if (ably != null) - ably.close(); - Defaults.realtimeRequestTimeout = oldRealtimeTimeout; - } - } - /* * Spec: RTN11d * Checks that all channels become if the state is CLOSED transitions all the channels to diff --git a/lib/src/test/kotlin/io/ably/lib/uts/unit/realtime/ChannelsCollectionTest.kt b/lib/src/test/kotlin/io/ably/lib/uts/unit/realtime/ChannelsCollectionTest.kt new file mode 100644 index 000000000..86461a8b8 --- /dev/null +++ b/lib/src/test/kotlin/io/ably/lib/uts/unit/realtime/ChannelsCollectionTest.kt @@ -0,0 +1,129 @@ +package io.ably.lib.uts.unit.realtime + +import io.ably.lib.uts.infra.unit.* +import io.ably.lib.realtime.ChannelState +import io.ably.lib.realtime.ConnectionState +import io.ably.lib.types.AblyException +import io.ably.lib.types.ProtocolMessage +import io.ably.lib.uts.infra.awaitChannelState +import io.ably.lib.uts.infra.awaitState +import kotlinx.coroutines.test.runTest +import java.util.UUID +import java.util.concurrent.CopyOnWriteArrayList +import kotlin.test.* + +class ChannelsCollectionTest { + + /** + * @UTS realtime/unit/RTS4c/release-nonexistent-noop-0 + */ + @Test + fun `RTS4c - release on non-existent channel is no-op`() { + val channelName = "test-RTS4c-nonexistent-${UUID.randomUUID()}" + val client = TestRealtimeClient { autoConnect = false } + + client.channels.release(channelName) + + assertFalse(client.channels.containsKey(channelName)) + client.close() + } + + /** + * @UTS realtime/unit/RTS4d/release-removes-channel-0 + */ + @Test + fun `RTS4d - release removes an initialized channel`() { + val channelName = "test-RTS4d-${UUID.randomUUID()}" + val client = TestRealtimeClient { autoConnect = false } + + val channel = client.channels.get(channelName) + assertEquals(ChannelState.initialized, channel.state) + assertTrue(client.channels.containsKey(channelName)) + + client.channels.release(channelName) + + assertFalse(client.channels.containsKey(channelName)) + client.close() + } + + /** + * @UTS realtime/unit/RTS4d/release-after-detach-1 + */ + @Test + fun `RTS4d - release removes a channel once detached`() = runTest { + val channelName = "test-RTS4d-detached-${UUID.randomUUID()}" + lateinit var mockWs: MockWebSocket + mockWs = MockWebSocket { + onConnectionAttempt = { conn -> conn.respondWithSuccess(CONNECTED_MESSAGE) } + onMessageFromClient = { msg -> + when (msg.action) { + ProtocolMessage.Action.attach -> + mockWs.sendToClient(ProtocolMessage(ProtocolMessage.Action.attached, msg.channel)) + ProtocolMessage.Action.detach -> + mockWs.sendToClient(ProtocolMessage(ProtocolMessage.Action.detached, msg.channel)) + else -> Unit + } + } + } + val client = TestRealtimeClient { + autoConnect = false + install(mockWs) + } + val channel = client.channels.get(channelName) + + client.connect() + awaitState(client, ConnectionState.connected) + + channel.attach() + awaitChannelState(channel, ChannelState.attached) + channel.detach() + awaitChannelState(channel, ChannelState.detached) + + client.channels.release(channelName) + + assertFalse(client.channels.containsKey(channelName)) + client.close() + } + + /** + * @UTS realtime/unit/RTS4e/release-attached-fails-0 + */ + @Test + fun `RTS4e - release of an attached channel fails`() = runTest { + val channelName = "test-RTS4e-attached-${UUID.randomUUID()}" + val capturedDetachMessages = CopyOnWriteArrayList() + lateinit var mockWs: MockWebSocket + mockWs = MockWebSocket { + onConnectionAttempt = { conn -> conn.respondWithSuccess(CONNECTED_MESSAGE) } + onMessageFromClient = { msg -> + when (msg.action) { + ProtocolMessage.Action.attach -> + mockWs.sendToClient(ProtocolMessage(ProtocolMessage.Action.attached, msg.channel)) + ProtocolMessage.Action.detach -> capturedDetachMessages.add(msg) + else -> Unit + } + } + } + val client = TestRealtimeClient { + autoConnect = false + install(mockWs) + } + val channel = client.channels.get(channelName) + + client.connect() + awaitState(client, ConnectionState.connected) + + channel.attach() + awaitChannelState(channel, ChannelState.attached) + + val error = assertFailsWith { client.channels.release(channelName) }.errorInfo + + assertEquals(90011, error.code) + assertEquals(400, error.statusCode) + assertEquals(ChannelState.attached, channel.state) + assertTrue(client.channels.containsKey(channelName)) + assertSame(channel, client.channels.get(channelName)) + assertEquals(0, capturedDetachMessages.size) + client.close() + } +} diff --git a/liveobjects/src/test/kotlin/io/ably/lib/liveobjects/integration/setup/IntegrationTest.kt b/liveobjects/src/test/kotlin/io/ably/lib/liveobjects/integration/setup/IntegrationTest.kt index 7fcfa8345..18509f1a2 100644 --- a/liveobjects/src/test/kotlin/io/ably/lib/liveobjects/integration/setup/IntegrationTest.kt +++ b/liveobjects/src/test/kotlin/io/ably/lib/liveobjects/integration/setup/IntegrationTest.kt @@ -59,9 +59,8 @@ abstract class IntegrationTest { @After fun afterEach() { for (ablyRealtime in realtimeClients.values) { - for ((channelName, channel) in ablyRealtime.channels.entrySet()) { + for (channel in ablyRealtime.channels.values()) { channel.off() - ablyRealtime.channels.release(channelName) } ablyRealtime.close() } diff --git a/pubsub-adapter/src/main/kotlin/com/ably/pubsub/Channels.kt b/pubsub-adapter/src/main/kotlin/com/ably/pubsub/Channels.kt index cdc989924..6189dea14 100644 --- a/pubsub-adapter/src/main/kotlin/com/ably/pubsub/Channels.kt +++ b/pubsub-adapter/src/main/kotlin/com/ably/pubsub/Channels.kt @@ -39,7 +39,7 @@ public interface Channels : Iterable { /** * Releases a [Channel] object, deleting it, and enabling it to be garbage collected. * It also removes any listeners associated with the channel. - * To release a channel, the [ChannelState] must be `INITIALIZED`, `DETACHED`, or `FAILED`. + * A realtime channel can only be released when it is in the `INITIALIZED`, `DETACHED`, or `FAILED` state. *

* Spec: RSN4, RTS4 * @param name The channel name.