http://git-wip-us.apache.org/repos/asf/geode/blob/01132014/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetFunctionAttribute.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetFunctionAttribute.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetFunctionAttribute.java index 8ec16ef..f56a4d9 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetFunctionAttribute.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetFunctionAttribute.java @@ -34,15 +34,15 @@ public class GetFunctionAttribute extends BaseCommand { } @Override - public void cmdExecute(Message msg, ServerConnection servConn, long start) throws IOException { - servConn.setAsTrue(REQUIRES_RESPONSE); - String functionId = msg.getPart(0).getString(); + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start) throws IOException { + serverConnection.setAsTrue(REQUIRES_RESPONSE); + String functionId = clientMessage.getPart(0).getString(); if (functionId == null) { String message = LocalizedStrings.GetFunctionAttribute_THE_INPUT_0_FOR_GET_FUNCTION_ATTRIBUTE_REQUEST_IS_NULL .toLocalizedString("functionId"); - logger.warn("{}: {}", servConn.getName(), message); - sendError(msg, message, servConn); + logger.warn("{}: {}", serverConnection.getName(), message); + sendError(clientMessage, message, serverConnection); return; } @@ -52,8 +52,8 @@ public class GetFunctionAttribute extends BaseCommand { message = LocalizedStrings.GetFunctionAttribute_THE_FUNCTION_IS_NOT_REGISTERED_FOR_FUNCTION_ID_0 .toLocalizedString(functionId); - logger.warn("{}: {}", servConn.getName(), message); - sendError(msg, message, servConn); + logger.warn("{}: {}", serverConnection.getName(), message); + sendError(clientMessage, message, serverConnection); return; } @@ -61,7 +61,7 @@ public class GetFunctionAttribute extends BaseCommand { functionAttributes[0] = (byte) (function.hasResult() ? 1 : 0); functionAttributes[1] = (byte) (function.isHA() ? 1 : 0); functionAttributes[2] = (byte) (function.optimizeForWrite() ? 1 : 0); - writeResponseWithFunctionAttribute(functionAttributes, msg, servConn); + writeResponseWithFunctionAttribute(functionAttributes, clientMessage, serverConnection); } private void sendError(Message msg, String message, ServerConnection servConn)
http://git-wip-us.apache.org/repos/asf/geode/blob/01132014/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPDXEnumById.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPDXEnumById.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPDXEnumById.java index 54a21ed..cc7dd05 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPDXEnumById.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPDXEnumById.java @@ -36,32 +36,32 @@ public class GetPDXEnumById extends BaseCommand { private GetPDXEnumById() {} @Override - public void cmdExecute(Message msg, ServerConnection servConn, long start) + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start) throws IOException, ClassNotFoundException { - servConn.setAsTrue(REQUIRES_RESPONSE); + serverConnection.setAsTrue(REQUIRES_RESPONSE); if (logger.isDebugEnabled()) { - logger.debug("{}: Received get pdx enum by id request ({} parts) from {}", servConn.getName(), - msg.getNumberOfParts(), servConn.getSocketString()); + logger.debug("{}: Received get pdx enum by id request ({} parts) from {}", serverConnection.getName(), + clientMessage.getNumberOfParts(), serverConnection.getSocketString()); } - int enumId = msg.getPart(0).getInt(); + int enumId = clientMessage.getPart(0).getInt(); EnumInfo result; try { - InternalCache cache = servConn.getCache(); + InternalCache cache = serverConnection.getCache(); TypeRegistry registry = cache.getPdxRegistry(); result = registry.getEnumInfoById(enumId); } catch (Exception e) { - writeException(msg, e, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } - Message responseMsg = servConn.getResponseMessage(); + Message responseMsg = serverConnection.getResponseMessage(); responseMsg.setMessageType(MessageType.RESPONSE); responseMsg.setNumberOfParts(1); - responseMsg.setTransactionId(msg.getTransactionId()); + responseMsg.setTransactionId(clientMessage.getTransactionId()); responseMsg.addObjPart(result); - responseMsg.send(servConn); - servConn.setAsTrue(RESPONDED); + responseMsg.send(serverConnection); + serverConnection.setAsTrue(RESPONDED); } } http://git-wip-us.apache.org/repos/asf/geode/blob/01132014/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPDXIdForEnum.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPDXIdForEnum.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPDXIdForEnum.java index 1b21383..7bf5b4f 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPDXIdForEnum.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPDXIdForEnum.java @@ -36,33 +36,33 @@ public class GetPDXIdForEnum extends BaseCommand { private GetPDXIdForEnum() {} @Override - public void cmdExecute(Message msg, ServerConnection servConn, long start) + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start) throws IOException, ClassNotFoundException { - servConn.setAsTrue(REQUIRES_RESPONSE); + serverConnection.setAsTrue(REQUIRES_RESPONSE); if (logger.isDebugEnabled()) { logger.debug("{}: Received get pdx id for enum request ({} parts) from {}", - servConn.getName(), msg.getNumberOfParts(), servConn.getSocketString()); + serverConnection.getName(), clientMessage.getNumberOfParts(), serverConnection.getSocketString()); } - EnumInfo enumInfo = (EnumInfo) msg.getPart(0).getObject(); + EnumInfo enumInfo = (EnumInfo) clientMessage.getPart(0).getObject(); int enumId; try { - InternalCache cache = servConn.getCache(); + InternalCache cache = serverConnection.getCache(); TypeRegistry registry = cache.getPdxRegistry(); enumId = registry.defineEnum(enumInfo); } catch (Exception e) { - writeException(msg, e, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } - Message responseMsg = servConn.getResponseMessage(); + Message responseMsg = serverConnection.getResponseMessage(); responseMsg.setMessageType(MessageType.RESPONSE); responseMsg.setNumberOfParts(1); - responseMsg.setTransactionId(msg.getTransactionId()); + responseMsg.setTransactionId(clientMessage.getTransactionId()); responseMsg.addIntPart(enumId); - responseMsg.send(servConn); - servConn.setAsTrue(RESPONDED); + responseMsg.send(serverConnection); + serverConnection.setAsTrue(RESPONDED); } } http://git-wip-us.apache.org/repos/asf/geode/blob/01132014/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPDXIdForType.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPDXIdForType.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPDXIdForType.java index 2054196..e5dc5f0 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPDXIdForType.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPDXIdForType.java @@ -36,34 +36,34 @@ public class GetPDXIdForType extends BaseCommand { private GetPDXIdForType() {} @Override - public void cmdExecute(Message msg, ServerConnection servConn, long start) + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start) throws IOException, ClassNotFoundException { - servConn.setAsTrue(REQUIRES_RESPONSE); + serverConnection.setAsTrue(REQUIRES_RESPONSE); if (logger.isDebugEnabled()) { logger.debug("{}: Received get pdx id for type request ({} parts) from {}", - servConn.getName(), msg.getNumberOfParts(), servConn.getSocketString()); + serverConnection.getName(), clientMessage.getNumberOfParts(), serverConnection.getSocketString()); } - int noOfParts = msg.getNumberOfParts(); + int noOfParts = clientMessage.getNumberOfParts(); - PdxType type = (PdxType) msg.getPart(0).getObject(); + PdxType type = (PdxType) clientMessage.getPart(0).getObject(); int pdxId; try { - InternalCache cache = servConn.getCache(); + InternalCache cache = serverConnection.getCache(); TypeRegistry registry = cache.getPdxRegistry(); pdxId = registry.defineType(type); } catch (Exception e) { - writeException(msg, e, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } - Message responseMsg = servConn.getResponseMessage(); + Message responseMsg = serverConnection.getResponseMessage(); responseMsg.setMessageType(MessageType.RESPONSE); responseMsg.setNumberOfParts(1); - responseMsg.setTransactionId(msg.getTransactionId()); + responseMsg.setTransactionId(clientMessage.getTransactionId()); responseMsg.addIntPart(pdxId); - responseMsg.send(servConn); - servConn.setAsTrue(RESPONDED); + responseMsg.send(serverConnection); + serverConnection.setAsTrue(RESPONDED); } } http://git-wip-us.apache.org/repos/asf/geode/blob/01132014/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPDXTypeById.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPDXTypeById.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPDXTypeById.java index 2470893..032e8b3 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPDXTypeById.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPDXTypeById.java @@ -36,32 +36,32 @@ public class GetPDXTypeById extends BaseCommand { private GetPDXTypeById() {} @Override - public void cmdExecute(Message msg, ServerConnection servConn, long start) + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start) throws IOException, ClassNotFoundException { - servConn.setAsTrue(REQUIRES_RESPONSE); + serverConnection.setAsTrue(REQUIRES_RESPONSE); if (logger.isDebugEnabled()) { - logger.debug("{}: Received get pdx type by id request ({} parts) from {}", servConn.getName(), - msg.getNumberOfParts(), servConn.getSocketString()); + logger.debug("{}: Received get pdx type by id request ({} parts) from {}", serverConnection.getName(), + clientMessage.getNumberOfParts(), serverConnection.getSocketString()); } - int pdxId = msg.getPart(0).getInt(); + int pdxId = clientMessage.getPart(0).getInt(); PdxType type; try { - InternalCache cache = servConn.getCache(); + InternalCache cache = serverConnection.getCache(); TypeRegistry registry = cache.getPdxRegistry(); type = registry.getType(pdxId); } catch (Exception e) { - writeException(msg, e, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } - Message responseMsg = servConn.getResponseMessage(); + Message responseMsg = serverConnection.getResponseMessage(); responseMsg.setMessageType(MessageType.RESPONSE); responseMsg.setNumberOfParts(1); - responseMsg.setTransactionId(msg.getTransactionId()); + responseMsg.setTransactionId(clientMessage.getTransactionId()); responseMsg.addObjPart(type); - responseMsg.send(servConn); - servConn.setAsTrue(RESPONDED); + responseMsg.send(serverConnection); + serverConnection.setAsTrue(RESPONDED); } } http://git-wip-us.apache.org/repos/asf/geode/blob/01132014/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPdxEnums70.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPdxEnums70.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPdxEnums70.java index 19551c4..7753584 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPdxEnums70.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPdxEnums70.java @@ -36,31 +36,31 @@ public class GetPdxEnums70 extends BaseCommand { private GetPdxEnums70() {} @Override - public void cmdExecute(Message msg, ServerConnection servConn, long start) + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start) throws IOException, ClassNotFoundException { - servConn.setAsTrue(REQUIRES_RESPONSE); + serverConnection.setAsTrue(REQUIRES_RESPONSE); if (logger.isDebugEnabled()) { - logger.debug("{}: Received get pdx enums from {}", servConn.getName(), - servConn.getSocketString()); + logger.debug("{}: Received get pdx enums from {}", serverConnection.getName(), + serverConnection.getSocketString()); } Map<Integer, EnumInfo> enums; try { - InternalCache cache = servConn.getCache(); + InternalCache cache = serverConnection.getCache(); enums = cache.getPdxRegistry().enumMap(); } catch (Exception e) { - writeException(msg, e, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } - Message responseMsg = servConn.getResponseMessage(); + Message responseMsg = serverConnection.getResponseMessage(); responseMsg.setMessageType(MessageType.RESPONSE); responseMsg.setNumberOfParts(1); - responseMsg.setTransactionId(msg.getTransactionId()); + responseMsg.setTransactionId(clientMessage.getTransactionId()); responseMsg.addObjPart(enums); - responseMsg.send(servConn); - servConn.setAsTrue(RESPONDED); + responseMsg.send(serverConnection); + serverConnection.setAsTrue(RESPONDED); } } http://git-wip-us.apache.org/repos/asf/geode/blob/01132014/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPdxTypes70.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPdxTypes70.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPdxTypes70.java index cc96b8e..c31a375 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPdxTypes70.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetPdxTypes70.java @@ -36,30 +36,30 @@ public class GetPdxTypes70 extends BaseCommand { private GetPdxTypes70() {} @Override - public void cmdExecute(Message msg, ServerConnection servConn, long start) + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start) throws IOException, ClassNotFoundException { - servConn.setAsTrue(REQUIRES_RESPONSE); + serverConnection.setAsTrue(REQUIRES_RESPONSE); if (logger.isDebugEnabled()) { - logger.debug("{}: Received get pdx types from {}", servConn.getName(), - servConn.getSocketString()); + logger.debug("{}: Received get pdx types from {}", serverConnection.getName(), + serverConnection.getSocketString()); } Map<Integer, PdxType> types; try { - InternalCache cache = servConn.getCache(); + InternalCache cache = serverConnection.getCache(); types = cache.getPdxRegistry().typeMap(); } catch (Exception e) { - writeException(msg, e, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } - Message responseMsg = servConn.getResponseMessage(); + Message responseMsg = serverConnection.getResponseMessage(); responseMsg.setMessageType(MessageType.RESPONSE); responseMsg.setNumberOfParts(1); - responseMsg.setTransactionId(msg.getTransactionId()); + responseMsg.setTransactionId(clientMessage.getTransactionId()); responseMsg.addObjPart(types); - responseMsg.send(servConn); - servConn.setAsTrue(RESPONDED); + responseMsg.send(serverConnection); + serverConnection.setAsTrue(RESPONDED); } } http://git-wip-us.apache.org/repos/asf/geode/blob/01132014/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Invalid.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Invalid.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Invalid.java index 24d623a..314ba07 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Invalid.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Invalid.java @@ -36,12 +36,13 @@ public class Invalid extends BaseCommand { private Invalid() {} @Override - public void cmdExecute(Message msg, ServerConnection servConn, long start) throws IOException { + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start) throws IOException { logger.error( LocalizedMessage.create(LocalizedStrings.Invalid_0_INVALID_MESSAGE_TYPE_WITH_TX_1_FROM_2, - new Object[] {servConn.getName(), Integer.valueOf(msg.getTransactionId()), - servConn.getSocketString()})); - writeErrorResponse(msg, MessageType.INVALID, servConn); + new Object[] { + serverConnection.getName(), Integer.valueOf(clientMessage.getTransactionId()), + serverConnection.getSocketString()})); + writeErrorResponse(clientMessage, MessageType.INVALID, serverConnection); } http://git-wip-us.apache.org/repos/asf/geode/blob/01132014/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Invalidate.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Invalidate.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Invalidate.java index 42b2497..22bf6f4 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Invalidate.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Invalidate.java @@ -48,15 +48,15 @@ public class Invalidate extends BaseCommand { } @Override - public void cmdExecute(Message msg, ServerConnection servConn, long start) + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start) throws IOException, InterruptedException { Part regionNamePart = null, keyPart = null, callbackArgPart = null; String regionName = null; Object callbackArg = null, key = null; Part eventPart = null; StringBuffer errMessage = new StringBuffer(); - CacheServerStats stats = servConn.getCacheServerStats(); - servConn.setAsTrue(REQUIRES_RESPONSE); + CacheServerStats stats = serverConnection.getCacheServerStats(); + serverConnection.setAsTrue(REQUIRES_RESPONSE); { long oldStart = start; @@ -64,17 +64,17 @@ public class Invalidate extends BaseCommand { stats.incReadInvalidateRequestTime(start - oldStart); } // Retrieve the data from the message parts - regionNamePart = msg.getPart(0); - keyPart = msg.getPart(1); - eventPart = msg.getPart(2); + regionNamePart = clientMessage.getPart(0); + keyPart = clientMessage.getPart(1); + eventPart = clientMessage.getPart(2); // callbackArgPart = null; (redundant assignment) - if (msg.getNumberOfParts() > 3) { - callbackArgPart = msg.getPart(3); + if (clientMessage.getNumberOfParts() > 3) { + callbackArgPart = clientMessage.getPart(3); try { callbackArg = callbackArgPart.getObject(); } catch (Exception e) { - writeException(msg, e, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } } @@ -82,14 +82,14 @@ public class Invalidate extends BaseCommand { try { key = keyPart.getStringOrObject(); } catch (Exception e) { - writeException(msg, e, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } if (logger.isDebugEnabled()) { - logger.debug(servConn.getName() + ": Received invalidate request (" + msg.getPayloadLength() - + " bytes) from " + servConn.getSocketString() + " for region " + regionName + " key " - + key); + logger.debug(serverConnection.getName() + ": Received invalidate request (" + clientMessage.getPayloadLength() + + " bytes) from " + serverConnection.getSocketString() + " for region " + regionName + " key " + + key); } // Process the invalidate request @@ -108,23 +108,23 @@ public class Invalidate extends BaseCommand { .append(LocalizedStrings.BaseCommand__THE_INPUT_REGION_NAME_FOR_THE_0_REQUEST_IS_NULL .toLocalizedString("invalidate")); } - writeErrorResponse(msg, MessageType.DESTROY_DATA_ERROR, errMessage.toString(), servConn); - servConn.setAsTrue(RESPONDED); + writeErrorResponse(clientMessage, MessageType.DESTROY_DATA_ERROR, errMessage.toString(), serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } - LocalRegion region = (LocalRegion) servConn.getCache().getRegion(regionName); + LocalRegion region = (LocalRegion) serverConnection.getCache().getRegion(regionName); if (region == null) { String reason = LocalizedStrings.BaseCommand__0_WAS_NOT_FOUND_DURING_1_REQUEST .toLocalizedString(regionName, "invalidate"); - writeRegionDestroyedEx(msg, regionName, reason, servConn); - servConn.setAsTrue(RESPONDED); + writeRegionDestroyedEx(clientMessage, regionName, reason, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } // Invalidate the entry ByteBuffer eventIdPartsBuffer = ByteBuffer.wrap(eventPart.getSerializedForm()); long threadId = EventID.readEventIdPartsFromOptmizedByteArray(eventIdPartsBuffer); long sequenceId = EventID.readEventIdPartsFromOptmizedByteArray(eventIdPartsBuffer); - EventID eventId = new EventID(servConn.getEventMemberIDByteArray(), threadId, sequenceId); + EventID eventId = new EventID(serverConnection.getEventMemberIDByteArray(), threadId, sequenceId); Breadcrumbs.setEventId(eventId); @@ -134,7 +134,7 @@ public class Invalidate extends BaseCommand { // for integrated security this.securityService.authorizeRegionWrite(regionName, key.toString()); - AuthorizeRequest authzRequest = servConn.getAuthzRequest(); + AuthorizeRequest authzRequest = serverConnection.getAuthzRequest(); if (authzRequest != null) { InvalidateOperationContext invalidateContext = authzRequest.invalidateAuthorize(regionName, key, callbackArg); @@ -143,7 +143,7 @@ public class Invalidate extends BaseCommand { EventIDHolder clientEvent = new EventIDHolder(eventId); // msg.isRetry might be set by v7.0 and later clients - if (msg.isRetry()) { + if (clientMessage.isRetry()) { // if (logger.isDebugEnabled()) { // logger.debug("DEBUG: encountered isRetry in Invalidate"); // } @@ -157,9 +157,9 @@ public class Invalidate extends BaseCommand { } } - region.basicBridgeInvalidate(key, callbackArg, servConn.getProxyID(), true, clientEvent); + region.basicBridgeInvalidate(key, callbackArg, serverConnection.getProxyID(), true, clientEvent); tag = clientEvent.getVersionTag(); - servConn.setModificationInfo(true, regionName, key); + serverConnection.setModificationInfo(true, regionName, key); } catch (EntryNotFoundException e) { // Don't send an exception back to the client if this // exception happens. Just log it and continue. @@ -167,25 +167,25 @@ public class Invalidate extends BaseCommand { LocalizedStrings.BaseCommand_DURING_0_NO_ENTRY_WAS_FOUND_FOR_KEY_1, new Object[] {"invalidate", key})); } catch (RegionDestroyedException rde) { - writeException(msg, rde, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, rde, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } catch (Exception e) { // If an interrupted exception is thrown , rethrow it - checkForInterrupt(servConn, e); + checkForInterrupt(serverConnection, e); // If an exception occurs during the destroy, preserve the connection - writeException(msg, e, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); if (e instanceof GemFireSecurityException) { // Fine logging for security exceptions since these are already // logged by the security logger if (logger.isDebugEnabled()) { - logger.debug("{}: Unexpected Security exception", servConn.getName(), e); + logger.debug("{}: Unexpected Security exception", serverConnection.getName(), e); } } else { logger.warn(LocalizedMessage.create(LocalizedStrings.BaseCommand_0_UNEXPECTED_EXCEPTION, - servConn.getName()), e); + serverConnection.getName()), e); } return; } @@ -199,17 +199,17 @@ public class Invalidate extends BaseCommand { if (region instanceof PartitionedRegion) { PartitionedRegion pr = (PartitionedRegion) region; if (pr.getNetworkHopType() != PartitionedRegion.NETWORK_HOP_NONE) { - writeReplyWithRefreshMetadata(msg, servConn, pr, pr.getNetworkHopType(), tag); + writeReplyWithRefreshMetadata(clientMessage, serverConnection, pr, pr.getNetworkHopType(), tag); pr.clearNetworkHopData(); } else { - writeReply(msg, servConn, tag); + writeReply(clientMessage, serverConnection, tag); } } else { - writeReply(msg, servConn, tag); + writeReply(clientMessage, serverConnection, tag); } - servConn.setAsTrue(RESPONDED); + serverConnection.setAsTrue(RESPONDED); if (logger.isDebugEnabled()) { - logger.debug("{}: Sent invalidate response for region {} key {}", servConn.getName(), + logger.debug("{}: Sent invalidate response for region {} key {}", serverConnection.getName(), regionName, key); } stats.incWriteInvalidateResponseTime(DistributionStats.getStatTime() - start); http://git-wip-us.apache.org/repos/asf/geode/blob/01132014/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Invalidate70.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Invalidate70.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Invalidate70.java index 4ac5023..2531e1d 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Invalidate70.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Invalidate70.java @@ -60,7 +60,7 @@ public class Invalidate70 extends Invalidate { pr.getPrStats().incPRMetaDataSentCount(); replyMsg.send(servConn); if (logger.isTraceEnabled()) { - logger.trace("{}: rpl with REFRESH_METADAT tx: {}", servConn.getName(), + logger.trace("{}: rpl with REFRESH_METADATA tx: {}", servConn.getName(), origMsg.getTransactionId()); } } @@ -90,7 +90,7 @@ public class Invalidate70 extends Invalidate { logger.debug("response has no version tag"); } } - replyMsg.addBytesPart(OK_BYTES); // make old single-hop code happy by putting byte[]{0} here + replyMsg.addBytesPart(okBytes()); // make old single-hop code happy by putting byte[]{0} here replyMsg.send(servConn); if (logger.isTraceEnabled()) { logger.trace("{}: rpl tx: {} parts={}", servConn.getName(), origMsg.getTransactionId(), http://git-wip-us.apache.org/repos/asf/geode/blob/01132014/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/KeySet.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/KeySet.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/KeySet.java index 5f7388c..a35c4b0 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/KeySet.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/KeySet.java @@ -26,7 +26,6 @@ import java.util.Set; import org.apache.geode.cache.Region; import org.apache.geode.cache.operations.KeySetOperationContext; import org.apache.geode.internal.cache.LocalRegion; -import org.apache.geode.internal.cache.tier.CachedRegionHelper; import org.apache.geode.internal.cache.tier.Command; import org.apache.geode.internal.cache.tier.MessageType; import org.apache.geode.internal.cache.tier.sockets.BaseCommand; @@ -49,21 +48,21 @@ public class KeySet extends BaseCommand { } @Override - public void cmdExecute(Message msg, ServerConnection servConn, long start) + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start) throws IOException, InterruptedException { Part regionNamePart = null; String regionName = null; - servConn.setAsTrue(REQUIRES_RESPONSE); - servConn.setAsTrue(REQUIRES_CHUNKED_RESPONSE); + serverConnection.setAsTrue(REQUIRES_RESPONSE); + serverConnection.setAsTrue(REQUIRES_CHUNKED_RESPONSE); // Retrieve the region name from the message parts - regionNamePart = msg.getPart(0); + regionNamePart = clientMessage.getPart(0); regionName = regionNamePart.getString(); - ChunkedMessage chunkedResponseMsg = servConn.getChunkedResponseMessage(); + ChunkedMessage chunkedResponseMsg = serverConnection.getChunkedResponseMessage(); final boolean isDebugEnabled = logger.isDebugEnabled(); if (isDebugEnabled) { logger.debug("{}: Received key set request ({} bytes) from {} for region {}", - servConn.getName(), msg.getPayloadLength(), servConn.getSocketString(), regionName); + serverConnection.getName(), clientMessage.getPayloadLength(), serverConnection.getSocketString(), regionName); } // Process the key set request @@ -72,41 +71,41 @@ public class KeySet extends BaseCommand { // if (regionName == null) (can only be null) { message = LocalizedStrings.KeySet_0_THE_INPUT_REGION_NAME_FOR_THE_KEY_SET_REQUEST_IS_NULL - .toLocalizedString(servConn.getName()); + .toLocalizedString(serverConnection.getName()); logger.warn(LocalizedMessage.create( LocalizedStrings.KeySet_0_THE_INPUT_REGION_NAME_FOR_THE_KEY_SET_REQUEST_IS_NULL, - servConn.getName())); + serverConnection.getName())); } - writeKeySetErrorResponse(msg, MessageType.KEY_SET_DATA_ERROR, message, servConn); - servConn.setAsTrue(RESPONDED); + writeKeySetErrorResponse(clientMessage, MessageType.KEY_SET_DATA_ERROR, message, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } - LocalRegion region = (LocalRegion) servConn.getCache().getRegion(regionName); + LocalRegion region = (LocalRegion) serverConnection.getCache().getRegion(regionName); if (region == null) { String reason = LocalizedStrings.KeySet__0_WAS_NOT_FOUND_DURING_KEY_SET_REQUEST .toLocalizedString(regionName); - writeRegionDestroyedEx(msg, regionName, reason, servConn); - servConn.setAsTrue(RESPONDED); + writeRegionDestroyedEx(clientMessage, regionName, reason, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } try { this.securityService.authorizeRegionRead(regionName); } catch (NotAuthorizedException ex) { - writeChunkedException(msg, ex, false, servConn); - servConn.setAsTrue(RESPONDED); + writeChunkedException(clientMessage, ex, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } KeySetOperationContext keySetContext = null; - AuthorizeRequest authzRequest = servConn.getAuthzRequest(); + AuthorizeRequest authzRequest = serverConnection.getAuthzRequest(); if (authzRequest != null) { try { keySetContext = authzRequest.keySetAuthorize(regionName); } catch (NotAuthorizedException ex) { - writeChunkedException(msg, ex, false, servConn); - servConn.setAsTrue(RESPONDED); + writeChunkedException(clientMessage, ex, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } } @@ -117,20 +116,20 @@ public class KeySet extends BaseCommand { // Send header chunkedResponseMsg.setMessageType(MessageType.RESPONSE); - chunkedResponseMsg.setTransactionId(msg.getTransactionId()); + chunkedResponseMsg.setTransactionId(clientMessage.getTransactionId()); chunkedResponseMsg.sendHeader(); // Send chunk response try { - fillAndSendKeySetResponseChunks(region, regionName, keySetContext, servConn); - servConn.setAsTrue(RESPONDED); + fillAndSendKeySetResponseChunks(region, regionName, keySetContext, serverConnection); + serverConnection.setAsTrue(RESPONDED); } catch (Exception e) { // If an interrupted exception is thrown , rethrow it - checkForInterrupt(servConn, e); + checkForInterrupt(serverConnection, e); // Otherwise, write an exception message and continue - writeChunkedException(msg, e, false, servConn, servConn.getChunkedResponseMessage()); - servConn.setAsTrue(RESPONDED); + writeChunkedException(clientMessage, e, serverConnection, serverConnection.getChunkedResponseMessage()); + serverConnection.setAsTrue(RESPONDED); return; } @@ -138,7 +137,7 @@ public class KeySet extends BaseCommand { // logger.fine(getName() + ": Sent chunk (1 of 1) of register interest // response (" + chunkedResponseMsg.getBufferLength() + " bytes) for // region " + regionName + " key " + key); - logger.debug("{}: Sent key set response for the region {}", servConn.getName(), regionName); + logger.debug("{}: Sent key set response for the region {}", serverConnection.getName(), regionName); } // bserverStats.incLong(writeDestroyResponseTimeId, // DistributionStats.getStatTime() - start); @@ -160,7 +159,7 @@ public class KeySet extends BaseCommand { keySet = keySetContext.getKeySet(); } - List keyList = new ArrayList(maximumChunkSize); + List keyList = new ArrayList(MAXIMUM_CHUNK_SIZE); final boolean isTraceEnabled = logger.isTraceEnabled(); for (Iterator it = keySet.iterator(); it.hasNext();) { Object entryKey = it.next(); @@ -169,7 +168,7 @@ public class KeySet extends BaseCommand { logger.trace("{}: fillAndSendKeySetResponseKey <{}>; list size was {}; region: {}", servConn.getName(), entryKey, keyList.size(), region.getFullPath()); } - if (keyList.size() == maximumChunkSize) { + if (keyList.size() == MAXIMUM_CHUNK_SIZE) { // Send the chunk and clear the list sendKeySetResponseChunk(region, keyList, false, servConn); keyList.clear(); @@ -185,7 +184,7 @@ public class KeySet extends BaseCommand { chunkedResponseMsg.setNumberOfParts(1); chunkedResponseMsg.setLastChunk(lastChunk); - chunkedResponseMsg.addObjPart(list, zipValues); + chunkedResponseMsg.addObjPart(list, false); if (logger.isDebugEnabled()) { logger.debug("{}: Sending {} key set response chunk for region={}{}", servConn.getName(), http://git-wip-us.apache.org/repos/asf/geode/blob/01132014/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/MakePrimary.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/MakePrimary.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/MakePrimary.java index fc8fe44..0786990 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/MakePrimary.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/MakePrimary.java @@ -34,30 +34,30 @@ public class MakePrimary extends BaseCommand { private MakePrimary() {} @Override - public void cmdExecute(Message msg, ServerConnection servConn, long start) + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start) throws IOException, ClassNotFoundException { - servConn.setAsTrue(REQUIRES_RESPONSE); - Part isClientReadyPart = msg.getPart(0); + serverConnection.setAsTrue(REQUIRES_RESPONSE); + Part isClientReadyPart = clientMessage.getPart(0); byte[] isClientReadyPartBytes = (byte[]) isClientReadyPart.getObject(); boolean isClientReady = isClientReadyPartBytes[0] == 0x01; final boolean isDebugEnabled = logger.isDebugEnabled(); if (isDebugEnabled) { logger.debug("{}: Received make primary request ({} bytes) isClientReady={}: from {}", - servConn.getName(), msg.getPayloadLength(), isClientReady, servConn.getSocketString()); + serverConnection.getName(), clientMessage.getPayloadLength(), isClientReady, serverConnection.getSocketString()); } try { - servConn.getAcceptor().getCacheClientNotifier().makePrimary(servConn.getProxyID(), + serverConnection.getAcceptor().getCacheClientNotifier().makePrimary(serverConnection.getProxyID(), isClientReady); - writeReply(msg, servConn); - servConn.setAsTrue(RESPONDED); + writeReply(clientMessage, serverConnection); + serverConnection.setAsTrue(RESPONDED); if (isDebugEnabled) { - logger.debug("{}: Sent make primary response for {}", servConn.getName(), - servConn.getSocketString()); + logger.debug("{}: Sent make primary response for {}", serverConnection.getName(), + serverConnection.getSocketString()); } } catch (Exception e) { - writeException(msg, e, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); } } http://git-wip-us.apache.org/repos/asf/geode/blob/01132014/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ManagementCommand.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ManagementCommand.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ManagementCommand.java index b5e1ca3..9a79540 100755 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ManagementCommand.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ManagementCommand.java @@ -28,7 +28,7 @@ import org.apache.geode.internal.cache.tier.sockets.ServerConnection; public class ManagementCommand extends BaseCommand { @Override - public void cmdExecute(Message msg, ServerConnection servConn, long start) + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start) throws IOException, ClassNotFoundException, InterruptedException { // TODO Auto-generated method stub } http://git-wip-us.apache.org/repos/asf/geode/blob/01132014/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PeriodicAck.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PeriodicAck.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PeriodicAck.java index d301909..e57385f 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PeriodicAck.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PeriodicAck.java @@ -33,43 +33,43 @@ public class PeriodicAck extends BaseCommand { private PeriodicAck() {} @Override - public void cmdExecute(Message msg, ServerConnection servConn, long start) + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start) throws IOException, ClassNotFoundException { - servConn.setAsTrue(REQUIRES_RESPONSE); + serverConnection.setAsTrue(REQUIRES_RESPONSE); if (logger.isDebugEnabled()) { - logger.debug("{}: Received periodic ack request ({} bytes) from {}", servConn.getName(), - msg.getPayloadLength(), servConn.getSocketString()); + logger.debug("{}: Received periodic ack request ({} bytes) from {}", serverConnection.getName(), + clientMessage.getPayloadLength(), serverConnection.getSocketString()); } try { - int numEvents = msg.getNumberOfParts(); + int numEvents = clientMessage.getNumberOfParts(); boolean success = false; - CacheClientNotifier ccn = servConn.getAcceptor().getCacheClientNotifier(); - CacheClientProxy proxy = ccn.getClientProxy(servConn.getProxyID()); + CacheClientNotifier ccn = serverConnection.getAcceptor().getCacheClientNotifier(); + CacheClientProxy proxy = ccn.getClientProxy(serverConnection.getProxyID()); if (proxy != null) { proxy.getHARegionQueue().createAckedEventsMap(); for (int i = 0; i < numEvents; i++) { - Part eventIdPart = msg.getPart(i); - eventIdPart.setVersion(servConn.getClientVersion()); + Part eventIdPart = clientMessage.getPart(i); + eventIdPart.setVersion(serverConnection.getClientVersion()); EventID eid = (EventID) eventIdPart.getObject(); - success = ccn.processDispatchedMessage(servConn.getProxyID(), eid); + success = ccn.processDispatchedMessage(serverConnection.getProxyID(), eid); if (!success) break; } } if (success) { proxy.getHARegionQueue().setAckedEvents(); - writeReply(msg, servConn); - servConn.setAsTrue(RESPONDED); + writeReply(clientMessage, serverConnection); + serverConnection.setAsTrue(RESPONDED); } } catch (Exception e) { - writeException(msg, e, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); } if (logger.isDebugEnabled()) { - logger.debug("{}: Sent periodic ack response for {}", servConn.getName(), - servConn.getSocketString()); + logger.debug("{}: Sent periodic ack response for {}", serverConnection.getName(), + serverConnection.getSocketString()); } } http://git-wip-us.apache.org/repos/asf/geode/blob/01132014/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Ping.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Ping.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Ping.java index 7a12ce1..9755410 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Ping.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Ping.java @@ -39,35 +39,35 @@ public class Ping extends BaseCommand { private Ping() {} @Override - public void cmdExecute(Message msg, ServerConnection servConn, long start) throws IOException { + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start) throws IOException { final boolean isDebugEnabled = logger.isDebugEnabled(); if (isDebugEnabled) { - logger.debug("{}: rcv tx: {} from {} rcvTime: {}", servConn.getName(), msg.getTransactionId(), - servConn.getSocketString(), (DistributionStats.getStatTime() - start)); + logger.debug("{}: rcv tx: {} from {} rcvTime: {}", serverConnection.getName(), clientMessage.getTransactionId(), + serverConnection.getSocketString(), (DistributionStats.getStatTime() - start)); } ClientHealthMonitor chm = ClientHealthMonitor.getInstance(); if (chm != null) - chm.receivedPing(servConn.getProxyID()); - CachedRegionHelper crHelper = servConn.getCachedRegionHelper(); + chm.receivedPing(serverConnection.getProxyID()); + CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper(); - writeReply(msg, servConn); - servConn.setAsTrue(RESPONDED); + writeReply(clientMessage, serverConnection); + serverConnection.setAsTrue(RESPONDED); if (isDebugEnabled) { - logger.debug("{}: Sent ping reply to {}", servConn.getName(), servConn.getSocketString()); + logger.debug("{}: Sent ping reply to {}", serverConnection.getName(), serverConnection.getSocketString()); } } @Override - protected void writeReply(Message origMsg, ServerConnection servConn) throws IOException { - Message replyMsg = servConn.getReplyMessage(); - servConn.getCache().getCancelCriterion().checkCancelInProgress(null); + protected void writeReply(Message origMsg, ServerConnection serverConnection) throws IOException { + Message replyMsg = serverConnection.getReplyMessage(); + serverConnection.getCache().getCancelCriterion().checkCancelInProgress(null); replyMsg.setMessageType(MessageType.REPLY); replyMsg.setNumberOfParts(1); replyMsg.setTransactionId(origMsg.getTransactionId()); - replyMsg.addBytesPart(OK_BYTES); - replyMsg.send(servConn); + replyMsg.addBytesPart(okBytes()); + replyMsg.send(serverConnection); if (logger.isTraceEnabled()) { - logger.trace("{}: rpl tx: {}", servConn.getName(), origMsg.getTransactionId()); + logger.trace("{}: rpl tx: {}", serverConnection.getName(), origMsg.getTransactionId()); } } } http://git-wip-us.apache.org/repos/asf/geode/blob/01132014/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Put.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Put.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Put.java index 2a235c0..d724f66 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Put.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Put.java @@ -49,36 +49,36 @@ public class Put extends BaseCommand { @Override - public void cmdExecute(Message msg, ServerConnection servConn, long start) + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start) throws IOException, InterruptedException { Part regionNamePart = null, keyPart = null, valuePart = null, callbackArgPart = null; String regionName = null; Object callbackArg = null, key = null; Part eventPart = null; String errMessage = ""; - CachedRegionHelper crHelper = servConn.getCachedRegionHelper(); - CacheServerStats stats = servConn.getCacheServerStats(); + CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper(); + CacheServerStats stats = serverConnection.getCacheServerStats(); // requiresResponse = true; - servConn.setAsTrue(REQUIRES_RESPONSE); + serverConnection.setAsTrue(REQUIRES_RESPONSE); { long oldStart = start; start = DistributionStats.getStatTime(); stats.incReadPutRequestTime(start - oldStart); } // Retrieve the data from the message parts - regionNamePart = msg.getPart(0); - keyPart = msg.getPart(1); - valuePart = msg.getPart(2); - eventPart = msg.getPart(3); + regionNamePart = clientMessage.getPart(0); + keyPart = clientMessage.getPart(1); + valuePart = clientMessage.getPart(2); + eventPart = clientMessage.getPart(3); // callbackArgPart = null; (redundant assignment) - if (msg.getNumberOfParts() > 4) { - callbackArgPart = msg.getPart(4); + if (clientMessage.getNumberOfParts() > 4) { + callbackArgPart = clientMessage.getPart(4); try { callbackArg = callbackArgPart.getObject(); } catch (Exception e) { - writeException(msg, e, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } } @@ -87,14 +87,14 @@ public class Put extends BaseCommand { try { key = keyPart.getStringOrObject(); } catch (Exception e) { - writeException(msg, e, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } if (logger.isTraceEnabled()) { logger.trace("{}: Received put request ({} bytes) from {} for region {} key {} value {}", - servConn.getName(), msg.getPayloadLength(), servConn.getSocketString(), regionName, key, + serverConnection.getName(), clientMessage.getPayloadLength(), serverConnection.getSocketString(), regionName, key, valuePart); } @@ -102,28 +102,28 @@ public class Put extends BaseCommand { if (key == null || regionName == null) { if (key == null) { logger.warn(LocalizedMessage.create( - LocalizedStrings.Put_0_THE_INPUT_KEY_FOR_THE_PUT_REQUEST_IS_NULL, servConn.getName())); + LocalizedStrings.Put_0_THE_INPUT_KEY_FOR_THE_PUT_REQUEST_IS_NULL, serverConnection.getName())); errMessage = LocalizedStrings.Put_THE_INPUT_KEY_FOR_THE_PUT_REQUEST_IS_NULL.toLocalizedString(); } if (regionName == null) { logger.warn(LocalizedMessage.create( LocalizedStrings.Put_0_THE_INPUT_REGION_NAME_FOR_THE_PUT_REQUEST_IS_NULL, - servConn.getName())); + serverConnection.getName())); errMessage = LocalizedStrings.Put_THE_INPUT_REGION_NAME_FOR_THE_PUT_REQUEST_IS_NULL .toLocalizedString(); } - writeErrorResponse(msg, MessageType.PUT_DATA_ERROR, errMessage.toString(), servConn); - servConn.setAsTrue(RESPONDED); + writeErrorResponse(clientMessage, MessageType.PUT_DATA_ERROR, errMessage.toString(), serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } - LocalRegion region = (LocalRegion) servConn.getCache().getRegion(regionName); + LocalRegion region = (LocalRegion) serverConnection.getCache().getRegion(regionName); if (region == null) { String reason = LocalizedStrings.Put_REGION_WAS_NOT_FOUND_DURING_PUT_REQUEST.toLocalizedString(); - writeRegionDestroyedEx(msg, regionName, reason, servConn); - servConn.setAsTrue(RESPONDED); + writeRegionDestroyedEx(clientMessage, regionName, reason, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } @@ -131,18 +131,18 @@ public class Put extends BaseCommand { // Invalid to 'put' a null value in an existing key logger.info(LocalizedMessage.create( LocalizedStrings.Put_0_ATTEMPTED_TO_PUT_A_NULL_VALUE_FOR_EXISTING_KEY_1, - new Object[] {servConn.getName(), key})); + new Object[] { serverConnection.getName(), key})); errMessage = LocalizedStrings.Put_ATTEMPTED_TO_PUT_A_NULL_VALUE_FOR_EXISTING_KEY_0.toLocalizedString(); - writeErrorResponse(msg, MessageType.PUT_DATA_ERROR, errMessage, servConn); - servConn.setAsTrue(RESPONDED); + writeErrorResponse(clientMessage, MessageType.PUT_DATA_ERROR, errMessage, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } ByteBuffer eventIdPartsBuffer = ByteBuffer.wrap(eventPart.getSerializedForm()); long threadId = EventID.readEventIdPartsFromOptmizedByteArray(eventIdPartsBuffer); long sequenceId = EventID.readEventIdPartsFromOptmizedByteArray(eventIdPartsBuffer); - EventID eventId = new EventID(servConn.getEventMemberIDByteArray(), threadId, sequenceId); + EventID eventId = new EventID(serverConnection.getEventMemberIDByteArray(), threadId, sequenceId); try { byte[] value = valuePart.getSerializedForm(); @@ -150,7 +150,7 @@ public class Put extends BaseCommand { this.securityService.authorizeRegionWrite(regionName, key.toString()); - AuthorizeRequest authzRequest = servConn.getAuthzRequest(); + AuthorizeRequest authzRequest = serverConnection.getAuthzRequest(); if (authzRequest != null) { if (DynamicRegionFactory.regionIsDynamicRegionList(regionName)) { authzRequest.createRegionAuthorize((String) key); @@ -173,46 +173,46 @@ public class Put extends BaseCommand { // Create the null entry. Since the value is null, the value of the // isObject // the true after null doesn't matter and is not used. - result = region.basicBridgeCreate(key, null, true, callbackArg, servConn.getProxyID(), true, + result = region.basicBridgeCreate(key, null, true, callbackArg, serverConnection.getProxyID(), true, new EventIDHolder(eventId), false); } else { // Put the entry result = region.basicBridgePut(key, value, null, isObject, callbackArg, - servConn.getProxyID(), true, new EventIDHolder(eventId)); + serverConnection.getProxyID(), true, new EventIDHolder(eventId)); } if (result) { - servConn.setModificationInfo(true, regionName, key); + serverConnection.setModificationInfo(true, regionName, key); } else { StringId message = LocalizedStrings.PUT_0_FAILED_TO_PUT_ENTRY_FOR_REGION_1_KEY_2_VALUE_3; - Object[] messageArgs = new Object[] {servConn.getName(), regionName, key, valuePart}; + Object[] messageArgs = new Object[] { serverConnection.getName(), regionName, key, valuePart}; String s = message.toLocalizedString(messageArgs); logger.info(s); throw new Exception(s); } } catch (RegionDestroyedException rde) { - writeException(msg, rde, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, rde, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } catch (ResourceException re) { - writeException(msg, re, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, re, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } catch (Exception ce) { // If an interrupted exception is thrown , rethrow it - checkForInterrupt(servConn, ce); + checkForInterrupt(serverConnection, ce); // If an exception occurs during the put, preserve the connection - writeException(msg, ce, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, ce, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); if (ce instanceof GemFireSecurityException) { // Fine logging for security exceptions since these are already // logged by the security logger if (logger.isDebugEnabled()) { - logger.debug("{}: Unexpected Security exception", servConn.getName(), ce); + logger.debug("{}: Unexpected Security exception", serverConnection.getName(), ce); } } else { logger.warn(LocalizedMessage.create(LocalizedStrings.PUT_0_UNEXPECTED_EXCEPTION, - servConn.getName()), ce); + serverConnection.getName()), ce); } return; } finally { @@ -222,12 +222,12 @@ public class Put extends BaseCommand { } // Increment statistics and write the reply - writeReply(msg, servConn); + writeReply(clientMessage, serverConnection); - servConn.setAsTrue(RESPONDED); + serverConnection.setAsTrue(RESPONDED); if (logger.isDebugEnabled()) { logger.debug("{}: Sent put response back to {} for region {} key {} value {}", - servConn.getName(), servConn.getSocketString(), regionName, key, valuePart); + serverConnection.getName(), serverConnection.getSocketString(), regionName, key, valuePart); } stats.incWritePutResponseTime(DistributionStats.getStatTime() - start); } http://git-wip-us.apache.org/repos/asf/geode/blob/01132014/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Put61.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Put61.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Put61.java index d123ea2..3f9a72e 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Put61.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Put61.java @@ -55,7 +55,7 @@ public class Put61 extends BaseCommand { } @Override - public void cmdExecute(Message msg, ServerConnection servConn, long p_start) + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long p_start) throws IOException, InterruptedException { long start = p_start; Part regionNamePart = null, keyPart = null, valuePart = null, callbackArgPart = null; @@ -64,36 +64,36 @@ public class Put61 extends BaseCommand { Part eventPart = null; StringBuffer errMessage = new StringBuffer(); boolean isDelta = false; - CachedRegionHelper crHelper = servConn.getCachedRegionHelper(); - CacheServerStats stats = servConn.getCacheServerStats(); + CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper(); + CacheServerStats stats = serverConnection.getCacheServerStats(); // requiresResponse = true; - servConn.setAsTrue(REQUIRES_RESPONSE); + serverConnection.setAsTrue(REQUIRES_RESPONSE); { long oldStart = start; start = DistributionStats.getStatTime(); stats.incReadPutRequestTime(start - oldStart); } // Retrieve the data from the message parts - regionNamePart = msg.getPart(0); - keyPart = msg.getPart(1); + regionNamePart = clientMessage.getPart(0); + keyPart = clientMessage.getPart(1); try { - isDelta = (Boolean) msg.getPart(2).getObject(); + isDelta = (Boolean) clientMessage.getPart(2).getObject(); } catch (Exception e) { - writeException(msg, MessageType.PUT_DELTA_ERROR, e, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, MessageType.PUT_DELTA_ERROR, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); // CachePerfStats not available here. return; } - valuePart = msg.getPart(3); - eventPart = msg.getPart(4); - if (msg.getNumberOfParts() > 5) { - callbackArgPart = msg.getPart(5); + valuePart = clientMessage.getPart(3); + eventPart = clientMessage.getPart(4); + if (clientMessage.getNumberOfParts() > 5) { + callbackArgPart = clientMessage.getPart(5); try { callbackArg = callbackArgPart.getObject(); } catch (Exception e) { - writeException(msg, e, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } } @@ -102,16 +102,16 @@ public class Put61 extends BaseCommand { try { key = keyPart.getStringOrObject(); } catch (Exception e) { - writeException(msg, e, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } final boolean isDebugEnabled = logger.isDebugEnabled(); if (isDebugEnabled) { logger.debug("{}: Received 6.1{}put request ({} bytes) from {} for region {} key {}", - servConn.getName(), (isDelta ? " delta " : " "), msg.getPayloadLength(), - servConn.getSocketString(), regionName, key); + serverConnection.getName(), (isDelta ? " delta " : " "), clientMessage.getPayloadLength(), + serverConnection.getSocketString(), regionName, key); } // Process the put request @@ -119,27 +119,27 @@ public class Put61 extends BaseCommand { if (key == null) { String putMsg = " The input key for the 6.1 put request is null"; if (isDebugEnabled) { - logger.debug("{}:{}", servConn.getName(), putMsg); + logger.debug("{}:{}", serverConnection.getName(), putMsg); } errMessage.append(putMsg); } if (regionName == null) { String putMsg = " The input region name for the 6.1 put request is null"; if (isDebugEnabled) { - logger.debug("{}:{}", servConn.getName(), putMsg); + logger.debug("{}:{}", serverConnection.getName(), putMsg); } errMessage.append(putMsg); } - writeErrorResponse(msg, MessageType.PUT_DATA_ERROR, errMessage.toString(), servConn); - servConn.setAsTrue(RESPONDED); + writeErrorResponse(clientMessage, MessageType.PUT_DATA_ERROR, errMessage.toString(), serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } - LocalRegion region = (LocalRegion) servConn.getCache().getRegion(regionName); + LocalRegion region = (LocalRegion) serverConnection.getCache().getRegion(regionName); if (region == null) { String reason = " was not found during 6.1 put request"; - writeRegionDestroyedEx(msg, regionName, reason, servConn); - servConn.setAsTrue(RESPONDED); + writeRegionDestroyedEx(clientMessage, regionName, reason, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } @@ -147,11 +147,11 @@ public class Put61 extends BaseCommand { // Invalid to 'put' a null value in an existing key String putMsg = " Attempted to 6.1 put a null value for existing key " + key; if (isDebugEnabled) { - logger.debug("{}:{}", servConn.getName(), putMsg); + logger.debug("{}:{}", serverConnection.getName(), putMsg); } errMessage.append(putMsg); - writeErrorResponse(msg, MessageType.PUT_DATA_ERROR, errMessage.toString(), servConn); - servConn.setAsTrue(RESPONDED); + writeErrorResponse(clientMessage, MessageType.PUT_DATA_ERROR, errMessage.toString(), serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } @@ -160,7 +160,7 @@ public class Put61 extends BaseCommand { ByteBuffer eventIdPartsBuffer = ByteBuffer.wrap(eventPart.getSerializedForm()); long threadId = EventID.readEventIdPartsFromOptmizedByteArray(eventIdPartsBuffer); long sequenceId = EventID.readEventIdPartsFromOptmizedByteArray(eventIdPartsBuffer); - EventID eventId = new EventID(servConn.getEventMemberIDByteArray(), threadId, sequenceId); + EventID eventId = new EventID(serverConnection.getEventMemberIDByteArray(), threadId, sequenceId); try { Object value = null; @@ -169,13 +169,13 @@ public class Put61 extends BaseCommand { } boolean isObject = valuePart.isObject(); boolean isMetaRegion = region.isUsedForMetaRegion(); - msg.setMetaRegion(isMetaRegion); + clientMessage.setMetaRegion(isMetaRegion); this.securityService.authorizeRegionWrite(regionName, key.toString()); AuthorizeRequest authzRequest = null; if (!isMetaRegion) { - authzRequest = servConn.getAuthzRequest(); + authzRequest = serverConnection.getAuthzRequest(); } if (authzRequest != null) { if (DynamicRegionFactory.regionIsDynamicRegionList(regionName)) { @@ -199,7 +199,7 @@ public class Put61 extends BaseCommand { // Create the null entry. Since the value is null, the value of the // isObject // the true after null doesn't matter and is not used. - result = region.basicBridgeCreate(key, null, true, callbackArg, servConn.getProxyID(), true, + result = region.basicBridgeCreate(key, null, true, callbackArg, serverConnection.getProxyID(), true, new EventIDHolder(eventId), false); } else { // Put the entry @@ -208,50 +208,50 @@ public class Put61 extends BaseCommand { delta = valuePart.getSerializedForm(); } result = region.basicBridgePut(key, value, delta, isObject, callbackArg, - servConn.getProxyID(), true, new EventIDHolder(eventId)); + serverConnection.getProxyID(), true, new EventIDHolder(eventId)); } if (result) { - servConn.setModificationInfo(true, regionName, key); + serverConnection.setModificationInfo(true, regionName, key); } else { - String message = servConn.getName() + ": Failed to 6.1 put entry for region " + regionName - + " key " + key + " value " + valuePart; + String message = serverConnection.getName() + ": Failed to 6.1 put entry for region " + regionName + + " key " + key + " value " + valuePart; if (isDebugEnabled) { logger.debug(message); } throw new Exception(message); } } catch (RegionDestroyedException rde) { - writeException(msg, rde, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, rde, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } catch (ResourceException re) { - writeException(msg, re, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, re, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } catch (InvalidDeltaException ide) { logger.info(LocalizedMessage.create( LocalizedStrings.UpdateOperation_ERROR_APPLYING_DELTA_FOR_KEY_0_OF_REGION_1, new Object[] {key, regionName})); - writeException(msg, MessageType.PUT_DELTA_ERROR, ide, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, MessageType.PUT_DELTA_ERROR, ide, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); region.getCachePerfStats().incDeltaFullValuesRequested(); return; } catch (Exception ce) { // If an interrupted exception is thrown , rethrow it - checkForInterrupt(servConn, ce); + checkForInterrupt(serverConnection, ce); // If an exception occurs during the put, preserve the connection - writeException(msg, ce, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, ce, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); if (ce instanceof GemFireSecurityException) { // Fine logging for security exceptions since these are already // logged by the security logger if (isDebugEnabled) { - logger.debug("{}: Unexpected Security exception", servConn.getName(), ce); + logger.debug("{}: Unexpected Security exception", serverConnection.getName(), ce); } } else if (isDebugEnabled) { - logger.debug("{}: Unexpected Exception", servConn.getName(), ce); + logger.debug("{}: Unexpected Exception", serverConnection.getName(), ce); } return; } finally { @@ -264,18 +264,18 @@ public class Put61 extends BaseCommand { if (region instanceof PartitionedRegion) { PartitionedRegion pr = (PartitionedRegion) region; if (pr.getNetworkHopType() != PartitionedRegion.NETWORK_HOP_NONE) { - writeReplyWithRefreshMetadata(msg, servConn, pr, pr.getNetworkHopType()); + writeReplyWithRefreshMetadata(clientMessage, serverConnection, pr, pr.getNetworkHopType()); pr.clearNetworkHopData(); } else { - writeReply(msg, servConn); + writeReply(clientMessage, serverConnection); } } else { - writeReply(msg, servConn); + writeReply(clientMessage, serverConnection); } - servConn.setAsTrue(RESPONDED); + serverConnection.setAsTrue(RESPONDED); if (isDebugEnabled) { logger.debug("{}: Sent 6.1 put response back to {} for region {} key {} value {}", - servConn.getName(), servConn.getSocketString(), regionName, key, valuePart); + serverConnection.getName(), serverConnection.getSocketString(), regionName, key, valuePart); } stats.incWritePutResponseTime(DistributionStats.getStatTime() - start); }
