geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From r..@apache.org
Subject [26/51] [partial] incubator-geode git commit: WAN and CQ code drop under the Pivotal SGA
Date Wed, 20 Jan 2016 02:22:32 GMT
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/security/ClientPostAuthorizationDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/security/ClientPostAuthorizationDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/security/ClientPostAuthorizationDUnitTest.java
new file mode 100644
index 0000000..91df070
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/security/ClientPostAuthorizationDUnitTest.java
@@ -0,0 +1,390 @@
+/*=========================================================================
+ * Copyright (c) 2010-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * one or more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+package com.gemstone.gemfire.security;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+import java.util.Random;
+
+import security.AuthzCredentialGenerator;
+import security.CredentialGenerator;
+import com.gemstone.gemfire.cache.operations.OperationContext.OperationCode;
+import com.gemstone.gemfire.internal.AvailablePort;
+
+import dunit.Host;
+
+/**
+ * Tests for authorization from client to server. This tests for authorization
+ * with post-process callbacks in case return values of operations and for
+ * notifications along-with failover.
+ * 
+ * @author sumedh
+ * @since 5.5
+ */
+public class ClientPostAuthorizationDUnitTest extends
+    ClientAuthorizationTestBase {
+
+
+  /** constructor */
+  public ClientPostAuthorizationDUnitTest(String name) {
+    super(name);
+  }
+
+  public void setUp() throws Exception {
+
+    super.setUp();
+    final Host host = Host.getHost(0);
+    server1 = host.getVM(0);
+    server2 = host.getVM(1);
+    client1 = host.getVM(2);
+    client2 = host.getVM(3);
+
+    server1.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+        new Object[] { serverExpectedExceptions });
+    server2.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+        new Object[] { serverExpectedExceptions });
+    client2.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+        new Object[] { clientExpectedExceptions });
+    SecurityTestUtil.registerExpectedExceptions(clientExpectedExceptions);
+  }
+
+  // Region: Tests
+
+  public void testAllPostOps() {
+
+    OperationWithAction[] allOps = {
+        // Test CREATE and verify with a GET
+        new OperationWithAction(OperationCode.PUT),
+        new OperationWithAction(OperationCode.GET, 2, OpFlags.CHECK_NOKEY, 4),
+        new OperationWithAction(OperationCode.GET, 3, OpFlags.CHECK_NOKEY
+            | OpFlags.CHECK_NOTAUTHZ, 4),
+
+        // OPBLOCK_END indicates end of an operation block that needs to
+        // be executed on each server when doing failover
+        OperationWithAction.OPBLOCK_END,
+
+        // Test UPDATE and verify with a GET
+        new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN
+            | OpFlags.USE_NEWVAL, 4),
+        new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+            | OpFlags.USE_NEWVAL, 4),
+        new OperationWithAction(OperationCode.GET, 3,
+            OpFlags.USE_OLDCONN | OpFlags.CHECK_NOKEY | OpFlags.USE_NEWVAL
+                | OpFlags.CHECK_NOTAUTHZ, 4),
+
+        OperationWithAction.OPBLOCK_END,
+
+        // Test UPDATE and verify with a KEY_SET
+        new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN, 6),
+        new OperationWithAction(OperationCode.KEY_SET, 2, OpFlags.NONE, 6),
+        new OperationWithAction(OperationCode.KEY_SET, 3,
+            OpFlags.CHECK_NOTAUTHZ, 6),
+
+        OperationWithAction.OPBLOCK_END,
+
+        // Test UPDATE and verify with a QUERY
+        new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN
+            | OpFlags.USE_NEWVAL, 7),
+        new OperationWithAction(OperationCode.QUERY, 2, OpFlags.USE_NEWVAL, 7),
+        new OperationWithAction(OperationCode.QUERY, 3, OpFlags.USE_NEWVAL
+            | OpFlags.CHECK_NOTAUTHZ, 7),
+
+        OperationWithAction.OPBLOCK_END,
+
+        // Test UPDATE and verify with a EXECUTE_CQ initial results
+        new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN, 8),
+        new OperationWithAction(OperationCode.EXECUTE_CQ, 2, OpFlags.NONE, 8),
+        new OperationWithAction(OperationCode.EXECUTE_CQ, 3,
+            OpFlags.CHECK_NOTAUTHZ, 8),
+
+        OperationWithAction.OPBLOCK_END };
+
+    Iterator iter = getDummyGeneratorCombos().iterator();
+    while (iter.hasNext()) {
+      AuthzCredentialGenerator gen = (AuthzCredentialGenerator)iter.next();
+      CredentialGenerator cGen = gen.getCredentialGenerator();
+      Properties extraAuthProps = cGen.getSystemProperties();
+      Properties javaProps = cGen.getJavaProperties();
+      Properties extraAuthzProps = gen.getSystemProperties();
+      String authenticator = cGen.getAuthenticator();
+      String authInit = cGen.getAuthInit();
+      String accessor = gen.getAuthorizationCallback();
+      TestAuthzCredentialGenerator tgen = new TestAuthzCredentialGenerator(gen);
+
+      getLogWriter().info("testAllPostOps: Using authinit: " + authInit);
+      getLogWriter().info(
+          "testAllPostOps: Using authenticator: " + authenticator);
+      getLogWriter().info("testAllPostOps: Using accessor: " + accessor);
+
+      // Start servers with all required properties
+      Properties serverProps = buildProperties(authenticator, accessor, true,
+          extraAuthProps, extraAuthzProps);
+      // Get ports for the servers
+      Integer port1 = new Integer(AvailablePort
+          .getRandomAvailablePort(AvailablePort.SOCKET));
+      Integer port2 = new Integer(AvailablePort
+          .getRandomAvailablePort(AvailablePort.SOCKET));
+
+      // Close down any running servers
+      server1.invoke(SecurityTestUtil.class, "closeCache");
+      server2.invoke(SecurityTestUtil.class, "closeCache");
+
+      // Perform all the ops on the clients
+      List opBlock = new ArrayList();
+      Random rnd = new Random();
+      for (int opNum = 0; opNum < allOps.length; ++opNum) {
+        // Start client with valid credentials as specified in
+        // OperationWithAction
+        OperationWithAction currentOp = allOps[opNum];
+        if (currentOp.equals(OperationWithAction.OPBLOCK_END)
+            || currentOp.equals(OperationWithAction.OPBLOCK_NO_FAILOVER)) {
+          // End of current operation block; execute all the operations
+          // on the servers with failover
+          if (opBlock.size() > 0) {
+            // Start the first server and execute the operation block
+            server1.invoke(ClientAuthorizationTestBase.class,
+                "createCacheServer", new Object[] {
+                    SecurityTestUtil.getLocatorPort(), port1, serverProps,
+                    javaProps });
+            server2.invoke(SecurityTestUtil.class, "closeCache");
+            executeOpBlock(opBlock, port1, port2, authInit, extraAuthProps,
+                extraAuthzProps, tgen, rnd);
+            if (!currentOp.equals(OperationWithAction.OPBLOCK_NO_FAILOVER)) {
+              // Failover to the second server and run the block again
+              server2.invoke(ClientAuthorizationTestBase.class,
+                  "createCacheServer", new Object[] {
+                      SecurityTestUtil.getLocatorPort(), port2, serverProps,
+                      javaProps });
+              server1.invoke(SecurityTestUtil.class, "closeCache");
+              executeOpBlock(opBlock, port1, port2, authInit, extraAuthProps,
+                  extraAuthzProps, tgen, rnd);
+            }
+            opBlock.clear();
+          }
+        }
+        else {
+          currentOp.setOpNum(opNum);
+          opBlock.add(currentOp);
+        }
+      }
+    }
+  }
+
+  public void testAllOpsNotifications() {
+
+    OperationWithAction[] allOps = {
+        // Test CREATE and verify with a GET
+        new OperationWithAction(OperationCode.REGISTER_INTEREST,
+            OperationCode.GET, 2, OpFlags.USE_REGEX
+                | OpFlags.REGISTER_POLICY_NONE, 8),
+        new OperationWithAction(OperationCode.REGISTER_INTEREST,
+            OperationCode.GET, 3, OpFlags.USE_REGEX
+                | OpFlags.REGISTER_POLICY_NONE | OpFlags.USE_NOTAUTHZ, 8),
+        new OperationWithAction(OperationCode.PUT),
+        new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+            | OpFlags.LOCAL_OP, 4),
+        new OperationWithAction(OperationCode.GET, 3, OpFlags.USE_OLDCONN
+            | OpFlags.LOCAL_OP | OpFlags.CHECK_FAIL, 4),
+
+        // OPBLOCK_END indicates end of an operation block that needs to
+        // be executed on each server when doing failover
+        OperationWithAction.OPBLOCK_END,
+
+        // Test UPDATE and verify with a GET
+        new OperationWithAction(OperationCode.REGISTER_INTEREST,
+            OperationCode.GET, 2, OpFlags.USE_REGEX
+                | OpFlags.REGISTER_POLICY_NONE, 8),
+        new OperationWithAction(OperationCode.REGISTER_INTEREST,
+            OperationCode.GET, 3, OpFlags.USE_REGEX
+                | OpFlags.REGISTER_POLICY_NONE | OpFlags.USE_NOTAUTHZ, 8),
+        new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN
+            | OpFlags.USE_NEWVAL, 4),
+        new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+            | OpFlags.LOCAL_OP | OpFlags.USE_NEWVAL, 4),
+        new OperationWithAction(OperationCode.GET, 3, OpFlags.USE_OLDCONN
+            | OpFlags.LOCAL_OP | OpFlags.USE_NEWVAL | OpFlags.CHECK_FAIL, 4),
+
+        OperationWithAction.OPBLOCK_END,
+
+        // Test DESTROY and verify with GET that keys should not exist
+        new OperationWithAction(OperationCode.PUT, 3, OpFlags.NONE, 8),
+        new OperationWithAction(OperationCode.REGISTER_INTEREST,
+            OperationCode.GET, 2, OpFlags.USE_REGEX, 8),
+        new OperationWithAction(OperationCode.REGISTER_INTEREST, 3,
+            OpFlags.USE_REGEX | OpFlags.USE_OLDCONN | OpFlags.REGISTER_POLICY_NONE, 8),
+        // registerInterest now clears the keys, so a dummy put to add
+        // those keys back for the case when updates should not come
+        new OperationWithAction(OperationCode.PUT, 3, OpFlags.USE_OLDCONN, 8),
+        new OperationWithAction(OperationCode.DESTROY, 1, OpFlags.USE_OLDCONN,
+            4),
+        new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+            | OpFlags.LOCAL_OP | OpFlags.CHECK_FAIL, 4),
+        new OperationWithAction(OperationCode.GET, 3, OpFlags.USE_OLDCONN
+            | OpFlags.LOCAL_OP, 4),
+        // Repopulate the region
+        new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN, 8),
+
+        OperationWithAction.OPBLOCK_END,
+
+        // Do REGION_CLEAR and check with GET
+        new OperationWithAction(OperationCode.PUT, 3, OpFlags.NONE, 8),
+        new OperationWithAction(OperationCode.REGISTER_INTEREST,
+            OperationCode.GET, 2, OpFlags.USE_ALL_KEYS, 1),
+        new OperationWithAction(OperationCode.REGISTER_INTEREST, 3,
+            OpFlags.USE_ALL_KEYS | OpFlags.USE_OLDCONN | OpFlags.REGISTER_POLICY_NONE, 1),
+        // registerInterest now clears the keys, so a dummy put to add
+        // those keys back for the case when updates should not come
+        new OperationWithAction(OperationCode.PUT, 3, OpFlags.USE_OLDCONN, 8),
+        new OperationWithAction(OperationCode.REGION_CLEAR, 1,
+            OpFlags.USE_OLDCONN, 1),
+        new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+            | OpFlags.LOCAL_OP | OpFlags.CHECK_FAIL, 8),
+        new OperationWithAction(OperationCode.GET, 3, OpFlags.USE_OLDCONN
+            | OpFlags.LOCAL_OP, 8),
+        // Repopulate the region
+        new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN, 8),
+
+        OperationWithAction.OPBLOCK_END,
+
+        // Do REGION_CREATE and check with CREATE/GET
+        new OperationWithAction(OperationCode.REGISTER_INTEREST,
+            OperationCode.GET, 2, OpFlags.USE_ALL_KEYS | OpFlags.ENABLE_DRF, 1),
+        new OperationWithAction(OperationCode.REGISTER_INTEREST,
+            OperationCode.GET, 3, OpFlags.USE_ALL_KEYS | OpFlags.ENABLE_DRF
+                | OpFlags.USE_NOTAUTHZ | OpFlags.REGISTER_POLICY_NONE, 1),
+        new OperationWithAction(OperationCode.REGION_CREATE, 1,
+            OpFlags.ENABLE_DRF, 1),
+        new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN
+            | OpFlags.USE_SUBREGION, 4),
+        new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+            | OpFlags.LOCAL_OP | OpFlags.USE_SUBREGION
+            | OpFlags.NO_CREATE_SUBREGION, 4),
+        new OperationWithAction(OperationCode.GET, 3, OpFlags.USE_OLDCONN
+            | OpFlags.LOCAL_OP | OpFlags.USE_SUBREGION
+            | OpFlags.NO_CREATE_SUBREGION | OpFlags.CHECK_NOREGION, 4),
+
+        // Do REGION_DESTROY of the sub-region and check with GET
+        new OperationWithAction(OperationCode.REGION_DESTROY, 1,
+            OpFlags.USE_OLDCONN | OpFlags.USE_SUBREGION, 1),
+        new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+            | OpFlags.LOCAL_OP | OpFlags.USE_SUBREGION
+            | OpFlags.NO_CREATE_SUBREGION | OpFlags.CHECK_NOREGION, 4),
+        new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+            | OpFlags.USE_SUBREGION | OpFlags.CHECK_NOKEY
+            | OpFlags.CHECK_EXCEPTION, 4),
+        new OperationWithAction(OperationCode.GET, 3, OpFlags.USE_OLDCONN
+            | OpFlags.LOCAL_OP | OpFlags.USE_SUBREGION
+            | OpFlags.NO_CREATE_SUBREGION | OpFlags.CHECK_NOREGION, 4),
+        new OperationWithAction(OperationCode.GET, 3, OpFlags.USE_OLDCONN
+            | OpFlags.USE_SUBREGION | OpFlags.CHECK_NOKEY
+            | OpFlags.CHECK_EXCEPTION, 4),
+
+        OperationWithAction.OPBLOCK_END,
+
+        // Do REGION_DESTROY of the region and check with GET
+        new OperationWithAction(OperationCode.PUT, 3, OpFlags.NONE, 8),
+        new OperationWithAction(OperationCode.REGISTER_INTEREST,
+            OperationCode.GET, 2, OpFlags.USE_ALL_KEYS, 1),
+        new OperationWithAction(OperationCode.REGISTER_INTEREST, 3,
+            OpFlags.USE_ALL_KEYS | OpFlags.USE_OLDCONN | OpFlags.REGISTER_POLICY_NONE, 1),
+        // registerInterest now clears the keys, so a dummy put to add
+        // those keys back for the case when updates should not come
+        new OperationWithAction(OperationCode.PUT, 3, OpFlags.USE_OLDCONN, 8),
+        new OperationWithAction(OperationCode.REGION_DESTROY, 1, OpFlags.NONE,
+            1),
+        new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+            | OpFlags.LOCAL_OP | OpFlags.CHECK_NOREGION, 4),
+        new OperationWithAction(OperationCode.GET, 3, OpFlags.USE_OLDCONN
+            | OpFlags.LOCAL_OP, 4),
+
+        OperationWithAction.OPBLOCK_NO_FAILOVER };
+
+      AuthzCredentialGenerator gen = getXmlAuthzGenerator();
+      getLogWriter().info("Executing opblocks with credential generator " + gen);
+      CredentialGenerator cGen = gen.getCredentialGenerator();
+      Properties extraAuthProps = cGen.getSystemProperties();
+      Properties javaProps = cGen.getJavaProperties();
+      Properties extraAuthzProps = gen.getSystemProperties();
+      String authenticator = cGen.getAuthenticator();
+      String authInit = cGen.getAuthInit();
+      String accessor = gen.getAuthorizationCallback();
+      TestAuthzCredentialGenerator tgen = new TestAuthzCredentialGenerator(gen);
+
+      getLogWriter().info(
+          "testAllOpsNotifications: Using authinit: " + authInit);
+      getLogWriter().info(
+          "testAllOpsNotifications: Using authenticator: " + authenticator);
+      getLogWriter().info(
+          "testAllOpsNotifications: Using accessor: " + accessor);
+
+      // Start servers with all required properties
+      Properties serverProps = buildProperties(authenticator, accessor, true,
+          extraAuthProps, extraAuthzProps);
+      // Get ports for the servers
+      Integer port1 = new Integer(AvailablePort
+          .getRandomAvailablePort(AvailablePort.SOCKET));
+      Integer port2 = new Integer(AvailablePort
+          .getRandomAvailablePort(AvailablePort.SOCKET));
+
+      // Perform all the ops on the clients
+      List opBlock = new ArrayList();
+      Random rnd = new Random();
+      for (int opNum = 0; opNum < allOps.length; ++opNum) {
+        // Start client with valid credentials as specified in
+        // OperationWithAction
+        OperationWithAction currentOp = allOps[opNum];
+        if (currentOp.equals(OperationWithAction.OPBLOCK_END)
+            || currentOp.equals(OperationWithAction.OPBLOCK_NO_FAILOVER)) {
+          // End of current operation block; execute all the operations
+          // on the servers with failover
+          if (opBlock.size() > 0) {
+            // Start the first server and execute the operation block
+            server1.invoke(ClientAuthorizationTestBase.class,
+                "createCacheServer", new Object[] {
+                    SecurityTestUtil.getLocatorPort(), port1, serverProps,
+                    javaProps });
+            server2.invoke(SecurityTestUtil.class, "closeCache");
+            executeOpBlock(opBlock, port1, port2, authInit, extraAuthProps,
+                extraAuthzProps, tgen, rnd);
+            if (!currentOp.equals(OperationWithAction.OPBLOCK_NO_FAILOVER)) {
+              // Failover to the second server and run the block again
+              server2.invoke(ClientAuthorizationTestBase.class,
+                  "createCacheServer", new Object[] {
+                      SecurityTestUtil.getLocatorPort(), port2, serverProps,
+                      javaProps });
+              server1.invoke(SecurityTestUtil.class, "closeCache");
+              executeOpBlock(opBlock, port1, port2, authInit, extraAuthProps,
+                  extraAuthzProps, tgen, rnd);
+            }
+            opBlock.clear();
+          }
+        }
+        else {
+          currentOp.setOpNum(opNum);
+          opBlock.add(currentOp);
+        }
+      }
+  }
+
+  // End Region: Tests
+
+  public void tearDown2() throws Exception {
+
+    super.tearDown2();
+    // close the clients first
+    client1.invoke(SecurityTestUtil.class, "closeCache");
+    client2.invoke(SecurityTestUtil.class, "closeCache");
+    SecurityTestUtil.closeCache();
+    // then close the servers
+    server1.invoke(SecurityTestUtil.class, "closeCache");
+    server2.invoke(SecurityTestUtil.class, "closeCache");
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/security/MultiuserAPIDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/security/MultiuserAPIDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/security/MultiuserAPIDUnitTest.java
new file mode 100644
index 0000000..9c5d18a
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/security/MultiuserAPIDUnitTest.java
@@ -0,0 +1,382 @@
+/*=========================================================================
+ * Copyright (c) 2010-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * one or more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+package com.gemstone.gemfire.security;
+
+import hydra.Log;
+
+import java.io.IOException;
+import java.util.Properties;
+
+import javax.net.ssl.SSLException;
+import javax.net.ssl.SSLHandshakeException;
+
+import security.CredentialGenerator;
+
+import com.gemstone.gemfire.cache.Region;
+import com.gemstone.gemfire.cache.client.Pool;
+import com.gemstone.gemfire.cache.execute.FunctionService;
+import com.gemstone.gemfire.cache.query.CqAttributesFactory;
+import com.gemstone.gemfire.cache.query.CqException;
+import com.gemstone.gemfire.cache.query.CqQuery;
+import com.gemstone.gemfire.cache.query.Query;
+import com.gemstone.gemfire.distributed.internal.DistributionConfig;
+import com.gemstone.gemfire.internal.cache.GemFireCacheImpl;
+import com.gemstone.gemfire.internal.cache.PoolManagerImpl;
+
+import dunit.Host;
+import dunit.VM;
+import security.DummyCredentialGenerator;
+
+public class MultiuserAPIDUnitTest extends ClientAuthorizationTestBase {
+
+  /** constructor */
+  public MultiuserAPIDUnitTest(String name) {
+    super(name);
+  }
+
+  private VM server1 = null;
+
+  private VM server2 = null;
+
+  private VM client1 = null;
+
+  private VM client2 = null;
+
+  private static final String[] serverExpectedExceptions = {
+      AuthenticationRequiredException.class.getName(),
+      AuthenticationFailedException.class.getName(),
+      GemFireSecurityException.class.getName(),
+      ClassNotFoundException.class.getName(), IOException.class.getName(),
+      SSLException.class.getName(), SSLHandshakeException.class.getName()};
+
+  private static final String[] clientExpectedExceptions = {
+      AuthenticationRequiredException.class.getName(),
+      AuthenticationFailedException.class.getName(),
+      SSLHandshakeException.class.getName()};
+
+  public void setUp() throws Exception {
+    super.setUp();
+    final Host host = Host.getHost(0);
+    server1 = host.getVM(0);
+    server2 = host.getVM(1);
+    client1 = host.getVM(2);
+    client2 = host.getVM(3);
+
+    server1.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+        new Object[] {serverExpectedExceptions});
+    server2.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+        new Object[] {serverExpectedExceptions});
+    client1.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+        new Object[] {clientExpectedExceptions});
+    client2.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+        new Object[] {clientExpectedExceptions});
+  }
+
+  public static Integer createCacheServer(Object dsPort, Object locatorString,
+      Object authenticator, Object extraProps, Object javaProps) {
+
+    Properties authProps;
+    if (extraProps == null) {
+      authProps = new Properties();
+    } else {
+      authProps = (Properties)extraProps;
+    }
+    if (authenticator != null) {
+      authProps.setProperty(
+          DistributionConfig.SECURITY_CLIENT_AUTHENTICATOR_NAME, authenticator
+              .toString());
+    }
+    return SecurityTestUtil.createCacheServer(authProps, javaProps,
+        (Integer)dsPort, (String)locatorString, null, new Integer(
+            SecurityTestUtil.NO_EXCEPTION));
+  }
+
+  private static void createCacheClient(Object authInit, Properties authProps,
+      Properties javaProps, Integer[] ports, Object numConnections,
+      Boolean multiUserMode, Integer expectedResult) {
+
+    String authInitStr = (authInit == null ? null : authInit.toString());
+    SecurityTestUtil.createCacheClient(authInitStr, authProps, javaProps,
+        ports, numConnections, multiUserMode.toString(), expectedResult);
+  }
+
+  public static void createCacheClient(Object authInit, Object authProps,
+      Object javaProps, Integer port1, Integer port2, Object numConnections,
+      Boolean multiUserMode, Integer expectedResult) {
+
+    createCacheClient(authInit, (Properties)authProps, (Properties)javaProps,
+        new Integer[] {port1, port2}, numConnections, multiUserMode,
+        expectedResult);
+  }
+
+  public static void registerAllInterest() {
+    Region region = SecurityTestUtil.getCache().getRegion(
+        SecurityTestUtil.regionName);
+    assertNotNull(region);
+    region.registerInterestRegex(".*");
+  }
+
+  private void setUpVMs(CredentialGenerator gen, Boolean multiUser) {
+    Properties extraProps = gen.getSystemProperties();
+    Properties javaProps = gen.getJavaProperties();
+    String authenticator = gen.getAuthenticator();
+    String authInit = gen.getAuthInit();
+
+    getLogWriter().info(
+        "testValidCredentials: Using scheme: " + gen.classCode());
+    getLogWriter().info(
+        "testValidCredentials: Using authenticator: " + authenticator);
+    getLogWriter().info("testValidCredentials: Using authinit: " + authInit);
+
+    // Start the servers
+    Integer locPort1 = SecurityTestUtil.getLocatorPort();
+    Integer locPort2 = SecurityTestUtil.getLocatorPort();
+    String locString = SecurityTestUtil.getLocatorString();
+    Integer port1 = (Integer)server1.invoke(MultiuserAPIDUnitTest.class,
+        "createCacheServer", new Object[] {locPort1, locString, authenticator,
+            extraProps, javaProps});
+    Integer port2 = (Integer)server2.invoke(MultiuserAPIDUnitTest.class,
+        "createCacheServer", new Object[] {locPort2, locString, authenticator,
+            extraProps, javaProps});
+
+    // Start the clients with valid credentials
+    Properties credentials1 = gen.getValidCredentials(1);
+    Properties javaProps1 = gen.getJavaProperties();
+    getLogWriter().info(
+        "testValidCredentials: For first client credentials: " + credentials1
+            + " : " + javaProps1);
+    Properties credentials2 = gen.getValidCredentials(2);
+    Properties javaProps2 = gen.getJavaProperties();
+    getLogWriter().info(
+        "testValidCredentials: For second client credentials: " + credentials2
+            + " : " + javaProps2);
+    client1.invoke(MultiuserAPIDUnitTest.class, "createCacheClient",
+        new Object[] {authInit, credentials1, javaProps1, port1, port2, null,
+            multiUser, new Integer(SecurityTestUtil.NO_EXCEPTION)});
+  }
+
+  public void testSingleUserUnsupportedAPIs() {
+      // Start servers
+      // Start clients with multiuser-authentication set to false
+      setUpVMs(new DummyCredentialGenerator(), Boolean.FALSE);
+      client1.invoke(MultiuserAPIDUnitTest.class, "verifyDisallowedOps",
+          new Object[] {Boolean.FALSE});
+  }
+
+  public void testMultiUserUnsupportedAPIs() {
+      // Start servers.
+      // Start clients with multiuser-authentication set to true.
+      setUpVMs(new DummyCredentialGenerator(), Boolean.TRUE);
+      client1.invoke(MultiuserAPIDUnitTest.class, "verifyDisallowedOps",
+          new Object[] {Boolean.TRUE});
+  }
+
+  public static void verifyDisallowedOps(Boolean multiuserMode) {
+    String op = "unknown";
+    boolean success = false;
+    if (!multiuserMode) {
+      success = false;
+      try {
+        // Attempt cache.createAuthenticatedCacheView() and expect an exception, fail otherwise
+        op = "Pool.createSecureUserCache()";
+        GemFireCacheImpl.getInstance().createAuthenticatedView(new Properties(), "testPool");
+      } catch (IllegalStateException uoe) {
+        Log.getLogWriter().info(op + ": Got expected exception: " + uoe);
+        success = true;
+      } catch (Exception e) {
+        fail("Got unexpected exception while doing " + op, e);
+      }
+      if (!success) {
+        fail("Did not get exception while doing " + op);
+      }
+    } else { // multiuser mode
+      Region realRegion = GemFireCacheImpl.getInstance().getRegion(
+          SecurityTestUtil.regionName);
+      Region proxyRegion = SecurityTestUtil.proxyCaches[0]
+          .getRegion(SecurityTestUtil.regionName);
+      Pool pool = PoolManagerImpl.getPMI().find("testPool");
+      for (int i = 0; i <= 27; i++) {
+        success = false;
+        try {
+          switch (i) {
+            // Attempt (real) Region.create/put/get/containsKeyOnServer/destroy/
+            // destroyRegion/clear/remove/registerInterest/unregisterInterest()
+            // and expect an exception, fail otherwise.
+            case 0:
+              op = "Region.create()";
+              realRegion.create("key", "value");
+              break;
+            case 1:
+              op = "Region.put()";
+              realRegion.put("key", "value");
+              break;
+            case 2:
+              op = "Region.get()";
+              realRegion.get("key");
+              break;
+            case 3:
+              op = "Region.containsKeyOnServer()";
+              realRegion.containsKeyOnServer("key");
+              break;
+            case 4:
+              op = "Region.remove()";
+              realRegion.remove("key");
+              break;
+            case 5:
+              op = "Region.destroy()";
+              realRegion.destroy("key");
+              break;
+            case 6:
+              op = "Region.destroyRegion()";
+              realRegion.destroyRegion();
+              break;
+            case 7:
+              op = "Region.registerInterest()";
+              realRegion.registerInterest("key");
+              break;
+            // case 8:
+            // op = "Region.unregisterInterest()";
+            // realRegion.unregisterInterest("key");
+            // break;
+            case 8:
+              op = "Region.clear()";
+              realRegion.clear();
+              break;
+            // Attempt ProxyRegion.createSubregion/forceRolling/
+            // getAttributesMutator/registerInterest/loadSnapShot/saveSnapshot/
+            // setUserAttribute/unregisterInterest/writeToDisk
+            // and expect an exception, fail otherwise.
+            case 9:
+              op = "ProxyRegion.createSubregion()";
+              proxyRegion.createSubregion("subregion",
+                  null);
+              break;
+            case 10:
+              op = "ProxyRegion.forceRolling()";
+              proxyRegion.forceRolling();
+              break;
+            case 11:
+              op = "ProxyRegion.getAttributesMutator()";
+              proxyRegion.getAttributesMutator();
+              break;
+            case 12:
+              op = "ProxyRegion.registerInterest()";
+              proxyRegion.registerInterest("key");
+              break;
+            case 13:
+              op = "ProxyRegion.loadSnapshot()";
+              proxyRegion.loadSnapshot(null);
+              break;
+            case 14:
+              op = "ProxyRegion.saveSnapshot()";
+              proxyRegion.saveSnapshot(null);
+              break;
+            case 15:
+              op = "ProxyRegion.setUserAttribute()";
+              proxyRegion.setUserAttribute(null);
+              break;
+            case 16:
+              op = "ProxyRegion.unregisterInterestRegex()";
+              proxyRegion.unregisterInterestRegex("*");
+              break;
+            // Attempt FunctionService.onRegion/onServer/s(pool) and expect an
+            // exception, fail otherwise.
+            case 17:
+              op = "FunctionService.onRegion()";
+              FunctionService.onRegion(realRegion);
+              break;
+            case 18:
+              op = "FunctionService.onServer(pool)";
+              FunctionService.onServer(pool);
+              break;
+            case 19:
+              op = "FunctionService.onServers(pool)";
+              FunctionService.onServers(pool);
+              break;
+            // Attempt
+            // QueryService.newQuery().execute()/newCq().execute/executeWithInitialResults()
+            case 20:
+              op = "QueryService.newQuery.execute()";
+              Query query = pool.getQueryService().newQuery(
+                  "SELECT * FROM /" + SecurityTestUtil.regionName);
+              query.execute();
+              break;
+            case 21:
+              op = "QueryService.newCq.execute()";
+              CqQuery cqQuery = pool.getQueryService().newCq(
+                  "SELECT * FROM /" + SecurityTestUtil.regionName,
+                  new CqAttributesFactory().create());
+              try {
+                cqQuery.execute();
+              } catch (CqException ce) {
+                throw (Exception)ce.getCause();
+              }
+              break;
+            case 22:
+              op = "QueryService.newCq.executeWithInitialResults()";
+              cqQuery = pool.getQueryService().newCq(
+                  "SELECT * FROM /" + SecurityTestUtil.regionName,
+                  new CqAttributesFactory().create());
+              try {
+                cqQuery.executeWithInitialResults();
+              } catch (CqException ce) {
+                throw (Exception)ce.getCause();
+              }
+              break;
+            // Attempt ProxyQueryService.getIndex/createIndex/removeIndex() and
+            // expect an exception, fail otherwise.
+            case 23:
+              op = "ProxyQueryService().getIndexes()";
+              SecurityTestUtil.proxyCaches[0].getQueryService()
+                  .getIndexes(null);
+              break;
+            case 24:
+              op = "ProxyQueryService().createIndex()";
+              SecurityTestUtil.proxyCaches[0].getQueryService().createIndex(
+                  null, null, null );
+              break;
+            case 25:
+              op = "ProxyQueryService().removeIndexes()";
+              SecurityTestUtil.proxyCaches[0].getQueryService().removeIndexes();
+              break;
+            case 26:
+              op = "ProxyRegion.localDestroy()";
+              proxyRegion.localDestroy("key");
+              break;
+            case 27:
+              op = "ProxyRegion.localInvalidate()";
+              proxyRegion.localInvalidate("key");
+              break;
+            default:
+              fail("Unknown op code: " + i);
+              break;
+          }
+        } catch (UnsupportedOperationException uoe) {
+          Log.getLogWriter().info(op + ": Got expected exception: " + uoe);
+          success = true;
+        } catch (Exception e) {
+          fail("Got unexpected exception while doing " + op, e);
+        }
+        if (!success) {
+          fail("Did not get exception while doing " + op);
+        }
+      }
+    }
+  }
+
+  public void tearDown2() throws Exception {
+    super.tearDown2();
+    // close the clients first
+    client1.invoke(SecurityTestUtil.class, "closeCache");
+    client2.invoke(SecurityTestUtil.class, "closeCache");
+    // then close the servers
+    server1.invoke(SecurityTestUtil.class, "closeCache");
+    server2.invoke(SecurityTestUtil.class, "closeCache");
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/security/MultiuserDurableCQAuthzDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/security/MultiuserDurableCQAuthzDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/security/MultiuserDurableCQAuthzDUnitTest.java
new file mode 100644
index 0000000..99c55b1
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/security/MultiuserDurableCQAuthzDUnitTest.java
@@ -0,0 +1,484 @@
+/*=========================================================================
+ * Copyright (c) 2010-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * one or more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+/**
+ * 
+ */
+package com.gemstone.gemfire.security;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Random;
+
+import security.AuthzCredentialGenerator;
+import security.CredentialGenerator;
+
+import com.gemstone.gemfire.cache.Region;
+import com.gemstone.gemfire.cache.operations.OperationContext.OperationCode;
+import com.gemstone.gemfire.cache.query.CqAttributes;
+import com.gemstone.gemfire.cache.query.CqAttributesFactory;
+import com.gemstone.gemfire.cache.query.CqListener;
+import com.gemstone.gemfire.cache.query.CqQuery;
+import com.gemstone.gemfire.cache.query.QueryService;
+import com.gemstone.gemfire.cache.query.SelectResults;
+import com.gemstone.gemfire.cache.query.cq.dunit.CqQueryTestListener;
+import com.gemstone.gemfire.cache.query.internal.cq.ClientCQImpl;
+import com.gemstone.gemfire.distributed.internal.InternalDistributedSystem;
+import com.gemstone.gemfire.internal.AvailablePort;
+import com.gemstone.gemfire.internal.cache.GemFireCacheImpl;
+import com.gemstone.gemfire.internal.logging.InternalLogWriter;
+
+import dunit.Host;
+import dunit.SerializableRunnable;
+
+/**
+ * @author ashetkar
+ *
+ */
+public class MultiuserDurableCQAuthzDUnitTest extends
+    ClientAuthorizationTestBase {
+  
+  public static final Map<String, String> cqNameToQueryStrings = new HashMap<String, String>();
+
+  static {
+    cqNameToQueryStrings.put("CQ_0", "SELECT * FROM ");
+    cqNameToQueryStrings.put("CQ_1", "SELECT * FROM ");
+  }
+
+  public MultiuserDurableCQAuthzDUnitTest(String name) {
+    super(name);
+  }
+
+  public void setUp() throws Exception {
+    super.setUp();
+    getSystem();
+    invokeInEveryVM(new SerializableRunnable("getSystem") {
+      public void run() {
+        getSystem();
+      }
+    });
+
+    final Host host = Host.getHost(0);
+    server1 = host.getVM(0);
+    server2 = host.getVM(1);
+    client1 = host.getVM(2);
+    client2 = host.getVM(3);
+
+    server1.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+        new Object[] { serverExpectedExceptions });
+    server2.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+        new Object[] { serverExpectedExceptions });
+    client2.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+        new Object[] { clientExpectedExceptions });
+    SecurityTestUtil.registerExpectedExceptions(clientExpectedExceptions);
+  }
+
+  public void testCQForDurableClientsWithDefaultClose() throws Exception {
+    /*
+     *  1. Start a server.
+     *  2. Start a durable client in mulituser secure mode.
+     *  3. Create two users registering unique durable CQs on server.
+     *  4. Invoke GemFireCache.close() at client.
+     *  5. Put some events on server satisfying both the CQs.
+     *  6. Up the client and the two users.
+     *  7. Confirm that the users receive the events which were enqueued at server while they were away.
+     *  8. Same for ProxyCache.close()
+     */
+    Integer numOfUsers = 2;
+    Integer numOfPuts = 5;
+    Boolean[] postAuthzAllowed = new Boolean[] {Boolean.TRUE, Boolean.TRUE};
+
+    doTest(numOfUsers, numOfPuts, postAuthzAllowed,
+          getXmlAuthzGenerator(), null);
+  }
+
+  public void testCQForDurableClientsWithCloseKeepAliveTrue() throws Exception {
+    /*
+     *  1. Start a server.
+     *  2. Start a durable client in mulituser secure mode.
+     *  3. Create two users registering unique durable CQs on server.
+     *  4. Invoke GemFireCache.close(false) at client.
+     *  5. Put some events on server satisfying both the CQs.
+     *  6. Up the client and the two users.
+     *  7. Observer the behaviour.
+     *  8. Same for ProxyCache.close(false)
+     */
+    Integer numOfUsers = 2;
+    Integer numOfPuts = 5;
+    Boolean[] postAuthzAllowed = new Boolean[] {Boolean.TRUE, Boolean.TRUE};
+
+    doTest(numOfUsers, numOfPuts, postAuthzAllowed,
+          getXmlAuthzGenerator(), Boolean.TRUE);
+  }
+
+  public void testCQForDurableClientsWithCloseKeepAliveFalse() throws Exception {
+    /*
+     *  1. Start a server.
+     *  2. Start a durable client in mulituser secure mode.
+     *  3. Create two users registering unique durable CQs on server.
+     *  4. Invoke GemFireCache.close(true) at client.
+     *  5. Put some events on server satisfying both the CQs.
+     *  6. Up the client and the two users.
+     *  7. Observer the behaviour.
+     *  8. Same for ProxyCache.close(true)
+     */
+    Integer numOfUsers = 2;
+    Integer numOfPuts = 5;
+    Boolean[] postAuthzAllowed = new Boolean[] {Boolean.TRUE, Boolean.TRUE};
+
+    doTest(numOfUsers, numOfPuts, postAuthzAllowed,
+              getXmlAuthzGenerator(), Boolean.FALSE);
+  }
+
+  private void doTest(Integer numOfUsers, Integer numOfPuts,
+      Boolean[] postAuthzAllowed, AuthzCredentialGenerator gen, Boolean keepAlive)
+      throws Exception {
+    CredentialGenerator cGen = gen.getCredentialGenerator();
+    Properties extraAuthProps = cGen.getSystemProperties();
+    Properties javaProps = cGen.getJavaProperties();
+    Properties extraAuthzProps = gen.getSystemProperties();
+    String authenticator = cGen.getAuthenticator();
+    String accessor = gen.getAuthorizationCallback();
+    String authInit = cGen.getAuthInit();
+    TestAuthzCredentialGenerator tgen = new TestAuthzCredentialGenerator(gen);
+
+    Properties serverProps = buildProperties(authenticator, accessor, true,
+        extraAuthProps, extraAuthzProps);
+
+    Properties opCredentials;
+    cGen = tgen.getCredentialGenerator();
+    Properties javaProps2 = null;
+    if (cGen != null) {
+      javaProps2 = cGen.getJavaProperties();
+    }
+
+    int[] indices = new int[numOfPuts];
+    for (int index = 0; index < numOfPuts; ++index) {
+      indices[index] = index;
+    }
+
+    Random rnd = new Random();
+    Properties[] authProps = new Properties[numOfUsers];
+    String durableClientId = "multiuser_durable_client_1";
+    Properties client2Credentials = null;
+    for (int i = 0; i < numOfUsers; i++) {
+      int rand = rnd.nextInt(100) + 1;
+      if (postAuthzAllowed[i]) {
+        opCredentials = tgen.getAllowedCredentials(new OperationCode[] {
+            OperationCode.EXECUTE_CQ, OperationCode.GET}, // For callback, GET should be allowed
+            new String[] {regionName}, indices, rand);
+      } else {
+        opCredentials = tgen.getDisallowedCredentials(new OperationCode[] {
+            OperationCode.GET}, // For callback, GET should be disallowed
+            new String[] {regionName}, indices, rand);
+      }
+      authProps[i] = SecurityTestUtil.concatProperties(new Properties[] {
+          opCredentials, extraAuthProps, extraAuthzProps});
+
+      if (client2Credentials == null) {
+        client2Credentials = tgen.getAllowedCredentials(new OperationCode[] {
+            OperationCode.PUT},
+            new String[] {regionName}, indices, rand);
+      }
+    }
+
+    // Get ports for the servers
+    Integer port1 = new Integer(AvailablePort
+        .getRandomAvailablePort(AvailablePort.SOCKET));
+    Integer port2 = new Integer(AvailablePort
+        .getRandomAvailablePort(AvailablePort.SOCKET));
+    Integer locatorPort = new Integer(AvailablePort
+        .getRandomAvailablePort(AvailablePort.SOCKET));
+    // Close down any running servers
+    server1.invoke(SecurityTestUtil.class, "closeCache");
+    server2.invoke(SecurityTestUtil.class, "closeCache");
+
+    server1.invoke(MultiuserDurableCQAuthzDUnitTest.class,
+        "createServerCache", new Object[] {serverProps, javaProps, locatorPort, port1});
+    client1.invoke(MultiuserDurableCQAuthzDUnitTest.class,
+        "createClientCache", new Object[] {javaProps2, authInit, authProps,
+            new Integer[] {port1, port2}, numOfUsers, durableClientId, postAuthzAllowed});
+
+//    client2.invoke(SecurityTestUtil.class, "createCacheClient",
+//        new Object[] {authInit, client2Credentials, javaProps2,
+//            new Integer[] {port1, port2}, null, SecurityTestUtil.NO_EXCEPTION});
+
+    client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "createCQ",
+        new Object[] {numOfUsers, Boolean.TRUE});
+    client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "executeCQ",
+        new Object[] {numOfUsers, new Boolean[] {false, false}, numOfPuts,
+            new String[numOfUsers]});
+    client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "readyForEvents");
+
+    if (keepAlive == null) {
+      client1.invoke(SecurityTestUtil.class, "closeCache");
+    } else {
+      client1.invoke(SecurityTestUtil.class, "closeCache",
+          new Object[] {keepAlive});
+    }
+
+    server1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "doPuts",
+        new Object[] {numOfPuts, Boolean.TRUE/* put last key */});
+
+    client1.invoke(MultiuserDurableCQAuthzDUnitTest.class,
+        "createClientCache", new Object[] {javaProps2, authInit, authProps,
+            new Integer[] {port1, port2}, numOfUsers, durableClientId, postAuthzAllowed});
+    client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "createCQ",
+        new Object[] {numOfUsers, Boolean.TRUE});
+    client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "executeCQ",
+        new Object[] {numOfUsers, new Boolean[] {false, false}, numOfPuts,
+            new String[numOfUsers]});
+    client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "readyForEvents");
+
+    if (!postAuthzAllowed[0] || keepAlive == null || !keepAlive) {
+      // Don't wait as no user is authorized to receive cq events.
+      Thread.sleep(1000);
+    } else {
+      client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "waitForLastKey",
+          new Object[] {Integer.valueOf(0), Boolean.TRUE});
+    }
+    Integer numOfCreates = (keepAlive == null) ? 0
+        : (keepAlive) ? (numOfPuts + 1/* last key */) : 0;
+    client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "checkCQListeners",
+        new Object[] {numOfUsers, postAuthzAllowed, numOfCreates, 0});
+
+    client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "proxyCacheClose",
+        new Object[] {new Integer[] {0, 1}, keepAlive});
+
+    client1.invoke(SecurityTestUtil.class, "createProxyCache",
+        new Object[] {new Integer[] {0, 1}, authProps});
+
+    client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "createCQ",
+        new Object[] {numOfUsers, Boolean.TRUE});
+    client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "executeCQ",
+        new Object[] {numOfUsers, new Boolean[] {false, false}, numOfPuts,
+            new String[numOfUsers]});
+
+    server1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "doPuts",
+        new Object[] {numOfPuts, Boolean.TRUE/* put last key */});
+
+    if (!postAuthzAllowed[0] || keepAlive == null || !keepAlive) {
+      // Don't wait as no user is authorized to receive cq events.
+      Thread.sleep(1000);
+    } else {
+      client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "waitForLastKey",
+          new Object[] {Integer.valueOf(0), Boolean.FALSE});
+    }
+    Integer numOfUpdates = numOfPuts + 1;
+    client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "checkCQListeners",
+        new Object[] {numOfUsers, postAuthzAllowed, 0, numOfUpdates});
+  }
+
+  public static void createServerCache(Properties serverProps,
+      Properties javaProps, Integer locatorPort, Integer serverPort) {
+    SecurityTestUtil.createCacheServer((Properties)serverProps, javaProps,
+        locatorPort, null, serverPort, Boolean.TRUE, new Integer(
+            SecurityTestUtil.NO_EXCEPTION));
+  }
+
+  public static void createClientCache(Properties javaProps,
+      String authInit, Properties[] authProps, Integer ports[],
+      Integer numOfUsers, Boolean[] postAuthzAllowed) {
+    SecurityTestUtil.createCacheClientForMultiUserMode(numOfUsers, authInit,
+        authProps, javaProps, ports, null, Boolean.FALSE,
+        SecurityTestUtil.NO_EXCEPTION);
+  }
+
+  public static void readyForEvents() {
+    GemFireCacheImpl.getInstance().readyForEvents();
+  }
+
+  public static void createClientCache(Properties javaProps,
+      String authInit, Properties[] authProps, Integer ports[],
+      Integer numOfUsers, String durableId, Boolean[] postAuthzAllowed) {
+    SecurityTestUtil.createCacheClientForMultiUserMode(numOfUsers, authInit,
+        authProps, javaProps, ports, null, Boolean.FALSE, durableId,
+        SecurityTestUtil.NO_EXCEPTION);
+  }
+
+  public static void createCQ(Integer num) {
+    createCQ(num, false);
+  }
+
+  public static void createCQ(Integer num, Boolean isDurable) {
+    for (int i = 0; i < num; i++) {
+      QueryService cqService = SecurityTestUtil.proxyCaches[i].getQueryService();
+      String cqName = "CQ_" + i;
+      String queryStr = cqNameToQueryStrings.get(cqName)
+          + SecurityTestUtil.proxyCaches[i].getRegion(regionName).getFullPath();
+      // Create CQ Attributes.
+      CqAttributesFactory cqf = new CqAttributesFactory();
+      CqListener[] cqListeners = {new CqQueryTestListener(getLogWriter())};
+      ((CqQueryTestListener)cqListeners[0]).cqName = cqName;
+
+      cqf.initCqListeners(cqListeners);
+      CqAttributes cqa = cqf.create();
+
+      // Create CQ.
+      try {
+        CqQuery cq1 = cqService.newCq(cqName, queryStr, cqa, isDurable);
+        assertTrue("newCq() state mismatch", cq1.getState().isStopped());
+      } catch (Exception ex) {
+        AssertionError err = new AssertionError("Failed to create CQ " + cqName
+            + " . ");
+        err.initCause(ex);
+        getLogWriter().info("CqService is :" + cqService, err);
+        throw err;
+      }
+    }
+  }
+
+  public static void executeCQ(Integer num, Boolean[] initialResults,
+      Integer expectedResultsSize, String[] expectedErr) {
+    InternalLogWriter logWriter = InternalDistributedSystem.getStaticInternalLogWriter();
+    for (int i = 0; i < num; i++) {
+      try {
+        if (expectedErr[i] != null) {
+          logWriter.info(
+              "<ExpectedException action=add>" + expectedErr[i]
+                  + "</ExpectedException>");
+        }
+        CqQuery cq1 = null;
+        String cqName = "CQ_" + i;
+        String queryStr = cqNameToQueryStrings.get(cqName)
+            + SecurityTestUtil.proxyCaches[i].getRegion(regionName)
+                .getFullPath();
+        QueryService cqService = SecurityTestUtil.proxyCaches[i]
+            .getQueryService();
+
+        // Get CqQuery object.
+        try {
+          cq1 = cqService.getCq(cqName);
+          if (cq1 == null) {
+            getLogWriter().info(
+                "Failed to get CqQuery object for CQ name: " + cqName);
+            fail("Failed to get CQ " + cqName);
+          } else {
+            getLogWriter().info("Obtained CQ, CQ name: " + cq1.getName());
+            assertTrue("newCq() state mismatch", cq1.getState().isStopped());
+          }
+        } catch (Exception ex) {
+          getLogWriter().info("CqService is :" + cqService);
+          getLogWriter().error(ex);
+          AssertionError err = new AssertionError("Failed to execute CQ "
+              + cqName);
+          err.initCause(ex);
+          throw err;
+        }
+
+        if (initialResults[i]) {
+          SelectResults cqResults = null;
+
+          try {
+            cqResults = cq1.executeWithInitialResults();
+          } catch (Exception ex) {
+            getLogWriter().info("CqService is: " + cqService);
+            ex.printStackTrace();
+            AssertionError err = new AssertionError("Failed to execute CQ "
+                + cqName);
+            err.initCause(ex);
+            throw err;
+          }
+          getLogWriter().info("initial result size = " + cqResults.size());
+          assertTrue("executeWithInitialResults() state mismatch", cq1
+              .getState().isRunning());
+          if (expectedResultsSize >= 0) {
+            assertEquals("unexpected results size", expectedResultsSize
+                .intValue(), cqResults.size());
+          }
+        } else {
+          try {
+            cq1.execute();
+          } catch (Exception ex) {
+            AssertionError err = new AssertionError("Failed to execute CQ "
+                + cqName);
+            err.initCause(ex);
+            if (expectedErr == null) {
+              getLogWriter().info("CqService is: " + cqService, err);
+            }
+            throw err;
+          }
+          assertTrue("execute() state mismatch", cq1.getState().isRunning());
+        }
+      } finally {
+        if (expectedErr[i] != null) {
+          logWriter.info(
+              "<ExpectedException action=remove>" + expectedErr[i]
+                  + "</ExpectedException>");
+        }
+      }
+    }
+  }
+
+  public static void doPuts(Integer num, Boolean putLastKey) {
+    Region region = GemFireCacheImpl.getInstance().getRegion(regionName);
+    for (int i = 0; i < num; i++) {
+      region.put("CQ_key"+i, "CQ_value"+i);
+    }
+    if (putLastKey) {
+      region.put("LAST_KEY", "LAST_KEY");
+    }
+  }
+
+  public static void putLastKey() {
+    Region region = GemFireCacheImpl.getInstance().getRegion(regionName);
+    region.put("LAST_KEY", "LAST_KEY");
+  }
+
+  public static void waitForLastKey(Integer cqIndex, Boolean isCreate) {
+    String cqName = "CQ_" + cqIndex;
+    QueryService qService = SecurityTestUtil.proxyCaches[cqIndex].getQueryService();
+    ClientCQImpl cqQuery = (ClientCQImpl)qService.getCq(cqName);
+    if (isCreate) {
+      ((CqQueryTestListener)cqQuery.getCqListeners()[cqIndex])
+          .waitForCreated("LAST_KEY");
+    } else {
+      ((CqQueryTestListener)cqQuery.getCqListeners()[cqIndex])
+          .waitForUpdated("LAST_KEY");
+    }
+  }
+
+  public static void checkCQListeners(Integer numOfUsers,
+      Boolean[] expectedListenerInvocation, Integer createEventsSize,
+      Integer updateEventsSize) {
+    for (int i = 0; i < numOfUsers; i++) {
+      String cqName = "CQ_" + i;
+      QueryService qService = SecurityTestUtil.proxyCaches[i].getQueryService();
+      ClientCQImpl cqQuery = (ClientCQImpl)qService.getCq(cqName);
+      if (expectedListenerInvocation[i]) {
+        for (CqListener listener : cqQuery.getCqListeners()) {
+          assertEquals(createEventsSize.intValue(),
+              ((CqQueryTestListener)listener).getCreateEventCount());
+          assertEquals(updateEventsSize.intValue(),
+              ((CqQueryTestListener)listener).getUpdateEventCount());
+        }
+      } else {
+        for (CqListener listener : cqQuery.getCqListeners()) {
+          assertEquals(0, ((CqQueryTestListener)listener).getTotalEventCount());
+        }
+      }
+    }
+  }
+
+  public static void proxyCacheClose(Integer[] userIndices) {
+    proxyCacheClose(userIndices, null);
+  }
+
+  public static void proxyCacheClose(Integer[] userIndices, Boolean keepAliveFlags) {
+    if (keepAliveFlags != null) {
+      for (int i : userIndices) {
+        SecurityTestUtil.proxyCaches[i].close(keepAliveFlags);
+      }
+    } else {
+      for (int i : userIndices) {
+        SecurityTestUtil.proxyCaches[i].close();
+      }
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/resources/.keepme
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/resources/.keepme b/gemfire-cq/src/test/resources/.keepme
new file mode 100644
index 0000000..e69de29

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/excludedClasses.txt
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/excludedClasses.txt b/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/excludedClasses.txt
new file mode 100644
index 0000000..c55df1d
--- /dev/null
+++ b/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/excludedClasses.txt
@@ -0,0 +1,2 @@
+# e.g.,
+#com/gemstone/gemfire/cache/query/CqAttributesFactory

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/openBugs.txt
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/openBugs.txt b/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/openBugs.txt
new file mode 100644
index 0000000..6c40e24
--- /dev/null
+++ b/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/openBugs.txt
@@ -0,0 +1,21 @@
+# This is a list of classes excluded due to open bugs about their having
+# incompatible changes.  There should be no entries in this file at the
+# time of a product release.
+
+# Each entry should be a bug number followed by a comma and the
+# full class name.  The package components can be delimited with a period
+# or a comma.  Don't include ".class" or ".java" at the end of the name.
+
+# example:  50174,com/gemstone/gemfire/distributed/internal/StartupResponseWithVersionMessage
+# example:  50175,com.gemstone.org.jgroups.Message$Header
+
+
+# ~~~~~~~~~~~~~~~~~~~ DataSerializables ~~~~~~~~~~~~~~~~~~~~~~~~~
+# these are failures from testDataSerializables
+
+
+# ~~~~~~~~~~~~~~~~~~~ Serializables ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+# these are failures from testSerializables
+
+
+

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedDataSerializables.txt
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedDataSerializables.txt b/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedDataSerializables.txt
new file mode 100644
index 0000000..eb9d8d6
--- /dev/null
+++ b/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedDataSerializables.txt
@@ -0,0 +1,4 @@
+com/gemstone/gemfire/cache/query/internal/cq/ServerCQImpl,2
+fromData,64,2ab40046594dc22ab400462bb80099b6009ab6007f2cc3a700084e2cc32dbf2a2bb8009bb6009cb5009d2a2bb8009eb5005d2a2bb9009f0100b800a0b50008b1
+toData,47,2ab40046b60047b800a12bb800a22ab4009db8007c2bb800a32ab4005d2bb800a42b2ab40008b600a5b900a60300b1
+

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedSerializables.txt
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedSerializables.txt b/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedSerializables.txt
new file mode 100755
index 0000000..ddd9718
--- /dev/null
+++ b/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedSerializables.txt
@@ -0,0 +1 @@
+com/gemstone/gemfire/cache/query/internal/cq/CqConflatable,true,-7215022132135862557,conflate:boolean,id:com/gemstone/gemfire/internal/cache/EventID,key:java/lang/Object,regionname:java/lang/String,value:java/lang/Object

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/resources/com/gemstone/gemfire/internal/cache/tier/sockets/durablecq-client-cache.xml
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/resources/com/gemstone/gemfire/internal/cache/tier/sockets/durablecq-client-cache.xml b/gemfire-cq/src/test/resources/com/gemstone/gemfire/internal/cache/tier/sockets/durablecq-client-cache.xml
new file mode 100644
index 0000000..afde78f
--- /dev/null
+++ b/gemfire-cq/src/test/resources/com/gemstone/gemfire/internal/cache/tier/sockets/durablecq-client-cache.xml
@@ -0,0 +1,21 @@
+<?xml version="1.0"?>
+ <!DOCTYPE client-cache PUBLIC
+    "-//GemStone Systems, Inc.//GemFire Declarative Caching 6.6//EN"
+    "http://www.gemstone.com/dtd/cache6_6.dtd">
+    
+
+<client-cache>
+    <pool 
+    name="client" 
+    subscription-enabled="true" 
+    load-conditioning-interval="60000"
+    read-timeout="30000" retry-attempts="5" >
+        <server host="localhost" port="10188" />
+    </pool>
+
+    <region name="testReadyForEventsNotCalledImplicitlyWithCacheXML_region" >
+    <region-attributes id="testReadyForEventsNotCalledImplicitlyWithCacheXML_region" statistics-enabled="true"  pool-name="client" refid="PROXY">
+        <cache-listener><class-name>com.gemstone.gemfire.internal.cache.tier.sockets.CacheServerTestUtil$ControlListener</class-name></cache-listener>
+    </region-attributes>
+    </region>
+</client-cache>

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/resources/com/gemstone/gemfire/internal/cache/tier/sockets/durablecq-server-cache.xml
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/resources/com/gemstone/gemfire/internal/cache/tier/sockets/durablecq-server-cache.xml b/gemfire-cq/src/test/resources/com/gemstone/gemfire/internal/cache/tier/sockets/durablecq-server-cache.xml
new file mode 100644
index 0000000..2d58506
--- /dev/null
+++ b/gemfire-cq/src/test/resources/com/gemstone/gemfire/internal/cache/tier/sockets/durablecq-server-cache.xml
@@ -0,0 +1,16 @@
+<?xml version="1.0"?>
+<!DOCTYPE cache PUBLIC
+    "-//GemStone Systems, Inc.//GemFire Declarative Caching 6.6//EN"
+    "http://www.gemstone.com/dtd/cache6_6.dtd">
+
+<cache is-server="true">
+    <!-- Define this cache server -->
+    <cache-server port="10188" >
+        <client-subscription eviction-policy="entry" capacity="1000"/>
+    </cache-server>
+    <region name="testReadyForEventsNotCalledImplicitlyWithCacheXML_region">
+        <region-attributes  data-policy="partition" enable-subscription-conflation="false">
+             <partition-attributes redundant-copies="0" total-num-buckets="113"/>
+        </region-attributes>
+    </region>
+</cache>

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/resources/cq/statistics/cqStats.bt
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/resources/cq/statistics/cqStats.bt b/gemfire-cq/src/test/resources/cq/statistics/cqStats.bt
new file mode 100644
index 0000000..ea6685c
--- /dev/null
+++ b/gemfire-cq/src/test/resources/cq/statistics/cqStats.bt
@@ -0,0 +1,3 @@
+cq/statistics/cqStatsHCT.conf
+edgeHosts =1 edgeVMsPerHost = 1 edgeThreadsPerVM = 1
+bridgeHosts =1 bridgeVMsPerHost = 1 bridgeThreadsPerVM = 1

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/resources/cq/statistics/cqStatsHCT.conf
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/resources/cq/statistics/cqStatsHCT.conf b/gemfire-cq/src/test/resources/cq/statistics/cqStatsHCT.conf
new file mode 100644
index 0000000..4623a5d
--- /dev/null
+++ b/gemfire-cq/src/test/resources/cq/statistics/cqStatsHCT.conf
@@ -0,0 +1,100 @@
+hydra.Prms-testRequirement = "Test CQ with a variety of operations with careful validation and serial execution in a bridge configuration";
+hydra.Prms-testDescription = "
+This test executes operations on entries on a region and carefully
+validates for correctness in the CQs. 
+The test uses serial round robin; the first thread
+in the round does a random operation, then all other threads in the round
+verify their view of the operation. After the last thread in the round verifies,
+it then becomes the first in the next round, thus the thread doing the random
+operation changes for each round. 
+";
+
+INCLUDE $JTESTS/hydraconfig/hydraparams1.inc;
+INCLUDE $JTESTS/hydraconfig/topology_hct.inc;
+
+hydra.VmPrms-extraClassPaths =
+   fcn "hydra.TestConfigFcns.duplicate(\"$GEMFIRE/lib/antlr.jar\", ${bridgeHosts})" ncf;
+
+THREADGROUP bridgeThreads
+  totalThreads = fcn
+                 ${bridgeHosts} * ${bridgeVMsPerHost} * ${bridgeThreadsPerVM}
+                 ncf
+  clientNames  = fcn "hydra.TestConfigFcns.generateNames
+                      (\"bridge\", ${bridgeHosts}, true)"
+                 ncf;
+THREADGROUP edgeThreads
+  totalThreads = fcn
+                 ${edgeHosts} * ${edgeVMsPerHost} * ${edgeThreadsPerVM}
+                 ncf
+  clientNames  = fcn "hydra.TestConfigFcns.generateNames
+                      (\"edge\", ${edgeHosts}, true)"
+                 ncf;         
+
+INITTASK     taskClass   = cq.statistics.CQStatisticsTest  taskMethod = HydraTask_initializeBridgeServer
+             threadGroups = bridgeThreads
+             runMode = always;
+
+INITTASK     taskClass   = cq.statistics.CQStatisticsTest  taskMethod = HydraTask_initializeClient
+             threadGroups = edgeThreads
+             runMode = always;
+
+INITTASK         taskClass   = cq.statistics.CQStatisticsTest  taskMethod = HydraTask_verifyCqStatistics
+             threadGroups = edgeThreads;
+
+hydra.GemFirePrms-stopSystemsAfterTest = true;
+
+hydra.Prms-totalTaskTimeSec = 1200;
+hydra.Prms-maxResultWaitSec = 600;
+hydra.Prms-serialExecution = true;
+hydra.Prms-roundRobin = true;
+
+INCLUDE $JTESTS/util/randomValues.inc;
+util.RandomValuesPrms-objectType = byte[];
+util.RandomValuesPrms-elementSize = 1000;
+util.RandomValuesPrms-borderCasePercentage = 0;
+util.ValueHolderPrms-useExtraObject = true;
+hydra.GemFirePrms-conserveSockets = ONEOF true false FOENO;
+
+hydra.CachePrms-names           = cache1;
+hydra.CachePrms-searchTimeout   = 600;
+
+hydra.Prms-useFixedRandomInMaster= true;
+hydra.RegionPrms-names          = clientRegion            serverRegion;
+hydra.RegionPrms-regionName     = testRegion              testRegion;
+hydra.RegionPrms-scope          = local                   ack;
+hydra.RegionPrms-poolName       = edgeDescript            none;
+hydra.RegionPrms-dataPolicy     = normal                  replicate;
+hydra.RegionPrms-cacheListeners = util.SilenceListener,   hct.BridgeEventListener;
+hydra.RegionPrms-partitionName  = none                    none;
+
+cq.CQUtilPrms-numOpsPerTask = 100;
+
+util.CachePrms-useDeclarativeXmlFile = ONEOF true false FOENO;
+util.TestHelperPrms-minTaskGranularitySec = 30;
+
+cq.CQUtilPrms-serverEntryOperations = ONEOF add add update invalidate get destroy FOENO; 
+cq.CQUtilPrms-clientEntryOperations = ONEOF add add update get destroy FOENO; 
+cq.CQUtilPrms-upperThreshold = 500;
+cq.CQUtilPrms-upperThresholdServerOperations = ONEOF destroy FOENO;
+cq.CQUtilPrms-upperThresholdClientOperations = ONEOF destroy FOENO;
+cq.CQUtilPrms-lowerThreshold = 0;
+cq.CQUtilPrms-lowerThresholdServerOperations = add;
+cq.CQUtilPrms-lowerThresholdClientOperations = add;
+cq.CQUtilPrms-queryDepth = 7;
+cq.CQUtilPrms-numQueriesPerClientVM = 100; 
+cq.CQUtilPrms-QueryServicePoolName = qservice;
+cq.CQUtilPrms-QueryServiceUsingPool = ONEOF true false FOENO;
+
+// define the edge clients
+hydra.PoolPrms-names                       = edgeDescript qservice;
+hydra.PoolPrms-minConnections        = 2;
+hydra.PoolPrms-subscriptionEnabled = true;
+hydra.PoolPrms-threadLocalConnections         = true;
+hydra.PoolPrms-readTimeout                 = 800000;
+hydra.PoolPrms-subscriptionRedundancy             = RANGE 0
+                                                   fcn ${bridgeHosts} * ${bridgeVMsPerHost} ncf
+                                             EGNAR;
+
+// define the bridge servers
+hydra.BridgePrms-names                = bridge;
+

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-wan/build.gradle
----------------------------------------------------------------------
diff --git a/gemfire-wan/build.gradle b/gemfire-wan/build.gradle
new file mode 100644
index 0000000..eba2539
--- /dev/null
+++ b/gemfire-wan/build.gradle
@@ -0,0 +1,5 @@
+dependencies {
+  provided project(':gemfire-core')
+  provided project(path: ':gemfire-core', configuration: 'testOutput')
+  provided project(path: ':gemfire-junit', configuration: 'testOutput')
+}

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-wan/src/main/java/.keepme
----------------------------------------------------------------------
diff --git a/gemfire-wan/src/main/java/.keepme b/gemfire-wan/src/main/java/.keepme
new file mode 100644
index 0000000..e69de29

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/GatewaySenderBatchOp.java
----------------------------------------------------------------------
diff --git a/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/GatewaySenderBatchOp.java b/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/GatewaySenderBatchOp.java
new file mode 100644
index 0000000..15be615
--- /dev/null
+++ b/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/GatewaySenderBatchOp.java
@@ -0,0 +1,304 @@
+/*=========================================================================
+ * Copyright (c) 2002-2014, Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+package com.gemstone.gemfire.cache.client.internal;
+
+import com.gemstone.gemfire.InternalGemFireError;
+import com.gemstone.gemfire.cache.client.ServerOperationException;
+import com.gemstone.gemfire.internal.Version;
+import com.gemstone.gemfire.internal.cache.EventID;
+import com.gemstone.gemfire.internal.cache.tier.MessageType;
+import com.gemstone.gemfire.internal.cache.tier.sockets.ChunkedMessage;
+import com.gemstone.gemfire.internal.cache.tier.sockets.Message;
+import com.gemstone.gemfire.internal.cache.tier.sockets.Part;
+import com.gemstone.gemfire.internal.cache.wan.BatchException70;
+import com.gemstone.gemfire.internal.cache.wan.GatewaySenderEventImpl;
+import com.gemstone.gemfire.internal.cache.wan.GatewaySenderEventRemoteDispatcher;
+import com.gemstone.gemfire.internal.cache.wan.GatewaySenderEventRemoteDispatcher.GatewayAck;
+import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
+import com.gemstone.gemfire.internal.logging.LogService;
+
+import java.net.SocketTimeoutException;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.logging.log4j.Logger;
+
+public class GatewaySenderBatchOp {
+  
+  private static final Logger logger = LogService.getLogger();
+  
+  /**
+   * Send a list of gateway events to a server to execute
+   * using connections from the given pool
+   * to communicate with the server.
+   * @param con the connection to send the message on.
+   * @param pool the pool to use to communicate with the server.
+   * @param events list of gateway events
+   * @param batchId the ID of this batch
+   * @param removeFromQueueOnException true if the events should be processed even after some exception
+   */
+  public static void executeOn(Connection con, ExecutablePool pool, List events, int batchId, boolean removeFromQueueOnException, boolean isRetry)
+  {
+    AbstractOp op = null;
+    //System.out.println("Version: "+con.getWanSiteVersion());
+    if (Version.GFE_651.compareTo(con.getWanSiteVersion()) >= 0) {
+      op = new GatewaySenderGFEBatchOpImpl(events, batchId, removeFromQueueOnException, con.getDistributedSystemId(), isRetry);
+    } else {
+      // Default should create a batch of server version (ACCEPTOR.VERSION)
+      op = new GatewaySenderGFEBatchOpImpl(events, batchId, removeFromQueueOnException, con.getDistributedSystemId(), isRetry);     
+    }
+    pool.executeOn(con, op, true/*timeoutFatal*/);
+  }
+  
+  
+  public static Object executeOn(Connection con, ExecutablePool pool)
+  {
+    AbstractOp op = null;
+    //System.out.println("Version: "+con.getWanSiteVersion());
+    // [sumedh] both cases are now same; why switch-case?
+    if (Version.GFE_651.compareTo(con.getWanSiteVersion()) >= 0) {
+      op = new GatewaySenderGFEBatchOpImpl();
+    } else {
+      // Default should create a batch of server version (ACCEPTOR.VERSION)
+      op = new GatewaySenderGFEBatchOpImpl();
+    }
+    return pool.executeOn(con, op, true/*timeoutFatal*/);
+  }
+                                                               
+  private GatewaySenderBatchOp() {
+    // no instances allowed
+  }
+  
+  static class GatewaySenderGFEBatchOpImpl extends AbstractOp {
+    
+    /**
+     * @throws com.gemstone.gemfire.SerializationException if serialization fails
+     */
+    public GatewaySenderGFEBatchOpImpl(List events, int batchId, boolean removeFromQueueOnException, int dsId, boolean isRetry)  {
+      super(MessageType.GATEWAY_RECEIVER_COMMAND, calcPartCount(events));
+      if (isRetry) {
+        getMessage().setIsRetry();
+      }
+      getMessage().addIntPart(events.size());
+      getMessage().addIntPart(batchId);
+      getMessage().addIntPart(dsId);
+      getMessage().addBytesPart(
+          new byte[] { removeFromQueueOnException ? (byte)1 : (byte)0 });
+      // Add each event
+      for (Iterator i = events.iterator(); i.hasNext();) {
+        GatewaySenderEventImpl event = (GatewaySenderEventImpl)i.next();
+        // Add action
+        int action = event.getAction();
+        getMessage().addIntPart(action);
+        { // Add posDup flag
+          byte posDupByte = (byte)(event.getPossibleDuplicate()?0x01:0x00);
+          getMessage().addBytesPart(new byte[] {posDupByte});
+        }
+        if (action >= 0 && action <= 3) {
+          // 0 = create
+          // 1 = update
+          // 2 = destroy
+          String regionName = event.getRegionPath();
+          EventID eventId = event.getEventId();
+          Object key = event.getKey();
+          Object callbackArg = event.getSenderCallbackArgument();
+
+          // Add region name
+          getMessage().addStringPart(regionName);
+          // Add event id
+          getMessage().addObjPart(eventId);
+          // Add key
+          getMessage().addStringOrObjPart(key);
+          if (action < 2 /* it is 0 or 1 */) {
+            byte[] value = event.getSerializedValue();
+            byte valueIsObject = event.getValueIsObject();;
+            // Add value (which is already a serialized byte[])
+            getMessage().addRawPart(value, (valueIsObject == 0x01));
+          }
+          // Add callback arg if necessary
+          if (callbackArg == null) {
+            getMessage().addBytesPart(new byte[] {0x00});
+          } else {
+            getMessage().addBytesPart(new byte[] {0x01});
+            getMessage().addObjPart(callbackArg);
+          }
+          getMessage().addLongPart(event.getVersionTimeStamp());
+        }
+      }
+    }
+
+    public GatewaySenderGFEBatchOpImpl() {
+      super(MessageType.GATEWAY_RECEIVER_COMMAND, 0);
+    }
+
+    @Override
+    public Object attempt(Connection cnx) throws Exception {
+      if (getMessage().getNumberOfParts() == 0) {
+        return attemptRead(cnx);
+      }
+      this.failed = true;
+      this.timedOut = false;
+      long start = startAttempt(cnx.getStats());
+      try {
+        try {
+          attemptSend(cnx);
+          this.failed = false;
+        } finally {
+          endSendAttempt(cnx.getStats(), start);
+        }
+      } finally {
+        endAttempt(cnx.getStats(), start);
+      }
+      return this.failed;
+    }
+    
+    private Object attemptRead(Connection cnx) throws Exception {
+      this.failed = true;
+      try {
+        Object result = attemptReadResponse(cnx);
+        this.failed = false;
+        return result;
+      } catch (SocketTimeoutException ste) {
+        this.failed = false;
+        this.timedOut = true;
+        throw ste;
+      } catch (Exception e) {
+        throw e;
+      }
+    }
+    
+    
+    /**
+     * Attempts to read a response to this operation by reading it from the
+     * given connection, and returning it.
+     * @param cnx the connection to read the response from
+     * @return the result of the operation
+     *         or <code>null</code> if the operation has no result.
+     * @throws Exception if the execute failed
+     */
+    protected Object attemptReadResponse(Connection cnx) throws Exception {
+      Message msg = createResponseMessage();
+      if (msg != null) {
+        msg.setComms(cnx.getSocket(), cnx.getInputStream(),
+            cnx.getOutputStream(),
+            ((ConnectionImpl)cnx).getCommBufferForAsyncRead(), cnx.getStats());
+        if (msg instanceof ChunkedMessage) {
+          try {
+            return processResponse(msg, cnx);
+          } finally {
+            msg.unsetComms();
+            // TODO (ashetkar) Handle the case when we fail to read the
+            // connection id.
+            processSecureBytes(cnx, msg);
+          }
+        }
+
+        try {
+          msg.recv();
+        } finally {
+          msg.unsetComms();
+          processSecureBytes(cnx, msg);
+        }
+        return processResponse(msg, cnx);
+      }
+      
+      return null;
+    }
+    
+    
+    private static int calcPartCount(List events) {
+      int numberOfParts = 4; // for the number of events and the batchId
+      for (Iterator i = events.iterator(); i.hasNext();) {
+        GatewaySenderEventImpl event = (GatewaySenderEventImpl)i.next();
+        numberOfParts += event.getNumberOfParts();
+      }
+      return numberOfParts;
+    }
+
+    @Override
+    protected void processSecureBytes(Connection cnx, Message message)
+        throws Exception {
+    }
+
+    @Override
+    protected boolean needsUserId() {
+      return false;
+    }
+
+    @Override
+    protected void sendMessage(Connection cnx) throws Exception {
+      getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+      getMessage().send(false);
+    }
+
+    @Override
+    protected Object processResponse(Message msg) throws Exception {
+      GatewayAck ack = null;
+      try {
+        // Read the header which describes the type of message following
+        switch (msg.getMessageType()) {
+        case MessageType.REPLY:
+          // Read the chunk
+          int batchId = msg.getPart(0).getInt();
+          int numEvents = msg.getPart(1).getInt();
+          ack = new GatewayAck(batchId, numEvents);
+          break;
+        case MessageType.EXCEPTION:
+          Part part0 = msg.getPart(0);
+
+          Object obj = part0.getObject();
+          if (obj instanceof List) {
+            List<BatchException70> l = (List<BatchException70>)part0.getObject();
+
+            if (logger.isDebugEnabled()) {
+              logger.debug("We got an exception from the GatewayReceiver. MessageType : {} obj :{}", msg.getMessageType(), obj);
+            }
+            // don't throw Exception but set it in the Ack
+            BatchException70 be = new BatchException70(l);
+            ack = new GatewayAck(be, l.get(0).getBatchId());
+
+          } else if (obj instanceof Throwable) {
+            String s = ": While reading Ack from receiver "
+                + ((Throwable)obj).getMessage();
+            throw new ServerOperationException(s, (Throwable)obj);
+          }
+          break;
+        default:
+          throw new InternalGemFireError(
+              LocalizedStrings.Op_UNKNOWN_MESSAGE_TYPE_0
+                  .toLocalizedString(Integer.valueOf(msg.getMessageType())));
+        }
+      } finally {
+        msg.clear();
+      }
+      return ack;
+    }
+    
+    @Override
+    protected boolean isErrorResponse(int msgType) {
+      return false;
+    }
+    @Override
+    protected long startAttempt(ConnectionStats stats) {
+      return stats.startGatewayBatch();
+    }
+    @Override
+    protected void endSendAttempt(ConnectionStats stats, long start) {
+      stats.endGatewayBatchSend(start, hasFailed());
+    }
+    @Override
+    protected void endAttempt(ConnectionStats stats, long start) {
+      stats.endGatewayBatch(start, hasTimedOut(), hasFailed());
+    }
+    
+    @Override
+    public boolean isGatewaySenderOp() {
+      return true;
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/SenderProxy.java
----------------------------------------------------------------------
diff --git a/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/SenderProxy.java b/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/SenderProxy.java
new file mode 100644
index 0000000..d2030b3
--- /dev/null
+++ b/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/SenderProxy.java
@@ -0,0 +1,34 @@
+/*=========================================================================
+ * Copyright (c) 2002-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+package com.gemstone.gemfire.cache.client.internal;
+
+import java.util.List;
+
+import com.gemstone.gemfire.cache.query.SelectResults;
+import com.gemstone.gemfire.distributed.internal.ServerLocation;
+
+/**
+ * Used to send operations from a sender to a receiver.
+ * @author skumar
+ * @since 8.1
+ */
+public class SenderProxy extends ServerProxy{
+  public SenderProxy(InternalPool pool) {
+    super(pool);
+  }
+
+  public void dispatchBatch_NewWAN(Connection con, List events, int batchId, boolean removeFromQueueOnException, boolean isRetry)
+  {
+    GatewaySenderBatchOp.executeOn(con, this.pool, events, batchId, removeFromQueueOnException, isRetry);
+  }
+  
+  public Object receiveAckFromReceiver(Connection con)
+  {
+    return GatewaySenderBatchOp.executeOn(con, this.pool);
+  }
+}


Mime
View raw message