You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by bo...@apache.org on 2016/07/06 16:25:40 UTC
incubator-geode git commit: GEODE-1610: ClientSession register
interest now enqueues initial value
Repository: incubator-geode
Updated Branches:
refs/heads/develop cf30f6cc2 -> c065b6aef
GEODE-1610: ClientSession register interest now enqueues initial value
Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/c065b6ae
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/c065b6ae
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/c065b6ae
Branch: refs/heads/develop
Commit: c065b6aef47016a75a06c4bb206531f2178ddd4a
Parents: cf30f6c
Author: Barry Oglesby <bo...@pivotal.io>
Authored: Tue Jul 5 17:36:16 2016 -0700
Committer: Barry Oglesby <bo...@pivotal.io>
Committed: Tue Jul 5 17:37:42 2016 -0700
----------------------------------------------------------------------
.../cache/tier/sockets/CacheClientProxy.java | 105 +++---
.../tier/sockets/ClientInterestMessageImpl.java | 40 +++
.../cache/tier/sockets/CacheServerTestUtil.java | 43 ++-
.../sockets/DurableClientSimpleDUnitTest.java | 357 ++++++-------------
.../tier/sockets/DurableClientTestCase.java | 158 +++-----
5 files changed, 313 insertions(+), 390 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/c065b6ae/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheClientProxy.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheClientProxy.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheClientProxy.java
index c4b48f4..4269e7f 100755
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheClientProxy.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheClientProxy.java
@@ -1084,27 +1084,7 @@ public class CacheClientProxy implements ClientSession {
// Enqueue the initial value message for the client if necessary
if (policy == InterestResultPolicy.KEYS_VALUES) {
- Get70 request = (Get70)Get70.getCommand();
- LocalRegion lr = (LocalRegion) this._cache.getRegion(regionName);
- Get70.Entry entry = request.getValueAndIsObject(lr, keyOfInterest, null,
- null);
- boolean isObject = entry.isObject;
- byte[] value = null;
- if (entry.value instanceof byte[]) {
- value = (byte[])entry.value;
- } else {
- try {
- value = CacheServerHelper.serialize(entry.value);
- } catch (IOException e) {
- logger.warn(LocalizedMessage.create(LocalizedStrings.CacheClientProxy_THE_FOLLOWING_EXCEPTION_OCCURRED_0, entry.value), e);
- }
- }
- VersionTag tag = entry.versionTag;
- ClientUpdateMessage updateMessage = new ClientUpdateMessageImpl(
- EnumListenerEvent.AFTER_CREATE, lr, keyOfInterest, value, null,
- (isObject ? (byte) 0x01 : (byte) 0x00), null, this.proxyID,
- new EventID(this._cache.getDistributedSystem()), tag);
- CacheClientNotifier.routeSingleClientMessage(updateMessage, this.proxyID);
+ enqueueInitialValue(null, regionName, keyOfInterest);
}
// Add the client to the region's filters
//addFilterRegisteredClients(regionName, keyOfInterest);
@@ -1136,16 +1116,58 @@ public class CacheClientProxy implements ClientSession {
}
// Enqueue the interest registration message for the client.
+ enqueueInterestRegistrationMessage(message);
+ }
+
+ private void enqueueInitialValue(ClientInterestMessageImpl clientInterestMessage, String regionName, Object keyOfInterest) {
+ // Get the initial value
+ Get70 request = (Get70)Get70.getCommand();
+ LocalRegion lr = (LocalRegion) this._cache.getRegion(regionName);
+ Get70.Entry entry = request.getValueAndIsObject(lr, keyOfInterest, null, null);
+ boolean isObject = entry.isObject;
+ byte[] value = null;
+
+ // If the initial value is not null, add it to the client's queue
+ if (entry.value != null) {
+ if (entry.value instanceof byte[]) {
+ value = (byte[])entry.value;
+ } else {
+ try {
+ value = CacheServerHelper.serialize(entry.value);
+ } catch (IOException e) {
+ logger.warn(LocalizedMessage.create(LocalizedStrings.CacheClientProxy_THE_FOLLOWING_EXCEPTION_OCCURRED_0, entry.value), e);
+ }
+ }
+ VersionTag tag = entry.versionTag;
+
+ // Initialize the event id.
+ EventID eventId = null;
+ if (clientInterestMessage == null) {
+ // If the clientInterestMessage is null, create a new event id
+ eventId = new EventID(this._cache.getDistributedSystem());
+ } else {
+ // If the clientInterestMessage is not null, base the event id off its event id to fix GEM-794.
+ // This will cause the updateMessage created below to have the same event id as the one created
+ // in the primary.
+ eventId = new EventID(clientInterestMessage.getEventId(), 1);
+ }
+ ClientUpdateMessage updateMessage = new ClientUpdateMessageImpl(
+ EnumListenerEvent.AFTER_CREATE, lr, keyOfInterest, value, null,
+ (isObject ? (byte) 0x01 : (byte) 0x00), null, this.proxyID,
+ eventId, tag);
+ CacheClientNotifier.routeSingleClientMessage(updateMessage, this.proxyID);
+ }
+ }
+
+ private void enqueueInterestRegistrationMessage(ClientInterestMessageImpl message) {
+ // Enqueue the interest registration message for the client.
// If the client is not 7.0.1 or greater and the key of interest is a list,
// then create an individual message for each entry in the list since the
// client doesn't support a ClientInterestMessageImpl containing a list.
if (Version.GFE_701.compareTo(this.clientVersion) > 0
- && keyOfInterest instanceof List) {
- for (Iterator i = ((List) keyOfInterest).iterator(); i.hasNext();) {
- this._messageDispatcher.enqueueMessage(new ClientInterestMessageImpl(
- new EventID(this._cache.getDistributedSystem()), regionName,
- i.next(), interestType, policy.getOrdinal(), isDurable, !receiveValues,
- ClientInterestMessageImpl.REGISTER));
+ && message.getKeyOfInterest() instanceof List) {
+ for (Iterator i = ((List) message.getKeyOfInterest()).iterator(); i.hasNext();) {
+ this._messageDispatcher.enqueueMessage(new ClientInterestMessageImpl(message, i.next()));
}
} else {
this._messageDispatcher.enqueueMessage(message);
@@ -1205,20 +1227,7 @@ public class CacheClientProxy implements ClientSession {
}
// Enqueue the interest unregistration message for the client.
- // If the client is not 7.0.1 or greater and the key of interest is a list,
- // then create an individual message for each entry in the list since the
- // client doesn't support a ClientInterestMessageImpl containing a list.
- if (Version.GFE_701.compareTo(this.clientVersion) > 0
- && keyOfInterest instanceof List) {
- for (Iterator i = ((List) keyOfInterest).iterator(); i.hasNext();) {
- this._messageDispatcher.enqueueMessage(new ClientInterestMessageImpl(
- new EventID(this._cache.getDistributedSystem()), regionName,
- i.next(), interestType, (byte) 0, isDurable, !receiveValues,
- ClientInterestMessageImpl.UNREGISTER));
- }
- } else {
- this._messageDispatcher.enqueueMessage(message);
- }
+ enqueueInterestRegistrationMessage(message);
}
protected void notifySecondariesOfInterestChange(ClientInterestMessageImpl message) {
@@ -1410,6 +1419,7 @@ public class CacheClientProxy implements ClientSession {
/** sent by the cache client notifier when there is an interest registration change */
protected void processInterestMessage(ClientInterestMessageImpl message) {
+ // Register or unregister interest depending on the interest type
int interestType = message.getInterestType();
String regionName = message.getRegionName();
Object key = message.getKeyOfInterest();
@@ -1460,7 +1470,18 @@ public class CacheClientProxy implements ClientSession {
.append(InterestType.getString(message.getInterestType()));
logger.debug(buffer.toString());
}
- }
+ }
+
+ // Enqueue the interest message in this secondary proxy (fix for bug #52088)
+ enqueueInterestRegistrationMessage(message);
+
+ // Enqueue the initial value if the message is register on a key that is not a list (fix for bug #52088)
+ if (message.isRegister()
+ && message.getInterestType() == InterestType.KEY
+ && !(key instanceof List)
+ && InterestResultPolicy.fromOrdinal(message.getInterestResultPolicy()) == InterestResultPolicy.KEYS_VALUES) {
+ enqueueInitialValue(message, regionName, key);
+ }
}
private boolean postDeliverAuthCheckPassed(ClientUpdateMessage clientMessage) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/c065b6ae/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/ClientInterestMessageImpl.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/ClientInterestMessageImpl.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/ClientInterestMessageImpl.java
index c7fd61a..ee53ef2 100755
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/ClientInterestMessageImpl.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/ClientInterestMessageImpl.java
@@ -18,6 +18,7 @@
package com.gemstone.gemfire.internal.cache.tier.sockets;
import com.gemstone.gemfire.DataSerializer;
+import com.gemstone.gemfire.cache.CacheFactory;
import com.gemstone.gemfire.internal.DataSerializableFixedID;
import com.gemstone.gemfire.internal.Version;
import com.gemstone.gemfire.internal.cache.EventID;
@@ -112,6 +113,17 @@ public class ClientInterestMessageImpl implements ClientMessage {
this.action = action;
}
+ public ClientInterestMessageImpl(ClientInterestMessageImpl message, Object keyOfInterest) {
+ this.eventId = new EventID(CacheFactory.getAnyInstance().getDistributedSystem());
+ this.regionName = message.regionName;
+ this.keyOfInterest = keyOfInterest;
+ this.interestType = message.interestType;
+ this.interestResultPolicy = message.interestResultPolicy;
+ this.isDurable = message.isDurable;
+ this.forUpdatesAsInvalidates = message.forUpdatesAsInvalidates;
+ this.action = message.action;
+ }
+
/**
* Default constructor.
*/
@@ -231,6 +243,10 @@ public class ClientInterestMessageImpl implements ClientMessage {
return this.interestType;
}
+ public byte getInterestResultPolicy() {
+ return this.interestResultPolicy;
+ }
+
public boolean getIsDurable() {
return this.isDurable;
}
@@ -265,6 +281,30 @@ public class ClientInterestMessageImpl implements ClientMessage {
public void setLatestValue(Object value){
}
+
+ public String toString() {
+ return new StringBuilder()
+ .append(getClass().getSimpleName())
+ .append("[")
+ .append("eventId=")
+ .append(this.eventId)
+ .append("; regionName=")
+ .append(this.regionName)
+ .append("; keyOfInterest=")
+ .append(this.keyOfInterest)
+ .append("; isDurable=")
+ .append(this.isDurable)
+ .append("; forUpdatesAsInvalidates=")
+ .append(this.forUpdatesAsInvalidates)
+ .append("; interestType=")
+ .append(this.interestType)
+ .append("; interestResultPolicy=")
+ .append(this.interestResultPolicy)
+ .append("; action=")
+ .append(this.action)
+ .append("]")
+ .toString();
+ }
@Override
public Version[] getSerializationVersions() {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/c065b6ae/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheServerTestUtil.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheServerTestUtil.java b/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheServerTestUtil.java
index 3d9c039..b0520f8 100755
--- a/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheServerTestUtil.java
+++ b/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheServerTestUtil.java
@@ -26,6 +26,7 @@ import java.net.URL;
import java.util.Enumeration;
import java.util.Iterator;
import java.util.LinkedList;
+import java.util.List;
import java.util.Properties;
import com.gemstone.gemfire.cache.AttributesFactory;
@@ -505,14 +506,19 @@ public class CacheServerTestUtil extends JUnit4DistributedTestCase {
this.arg = ee.getCallbackArgument();
this.type = type;
}
- @Override
+ public boolean isCreate() {
+ return this.type == TYPE_CREATE;
+ }
public String toString() {
return "EventWrapper: event=" + event + ", type=" + type;
}
}
public static class ControlListener extends CacheListenerAdapter implements Declarable {
- public final LinkedList events = new LinkedList();
+ public final LinkedList<EventWrapper> events = new LinkedList();
+ public final LinkedList<EntryEvent> createEvents = new LinkedList();
+ public final LinkedList<EntryEvent> updateEvents = new LinkedList();
+ public final LinkedList<EntryEvent> destroyEvents = new LinkedList();
public final Object CONTROL_LOCK = new Object();
//added to test creation of cache from xml
@@ -521,10 +527,18 @@ public class CacheServerTestUtil extends JUnit4DistributedTestCase {
}
public boolean waitWhileNotEnoughEvents(long sleepMs, int eventCount) {
+ return waitWhileNotEnoughEvents(sleepMs, eventCount, -1);
+ }
+
+ public boolean waitWhileNotEnoughEvents(long sleepMs, int eventCount, int eventType) {
+ return waitWhileNotEnoughEvents(sleepMs, eventCount, getEvents(eventType));
+ }
+
+ public boolean waitWhileNotEnoughEvents(long sleepMs, int eventCount, List eventsToCheck) {
long maxMillis = System.currentTimeMillis() + sleepMs;
synchronized(this.CONTROL_LOCK) {
try {
- while (this.events.size() < eventCount) {
+ while (eventsToCheck.size() < eventCount) {
long waitMillis = maxMillis - System.currentTimeMillis();
if (waitMillis < 10) {
break;
@@ -534,14 +548,33 @@ public class CacheServerTestUtil extends JUnit4DistributedTestCase {
} catch (InterruptedException abort) {
fail("interrupted");
}
- return !this.events.isEmpty();
+ return !eventsToCheck.isEmpty();
} // synchronized
}
+ public List getEvents(int eventType) {
+ List eventsToCheck = null;
+ switch (eventType) {
+ case TYPE_CREATE:
+ eventsToCheck = this.createEvents;
+ break;
+ case TYPE_UPDATE:
+ eventsToCheck = this.updateEvents;
+ break;
+ case TYPE_DESTROY:
+ eventsToCheck = this.destroyEvents;
+ break;
+ default:
+ eventsToCheck = this.events;
+ }
+ return eventsToCheck;
+ }
+
@Override
public void afterCreate(EntryEvent e) {
synchronized(this.CONTROL_LOCK) {
this.events.add(new EventWrapper(e, TYPE_CREATE));
+ this.createEvents.add(e);
this.CONTROL_LOCK.notifyAll();
}
}
@@ -550,6 +583,7 @@ public class CacheServerTestUtil extends JUnit4DistributedTestCase {
public void afterUpdate(EntryEvent e) {
synchronized(this.CONTROL_LOCK) {
this.events.add(new EventWrapper(e, TYPE_UPDATE));
+ this.updateEvents.add(e);
this.CONTROL_LOCK.notifyAll();
}
}
@@ -566,6 +600,7 @@ public class CacheServerTestUtil extends JUnit4DistributedTestCase {
public void afterDestroy(EntryEvent e) {
synchronized(this.CONTROL_LOCK) {
this.events.add(new EventWrapper(e, TYPE_DESTROY));
+ this.destroyEvents.add(e);
this.CONTROL_LOCK.notifyAll();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/c065b6ae/geode-cq/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/DurableClientSimpleDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/DurableClientSimpleDUnitTest.java b/geode-cq/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/DurableClientSimpleDUnitTest.java
index f2afb1f..7bfd1eb 100644
--- a/geode-cq/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/DurableClientSimpleDUnitTest.java
+++ b/geode-cq/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/DurableClientSimpleDUnitTest.java
@@ -16,10 +16,16 @@
*/
package com.gemstone.gemfire.internal.cache.tier.sockets;
+import static com.gemstone.gemfire.internal.cache.tier.sockets.CacheServerTestUtil.TYPE_CREATE;
import static com.gemstone.gemfire.test.dunit.Assert.*;
import java.util.Iterator;
+import java.util.concurrent.TimeUnit;
+import com.gemstone.gemfire.cache.ClientSession;
+import com.gemstone.gemfire.internal.cache.ha.HARegionQueue;
+import com.gemstone.gemfire.internal.cache.ha.HARegionQueueStats;
+import com.jayway.awaitility.Awaitility;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.experimental.categories.Category;
@@ -100,37 +106,11 @@ public class DurableClientSimpleDUnitTest extends DurableClientTestCase {
// Publish some entries
final int numberOfEntries = 10;
- this.publisherClientVM.invoke(new CacheSerializableRunnable("Register interest") {
- @Override
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
+ publishEntries(numberOfEntries);
- // Publish some entries
- for (int i=0; i<numberOfEntries; i++) {
- String keyAndValue = String.valueOf(i);
- region.put(keyAndValue, keyAndValue);
- }
- }
- });
-
// Verify the durable client received the updates
- this.durableClientVM.invoke(new CacheSerializableRunnable("Verify updates") {
- @Override
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
+ verifyDurableClientEvents(this.durableClientVM, numberOfEntries);
- // Get the listener and wait for the appropriate number of events
- CacheServerTestUtil.ControlListener listener = (CacheServerTestUtil.ControlListener) region
- .getAttributes().getCacheListeners()[0];
- listener.waitWhileNotEnoughEvents(30000, numberOfEntries);
- assertEquals(numberOfEntries, listener.events.size());
- }
- });
-
// Stop the durable client
this.durableClientVM.invoke(() -> CacheServerTestUtil.closeCache());
@@ -288,37 +268,11 @@ public class DurableClientSimpleDUnitTest extends DurableClientTestCase {
// Publish some entries
final int numberOfEntries = 10;
- this.publisherClientVM.invoke(new CacheSerializableRunnable("Register interest") {
- @Override
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
-
- // Publish some entries
- for (int i=0; i<numberOfEntries; i++) {
- String keyAndValue = String.valueOf(i);
- region.put(keyAndValue, keyAndValue);
- }
- }
- });
+ publishEntries(numberOfEntries);
// Verify the durable client received the updates
- this.durableClientVM.invoke(new CacheSerializableRunnable("Verify updates") {
- @Override
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
+ verifyDurableClientEvents(this.durableClientVM, numberOfEntries);
- // Get the listener and wait for the appropriate number of events
- CacheServerTestUtil.ControlListener listener = (CacheServerTestUtil.ControlListener) region
- .getAttributes().getCacheListeners()[0];
- listener.waitWhileNotEnoughEvents(30000, numberOfEntries);
- assertEquals(numberOfEntries, listener.events.size());
- }
- });
-
// Stop the publisher client
this.publisherClientVM.invoke(() -> CacheServerTestUtil.closeCache());
@@ -467,20 +421,7 @@ public class DurableClientSimpleDUnitTest extends DurableClientTestCase {
// Publish some entries
final int numberOfEntries = 10;
- this.publisherClientVM.invoke(new CacheSerializableRunnable("Register interest") {
- @Override
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
-
- // Publish some entries
- for (int i=0; i<numberOfEntries; i++) {
- String keyAndValue = String.valueOf(i);
- region.put(keyAndValue, keyAndValue);
- }
- }
- });
+ publishEntries(numberOfEntries);
// Verify the durable client received the updates
this.durableClientVM.invoke(new CacheSerializableRunnable("Verify updates") {
@@ -519,20 +460,7 @@ public class DurableClientSimpleDUnitTest extends DurableClientTestCase {
});
// Publish some more entries
- this.publisherClientVM.invoke(new CacheSerializableRunnable("Register interest") {
- @Override
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
-
- // Publish some entries
- for (int i=0; i<numberOfEntries; i++) {
- String keyAndValue = String.valueOf(i);
- region.put(keyAndValue, keyAndValue);
- }
- }
- });
+ publishEntries(numberOfEntries);
// Re-start the durable client
this.durableClientVM.invoke(() -> CacheServerTestUtil.createCacheClient(getClientPool(NetworkUtils.getServerHostName(durableClientVM.getHost()), server1Port, server2Port, true), regionName, getClientDistributedSystemProperties(durableClientId), Boolean.TRUE));
@@ -682,53 +610,14 @@ public class DurableClientSimpleDUnitTest extends DurableClientTestCase {
// Publish some entries
final int numberOfEntries = 10;
- this.publisherClientVM.invoke(new CacheSerializableRunnable("Register interest") {
- @Override
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
-
- // Publish some entries
- for (int i=0; i<numberOfEntries; i++) {
- String keyAndValue = String.valueOf(i);
- region.put(keyAndValue, keyAndValue);
- }
- }
- });
+ publishEntries(numberOfEntries);
// Verify durable client 1 received the updates
- this.durableClientVM.invoke(new CacheSerializableRunnable("Verify updates") {
- @Override
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
-
- // Get the listener and wait for the appropriate number of events
- CacheServerTestUtil.ControlListener listener = (CacheServerTestUtil.ControlListener) region
- .getAttributes().getCacheListeners()[0];
- listener.waitWhileNotEnoughEvents(30000, numberOfEntries);
- assertEquals(numberOfEntries, listener.events.size());
- }
- });
+ verifyDurableClientEvents(this.durableClientVM, numberOfEntries);
// Verify durable client 2 received the updates
- durableClient2VM.invoke(new CacheSerializableRunnable("Verify updates") {
- @Override
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
+ verifyDurableClientEvents(durableClient2VM, numberOfEntries);
- // Get the listener and wait for the appropriate number of events
- CacheServerTestUtil.ControlListener listener = (CacheServerTestUtil.ControlListener) region
- .getAttributes().getCacheListeners()[0];
- listener.waitWhileNotEnoughEvents(30000, numberOfEntries);
- assertEquals(numberOfEntries, listener.events.size());
- }
- });
-
// ARB: Wait for queue ack to arrive at server.
try {
Thread.sleep(1000);
@@ -768,20 +657,7 @@ public class DurableClientSimpleDUnitTest extends DurableClientTestCase {
});
// Publish some more entries
- this.publisherClientVM.invoke(new CacheSerializableRunnable("Register interest") {
- @Override
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
-
- // Publish some entries
- for (int i=0; i<numberOfEntries; i++) {
- String keyAndValue = String.valueOf(i);
- region.put(keyAndValue, keyAndValue);
- }
- }
- });
+ publishEntries(numberOfEntries);
try {
Thread.sleep(1000);
@@ -830,36 +706,10 @@ public class DurableClientSimpleDUnitTest extends DurableClientTestCase {
});
// Verify durable client 1 received the updates held for it on the server
- this.durableClientVM.invoke(new CacheSerializableRunnable("Verify updates") {
- @Override
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
-
- // Get the listener and wait for the appropriate number of events
- CacheServerTestUtil.ControlListener listener = (CacheServerTestUtil.ControlListener) region
- .getAttributes().getCacheListeners()[0];
- listener.waitWhileNotEnoughEvents(30000, numberOfEntries);
- assertEquals(numberOfEntries, listener.events.size());
- }
- });
+ verifyDurableClientEvents(this.durableClientVM, numberOfEntries);
// Verify durable client 2 received the updates held for it on the server
- durableClient2VM.invoke(new CacheSerializableRunnable("Verify updates") {
- @Override
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
-
- // Get the listener and wait for the appropriate number of events
- CacheServerTestUtil.ControlListener listener = (CacheServerTestUtil.ControlListener) region
- .getAttributes().getCacheListeners()[0];
- listener.waitWhileNotEnoughEvents(30000, numberOfEntries);
- assertEquals(numberOfEntries, listener.events.size());
- }
- });
+ verifyDurableClientEvents(durableClient2VM, numberOfEntries);
// Stop durable client 1
this.durableClientVM.invoke(() -> CacheServerTestUtil.closeCache());
@@ -965,20 +815,7 @@ public class DurableClientSimpleDUnitTest extends DurableClientTestCase {
// Publish some entries
final int numberOfEntries = 10;
- this.publisherClientVM.invoke(new CacheSerializableRunnable("Register interest") {
- @Override
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
-
- // Publish some entries
- for (int i=0; i<numberOfEntries; i++) {
- String keyAndValue = String.valueOf(i);
- region.put(keyAndValue, keyAndValue);
- }
- }
- });
+ publishEntries(numberOfEntries);
try {
Thread.sleep(1000);
@@ -1050,20 +887,7 @@ public class DurableClientSimpleDUnitTest extends DurableClientTestCase {
assertEquals(server1HARegionQueueName, server2HARegionQueueName);
// Verify the durable client received the updates
- this.durableClientVM.invoke(new CacheSerializableRunnable("Verify updates") {
- @Override
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
-
- // Get the listener and wait for the appropriate number of events
- CacheServerTestUtil.ControlListener listener = (CacheServerTestUtil.ControlListener) region
- .getAttributes().getCacheListeners()[0];
- listener.waitWhileNotEnoughEvents(30000, numberOfEntries);
- assertEquals(numberOfEntries, listener.events.size());
- }
- });
+ verifyDurableClientEvents(this.durableClientVM, numberOfEntries);
// Stop the durable client
this.durableClientVM.invoke(() -> CacheServerTestUtil.closeCache());
@@ -1235,20 +1059,7 @@ public class DurableClientSimpleDUnitTest extends DurableClientTestCase {
// Publish some entries
final int numberOfEntries = 10;
- this.publisherClientVM.invoke(new CacheSerializableRunnable("publish updates") {
- @Override
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
-
- // Publish some entries
- for (int i=0; i<numberOfEntries; i++) {
- String keyAndValue = String.valueOf(i);
- region.put(keyAndValue, keyAndValue);
- }
- }
- });
+ publishEntries(numberOfEntries);
// Verify the durable client received the updates
this.durableClientVM.invoke(new CacheSerializableRunnable("Verify updates") {
@@ -1288,20 +1099,7 @@ public class DurableClientSimpleDUnitTest extends DurableClientTestCase {
});
// Publish some more entries
- this.publisherClientVM.invoke(new CacheSerializableRunnable("Publish additional updates") {
- @Override
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
-
- // Publish some entries
- for (int i=0; i<numberOfEntries; i++) {
- String keyAndValue = String.valueOf(i);
- region.put(keyAndValue, keyAndValue + "lkj");
- }
- }
- });
+ publishEntries(numberOfEntries);
this.publisherClientVM.invoke(() -> CacheServerTestUtil.closeCache());
@@ -1491,20 +1289,7 @@ public class DurableClientSimpleDUnitTest extends DurableClientTestCase {
// Publish some entries
final int numberOfEntries = 10;
- this.publisherClientVM.invoke(new CacheSerializableRunnable("publish updates") {
- @Override
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
-
- // Publish some entries
- for (int i=0; i<numberOfEntries; i++) {
- String keyAndValue = String.valueOf(i);
- region.put(keyAndValue, keyAndValue);
- }
- }
- });
+ publishEntries(numberOfEntries);
// Verify the durable client received the updates
this.durableClientVM.invoke(new CacheSerializableRunnable("Verify updates") {
@@ -3459,5 +3244,99 @@ public class DurableClientSimpleDUnitTest extends DurableClientTestCase {
// Stop server 2
this.server2VM.invoke(() -> CacheServerTestUtil.closeCache());
}
-
+
+ @Test
+ public void testDurableClientReceivedClientSessionInitialValue() {
+ // Start server 1
+ int server1Port = this.server1VM.invoke(() -> CacheServerTestUtil.createCacheServer(regionName, new Boolean(true)));
+
+ // Start server 2
+ int server2Port = this.server2VM.invoke(() -> CacheServerTestUtil.createCacheServer(regionName, new Boolean(true)));
+
+ // Start normal publisher client
+ this.publisherClientVM.invoke(() -> CacheServerTestUtil.createCacheClient(
+ getClientPool(NetworkUtils.getServerHostName(this.publisherClientVM.getHost()), server1Port, server2Port, false),
+ this.regionName));
+
+ // Create an entry
+ publishEntries(1);
+
+ // Start a durable client with the ControlListener
+ final String durableClientId = getName() + "_client";
+ final int durableClientTimeout = 60; // keep the client alive for 60 seconds
+ restartDurableClient(new Object[] {
+ getClientPool(NetworkUtils.getServerHostName(this.durableClientVM.getHost()),server1Port, server2Port, true),
+ regionName,
+ getClientDistributedSystemProperties(durableClientId, durableClientTimeout),
+ true
+ });
+
+ // Use ClientSession on the server to register interest in entry key on behalf of durable client
+ boolean server1IsPrimary=false, server2IsPrimary=false;
+ boolean registered = this.server1VM
+ .invoke(() -> DurableClientSimpleDUnitTest.registerInterestWithClientSession(durableClientId, this.regionName, String.valueOf(0)));
+ if (registered) {
+ server1IsPrimary = true;
+ } else {
+ registered = this.server2VM
+ .invoke(() -> DurableClientSimpleDUnitTest.registerInterestWithClientSession(durableClientId, this.regionName, String.valueOf(0)));
+ if (registered) {
+ server2IsPrimary = true;
+ } else {
+ fail("ClientSession interest registration failed to occur in either server.");
+ }
+ }
+
+ // Verify durable client received create event
+ verifyDurableClientEvents(this.durableClientVM, 1, TYPE_CREATE);
+
+ // Wait for QRM to be processed on the secondary
+ waitForEventsRemovedByQueueRemovalMessage(server1IsPrimary ? this.server2VM : this.server1VM, durableClientId, 2);
+
+ // Stop durable client
+ disconnectDurableClient(true);
+
+ // Restart durable client
+ restartDurableClient(new Object[] {
+ getClientPool(NetworkUtils.getServerHostName(this.durableClientVM.getHost()),server1Port, server2Port, true),
+ regionName,
+ getClientDistributedSystemProperties(durableClientId, durableClientTimeout),
+ true
+ });
+
+ // Verify durable client does not receive create event
+ verifyNoDurableClientEvents(this.durableClientVM, 1, TYPE_CREATE);
+
+ // Stop the durable client
+ this.durableClientVM.invoke(() -> CacheServerTestUtil.closeCache());
+
+ // Stop server 1
+ this.server1VM.invoke(() -> CacheServerTestUtil.closeCache());
+
+ // Stop server 2
+ this.server2VM.invoke(() -> CacheServerTestUtil.closeCache());
+ }
+
+ public static boolean registerInterestWithClientSession(String durableClientId, String regionName, Object keyOfInterest) {
+ ClientSession session = getBridgeServer().getClientSession(durableClientId);
+ boolean registered = false;
+ if (session.isPrimary()) {
+ session.registerInterest(regionName, keyOfInterest, InterestResultPolicy.KEYS_VALUES, true, true);
+ registered = true;
+ }
+ return registered;
+ }
+
+ private void waitForEventsRemovedByQueueRemovalMessage(VM secondaryServerVM, final String durableClientId, final int numEvents) {
+ secondaryServerVM.invoke(() -> DurableClientSimpleDUnitTest.waitForEventsRemovedByQueueRemovalMessage(durableClientId, numEvents));
+ }
+
+ private static void waitForEventsRemovedByQueueRemovalMessage(String durableClientId, final int numEvents) {
+ CacheClientNotifier ccn = CacheClientNotifier.getInstance();
+ CacheClientProxy ccp = ccn.getClientProxy(durableClientId);
+ HARegionQueue harq = ccp.getHARegionQueue();
+ HARegionQueueStats harqStats = harq.getStatistics();
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> assertEquals("Expected queue removal messages: " +
+ numEvents + " but actual messages: " + harqStats.getEventsRemovedByQrm(), numEvents, harqStats.getEventsRemovedByQrm()));
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/c065b6ae/geode-cq/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/DurableClientTestCase.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/DurableClientTestCase.java b/geode-cq/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/DurableClientTestCase.java
index cc54a47..b1dcc58 100755
--- a/geode-cq/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/DurableClientTestCase.java
+++ b/geode-cq/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/DurableClientTestCase.java
@@ -594,19 +594,7 @@ public class DurableClientTestCase extends JUnit4DistributedTestCase {
// Publish some entries
final int numberOfEntries = 1;
- this.publisherClientVM.invoke(new CacheSerializableRunnable("Publish entries") {
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
-
- // Publish some entries
- for (int i=0; i<numberOfEntries; i++) {
- String keyAndValue = String.valueOf(i);
- region.put(keyAndValue, keyAndValue);
- }
- }
- });
+ publishEntries(numberOfEntries);
// Verify the durable client received the updates
this.verifyListenerUpdates(numberOfEntries);
@@ -639,20 +627,8 @@ public class DurableClientTestCase extends JUnit4DistributedTestCase {
});
// Publish some more entries
- this.publisherClientVM.invoke(new CacheSerializableRunnable("Register interest") {
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
+ publishEntries(numberOfEntries);
- // Publish some entries
- for (int i=0; i<numberOfEntries; i++) {
- String keyAndValue = String.valueOf(i);
- region.put(keyAndValue, keyAndValue);
- }
- }
- });
-
try {
Thread.sleep(1000);
} catch (InterruptedException ex) {
@@ -722,7 +698,23 @@ public class DurableClientTestCase extends JUnit4DistributedTestCase {
// Stop the server
this.server1VM.invoke(() -> CacheServerTestUtil.closeCache());
}
-
+
+ protected void publishEntries(final int numberOfEntries) {
+ this.publisherClientVM.invoke(new CacheSerializableRunnable("Publish entries") {
+ public void run2() throws CacheException {
+ // Get the region
+ Region region = CacheServerTestUtil.getCache().getRegion(regionName);
+ assertNotNull(region);
+
+ // Publish some entries
+ for (int i=0; i<numberOfEntries; i++) {
+ String keyAndValue = String.valueOf(i);
+ region.put(keyAndValue, keyAndValue);
+ }
+ }
+ });
+ }
+
/**
* Test that a durable client correctly receives updates after it reconnects.
*/
@@ -763,19 +755,7 @@ public class DurableClientTestCase extends JUnit4DistributedTestCase {
// Publish some entries
final int numberOfEntries = 1;
- this.publisherClientVM.invoke(new CacheSerializableRunnable("Publish entries") {
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
-
- // Publish some entries
- for (int i=0; i<numberOfEntries; i++) {
- String keyAndValue = String.valueOf(i);
- region.put(keyAndValue, keyAndValue);
- }
- }
- });
+ publishEntries(numberOfEntries);
// Verify the durable client received the updates
this.verifyListenerUpdates(numberOfEntries);
@@ -1034,19 +1014,7 @@ public class DurableClientTestCase extends JUnit4DistributedTestCase {
});
// Verify the durable client received the updates
- this.durableClientVM.invoke(new CacheSerializableRunnable("Verify updates") {
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
-
- // Get the listener and wait for the appropriate number of events
- CacheServerTestUtil.ControlListener listener = (CacheServerTestUtil.ControlListener) region
- .getAttributes().getCacheListeners()[0];
- listener.waitWhileNotEnoughEvents(30000, numberOfEntries);
- assertEquals(numberOfEntries, listener.events.size());
- }
- });
+ verifyDurableClientEvents(this.durableClientVM, numberOfEntries);
// Stop the durable client
this.durableClientVM.invoke(() -> CacheServerTestUtil.closeCache());
@@ -1058,6 +1026,34 @@ public class DurableClientTestCase extends JUnit4DistributedTestCase {
this.server1VM.invoke(() -> CacheServerTestUtil.closeCache());
}
+ protected void verifyDurableClientEvents(VM durableClientVm, final int numberOfEntries) {
+ verifyDurableClientEvents(durableClientVm, numberOfEntries, -1);
+ }
+
+ protected void verifyDurableClientEvents(VM durableClientVm, final int numberOfEntries, final int eventType) {
+ verifyDurableClientEvents(durableClientVm, numberOfEntries, eventType, numberOfEntries);
+ }
+
+ protected void verifyNoDurableClientEvents(VM durableClientVm, final int numberOfEntries, final int eventType) {
+ verifyDurableClientEvents(durableClientVm, numberOfEntries, eventType, 0);
+ }
+
+ private void verifyDurableClientEvents(VM durableClientVm, final int numberOfEntries, final int eventType, final int expectedNumberOfEvents) {
+ durableClientVm.invoke(new CacheSerializableRunnable("Verify updates") {
+ public void run2() throws CacheException {
+ // Get the region
+ Region region = CacheServerTestUtil.getCache().getRegion(regionName);
+ assertNotNull(region);
+
+ // Get the listener and wait for the appropriate number of events
+ CacheServerTestUtil.ControlListener listener = (CacheServerTestUtil.ControlListener) region
+ .getAttributes().getCacheListeners()[0];
+ listener.waitWhileNotEnoughEvents(30000, numberOfEntries, eventType);
+ assertEquals(expectedNumberOfEvents, listener.getEvents(eventType).size());
+ }
+ });
+ }
+
@Ignore("TODO: This test is failing inconsistently, see bug 51258")
@Test
public void testDurableNonHAFailover() throws InterruptedException {
@@ -1135,19 +1131,7 @@ public class DurableClientTestCase extends JUnit4DistributedTestCase {
// Publish some entries
final int numberOfEntries = 1;
- this.publisherClientVM.invoke(new CacheSerializableRunnable("Publish entries") {
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
-
- // Publish some entries
- for (int i=0; i<numberOfEntries; i++) {
- String keyAndValue = String.valueOf(i);
- region.put(keyAndValue, keyAndValue);
- }
- }
- });
+ publishEntries(numberOfEntries);
// Verify the durable client received the updates
this.verifyListenerUpdates(numberOfEntries);
@@ -1164,19 +1148,7 @@ public class DurableClientTestCase extends JUnit4DistributedTestCase {
this.disconnectDurableClient(true);
// Publish updates during client downtime
- this.publisherClientVM.invoke(new CacheSerializableRunnable("Publish entries") {
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
-
- // Publish some entries
- for (int i=0; i<numberOfEntries; i++) {
- String keyAndValue = String.valueOf(i);
- region.put(keyAndValue, keyAndValue);
- }
- }
- });
+ publishEntries(numberOfEntries);
// Re-start the durable client that is kept alive on the server
@@ -1322,19 +1294,7 @@ public class DurableClientTestCase extends JUnit4DistributedTestCase {
// Publish some entries
final int numberOfEntries = 1;
- this.publisherClientVM.invoke(new CacheSerializableRunnable("Publish entries") {
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
-
- // Publish some entries
- for (int i=0; i<numberOfEntries; i++) {
- String keyAndValue = String.valueOf(i);
- region.put(keyAndValue, keyAndValue);
- }
- }
- });
+ publishEntries(numberOfEntries);
try {
java.lang.Thread.sleep(10000);
@@ -1365,19 +1325,7 @@ public class DurableClientTestCase extends JUnit4DistributedTestCase {
this.server1VM.invoke(() -> CacheServerTestUtil.closeCache());
// Publish updates during client downtime
- this.publisherClientVM.invoke(new CacheSerializableRunnable("Publish entries") {
- public void run2() throws CacheException {
- // Get the region
- Region region = CacheServerTestUtil.getCache().getRegion(regionName);
- assertNotNull(region);
-
- // Publish some entries
- for (int i=0; i<numberOfEntries; i++) {
- String keyAndValue = String.valueOf(i);
- region.put(keyAndValue, keyAndValue);
- }
- }
- });
+ publishEntries(numberOfEntries);
// Re-start the durable client that is kept alive on the server
this.restartDurableClient(new Object[] {