Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 22 additions & 10 deletions lib/src/main/java/io/ably/lib/realtime/AblyRealtime.java
Original file line number Diff line number Diff line change
Expand Up @@ -188,12 +188,14 @@ public interface Channels extends ReadOnlyMap<String, Channel> {
/**
* 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.
* <p>
* 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<String, Channel> implements Channels, ConnectionManager.Channels {
Expand Down Expand Up @@ -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);
Expand Down
11 changes: 3 additions & 8 deletions lib/src/main/java/io/ably/lib/realtime/ChannelBase.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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));
Expand All @@ -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();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
@@ -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<ProtocolMessage>()
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<AblyException> { 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()
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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()
}
Expand Down
2 changes: 1 addition & 1 deletion pubsub-adapter/src/main/kotlin/com/ably/pubsub/Channels.kt
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ public interface Channels<ChannelType> : Iterable<ChannelType> {
/**
* 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.
* <p>
* Spec: RSN4, RTS4
* @param name The channel name.
Expand Down
Loading