geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kl...@apache.org
Subject [01/14] geode git commit: GEODE-2632: refactoring preparations for SecurityService and BaseCommand changes
Date Wed, 31 May 2017 23:12:49 GMT
Repository: geode
Updated Branches:
  refs/heads/develop f7e1ba2dc -> d1ec508ee


http://git-wip-us.apache.org/repos/asf/geode/blob/d1ec508e/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/CloseCQ.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/CloseCQ.java
b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/CloseCQ.java
index ac9b5da..18929eb 100644
--- a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/CloseCQ.java
+++ b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/CloseCQ.java
@@ -44,30 +44,32 @@ public class CloseCQ extends BaseCQCommand {
   private CloseCQ() {}
 
   @Override
-  public void cmdExecute(Message msg, ServerConnection servConn, long start) throws IOException
{
-    CachedRegionHelper crHelper = servConn.getCachedRegionHelper();
-    ClientProxyMembershipID id = servConn.getProxyID();
-    CacheServerStats stats = servConn.getCacheServerStats();
+  public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start)
+      throws IOException {
+    CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper();
+    ClientProxyMembershipID id = serverConnection.getProxyID();
+    CacheServerStats stats = serverConnection.getCacheServerStats();
 
     // Based on MessageType.QUERY
     // Added by Rao 2/1/2007
-    servConn.setAsTrue(REQUIRES_RESPONSE);
-    servConn.setAsTrue(REQUIRES_CHUNKED_RESPONSE);
+    serverConnection.setAsTrue(REQUIRES_RESPONSE);
+    serverConnection.setAsTrue(REQUIRES_CHUNKED_RESPONSE);
 
     start = DistributionStats.getStatTime();
     // Retrieve the data from the message parts
-    String cqName = msg.getPart(0).getString();
+    String cqName = clientMessage.getPart(0).getString();
 
     if (logger.isDebugEnabled()) {
-      logger.debug("{}: Received close CQ request from {} cqName: {}", servConn.getName(),
-          servConn.getSocketString(), cqName);
+      logger.debug("{}: Received close CQ request from {} cqName: {}", serverConnection.getName(),
+          serverConnection.getSocketString(), cqName);
     }
 
     // Process the query request
     if (cqName == null) {
       String err =
           LocalizedStrings.CloseCQ_THE_CQNAME_FOR_THE_CQ_CLOSE_REQUEST_IS_NULL.toLocalizedString();
-      sendCqResponse(MessageType.CQDATAERROR_MSG_TYPE, err, msg.getTransactionId(), null,
servConn);
+      sendCqResponse(MessageType.CQDATAERROR_MSG_TYPE, err, clientMessage.getTransactionId(),
null,
+          serverConnection);
       return;
     }
 
@@ -85,7 +87,7 @@ public class CloseCQ extends BaseCQCommand {
       }
       InternalCqQuery cqQuery = cqService.getCq(serverCqName);
 
-      AuthorizeRequest authzRequest = servConn.getAuthzRequest();
+      AuthorizeRequest authzRequest = serverConnection.getAuthzRequest();
       if (authzRequest != null) {
         String queryStr = null;
         Set cqRegionNames = null;
@@ -102,22 +104,24 @@ public class CloseCQ extends BaseCQCommand {
       // getMembershipID());
       cqService.closeCq(cqName, id);
       if (cqQuery != null)
-        servConn.removeCq(cqName, cqQuery.isDurable());
+        serverConnection.removeCq(cqName, cqQuery.isDurable());
     } catch (CqException cqe) {
-      sendCqResponse(MessageType.CQ_EXCEPTION_TYPE, "", msg.getTransactionId(), cqe, servConn);
+      sendCqResponse(MessageType.CQ_EXCEPTION_TYPE, "", clientMessage.getTransactionId(),
cqe,
+          serverConnection);
       return;
     } catch (Exception e) {
       String err =
           LocalizedStrings.CloseCQ_EXCEPTION_WHILE_CLOSING_CQ_CQNAME_0.toLocalizedString(cqName);
-      sendCqResponse(MessageType.CQ_EXCEPTION_TYPE, err, msg.getTransactionId(), e, servConn);
+      sendCqResponse(MessageType.CQ_EXCEPTION_TYPE, err, clientMessage.getTransactionId(),
e,
+          serverConnection);
       return;
     }
 
     // Send OK to client
     sendCqResponse(MessageType.REPLY,
-        LocalizedStrings.CloseCQ_CQ_CLOSED_SUCCESSFULLY.toLocalizedString(), msg.getTransactionId(),
-        null, servConn);
-    servConn.setAsTrue(RESPONDED);
+        LocalizedStrings.CloseCQ_CQ_CLOSED_SUCCESSFULLY.toLocalizedString(),
+        clientMessage.getTransactionId(), null, serverConnection);
+    serverConnection.setAsTrue(RESPONDED);
 
     {
       long oldStart = start;

http://git-wip-us.apache.org/repos/asf/geode/blob/d1ec508e/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ.java
b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ.java
index 9bddbc7..86d53f5 100644
--- a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ.java
+++ b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ.java
@@ -52,28 +52,28 @@ public class ExecuteCQ extends BaseCQCommand {
   private ExecuteCQ() {}
 
   @Override
-  public void cmdExecute(Message msg, ServerConnection servConn, long start)
+  public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start)
       throws IOException, InterruptedException {
-    AcceptorImpl acceptor = servConn.getAcceptor();
-    CachedRegionHelper crHelper = servConn.getCachedRegionHelper();
-    ClientProxyMembershipID id = servConn.getProxyID();
-    CacheServerStats stats = servConn.getCacheServerStats();
+    AcceptorImpl acceptor = serverConnection.getAcceptor();
+    CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper();
+    ClientProxyMembershipID id = serverConnection.getProxyID();
+    CacheServerStats stats = serverConnection.getCacheServerStats();
 
-    servConn.setAsTrue(REQUIRES_RESPONSE);
-    servConn.setAsTrue(REQUIRES_CHUNKED_RESPONSE);
+    serverConnection.setAsTrue(REQUIRES_RESPONSE);
+    serverConnection.setAsTrue(REQUIRES_CHUNKED_RESPONSE);
 
     // Retrieve the data from the message parts
-    String cqName = msg.getPart(0).getString();
-    String cqQueryString = msg.getPart(1).getString();
-    int cqState = msg.getPart(2).getInt();
+    String cqName = clientMessage.getPart(0).getString();
+    String cqQueryString = clientMessage.getPart(1).getString();
+    int cqState = clientMessage.getPart(2).getInt();
 
-    Part isDurablePart = msg.getPart(3);
+    Part isDurablePart = clientMessage.getPart(3);
     byte[] isDurableByte = isDurablePart.getSerializedForm();
     boolean isDurable = (isDurableByte == null || isDurableByte[0] == 0) ? false : true;
     if (logger.isDebugEnabled()) {
-      logger.debug("{}: Received {} request from {} CqName: {} queryString: {}", servConn.getName(),
-          MessageType.getString(msg.getMessageType()), servConn.getSocketString(), cqName,
-          cqQueryString);
+      logger.debug("{}: Received {} request from {} CqName: {} queryString: {}",
+          serverConnection.getName(), MessageType.getString(clientMessage.getMessageType()),
+          serverConnection.getSocketString(), cqName, cqQueryString);
     }
 
     DefaultQueryService qService = null;
@@ -87,7 +87,7 @@ public class ExecuteCQ extends BaseCQCommand {
       qService = (DefaultQueryService) crHelper.getCache().getLocalQueryService();
 
       // Authorization check
-      AuthorizeRequest authzRequest = servConn.getAuthzRequest();
+      AuthorizeRequest authzRequest = serverConnection.getAuthzRequest();
       if (authzRequest != null) {
         query = qService.newQuery(cqQueryString);
         cqRegionNames = ((DefaultQuery) query).getRegionsInQuery(null);
@@ -108,10 +108,11 @@ public class ExecuteCQ extends BaseCQCommand {
       cqQuery = cqServiceForExec.executeCq(cqName, cqQueryString, cqState, id,
           acceptor.getCacheClientNotifier(), isDurable, false, 0, null);
     } catch (CqException cqe) {
-      sendCqResponse(MessageType.CQ_EXCEPTION_TYPE, "", msg.getTransactionId(), cqe, servConn);
+      sendCqResponse(MessageType.CQ_EXCEPTION_TYPE, "", clientMessage.getTransactionId(),
cqe,
+          serverConnection);
       return;
     } catch (Exception e) {
-      writeChunkedException(msg, e, false, servConn);
+      writeChunkedException(clientMessage, e, serverConnection);
       return;
     }
 
@@ -119,7 +120,7 @@ public class ExecuteCQ extends BaseCQCommand {
     boolean sendResults = false;
     boolean successQuery = false;
 
-    if (msg.getMessageType() == MessageType.EXECUTECQ_WITH_IR_MSG_TYPE) {
+    if (clientMessage.getMessageType() == MessageType.EXECUTECQ_WITH_IR_MSG_TYPE) {
       sendResults = true;
     }
 
@@ -130,8 +131,8 @@ public class ExecuteCQ extends BaseCQCommand {
         cqRegionNames = ((DefaultQuery) query).getRegionsInQuery(null);
       }
       ((DefaultQuery) query).setIsCqQuery(true);
-      successQuery = processQuery(msg, query, cqQueryString, cqRegionNames, start, cqQuery,
-          executeCQContext, servConn, sendResults);
+      successQuery = processQuery(clientMessage, query, cqQueryString, cqRegionNames, start,
+          cqQuery, executeCQContext, serverConnection, sendResults);
 
       // Update the CQ statistics.
       cqQuery.getVsdStats().setCqInitialResultsTime((DistributionStats.getStatTime()) - oldstart);
@@ -153,12 +154,12 @@ public class ExecuteCQ extends BaseCQCommand {
       // Send OK to client
       sendCqResponse(MessageType.REPLY,
           LocalizedStrings.ExecuteCQ_CQ_CREATED_SUCCESSFULLY.toLocalizedString(),
-          msg.getTransactionId(), null, servConn);
+          clientMessage.getTransactionId(), null, serverConnection);
 
       long start2 = DistributionStats.getStatTime();
       stats.incProcessCreateCqTime(start2 - oldstart);
     }
-    servConn.setAsTrue(RESPONDED);
+    serverConnection.setAsTrue(RESPONDED);
   }
 
 }

http://git-wip-us.apache.org/repos/asf/geode/blob/d1ec508e/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ61.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ61.java
b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ61.java
index de61445..fcc45de 100755
--- a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ61.java
+++ b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ61.java
@@ -60,31 +60,31 @@ public class ExecuteCQ61 extends BaseCQCommand {
   private ExecuteCQ61() {}
 
   @Override
-  public void cmdExecute(Message msg, ServerConnection servConn, long start)
+  public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start)
       throws IOException, InterruptedException {
-    AcceptorImpl acceptor = servConn.getAcceptor();
-    CachedRegionHelper crHelper = servConn.getCachedRegionHelper();
-    ClientProxyMembershipID id = servConn.getProxyID();
-    CacheServerStats stats = servConn.getCacheServerStats();
+    AcceptorImpl acceptor = serverConnection.getAcceptor();
+    CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper();
+    ClientProxyMembershipID id = serverConnection.getProxyID();
+    CacheServerStats stats = serverConnection.getCacheServerStats();
 
-    servConn.setAsTrue(REQUIRES_RESPONSE);
-    servConn.setAsTrue(REQUIRES_CHUNKED_RESPONSE);
+    serverConnection.setAsTrue(REQUIRES_RESPONSE);
+    serverConnection.setAsTrue(REQUIRES_CHUNKED_RESPONSE);
 
     // Retrieve the data from the message parts
-    String cqName = msg.getPart(0).getString();
-    String cqQueryString = msg.getPart(1).getString();
-    int cqState = msg.getPart(2).getInt();
+    String cqName = clientMessage.getPart(0).getString();
+    String cqQueryString = clientMessage.getPart(1).getString();
+    int cqState = clientMessage.getPart(2).getInt();
 
-    Part isDurablePart = msg.getPart(3);
+    Part isDurablePart = clientMessage.getPart(3);
     byte[] isDurableByte = isDurablePart.getSerializedForm();
     boolean isDurable = (isDurableByte == null || isDurableByte[0] == 0) ? false : true;
     // region data policy
-    Part regionDataPolicyPart = msg.getPart(msg.getNumberOfParts() - 1);
+    Part regionDataPolicyPart = clientMessage.getPart(clientMessage.getNumberOfParts() -
1);
     byte[] regionDataPolicyPartBytes = regionDataPolicyPart.getSerializedForm();
     if (logger.isDebugEnabled()) {
-      logger.debug("{}: Received {} request from {} CqName: {} queryString: {}", servConn.getName(),
-          MessageType.getString(msg.getMessageType()), servConn.getSocketString(), cqName,
-          cqQueryString);
+      logger.debug("{}: Received {} request from {} CqName: {} queryString: {}",
+          serverConnection.getName(), MessageType.getString(clientMessage.getMessageType()),
+          serverConnection.getSocketString(), cqName, cqQueryString);
     }
 
     // Check if the Server is running in NotifyBySubscription=true mode.
@@ -96,8 +96,8 @@ public class ExecuteCQ61 extends BaseCQCommand {
         String err =
             LocalizedStrings.ExecuteCQ_SERVER_NOTIFYBYSUBSCRIPTION_MODE_IS_SET_TO_FALSE_CQ_EXECUTION_IS_NOT_SUPPORTED_IN_THIS_MODE
                 .toLocalizedString();
-        sendCqResponse(MessageType.CQDATAERROR_MSG_TYPE, err, msg.getTransactionId(), null,
-            servConn);
+        sendCqResponse(MessageType.CQDATAERROR_MSG_TYPE, err, clientMessage.getTransactionId(),
+            null, serverConnection);
         return;
       }
     }
@@ -113,7 +113,7 @@ public class ExecuteCQ61 extends BaseCQCommand {
       qService = (DefaultQueryService) crHelper.getCache().getLocalQueryService();
 
       // Authorization check
-      AuthorizeRequest authzRequest = servConn.getAuthzRequest();
+      AuthorizeRequest authzRequest = serverConnection.getAuthzRequest();
       if (authzRequest != null) {
         query = qService.newQuery(cqQueryString);
         cqRegionNames = ((DefaultQuery) query).getRegionsInQuery(null);
@@ -141,16 +141,17 @@ public class ExecuteCQ61 extends BaseCQCommand {
       // registering cq with serverConnection so that when CCP will require auth info it
can access
       // that
       // registering cq auth before as possibility that you may get event
-      servConn.setCq(cqName, isDurable);
+      serverConnection.setCq(cqName, isDurable);
       cqQuery = (ServerCQImpl) cqServiceForExec.executeCq(cqName, cqQueryString, cqState,
id, ccn,
           isDurable, true, regionDataPolicyPartBytes[0], null);
     } catch (CqException cqe) {
-      sendCqResponse(MessageType.CQ_EXCEPTION_TYPE, "", msg.getTransactionId(), cqe, servConn);
-      servConn.removeCq(cqName, isDurable);
+      sendCqResponse(MessageType.CQ_EXCEPTION_TYPE, "", clientMessage.getTransactionId(),
cqe,
+          serverConnection);
+      serverConnection.removeCq(cqName, isDurable);
       return;
     } catch (Exception e) {
-      writeChunkedException(msg, e, false, servConn);
-      servConn.removeCq(cqName, isDurable);
+      writeChunkedException(clientMessage, e, serverConnection);
+      serverConnection.removeCq(cqName, isDurable);
       return;
     }
 
@@ -158,7 +159,7 @@ public class ExecuteCQ61 extends BaseCQCommand {
     boolean sendResults = false;
     boolean successQuery = false;
 
-    if (msg.getMessageType() == MessageType.EXECUTECQ_WITH_IR_MSG_TYPE) {
+    if (clientMessage.getMessageType() == MessageType.EXECUTECQ_WITH_IR_MSG_TYPE) {
       sendResults = true;
     }
 
@@ -173,8 +174,8 @@ public class ExecuteCQ61 extends BaseCQCommand {
           cqRegionNames = ((DefaultQuery) query).getRegionsInQuery(null);
         }
         ((DefaultQuery) query).setIsCqQuery(true);
-        successQuery = processQuery(msg, query, cqQueryString, cqRegionNames, start, cqQuery,
-            executeCQContext, servConn, sendResults);
+        successQuery = processQuery(clientMessage, query, cqQueryString, cqRegionNames, start,
+            cqQuery, executeCQContext, serverConnection, sendResults);
 
 
         // Update the CQ statistics.
@@ -203,12 +204,12 @@ public class ExecuteCQ61 extends BaseCQCommand {
       // Send OK to client
       sendCqResponse(MessageType.REPLY,
           LocalizedStrings.ExecuteCQ_CQ_CREATED_SUCCESSFULLY.toLocalizedString(),
-          msg.getTransactionId(), null, servConn);
+          clientMessage.getTransactionId(), null, serverConnection);
 
       long start2 = DistributionStats.getStatTime();
       stats.incProcessCreateCqTime(start2 - oldstart);
     }
-    servConn.setAsTrue(RESPONDED);
+    serverConnection.setAsTrue(RESPONDED);
 
   }
 

http://git-wip-us.apache.org/repos/asf/geode/blob/d1ec508e/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetCQStats.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetCQStats.java
b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetCQStats.java
index 69be347..0fab303 100644
--- a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetCQStats.java
+++ b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetCQStats.java
@@ -36,32 +36,34 @@ public class GetCQStats extends BaseCQCommand {
   private GetCQStats() {}
 
   @Override
-  public void cmdExecute(Message msg, ServerConnection servConn, long start) throws IOException
{
-    CachedRegionHelper crHelper = servConn.getCachedRegionHelper();
+  public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start)
+      throws IOException {
+    CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper();
 
-    CacheServerStats stats = servConn.getCacheServerStats();
+    CacheServerStats stats = serverConnection.getCacheServerStats();
 
-    servConn.setAsTrue(REQUIRES_RESPONSE);
-    servConn.setAsTrue(REQUIRES_CHUNKED_RESPONSE);
+    serverConnection.setAsTrue(REQUIRES_RESPONSE);
+    serverConnection.setAsTrue(REQUIRES_CHUNKED_RESPONSE);
 
     final boolean isDebugEnabled = logger.isDebugEnabled();
     if (isDebugEnabled) {
-      logger.debug("{}: Received close all client CQs request from {}", servConn.getName(),
-          servConn.getSocketString());
+      logger.debug("{}: Received close all client CQs request from {}", serverConnection.getName(),
+          serverConnection.getSocketString());
     }
 
     // Retrieve the data from the message parts
-    String cqName = msg.getPart(0).getString();
+    String cqName = clientMessage.getPart(0).getString();
 
     if (isDebugEnabled) {
-      logger.debug("{}: Received close CQ request from {} cqName: {}", servConn.getName(),
-          servConn.getSocketString(), cqName);
+      logger.debug("{}: Received close CQ request from {} cqName: {}", serverConnection.getName(),
+          serverConnection.getSocketString(), cqName);
     }
 
     // Process the query request
     if (cqName == null) {
       String err = "The cqName for the cq stats request is null";
-      sendCqResponse(MessageType.CQDATAERROR_MSG_TYPE, err, msg.getTransactionId(), null,
servConn);
+      sendCqResponse(MessageType.CQDATAERROR_MSG_TYPE, err, clientMessage.getTransactionId(),
null,
+          serverConnection);
       return;
     }
 
@@ -74,13 +76,14 @@ public class GetCQStats extends BaseCQCommand {
       cqService.start();
     } catch (Exception e) {
       String err = "Exception while Getting the CQ Statistics. ";
-      sendCqResponse(MessageType.CQ_EXCEPTION_TYPE, err, msg.getTransactionId(), e, servConn);
+      sendCqResponse(MessageType.CQ_EXCEPTION_TYPE, err, clientMessage.getTransactionId(),
e,
+          serverConnection);
       return;
     }
     // Send OK to client
-    sendCqResponse(MessageType.REPLY, "cq stats sent successfully.", msg.getTransactionId(),
null,
-        servConn);
-    servConn.setAsTrue(RESPONDED);
+    sendCqResponse(MessageType.REPLY, "cq stats sent successfully.",
+        clientMessage.getTransactionId(), null, serverConnection);
+    serverConnection.setAsTrue(RESPONDED);
 
     {
       long oldStart = start;

http://git-wip-us.apache.org/repos/asf/geode/blob/d1ec508e/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetDurableCQs.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetDurableCQs.java
b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetDurableCQs.java
index a2d201d..c14bd72 100755
--- a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetDurableCQs.java
+++ b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetDurableCQs.java
@@ -44,19 +44,20 @@ public class GetDurableCQs extends BaseCQCommand {
   private GetDurableCQs() {}
 
   @Override
-  public void cmdExecute(Message msg, ServerConnection servConn, long start)
+  public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start)
       throws IOException, InterruptedException {
-    AcceptorImpl acceptor = servConn.getAcceptor();
-    CachedRegionHelper crHelper = servConn.getCachedRegionHelper();
-    ClientProxyMembershipID id = servConn.getProxyID();
-    CacheServerStats stats = servConn.getCacheServerStats();
+    AcceptorImpl acceptor = serverConnection.getAcceptor();
+    CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper();
+    ClientProxyMembershipID id = serverConnection.getProxyID();
+    CacheServerStats stats = serverConnection.getCacheServerStats();
 
-    servConn.setAsTrue(REQUIRES_RESPONSE);
-    servConn.setAsTrue(REQUIRES_CHUNKED_RESPONSE);
+    serverConnection.setAsTrue(REQUIRES_RESPONSE);
+    serverConnection.setAsTrue(REQUIRES_CHUNKED_RESPONSE);
 
     if (logger.isDebugEnabled()) {
-      logger.debug("{}: Received {} request from {}", servConn.getName(),
-          MessageType.getString(msg.getMessageType()), servConn.getSocketString());
+      logger.debug("{}: Received {} request from {}", serverConnection.getName(),
+          MessageType.getString(clientMessage.getMessageType()),
+          serverConnection.getSocketString());
     }
 
     DefaultQueryService qService = null;
@@ -68,7 +69,7 @@ public class GetDurableCQs extends BaseCQCommand {
       this.securityService.authorizeClusterRead();
 
       // Authorization check
-      AuthorizeRequest authzRequest = servConn.getAuthzRequest();
+      AuthorizeRequest authzRequest = serverConnection.getAuthzRequest();
       if (authzRequest != null) {
         authzRequest.getDurableCQsAuthorize();
       }
@@ -76,34 +77,35 @@ public class GetDurableCQs extends BaseCQCommand {
       cqServiceForExec = qService.getCqService();
       List<String> durableCqs = cqServiceForExec.getAllDurableClientCqs(id);
 
-      ChunkedMessage chunkedResponseMsg = servConn.getChunkedResponseMessage();
+      ChunkedMessage chunkedResponseMsg = serverConnection.getChunkedResponseMessage();
       chunkedResponseMsg.setMessageType(MessageType.RESPONSE);
-      chunkedResponseMsg.setTransactionId(msg.getTransactionId());
+      chunkedResponseMsg.setTransactionId(clientMessage.getTransactionId());
       chunkedResponseMsg.sendHeader();
 
-      List durableCqList = new ArrayList(maximumChunkSize);
+      List durableCqList = new ArrayList(MAXIMUM_CHUNK_SIZE);
       final boolean isTraceEnabled = logger.isTraceEnabled();
       for (Iterator it = durableCqs.iterator(); it.hasNext();) {
         Object durableCqName = it.next();
         durableCqList.add(durableCqName);
         if (isTraceEnabled) {
-          logger.trace("{}: getDurableCqsResponse <{}>; list size was {}", servConn.getName(),
-              durableCqName, durableCqList.size());
+          logger.trace("{}: getDurableCqsResponse <{}>; list size was {}",
+              serverConnection.getName(), durableCqName, durableCqList.size());
         }
-        if (durableCqList.size() == maximumChunkSize) {
+        if (durableCqList.size() == MAXIMUM_CHUNK_SIZE) {
           // Send the chunk and clear the list
-          sendDurableCqsResponseChunk(durableCqList, false, servConn);
+          sendDurableCqsResponseChunk(durableCqList, false, serverConnection);
           durableCqList.clear();
         }
       }
       // Send the last chunk even if the list is of zero size.
-      sendDurableCqsResponseChunk(durableCqList, true, servConn);
+      sendDurableCqsResponseChunk(durableCqList, true, serverConnection);
 
     } catch (CqException cqe) {
-      sendCqResponse(MessageType.CQ_EXCEPTION_TYPE, "", msg.getTransactionId(), cqe, servConn);
+      sendCqResponse(MessageType.CQ_EXCEPTION_TYPE, "", clientMessage.getTransactionId(),
cqe,
+          serverConnection);
       return;
     } catch (Exception e) {
-      writeChunkedException(msg, e, false, servConn);
+      writeChunkedException(clientMessage, e, serverConnection);
       return;
     }
   }
@@ -114,7 +116,7 @@ public class GetDurableCQs extends BaseCQCommand {
 
     chunkedResponseMsg.setNumberOfParts(1);
     chunkedResponseMsg.setLastChunk(lastChunk);
-    chunkedResponseMsg.addObjPart(list, zipValues);
+    chunkedResponseMsg.addObjPart(list, false);
 
     if (logger.isDebugEnabled()) {
       logger.debug("{}: Sending {} durableCQs response chunk{}", servConn.getName(),

http://git-wip-us.apache.org/repos/asf/geode/blob/d1ec508e/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/MonitorCQ.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/MonitorCQ.java
b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/MonitorCQ.java
index a8fec9f..5f0118b 100644
--- a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/MonitorCQ.java
+++ b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/MonitorCQ.java
@@ -36,39 +36,42 @@ public class MonitorCQ extends BaseCQCommand {
   private MonitorCQ() {}
 
   @Override
-  public void cmdExecute(Message msg, ServerConnection servConn, long start) throws IOException
{
-    CachedRegionHelper crHelper = servConn.getCachedRegionHelper();
-    servConn.setAsTrue(REQUIRES_RESPONSE);
-    servConn.setAsTrue(REQUIRES_CHUNKED_RESPONSE);
+  public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start)
+      throws IOException {
+    CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper();
+    serverConnection.setAsTrue(REQUIRES_RESPONSE);
+    serverConnection.setAsTrue(REQUIRES_CHUNKED_RESPONSE);
 
-    int op = msg.getPart(0).getInt();
+    int op = clientMessage.getPart(0).getInt();
 
     if (op < 1) {
       // This should have been taken care at the client - remove?
       String err = LocalizedStrings.MonitorCQ__0_THE_MONITORCQ_OPERATION_IS_INVALID
-          .toLocalizedString(servConn.getName());
-      sendCqResponse(MessageType.CQDATAERROR_MSG_TYPE, err, msg.getTransactionId(), null,
servConn);
+          .toLocalizedString(serverConnection.getName());
+      sendCqResponse(MessageType.CQDATAERROR_MSG_TYPE, err, clientMessage.getTransactionId(),
null,
+          serverConnection);
       return;
     }
 
     String regionName = null;
-    if (msg.getNumberOfParts() == 2) {
+    if (clientMessage.getNumberOfParts() == 2) {
       // This will be enable/disable on region.
-      regionName = msg.getPart(1).getString();
+      regionName = clientMessage.getPart(1).getString();
       if (regionName == null) {
         // This should have been taken care at the client - remove?
         String err =
             LocalizedStrings.MonitorCQ__0_A_NULL_REGION_NAME_WAS_PASSED_FOR_MONITORCQ_OPERATION
-                .toLocalizedString(servConn.getName());
-        sendCqResponse(MessageType.CQDATAERROR_MSG_TYPE, err, msg.getTransactionId(), null,
-            servConn);
+                .toLocalizedString(serverConnection.getName());
+        sendCqResponse(MessageType.CQDATAERROR_MSG_TYPE, err, clientMessage.getTransactionId(),
+            null, serverConnection);
         return;
       }
     }
 
     if (logger.isDebugEnabled()) {
-      logger.debug("{}: Received MonitorCq request from {} op: {}{}", servConn.getName(),
-          servConn.getSocketString(), op, (regionName != null) ? " RegionName: " + regionName
: "");
+      logger.debug("{}: Received MonitorCq request from {} op: {}{}", serverConnection.getName(),
+          serverConnection.getSocketString(), op,
+          (regionName != null) ? " RegionName: " + regionName : "");
     }
 
     this.securityService.authorizeClusterRead();
@@ -85,12 +88,14 @@ public class MonitorCQ extends BaseCQCommand {
       throw new CqException(
           LocalizedStrings.CqService_INVALID_CQ_MONITOR_REQUEST_RECEIVED.toLocalizedString());
     } catch (CqException cqe) {
-      sendCqResponse(MessageType.CQ_EXCEPTION_TYPE, "", msg.getTransactionId(), cqe, servConn);
+      sendCqResponse(MessageType.CQ_EXCEPTION_TYPE, "", clientMessage.getTransactionId(),
cqe,
+          serverConnection);
       return;
     } catch (Exception e) {
       String err = LocalizedStrings.MonitorCQ_EXCEPTION_WHILE_HANDLING_THE_MONITOR_REQUEST_OP_IS_0
           .toLocalizedString(Integer.valueOf(op));
-      sendCqResponse(MessageType.CQ_EXCEPTION_TYPE, err, msg.getTransactionId(), e, servConn);
+      sendCqResponse(MessageType.CQ_EXCEPTION_TYPE, err, clientMessage.getTransactionId(),
e,
+          serverConnection);
       return;
     }
   }

http://git-wip-us.apache.org/repos/asf/geode/blob/d1ec508e/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/StopCQ.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/StopCQ.java
b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/StopCQ.java
index 94304d3..99fbef1 100644
--- a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/StopCQ.java
+++ b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/StopCQ.java
@@ -44,30 +44,32 @@ public class StopCQ extends BaseCQCommand {
   private StopCQ() {}
 
   @Override
-  public void cmdExecute(Message msg, ServerConnection servConn, long start) throws IOException
{
-    CachedRegionHelper crHelper = servConn.getCachedRegionHelper();
-    ClientProxyMembershipID id = servConn.getProxyID();
-    CacheServerStats stats = servConn.getCacheServerStats();
+  public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start)
+      throws IOException {
+    CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper();
+    ClientProxyMembershipID id = serverConnection.getProxyID();
+    CacheServerStats stats = serverConnection.getCacheServerStats();
 
     // Based on MessageType.QUERY
     // Added by Rao 2/1/2007
-    servConn.setAsTrue(REQUIRES_RESPONSE);
-    servConn.setAsTrue(REQUIRES_CHUNKED_RESPONSE);
+    serverConnection.setAsTrue(REQUIRES_RESPONSE);
+    serverConnection.setAsTrue(REQUIRES_CHUNKED_RESPONSE);
 
     start = DistributionStats.getStatTime();
     // Retrieve the data from the message parts
-    String cqName = msg.getPart(0).getString();
+    String cqName = clientMessage.getPart(0).getString();
 
     if (logger.isDebugEnabled()) {
-      logger.debug("{}: Received stop CQ request from {} cqName: {}", servConn.getName(),
-          servConn.getSocketString(), cqName);
+      logger.debug("{}: Received stop CQ request from {} cqName: {}", serverConnection.getName(),
+          serverConnection.getSocketString(), cqName);
     }
 
     // Process the query request
     if (cqName == null) {
       String err =
           LocalizedStrings.StopCQ_THE_CQNAME_FOR_THE_CQ_STOP_REQUEST_IS_NULL.toLocalizedString();
-      sendCqResponse(MessageType.CQDATAERROR_MSG_TYPE, err, msg.getTransactionId(), null,
servConn);
+      sendCqResponse(MessageType.CQDATAERROR_MSG_TYPE, err, clientMessage.getTransactionId(),
null,
+          serverConnection);
       return;
     }
 
@@ -86,7 +88,7 @@ public class StopCQ extends BaseCQCommand {
 
       this.securityService.authorizeDataManage();
 
-      AuthorizeRequest authzRequest = servConn.getAuthzRequest();
+      AuthorizeRequest authzRequest = serverConnection.getAuthzRequest();
       if (authzRequest != null) {
         String queryStr = null;
         Set cqRegionNames = null;
@@ -100,23 +102,25 @@ public class StopCQ extends BaseCQCommand {
       }
       cqService.stopCq(cqName, id);
       if (cqQuery != null)
-        servConn.removeCq(cqName, cqQuery.isDurable());
+        serverConnection.removeCq(cqName, cqQuery.isDurable());
     } catch (CqException cqe) {
-      sendCqResponse(MessageType.CQ_EXCEPTION_TYPE, "", msg.getTransactionId(), cqe, servConn);
+      sendCqResponse(MessageType.CQ_EXCEPTION_TYPE, "", clientMessage.getTransactionId(),
cqe,
+          serverConnection);
       return;
     } catch (Exception e) {
       String err =
           LocalizedStrings.StopCQ_EXCEPTION_WHILE_STOPPING_CQ_NAMED_0.toLocalizedString(cqName);
-      sendCqResponse(MessageType.CQ_EXCEPTION_TYPE, err, msg.getTransactionId(), e, servConn);
+      sendCqResponse(MessageType.CQ_EXCEPTION_TYPE, err, clientMessage.getTransactionId(),
e,
+          serverConnection);
       return;
     }
 
     // Send OK to client
     sendCqResponse(MessageType.REPLY,
-        LocalizedStrings.StopCQ_CQ_STOPPED_SUCCESSFULLY.toLocalizedString(), msg.getTransactionId(),
-        null, servConn);
+        LocalizedStrings.StopCQ_CQ_STOPPED_SUCCESSFULLY.toLocalizedString(),
+        clientMessage.getTransactionId(), null, serverConnection);
 
-    servConn.setAsTrue(RESPONDED);
+    serverConnection.setAsTrue(RESPONDED);
 
     {
       long oldStart = start;

http://git-wip-us.apache.org/repos/asf/geode/blob/d1ec508e/geode-junit/src/main/java/org/apache/geode/test/junit/categories/UnitTest.java
----------------------------------------------------------------------
diff --git a/geode-junit/src/main/java/org/apache/geode/test/junit/categories/UnitTest.java
b/geode-junit/src/main/java/org/apache/geode/test/junit/categories/UnitTest.java
index 5614212..58d12a0 100755
--- a/geode-junit/src/main/java/org/apache/geode/test/junit/categories/UnitTest.java
+++ b/geode-junit/src/main/java/org/apache/geode/test/junit/categories/UnitTest.java
@@ -20,7 +20,7 @@ package org.apache.geode.test.junit.categories;
  *
  * <p>
  * <ul>
- * A {@code UnitTest} should <bold>not<bold> do any of the following:
+ * A {@code UnitTest} should <strong>not</strong> do any of the following:
  * <li>communicate with a database
  * <li>communicate across the network
  * <li>access the file system

http://git-wip-us.apache.org/repos/asf/geode/blob/d1ec508e/geode-junit/src/main/java/org/apache/geode/test/junit/rules/serializable/SerializableErrorCollector.java
----------------------------------------------------------------------
diff --git a/geode-junit/src/main/java/org/apache/geode/test/junit/rules/serializable/SerializableErrorCollector.java
b/geode-junit/src/main/java/org/apache/geode/test/junit/rules/serializable/SerializableErrorCollector.java
new file mode 100644
index 0000000..5557f1b
--- /dev/null
+++ b/geode-junit/src/main/java/org/apache/geode/test/junit/rules/serializable/SerializableErrorCollector.java
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information
regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version
2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain
a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under
the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
express
+ * or implied. See the License for the specific language governing permissions and limitations
under
+ * the License.
+ */
+package org.apache.geode.test.junit.rules.serializable;
+
+import org.junit.rules.ErrorCollector;
+
+import java.io.Serializable;
+
+public class SerializableErrorCollector extends ErrorCollector implements Serializable {
+}

http://git-wip-us.apache.org/repos/asf/geode/blob/d1ec508e/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDUnitTest.java
b/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDUnitTest.java
index f403447..702e6c8 100644
--- a/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDUnitTest.java
+++ b/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDUnitTest.java
@@ -14,8 +14,10 @@
  */
 package org.apache.geode.internal.cache.wan.parallel;
 
+import static org.apache.geode.internal.cache.tier.sockets.Message.MAX_MESSAGE_SIZE_PROPERTY;
 import static org.apache.geode.test.dunit.Assert.*;
 
+import org.junit.Rule;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 
@@ -31,8 +33,8 @@ import org.apache.geode.test.dunit.IgnoredException;
 import org.apache.geode.test.dunit.LogWriterUtils;
 import org.apache.geode.test.dunit.RMIException;
 import org.apache.geode.test.dunit.Wait;
+import org.apache.geode.test.dunit.rules.DistributedRestoreSystemProperties;
 import org.apache.geode.test.junit.categories.DistributedTest;
-import org.apache.geode.test.junit.categories.FlakyTest;
 
 /**
  * DUnit test for operations on ParallelGatewaySender
@@ -40,6 +42,10 @@ import org.apache.geode.test.junit.categories.FlakyTest;
 @Category(DistributedTest.class)
 public class ParallelGatewaySenderOperationsDUnitTest extends WANTestBase {
 
+  @Rule
+  public DistributedRestoreSystemProperties restoreSystemProperties =
+      new DistributedRestoreSystemProperties();
+
   @Override
   protected final void postSetUpWANTestBase() throws Exception {
     IgnoredException.addIgnoredException("Broken pipe||Unexpected IOException");
@@ -582,13 +588,14 @@ public class ParallelGatewaySenderOperationsDUnitTest extends WANTestBase
{
 
   @Test
   public void testParallelGatewaySenderMessageTooLargeException() {
+    vm4.invoke(() -> System.setProperty(MAX_MESSAGE_SIZE_PROPERTY, String.valueOf(1024
* 1024)));
+
     Integer[] locatorPorts = createLNAndNYLocators();
     Integer lnPort = locatorPorts[0];
     Integer nyPort = locatorPorts[1];
 
     // Create and start sender with reduced maximum message size and 1 dispatcher thread
     String regionName = getTestMethodName() + "_PR";
-    vm4.invoke(() -> setMaximumMessageSize(1024 * 1024));
     vm4.invoke(() -> createCache(lnPort));
     vm4.invoke(() -> setNumDispatcherThreadsForTheRun(1));
     vm4.invoke(() -> createSender("ln", 2, true, 100, 100, false, false, null, false));
@@ -617,12 +624,6 @@ public class ParallelGatewaySenderOperationsDUnitTest extends WANTestBase
{
     ignoredGIOE.remove();
   }
 
-  private void setMaximumMessageSize(int maximumMessageSizeBytes) {
-    Message.MAX_MESSAGE_SIZE = maximumMessageSizeBytes;
-    LogWriterUtils.getLogWriter().info("Set gemfire.client.max-message-size: "
-        + System.getProperty(DistributionConfig.GEMFIRE_PREFIX + "client.max-message-size"));
-  }
-
   private void createSendersReceiversAndPartitionedRegion(Integer lnPort, Integer nyPort,
       boolean createAccessors, boolean startSenders) {
     // Note: This is a test-specific method used by several test to create


Mime
View raw message