Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix Member Presence #607

Merged
merged 6 commits into from
Jan 21, 2021
Merged

Fix Member Presence #607

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
27b2b06
Fixed wait for channel attach state and members presence sync
sacOO7 Sep 29, 2020
e0e7f79
Added test for consistent presence during intermittent detach cycles
sacOO7 Sep 29, 2020
11bc5c0
reformatted code in accordance with checkstyle
sacOO7 Oct 3, 2020
bb3ad1a
Merge branch 'main' into bugfix/467-missing-member-presence
sacOO7 Oct 3, 2020
7a04fe7
removed tab characters, removed star imports
sacOO7 Oct 4, 2020
626999e
Simplified loop for checking if syncing is complete
sacOO7 Jan 7, 2021
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
12 changes: 8 additions & 4 deletions lib/src/main/java/io/ably/lib/realtime/Presence.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -792,11 +792,15 @@ private class PresenceMap {
* state other than attached or attaching
*/
synchronized void waitForSync() throws AblyException, InterruptedException {
boolean syncIsComplete = false; /* temporary variable to avoid potential race conditions */
while((channel.state == ChannelState.attached || channel.state == ChannelState.attaching) &&
/* = (and not ==) is intentional */
!(syncIsComplete = (!syncInProgress && syncComplete)))
boolean syncIsComplete = false; /* temporary variable to avoid potential race conditions */
while (channel.state == ChannelState.attaching) {
wait();
}
if (channel.state == ChannelState.attached) {
while (!(syncIsComplete = (!syncInProgress && syncComplete))) {
sacOO7 marked this conversation as resolved.
Show resolved Hide resolved
wait();
}
}

/* invalid channel state */
int errorCode;
Expand Down
113 changes: 102 additions & 11 deletions lib/src/test/java/io/ably/lib/test/realtime/RealtimePresenceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,19 +6,50 @@
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.isOneOf;
import static org.hamcrest.Matchers.not;
import static org.junit.Assert.*;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.IOException;
import java.util.*;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.List;
import java.util.UUID;

import com.google.gson.JsonArray;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import io.ably.lib.debug.DebugOptions;
import io.ably.lib.realtime.*;
import io.ably.lib.realtime.AblyRealtime;
import io.ably.lib.realtime.Channel;
import io.ably.lib.realtime.ChannelEvent;
import io.ably.lib.realtime.ChannelState;
import io.ably.lib.realtime.ChannelStateListener;
import io.ably.lib.realtime.CompletionListener;
import io.ably.lib.realtime.ConnectionEvent;
import io.ably.lib.realtime.ConnectionState;
import io.ably.lib.realtime.ConnectionStateListener;
import io.ably.lib.realtime.Presence;
import io.ably.lib.test.common.Setup;
import io.ably.lib.types.*;
import io.ably.lib.types.AblyException;
import io.ably.lib.types.Capability;
import io.ably.lib.types.ChannelOptions;
import io.ably.lib.types.ClientOptions;
import io.ably.lib.types.ErrorInfo;
import io.ably.lib.types.PaginatedResult;
import io.ably.lib.types.Param;
import io.ably.lib.types.PresenceMessage;
import io.ably.lib.types.ProtocolMessage;
import io.ably.lib.util.Serialisation;
import org.junit.Before;
import org.junit.Rule;
Expand All @@ -37,9 +68,7 @@
import io.ably.lib.test.common.ParameterizedTest;
import io.ably.lib.test.util.MockWebsocketFactory;
import io.ably.lib.transport.ConnectionManager;
import io.ably.lib.transport.Defaults;
import io.ably.lib.types.PresenceMessage.Action;
import io.ably.lib.util.Log;

public class RealtimePresenceTest extends ParameterizedTest {

Expand Down Expand Up @@ -1508,7 +1537,7 @@ public Presence.PresenceListener setMessageStack(List<PresenceMessage> messageSt
leavePresenceWaiter.waitFor(ably1.options.clientId, Action.leave);

/* Validate that,
* - we received all actions
*- we received all actions
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This change (and a few others) add noise. And, actually, in this case I think reduces readability of the comment.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't usually add unnecessary code. I think this was added as a part of checkstyle run.

*/
assertThat(receivedMessageStack.size(), is(equalTo(4)));
for (PresenceMessage message : receivedMessageStack) {
Expand Down Expand Up @@ -1586,7 +1615,7 @@ public void onPresenceMessage(PresenceMessage message) {
} catch(InterruptedException e) {}

/* Validate that,
* - we received specific actions
*- we received specific actions
*/
assertThat(receivedMessageStack.size(), is(equalTo(3)));
for (PresenceMessage message : receivedMessageStack) {
Expand Down Expand Up @@ -1663,7 +1692,7 @@ public Presence.PresenceListener setMessageStack(List<PresenceMessage> messageSt
waiter.waitFor(ably1.options.clientId, Action.leave);

/* Validate that,
* - we received specific actions
*- we received specific actions
*/
assertThat(receivedMessageStack, is(not(empty())));
for (PresenceMessage message : receivedMessageStack) {
Expand Down Expand Up @@ -2598,7 +2627,7 @@ public void presence_state_change () {
try {
DebugOptions opts = new DebugOptions(testVars.keys[0].keyStr);
fillInOptions(opts);
opts.autoConnect = false; /* to queue presence messages */
opts.autoConnect = false; /* to queue presence messages */
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why this change? It's unrelated to the fix you've made as far as I can see. Same goes for one below. Just adds noise for reviewers. Please try to keep PRs clean. 🙂

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Checkstyle related change


final MockWebsocketFactory mockTransport = new MockWebsocketFactory();
opts.transportFactory = mockTransport;
Expand Down Expand Up @@ -2819,7 +2848,7 @@ public void presence_enter_without_permission() throws AblyException {
/* get first token */
Auth.TokenParams tokenParams = new Auth.TokenParams();
Capability capability = new Capability();
capability.addResource(channelName, "publish"); /* no presence permission! */
capability.addResource(channelName, "publish"); /* no presence permission! */
tokenParams.capability = capability.toString();
tokenParams.clientId = testClientId1;

Expand Down Expand Up @@ -3244,6 +3273,68 @@ public void presence_get() throws AblyException, InterruptedException {
}
}

/**
* Test Presence.get()
* check if parent channel is able to detect presence
* during intermittent detach cycles
*/

public void checkMembersWithChannelPresence(Channel testChannel) throws AblyException {
PresenceMessage[] presenceMessages = testChannel.presence.get(true);
testChannel.detach();
assertEquals("Members count with channel presence should be " + presenceMessages.length, presenceMessages.length, 1);
}

@Test
public void test_consistent_presence_for_members() {
AblyRealtime clientAbly1 = null;
TestChannel testChannel = new TestChannel();
try {
/* subscribe for presence events in the anonymous connection */
PresenceWaiter presenceWaiter = new PresenceWaiter(testChannel.realtimeChannel);
/* set up a connection with specific clientId */
ClientOptions client1Opts = new ClientOptions() {{
tokenDetails = token1;
clientId = testClientId1;
}};
fillInOptions(client1Opts);
clientAbly1 = new AblyRealtime(client1Opts);

(new ConnectionWaiter(clientAbly1.connection)).waitFor(ConnectionState.connected);
assertEquals("Verify connected state reached", clientAbly1.connection.state, ConnectionState.connected);

Channel client1Channel = clientAbly1.channels.get(testChannel.channelName);
client1Channel.attach();
(new ChannelWaiter(client1Channel)).waitFor(ChannelState.attached);
assertEquals("Verify attached state reached", client1Channel.state, ChannelState.attached);

String enterString = "Entering presence from child channel";

CompletionWaiter enterComplete = new CompletionWaiter();
client1Channel.presence.enter(enterString, enterComplete);
enterComplete.waitFor();

presenceWaiter.waitFor(testClientId1, Action.enter);
assertNotNull(presenceWaiter.contains(testClientId1, Action.enter));
assertEquals(presenceWaiter.receivedMessages.get(0).data, enterString);

int parent_detach_cycle = 6;
for (int cycle = 0; cycle < parent_detach_cycle ; cycle++) {
Thread.sleep(1000);
checkMembersWithChannelPresence(testChannel.realtimeChannel);
}

} catch(AblyException | InterruptedException e) {
e.printStackTrace();
fail("Unexpected exception running test: " + e.getMessage());
} finally {
if(clientAbly1 != null)
clientAbly1.close();
if(testChannel != null)
testChannel.dispose();
}
}

/**
* Authenticate using wildcard token, initialize AblyRealtime so clientId is not known a priori,
* call enter() without attaching first, start connection
Expand Down