Return-Path: X-Original-To: archive-asf-public-internal@cust-asf2.ponee.io Delivered-To: archive-asf-public-internal@cust-asf2.ponee.io Received: from cust-asf.ponee.io (cust-asf.ponee.io [163.172.22.183]) by cust-asf2.ponee.io (Postfix) with ESMTP id D1F29200C79 for ; Sat, 20 May 2017 00:00:45 +0200 (CEST) Received: by cust-asf.ponee.io (Postfix) id D058B160BD6; Fri, 19 May 2017 22:00:45 +0000 (UTC) Delivered-To: archive-asf-public@cust-asf.ponee.io Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by cust-asf.ponee.io (Postfix) with SMTP id 8A395160BD5 for ; Sat, 20 May 2017 00:00:43 +0200 (CEST) Received: (qmail 17976 invoked by uid 500); 19 May 2017 22:00:40 -0000 Mailing-List: contact commits-help@geode.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@geode.apache.org Delivered-To: mailing list commits@geode.apache.org Received: (qmail 16637 invoked by uid 99); 19 May 2017 22:00:39 -0000 Received: from git1-us-west.apache.org (HELO git1-us-west.apache.org) (140.211.11.23) by apache.org (qpsmtpd/0.29) with ESMTP; Fri, 19 May 2017 22:00:39 +0000 Received: by git1-us-west.apache.org (ASF Mail Server at git1-us-west.apache.org, from userid 33) id C957DF21AC; Fri, 19 May 2017 22:00:38 +0000 (UTC) Content-Type: text/plain; charset="us-ascii" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit From: klund@apache.org To: commits@geode.apache.org Date: Fri, 19 May 2017 22:00:58 -0000 Message-Id: In-Reply-To: References: X-Mailer: ASF-Git Admin Mailer Subject: [21/28] geode git commit: Cleanup BaseCommand archived-at: Fri, 19 May 2017 22:00:46 -0000 http://git-wip-us.apache.org/repos/asf/geode/blob/db81427f/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Put65.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Put65.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Put65.java index d53c89e..581aec6 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Put65.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Put65.java @@ -63,7 +63,7 @@ public class Put65 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; @@ -72,11 +72,11 @@ public class Put65 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(); @@ -84,50 +84,50 @@ public class Put65 extends BaseCommand { } // Retrieve the data from the message parts int idx = 0; - regionNamePart = msg.getPart(idx++); + regionNamePart = clientMessage.getPart(idx++); Operation operation; try { - operation = (Operation) msg.getPart(idx++).getObject(); + operation = (Operation) clientMessage.getPart(idx++).getObject(); if (operation == null) { // native clients send a null since the op is java-serialized operation = Operation.UPDATE; } } catch (ClassNotFoundException e) { - writeException(msg, e, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } - int flags = msg.getPart(idx++).getInt(); + int flags = clientMessage.getPart(idx++).getInt(); boolean requireOldValue = ((flags & 0x01) == 0x01); boolean haveExpectedOldValue = ((flags & 0x02) == 0x02); Object expectedOldValue = null; if (haveExpectedOldValue) { try { - expectedOldValue = msg.getPart(idx++).getObject(); + expectedOldValue = clientMessage.getPart(idx++).getObject(); } catch (ClassNotFoundException e) { - writeException(msg, e, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } } - keyPart = msg.getPart(idx++); + keyPart = clientMessage.getPart(idx++); try { - isDelta = ((Boolean) msg.getPart(idx).getObject()).booleanValue(); + isDelta = ((Boolean) clientMessage.getPart(idx).getObject()).booleanValue(); idx += 1; } 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(idx++); - eventPart = msg.getPart(idx++); - if (msg.getNumberOfParts() > idx) { - callbackArgPart = msg.getPart(idx++); + valuePart = clientMessage.getPart(idx++); + eventPart = clientMessage.getPart(idx++); + if (clientMessage.getNumberOfParts() > idx) { + callbackArgPart = clientMessage.getPart(idx++); try { callbackArg = callbackArgPart.getObject(); } catch (Exception e) { - writeException(msg, e, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } } @@ -136,8 +136,8 @@ public class Put65 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; } @@ -145,8 +145,8 @@ public class Put65 extends BaseCommand { if (isDebugEnabled) { logger.debug( "{}: Received {}put request ({} bytes) from {} for region {} key {} txId {} posdup: {}", - servConn.getName(), (isDelta ? " delta " : " "), msg.getPayloadLength(), - servConn.getSocketString(), regionName, key, msg.getTransactionId(), msg.isRetry()); + serverConnection.getName(), (isDelta ? " delta " : " "), clientMessage.getPayloadLength(), + serverConnection.getSocketString(), regionName, key, clientMessage.getTransactionId(), clientMessage.isRetry()); } // Process the put request @@ -154,27 +154,27 @@ public class Put65 extends BaseCommand { if (key == null) { String putMsg = " The input key for the 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 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 put request"; - writeRegionDestroyedEx(msg, regionName, reason, servConn); - servConn.setAsTrue(RESPONDED); + writeRegionDestroyedEx(clientMessage, regionName, reason, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } @@ -182,11 +182,11 @@ public class Put65 extends BaseCommand { // Invalid to 'put' a null value in an existing key String putMsg = " Attempted to 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; } @@ -195,12 +195,12 @@ public class Put65 extends BaseCommand { long sequenceId = EventID.readEventIdPartsFromOptmizedByteArray(eventIdPartsBuffer); EventIDHolder clientEvent = - new EventIDHolder(new EventID(servConn.getEventMemberIDByteArray(), threadId, sequenceId)); + new EventIDHolder(new EventID(serverConnection.getEventMemberIDByteArray(), threadId, sequenceId)); Breadcrumbs.setEventId(clientEvent.getEventId()); // 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 Put65"); // } @@ -226,13 +226,13 @@ public class Put65 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)) { @@ -257,7 +257,7 @@ public class Put65 extends BaseCommand { // to be publicly accessible. if (operation == Operation.PUT_IF_ABSENT) { // try { - if (msg.isRetry() && clientEvent.getVersionTag() != null) { + if (clientMessage.isRetry() && clientEvent.getVersionTag() != null) { // bug #46590 the operation was successful the last time since it // was applied to the cache, so return success and the recovered // version tag @@ -267,16 +267,16 @@ public class Put65 extends BaseCommand { } // invoke basicBridgePutIfAbsent anyway to ensure that the event is distributed to all // servers - bug #51664 - region.basicBridgePutIfAbsent(key, value, isObject, callbackArg, servConn.getProxyID(), + region.basicBridgePutIfAbsent(key, value, isObject, callbackArg, serverConnection.getProxyID(), true, clientEvent); oldValue = null; } else { oldValue = region.basicBridgePutIfAbsent(key, value, isObject, callbackArg, - servConn.getProxyID(), true, clientEvent); + serverConnection.getProxyID(), true, clientEvent); } sendOldValue = true; oldValueIsObject = true; - Version clientVersion = servConn.getClientVersion(); + Version clientVersion = serverConnection.getClientVersion(); if (oldValue instanceof CachedDeserializable) { oldValue = ((CachedDeserializable) oldValue).getSerializedValue(); } else if (oldValue instanceof byte[]) { @@ -299,7 +299,7 @@ public class Put65 extends BaseCommand { } else if (operation == Operation.REPLACE) { // try { if (requireOldValue) { // replace(, ) - if (msg.isRetry() && clientEvent.isConcurrencyConflict() + if (clientMessage.isRetry() && clientEvent.isConcurrencyConflict() && clientEvent.getVersionTag() != null) { if (isDebugEnabled) { logger.debug("replace(k,v) operation was successful last time with version {}", @@ -307,10 +307,10 @@ public class Put65 extends BaseCommand { } } oldValue = region.basicBridgeReplace(key, value, isObject, callbackArg, - servConn.getProxyID(), true, clientEvent); + serverConnection.getProxyID(), true, clientEvent); sendOldValue = !clientEvent.isConcurrencyConflict(); oldValueIsObject = true; - Version clientVersion = servConn.getClientVersion(); + Version clientVersion = serverConnection.getClientVersion(); if (oldValue instanceof CachedDeserializable) { oldValue = ((CachedDeserializable) oldValue).getSerializedValue(); } else if (oldValue instanceof byte[]) { @@ -330,8 +330,8 @@ public class Put65 extends BaseCommand { } else { // boolean replace(, , ) { boolean didPut; didPut = region.basicBridgeReplace(key, expectedOldValue, value, isObject, callbackArg, - servConn.getProxyID(), true, clientEvent); - if (msg.isRetry() && clientEvent.getVersionTag() != null) { + serverConnection.getProxyID(), true, clientEvent); + if (clientMessage.isRetry() && clientEvent.getVersionTag() != null) { if (isDebugEnabled) { logger.debug("replace(k,v,v) operation was successful last time with version {}", clientEvent.getVersionTag()); @@ -356,9 +356,9 @@ public class Put65 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, clientEvent, false); - if (msg.isRetry() && clientEvent.isConcurrencyConflict() + if (clientMessage.isRetry() && clientEvent.isConcurrencyConflict() && clientEvent.getVersionTag() != null) { result = true; if (isDebugEnabled) { @@ -372,16 +372,16 @@ public class Put65 extends BaseCommand { if (isDelta) { delta = valuePart.getSerializedForm(); } - TXManagerImpl txMgr = (TXManagerImpl) servConn.getCache().getCacheTransactionManager(); + TXManagerImpl txMgr = (TXManagerImpl) serverConnection.getCache().getCacheTransactionManager(); // bug 43068 - use create() if in a transaction and op is CREATE if (txMgr.getTXState() != null && operation.isCreate()) { result = region.basicBridgeCreate(key, (byte[]) value, isObject, callbackArg, - servConn.getProxyID(), true, clientEvent, true); + serverConnection.getProxyID(), true, clientEvent, true); } else { result = region.basicBridgePut(key, value, delta, isObject, callbackArg, - servConn.getProxyID(), true, clientEvent); + serverConnection.getProxyID(), true, clientEvent); } - if (msg.isRetry() && clientEvent.isConcurrencyConflict() + if (clientMessage.isRetry() && clientEvent.isConcurrencyConflict() && clientEvent.getVersionTag() != null) { if (isDebugEnabled) { logger.debug("put(k,v) operation was successful last time with version {}", @@ -391,46 +391,46 @@ public class Put65 extends BaseCommand { } } if (result) { - servConn.setModificationInfo(true, regionName, key); + serverConnection.setModificationInfo(true, regionName, key); } else { - String message = servConn.getName() + ": Failed to put entry for region " + regionName - + " key " + key + " value " + valuePart; + String message = serverConnection.getName() + ": Failed to 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 { @@ -443,21 +443,21 @@ public class Put65 extends BaseCommand { if (region instanceof PartitionedRegion) { PartitionedRegion pr = (PartitionedRegion) region; if (pr.getNetworkHopType() != PartitionedRegion.NETWORK_HOP_NONE) { - writeReplyWithRefreshMetadata(msg, servConn, pr, sendOldValue, oldValueIsObject, oldValue, + writeReplyWithRefreshMetadata(clientMessage, serverConnection, pr, sendOldValue, oldValueIsObject, oldValue, pr.getNetworkHopType(), clientEvent.getVersionTag()); pr.clearNetworkHopData(); } else { - writeReply(msg, servConn, sendOldValue, oldValueIsObject, oldValue, + writeReply(clientMessage, serverConnection, sendOldValue, oldValueIsObject, oldValue, clientEvent.getVersionTag()); } } else { - writeReply(msg, servConn, sendOldValue, oldValueIsObject, oldValue, + writeReply(clientMessage, serverConnection, sendOldValue, oldValueIsObject, oldValue, clientEvent.getVersionTag()); } - servConn.setAsTrue(RESPONDED); + serverConnection.setAsTrue(RESPONDED); if (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); @@ -471,7 +471,7 @@ public class Put65 extends BaseCommand { replyMsg.setMessageType(MessageType.REPLY); replyMsg.setNumberOfParts(sendOldValue ? 3 : 1); replyMsg.setTransactionId(origMsg.getTransactionId()); - replyMsg.addBytesPart(OK_BYTES); + replyMsg.addBytesPart(okBytes()); if (sendOldValue) { replyMsg.addIntPart(oldValueIsObject ? 1 : 0); replyMsg.addObjPart(oldValue); @@ -499,7 +499,7 @@ public class Put65 extends BaseCommand { replyMsg.send(servConn); pr.getPrStats().incPRMetaDataSentCount(); if (logger.isTraceEnabled()) { - logger.trace("{}: rpl with REFRESH_METADAT tx: {} parts={}", servConn.getName(), + logger.trace("{}: rpl with REFRESH_METADATA tx: {} parts={}", servConn.getName(), origMsg.getTransactionId(), replyMsg.getNumberOfParts()); } } http://git-wip-us.apache.org/repos/asf/geode/blob/db81427f/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Put70.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Put70.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Put70.java index 38eb7ef..395dbce 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Put70.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Put70.java @@ -62,7 +62,7 @@ public class Put70 extends Put65 { } replyMsg.setNumberOfParts(parts); replyMsg.setTransactionId(origMsg.getTransactionId()); - replyMsg.addBytesPart(OK_BYTES); + replyMsg.addBytesPart(okBytes()); replyMsg.addIntPart(flags); if (sendOldValue) { replyMsg.addObjPart(oldValue); @@ -114,7 +114,7 @@ public class Put70 extends Put65 { replyMsg.send(servConn); pr.getPrStats().incPRMetaDataSentCount(); if (logger.isTraceEnabled()) { - logger.trace("{}: rpl with REFRESH_METADAT tx: {} parts={}", servConn.getName(), + logger.trace("{}: rpl with REFRESH_METADATA tx: {} parts={}", servConn.getName(), origMsg.getTransactionId(), replyMsg.getNumberOfParts()); } } http://git-wip-us.apache.org/repos/asf/geode/blob/db81427f/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PutAll.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PutAll.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PutAll.java index 0bcfd1b..281f737 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PutAll.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PutAll.java @@ -59,7 +59,7 @@ public class PutAll extends BaseCommand { private PutAll() {} @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, numberOfKeysPart = null, keyPart = null, valuePart = null; String regionName = null; @@ -67,12 +67,12 @@ public class PutAll extends BaseCommand { Object key = null; Part eventPart = null; StringBuffer errMessage = new StringBuffer(); - CachedRegionHelper crHelper = servConn.getCachedRegionHelper(); - CacheServerStats stats = servConn.getCacheServerStats(); + CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper(); + CacheServerStats stats = serverConnection.getCacheServerStats(); boolean replyWithMetaData = false; // requiresResponse = true; - servConn.setAsTrue(REQUIRES_RESPONSE); + serverConnection.setAsTrue(REQUIRES_RESPONSE); { long oldStart = start; start = DistributionStats.getStatTime(); @@ -82,64 +82,64 @@ public class PutAll extends BaseCommand { try { // Retrieve the data from the message parts // part 0: region name - regionNamePart = msg.getPart(0); + regionNamePart = clientMessage.getPart(0); regionName = regionNamePart.getString(); if (regionName == null) { String putAllMsg = LocalizedStrings.PutAll_THE_INPUT_REGION_NAME_FOR_THE_PUTALL_REQUEST_IS_NULL .toLocalizedString(); - logger.warn("{}: {}", servConn.getName(), putAllMsg); + logger.warn("{}: {}", serverConnection.getName(), putAllMsg); errMessage.append(putAllMsg); - 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) crHelper.getRegion(regionName); if (region == null) { String reason = " was not found during put request"; - writeRegionDestroyedEx(msg, regionName, reason, servConn); - servConn.setAsTrue(RESPONDED); + writeRegionDestroyedEx(clientMessage, regionName, reason, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } // part 1: eventID - eventPart = msg.getPart(1); + eventPart = clientMessage.getPart(1); 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); // part 2: number of keys - numberOfKeysPart = msg.getPart(2); + numberOfKeysPart = clientMessage.getPart(2); numberOfKeys = numberOfKeysPart.getInt(); // building the map Map map = new LinkedHashMap(); // Map isObjectMap = new LinkedHashMap(); for (int i = 0; i < numberOfKeys; i++) { - keyPart = msg.getPart(3 + i * 2); + keyPart = clientMessage.getPart(3 + i * 2); key = keyPart.getStringOrObject(); if (key == null) { String putAllMsg = LocalizedStrings.PutAll_ONE_OF_THE_INPUT_KEYS_FOR_THE_PUTALL_REQUEST_IS_NULL .toLocalizedString(); - logger.warn("{}: {}", servConn.getName(), putAllMsg); + logger.warn("{}: {}", serverConnection.getName(), putAllMsg); errMessage.append(putAllMsg); - 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; } - valuePart = msg.getPart(3 + i * 2 + 1); + valuePart = clientMessage.getPart(3 + i * 2 + 1); if (valuePart.isNull()) { String putAllMsg = LocalizedStrings.PutAll_ONE_OF_THE_INPUT_VALUES_FOR_THE_PUTALL_REQUEST_IS_NULL .toLocalizedString(); - logger.warn("{}: {}", servConn.getName(), putAllMsg); + logger.warn("{}: {}", serverConnection.getName(), putAllMsg); errMessage.append(putAllMsg); - 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; } @@ -155,15 +155,15 @@ public class PutAll extends BaseCommand { // isObjectMap.put(key, new Boolean(isObject)); } // for - if (msg.getNumberOfParts() == (3 + 2 * numberOfKeys + 1)) {// it means optional timeout has + if (clientMessage.getNumberOfParts() == (3 + 2 * numberOfKeys + 1)) {// it means optional timeout has // been added - int timeout = msg.getPart(3 + 2 * numberOfKeys).getInt(); - servConn.setRequestSpecificTimeout(timeout); + int timeout = clientMessage.getPart(3 + 2 * numberOfKeys).getInt(); + serverConnection.setRequestSpecificTimeout(timeout); } this.securityService.authorizeRegionWrite(regionName); - AuthorizeRequest authzRequest = servConn.getAuthzRequest(); + AuthorizeRequest authzRequest = serverConnection.getAuthzRequest(); if (authzRequest != null) { if (DynamicRegionFactory.regionIsDynamicRegionList(regionName)) { authzRequest.createRegionAuthorize(regionName); @@ -179,41 +179,41 @@ public class PutAll extends BaseCommand { if (logger.isDebugEnabled()) { logger.debug("{}: Received putAll request ({} bytes) from {} for region {}", - servConn.getName(), msg.getPayloadLength(), servConn.getSocketString(), regionName); + serverConnection.getName(), clientMessage.getPayloadLength(), serverConnection.getSocketString(), regionName); } region.basicBridgePutAll(map, Collections.emptyMap(), - servConn.getProxyID(), eventId, false, null); + serverConnection.getProxyID(), eventId, false, null); 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(); replyWithMetaData = true; } } } 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 (PutAllPartialResultException pre) { - writeException(msg, pre, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, pre, 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); logger.warn(LocalizedMessage.create(LocalizedStrings.Generic_0_UNEXPECTED_EXCEPTION, - servConn.getName()), ce); + serverConnection.getName()), ce); return; } finally { long oldStart = start; @@ -223,12 +223,12 @@ public class PutAll extends BaseCommand { // Increment statistics and write the reply if (!replyWithMetaData) { - writeReply(msg, servConn); + writeReply(clientMessage, serverConnection); } - servConn.setAsTrue(RESPONDED); + serverConnection.setAsTrue(RESPONDED); if (logger.isDebugEnabled()) { - logger.debug("{}: Sent putAll response back to {} for region {}", servConn.getName(), - servConn.getSocketString(), regionName); + logger.debug("{}: Sent putAll response back to {} for region {}", serverConnection.getName(), + serverConnection.getSocketString(), regionName); } stats.incWritePutAllResponseTime(DistributionStats.getStatTime() - start); } http://git-wip-us.apache.org/repos/asf/geode/blob/db81427f/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PutAll70.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PutAll70.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PutAll70.java index c5fcbae..ae2de09 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PutAll70.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PutAll70.java @@ -62,7 +62,7 @@ public class PutAll70 extends BaseCommand { private PutAll70() {} @Override - public void cmdExecute(Message msg, ServerConnection servConn, long startp) + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long startp) throws IOException, InterruptedException { long start = startp; // copy this since we need to modify it Part regionNamePart = null, numberOfKeysPart = null, keyPart = null, valuePart = null; @@ -74,11 +74,11 @@ public class PutAll70 extends BaseCommand { VersionedObjectList response = null; StringBuffer errMessage = new StringBuffer(); - 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(); @@ -88,40 +88,40 @@ public class PutAll70 extends BaseCommand { try { // Retrieve the data from the message parts // part 0: region name - regionNamePart = msg.getPart(0); + regionNamePart = clientMessage.getPart(0); regionName = regionNamePart.getString(); if (regionName == null) { String putAllMsg = LocalizedStrings.PutAll_THE_INPUT_REGION_NAME_FOR_THE_PUTALL_REQUEST_IS_NULL .toLocalizedString(); - logger.warn("{}: {}", servConn.getName(), putAllMsg); + logger.warn("{}: {}", serverConnection.getName(), putAllMsg); errMessage.append(putAllMsg); - 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) crHelper.getRegion(regionName); if (region == null) { String reason = " was not found during put request"; - writeRegionDestroyedEx(msg, regionName, reason, servConn); - servConn.setAsTrue(RESPONDED); + writeRegionDestroyedEx(clientMessage, regionName, reason, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } // part 1: eventID - eventPart = msg.getPart(1); + eventPart = clientMessage.getPart(1); 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); // part 2: invoke callbacks (used by import) - Part callbacksPart = msg.getPart(2); + Part callbacksPart = clientMessage.getPart(2); boolean skipCallbacks = callbacksPart.getInt() == 1 ? true : false; // part 3: number of keys - numberOfKeysPart = msg.getPart(3); + numberOfKeysPart = clientMessage.getPart(3); numberOfKeys = numberOfKeysPart.getInt(); // building the map @@ -129,28 +129,28 @@ public class PutAll70 extends BaseCommand { Map retryVersions = new LinkedHashMap(); // Map isObjectMap = new LinkedHashMap(); for (int i = 0; i < numberOfKeys; i++) { - keyPart = msg.getPart(4 + i * 2); + keyPart = clientMessage.getPart(4 + i * 2); key = keyPart.getStringOrObject(); if (key == null) { String putAllMsg = LocalizedStrings.PutAll_ONE_OF_THE_INPUT_KEYS_FOR_THE_PUTALL_REQUEST_IS_NULL .toLocalizedString(); - logger.warn("{}: {}", servConn.getName(), putAllMsg); + logger.warn("{}: {}", serverConnection.getName(), putAllMsg); errMessage.append(putAllMsg); - 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; } - valuePart = msg.getPart(4 + i * 2 + 1); + valuePart = clientMessage.getPart(4 + i * 2 + 1); if (valuePart.isNull()) { String putAllMsg = LocalizedStrings.PutAll_ONE_OF_THE_INPUT_VALUES_FOR_THE_PUTALL_REQUEST_IS_NULL .toLocalizedString(); - logger.warn("{}: {}", servConn.getName(), putAllMsg); + logger.warn("{}: {}", serverConnection.getName(), putAllMsg); errMessage.append(putAllMsg); - 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; } @@ -170,7 +170,7 @@ public class PutAll70 extends BaseCommand { value = valuePart.getSerializedForm(); } // put serializedform for auth. It will be modified with auth callback - if (msg.isRetry()) { + if (clientMessage.isRetry()) { // Constuct the thread id/sequence id information for this element in the // put all map @@ -198,15 +198,15 @@ public class PutAll70 extends BaseCommand { // isObjectMap.put(key, new Boolean(isObject)); } // for - if (msg.getNumberOfParts() == (4 + 2 * numberOfKeys + 1)) {// it means optional timeout has + if (clientMessage.getNumberOfParts() == (4 + 2 * numberOfKeys + 1)) {// it means optional timeout has // been added - int timeout = msg.getPart(4 + 2 * numberOfKeys).getInt(); - servConn.setRequestSpecificTimeout(timeout); + int timeout = clientMessage.getPart(4 + 2 * numberOfKeys).getInt(); + serverConnection.setRequestSpecificTimeout(timeout); } this.securityService.authorizeRegionWrite(regionName); - AuthorizeRequest authzRequest = servConn.getAuthzRequest(); + AuthorizeRequest authzRequest = serverConnection.getAuthzRequest(); if (authzRequest != null) { if (DynamicRegionFactory.regionIsDynamicRegionList(regionName)) { authzRequest.createRegionAuthorize(regionName); @@ -231,10 +231,10 @@ public class PutAll70 extends BaseCommand { if (logger.isDebugEnabled()) { logger.debug("{}: Received putAll request ({} bytes) from {} for region {}", - servConn.getName(), msg.getPayloadLength(), servConn.getSocketString(), regionName); + serverConnection.getName(), clientMessage.getPayloadLength(), serverConnection.getSocketString(), regionName); } - response = region.basicBridgePutAll(map, retryVersions, servConn.getProxyID(), eventId, + response = region.basicBridgePutAll(map, retryVersions, serverConnection.getProxyID(), eventId, skipCallbacks, null); if (!region.getConcurrencyChecksEnabled()) { // the client only needs this if versioning is being used @@ -244,33 +244,33 @@ public class PutAll70 extends BaseCommand { if (region instanceof PartitionedRegion) { PartitionedRegion pr = (PartitionedRegion) region; if (pr.getNetworkHopType() != PartitionedRegion.NETWORK_HOP_NONE) { - writeReplyWithRefreshMetadata(msg, response, servConn, pr, pr.getNetworkHopType()); + writeReplyWithRefreshMetadata(clientMessage, response, serverConnection, pr, pr.getNetworkHopType()); pr.clearNetworkHopData(); replyWithMetaData = true; } } } 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 (PutAllPartialResultException pre) { - writeException(msg, pre, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, pre, 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 (logger.fineEnabled()) { logger.warn(LocalizedMessage.create(LocalizedStrings.Generic_0_UNEXPECTED_EXCEPTION, - servConn.getName()), ce); + serverConnection.getName()), ce); // } return; } finally { @@ -279,11 +279,11 @@ public class PutAll70 extends BaseCommand { stats.incProcessPutAllTime(start - oldStart); } if (logger.isDebugEnabled()) { - logger.debug("{}: Sending putAll70 response back to {} for region {}: {}", servConn.getName(), - servConn.getSocketString(), regionName, response); + logger.debug("{}: Sending putAll70 response back to {} for region {}: {}", serverConnection.getName(), + serverConnection.getSocketString(), regionName, response); } // Starting in 7.0.1 we do not send the keys back - if (response != null && Version.GFE_70.compareTo(servConn.getClientVersion()) < 0) { + if (response != null && Version.GFE_70.compareTo(serverConnection.getClientVersion()) < 0) { if (logger.isDebugEnabled()) { logger.debug("setting putAll keys to null"); } @@ -292,14 +292,14 @@ public class PutAll70 extends BaseCommand { // Increment statistics and write the reply if (!replyWithMetaData) { - writeReply(msg, response, servConn); + writeReply(clientMessage, response, serverConnection); } - servConn.setAsTrue(RESPONDED); + serverConnection.setAsTrue(RESPONDED); stats.incWritePutAllResponseTime(DistributionStats.getStatTime() - start); } @Override - protected void writeReply(Message origMsg, ServerConnection servConn) throws IOException { + protected void writeReply(Message origMsg, ServerConnection serverConnection) throws IOException { throw new UnsupportedOperationException(); } @@ -311,7 +311,7 @@ public class PutAll70 extends BaseCommand { replyMsg.setMessageType(MessageType.REPLY); replyMsg.setNumberOfParts(2); replyMsg.setTransactionId(origMsg.getTransactionId()); - replyMsg.addBytesPart(OK_BYTES); + replyMsg.addBytesPart(okBytes()); if (response != null) { response.clearObjects(); replyMsg.addObjPart(response); @@ -323,7 +323,7 @@ public class PutAll70 extends BaseCommand { } @Override - protected void writeReplyWithRefreshMetadata(Message origMsg, ServerConnection servConn, + protected void writeReplyWithRefreshMetadata(Message origMsg, ServerConnection serverConnection, PartitionedRegion pr, byte nwHop) throws IOException { throw new UnsupportedOperationException(); } @@ -343,7 +343,7 @@ public class PutAll70 extends BaseCommand { replyMsg.send(servConn); pr.getPrStats().incPRMetaDataSentCount(); if (logger.isTraceEnabled()) { - logger.trace("{}: rpl with REFRESH_METADAT tx: {}", servConn.getName(), + logger.trace("{}: rpl with REFRESH_METADATA tx: {}", servConn.getName(), origMsg.getTransactionId()); } } http://git-wip-us.apache.org/repos/asf/geode/blob/db81427f/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PutAll80.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PutAll80.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PutAll80.java index a6285ed..aed5926 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PutAll80.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PutAll80.java @@ -75,7 +75,7 @@ public class PutAll80 extends BaseCommand { } @Override - public void cmdExecute(Message msg, ServerConnection servConn, long startp) + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long startp) throws IOException, InterruptedException { long start = startp; // copy this since we need to modify it Part regionNamePart = null, numberOfKeysPart = null, keyPart = null, valuePart = null; @@ -87,12 +87,12 @@ public class PutAll80 extends BaseCommand { VersionedObjectList response = null; StringBuffer errMessage = new StringBuffer(); - CachedRegionHelper crHelper = servConn.getCachedRegionHelper(); - CacheServerStats stats = servConn.getCacheServerStats(); + CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper(); + CacheServerStats stats = serverConnection.getCacheServerStats(); // requiresResponse = true; - servConn.setAsTrue(REQUIRES_RESPONSE); - servConn.setAsTrue(REQUIRES_CHUNKED_RESPONSE); // new in 8.0 + serverConnection.setAsTrue(REQUIRES_RESPONSE); + serverConnection.setAsTrue(REQUIRES_CHUNKED_RESPONSE); // new in 8.0 { long oldStart = start; start = DistributionStats.getStatTime(); @@ -102,60 +102,60 @@ public class PutAll80 extends BaseCommand { try { // Retrieve the data from the message parts // part 0: region name - regionNamePart = msg.getPart(0); + regionNamePart = clientMessage.getPart(0); regionName = regionNamePart.getString(); if (regionName == null) { String putAllMsg = LocalizedStrings.PutAll_THE_INPUT_REGION_NAME_FOR_THE_PUTALL_REQUEST_IS_NULL .toLocalizedString(); - logger.warn("{}: {}", servConn.getName(), putAllMsg); + logger.warn("{}: {}", serverConnection.getName(), putAllMsg); errMessage.append(putAllMsg); - writeChunkedErrorResponse(msg, MessageType.PUT_DATA_ERROR, errMessage.toString(), servConn); - servConn.setAsTrue(RESPONDED); + writeChunkedErrorResponse(clientMessage, MessageType.PUT_DATA_ERROR, errMessage.toString(), serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } LocalRegion region = (LocalRegion) crHelper.getRegion(regionName); if (region == null) { String reason = " was not found during putAll request"; - writeRegionDestroyedEx(msg, regionName, reason, servConn); - servConn.setAsTrue(RESPONDED); + writeRegionDestroyedEx(clientMessage, regionName, reason, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } final int BASE_PART_COUNT = getBasePartCount(); // part 1: eventID - eventPart = msg.getPart(1); + eventPart = clientMessage.getPart(1); 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); // part 2: invoke callbacks (used by import) - Part callbacksPart = msg.getPart(2); + Part callbacksPart = clientMessage.getPart(2); boolean skipCallbacks = callbacksPart.getInt() == 1 ? true : false; // part 3: flags - int flags = msg.getPart(3).getInt(); + int flags = clientMessage.getPart(3).getInt(); boolean clientIsEmpty = (flags & PutAllOp.FLAG_EMPTY) != 0; boolean clientHasCCEnabled = (flags & PutAllOp.FLAG_CONCURRENCY_CHECKS) != 0; // part 4: number of keys - numberOfKeysPart = msg.getPart(4); + numberOfKeysPart = clientMessage.getPart(4); numberOfKeys = numberOfKeysPart.getInt(); - Object callbackArg = getOptionalCallbackArg(msg); + Object callbackArg = getOptionalCallbackArg(clientMessage); if (logger.isDebugEnabled()) { StringBuilder buffer = new StringBuilder(); - buffer.append(servConn.getName()).append(": Received ").append(this.putAllClassName()) - .append(" request from ").append(servConn.getSocketString()).append(" for region ") - .append(regionName).append(callbackArg != null ? (" callbackArg " + callbackArg) : "") - .append(" with ").append(numberOfKeys).append(" entries."); + buffer.append(serverConnection.getName()).append(": Received ").append(this.putAllClassName()) + .append(" request from ").append(serverConnection.getSocketString()).append(" for region ") + .append(regionName).append(callbackArg != null ? (" callbackArg " + callbackArg) : "") + .append(" with ").append(numberOfKeys).append(" entries."); logger.debug(buffer.toString()); } // building the map @@ -163,30 +163,28 @@ public class PutAll80 extends BaseCommand { Map retryVersions = new LinkedHashMap(); // Map isObjectMap = new LinkedHashMap(); for (int i = 0; i < numberOfKeys; i++) { - keyPart = msg.getPart(BASE_PART_COUNT + i * 2); + keyPart = clientMessage.getPart(BASE_PART_COUNT + i * 2); key = keyPart.getStringOrObject(); if (key == null) { String putAllMsg = LocalizedStrings.PutAll_ONE_OF_THE_INPUT_KEYS_FOR_THE_PUTALL_REQUEST_IS_NULL .toLocalizedString(); - logger.warn("{}: {}", servConn.getName(), putAllMsg); + logger.warn("{}: {}", serverConnection.getName(), putAllMsg); errMessage.append(putAllMsg); - writeChunkedErrorResponse(msg, MessageType.PUT_DATA_ERROR, errMessage.toString(), - servConn); - servConn.setAsTrue(RESPONDED); + writeChunkedErrorResponse(clientMessage, MessageType.PUT_DATA_ERROR, errMessage.toString(), serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } - valuePart = msg.getPart(BASE_PART_COUNT + i * 2 + 1); + valuePart = clientMessage.getPart(BASE_PART_COUNT + i * 2 + 1); if (valuePart.isNull()) { String putAllMsg = LocalizedStrings.PutAll_ONE_OF_THE_INPUT_VALUES_FOR_THE_PUTALL_REQUEST_IS_NULL .toLocalizedString(); - logger.warn("{}: {}", servConn.getName(), putAllMsg); + logger.warn("{}: {}", serverConnection.getName(), putAllMsg); errMessage.append(putAllMsg); - writeChunkedErrorResponse(msg, MessageType.PUT_DATA_ERROR, errMessage.toString(), - servConn); - servConn.setAsTrue(RESPONDED); + writeChunkedErrorResponse(clientMessage, MessageType.PUT_DATA_ERROR, errMessage.toString(), serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } @@ -206,7 +204,7 @@ public class PutAll80 extends BaseCommand { value = valuePart.getSerializedForm(); } // put serializedform for auth. It will be modified with auth callback - if (msg.isRetry()) { + if (clientMessage.isRetry()) { // Constuct the thread id/sequence id information for this element in the // put all map @@ -234,16 +232,16 @@ public class PutAll80 extends BaseCommand { // isObjectMap.put(key, new Boolean(isObject)); } // for - if (msg.getNumberOfParts() == (BASE_PART_COUNT + 2 * numberOfKeys + 1)) {// it means optional + if (clientMessage.getNumberOfParts() == (BASE_PART_COUNT + 2 * numberOfKeys + 1)) {// it means optional // timeout has been // added - int timeout = msg.getPart(BASE_PART_COUNT + 2 * numberOfKeys).getInt(); - servConn.setRequestSpecificTimeout(timeout); + int timeout = clientMessage.getPart(BASE_PART_COUNT + 2 * numberOfKeys).getInt(); + serverConnection.setRequestSpecificTimeout(timeout); } this.securityService.authorizeRegionWrite(regionName); - AuthorizeRequest authzRequest = servConn.getAuthzRequest(); + AuthorizeRequest authzRequest = serverConnection.getAuthzRequest(); if (authzRequest != null) { if (DynamicRegionFactory.regionIsDynamicRegionList(regionName)) { authzRequest.createRegionAuthorize(regionName); @@ -267,7 +265,7 @@ public class PutAll80 extends BaseCommand { */ } - response = region.basicBridgePutAll(map, retryVersions, servConn.getProxyID(), eventId, + response = region.basicBridgePutAll(map, retryVersions, serverConnection.getProxyID(), eventId, skipCallbacks, callbackArg); if (!region.getConcurrencyChecksEnabled() || clientIsEmpty || !clientHasCCEnabled) { // the client only needs this if versioning is being used and the client @@ -283,32 +281,32 @@ public class PutAll80 extends BaseCommand { if (region instanceof PartitionedRegion) { PartitionedRegion pr = (PartitionedRegion) region; if (pr.getNetworkHopType() != PartitionedRegion.NETWORK_HOP_NONE) { - writeReplyWithRefreshMetadata(msg, response, servConn, pr, pr.getNetworkHopType()); + writeReplyWithRefreshMetadata(clientMessage, response, serverConnection, pr, pr.getNetworkHopType()); pr.clearNetworkHopData(); replyWithMetaData = true; } } } catch (RegionDestroyedException rde) { - writeChunkedException(msg, rde, false, servConn); - servConn.setAsTrue(RESPONDED); + writeChunkedException(clientMessage, rde, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } catch (ResourceException re) { - writeChunkedException(msg, re, false, servConn); - servConn.setAsTrue(RESPONDED); + writeChunkedException(clientMessage, re, serverConnection); + serverConnection.setAsTrue(RESPONDED); return; } catch (PutAllPartialResultException pre) { - writeChunkedException(msg, pre, false, servConn); - servConn.setAsTrue(RESPONDED); + writeChunkedException(clientMessage, pre, 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 - writeChunkedException(msg, ce, false, servConn); - servConn.setAsTrue(RESPONDED); + writeChunkedException(clientMessage, ce, serverConnection); + serverConnection.setAsTrue(RESPONDED); logger.warn(LocalizedMessage.create(LocalizedStrings.Generic_0_UNEXPECTED_EXCEPTION, - servConn.getName()), ce); + serverConnection.getName()), ce); return; } finally { long oldStart = start; @@ -316,21 +314,21 @@ public class PutAll80 extends BaseCommand { stats.incProcessPutAllTime(start - oldStart); } if (logger.isDebugEnabled()) { - logger.debug("{}: Sending {} response back to {} for regin {} {}", servConn.getName(), - putAllClassName(), servConn.getSocketString(), regionName, + logger.debug("{}: Sending {} response back to {} for regin {} {}", serverConnection.getName(), + putAllClassName(), serverConnection.getSocketString(), regionName, (logger.isTraceEnabled() ? ": " + response : "")); } // Increment statistics and write the reply if (!replyWithMetaData) { - writeReply(msg, response, servConn); + writeReply(clientMessage, response, serverConnection); } - servConn.setAsTrue(RESPONDED); + serverConnection.setAsTrue(RESPONDED); stats.incWritePutAllResponseTime(DistributionStats.getStatTime() - start); } @Override - protected void writeReply(Message origMsg, ServerConnection servConn) throws IOException { + protected void writeReply(Message origMsg, ServerConnection serverConnection) throws IOException { throw new UnsupportedOperationException(); } @@ -351,7 +349,7 @@ public class PutAll80 extends BaseCommand { } replyMsg.sendHeader(); if (listSize > 0) { - int chunkSize = 2 * maximumChunkSize; + int chunkSize = 2 * MAXIMUM_CHUNK_SIZE; // Chunker will stream over the list in its toData method VersionedObjectList.Chunker chunk = new VersionedObjectList.Chunker(response, chunkSize, false, false); @@ -383,7 +381,7 @@ public class PutAll80 extends BaseCommand { } @Override - protected void writeReplyWithRefreshMetadata(Message origMsg, ServerConnection servConn, + protected void writeReplyWithRefreshMetadata(Message origMsg, ServerConnection serverConnection, PartitionedRegion pr, byte nwHop) throws IOException { throw new UnsupportedOperationException(); } @@ -411,7 +409,7 @@ public class PutAll80 extends BaseCommand { replyMsg.setLastChunk(false); replyMsg.sendChunk(servConn); - int chunkSize = 2 * maximumChunkSize; // maximumChunkSize + int chunkSize = 2 * MAXIMUM_CHUNK_SIZE; // MAXIMUM_CHUNK_SIZE // Chunker will stream over the list in its toData method VersionedObjectList.Chunker chunk = new VersionedObjectList.Chunker(response, chunkSize, false, false); @@ -437,7 +435,7 @@ public class PutAll80 extends BaseCommand { } pr.getPrStats().incPRMetaDataSentCount(); if (logger.isTraceEnabled()) { - logger.trace("{}: rpl with REFRESH_METADAT tx: {}", servConn.getName(), + logger.trace("{}: rpl with REFRESH_METADATA tx: {}", servConn.getName(), origMsg.getTransactionId()); } } http://git-wip-us.apache.org/repos/asf/geode/blob/db81427f/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PutUserCredentials.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PutUserCredentials.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PutUserCredentials.java index 198eed6..dc3de67 100755 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PutUserCredentials.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/PutUserCredentials.java @@ -32,39 +32,39 @@ public class PutUserCredentials 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 { - boolean isSecureMode = msg.isSecureMode(); + boolean isSecureMode = clientMessage.isSecureMode(); // if (!isSecureMode) // client has not send secuirty header, need to send exception and log this in security (file) if (isSecureMode) { - int numberOfParts = msg.getNumberOfParts(); + int numberOfParts = clientMessage.getNumberOfParts(); if (numberOfParts == 1) { // need to get credentials try { - servConn.setAsTrue(REQUIRES_RESPONSE); - byte[] uniqueId = servConn.setCredentials(msg); - writeResponse(uniqueId, null, msg, false, servConn); + serverConnection.setAsTrue(REQUIRES_RESPONSE); + byte[] uniqueId = serverConnection.setCredentials(clientMessage); + writeResponse(uniqueId, null, clientMessage, false, serverConnection); } catch (GemFireSecurityException gfse) { - if (servConn.getSecurityLogWriter().warningEnabled()) { - servConn.getSecurityLogWriter().warning(LocalizedStrings.ONE_ARG, servConn.getName() - + ": Security exception: " + gfse.toString() - + (gfse.getCause() != null ? ", caused by: " + gfse.getCause().toString() : "")); + if (serverConnection.getSecurityLogWriter().warningEnabled()) { + serverConnection.getSecurityLogWriter().warning(LocalizedStrings.ONE_ARG, serverConnection.getName() + + ": Security exception: " + gfse.toString() + + (gfse.getCause() != null ? ", caused by: " + gfse.getCause().toString() : "")); } - writeException(msg, gfse, false, servConn); + writeException(clientMessage, gfse, false, serverConnection); } catch (Exception ex) { - if (servConn.getLogWriter().warningEnabled()) { - servConn.getLogWriter().warning( + if (serverConnection.getLogWriter().warningEnabled()) { + serverConnection.getLogWriter().warning( LocalizedStrings.CacheClientNotifier_AN_EXCEPTION_WAS_THROWN_FOR_CLIENT_0_1, - new Object[] {servConn.getProxyID(), ""}, ex); + new Object[] { serverConnection.getProxyID(), ""}, ex); } - writeException(msg, ex, false, servConn); + writeException(clientMessage, ex, false, serverConnection); } finally { - servConn.setAsTrue(RESPONDED); + serverConnection.setAsTrue(RESPONDED); } } else { http://git-wip-us.apache.org/repos/asf/geode/blob/db81427f/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Query.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Query.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Query.java index d3c0393..8b5b35e 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Query.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Query.java @@ -43,38 +43,38 @@ public class Query extends BaseCommandQuery { protected Query() {} @Override - public void cmdExecute(Message msg, ServerConnection servConn, long start) + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start) throws IOException, InterruptedException { // Based on MessageType.DESTROY // Added by gregp 10/18/05 - 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 queryString = msg.getPart(0).getString(); + String queryString = clientMessage.getPart(0).getString(); // this is optional part for message specific timeout, which right now send by native client // need to take care while adding new message - if (msg.getNumberOfParts() == 3) { - int timeout = msg.getPart(2).getInt(); - servConn.setRequestSpecificTimeout(timeout); + if (clientMessage.getNumberOfParts() == 3) { + int timeout = clientMessage.getPart(2).getInt(); + serverConnection.setRequestSpecificTimeout(timeout); } if (logger.isDebugEnabled()) { - logger.debug("{}: Received query request from {} queryString: {}", servConn.getName(), - servConn.getSocketString(), queryString); + logger.debug("{}: Received query request from {} queryString: {}", serverConnection.getName(), + serverConnection.getSocketString(), queryString); } try { // Create query QueryService queryService = - servConn.getCachedRegionHelper().getCache().getLocalQueryService(); + serverConnection.getCachedRegionHelper().getCache().getLocalQueryService(); org.apache.geode.cache.query.Query query = queryService.newQuery(queryString); Set regionNames = ((DefaultQuery) query).getRegionsInQuery(null); // Authorization check QueryOperationContext queryContext = null; - AuthorizeRequest authzRequest = servConn.getAuthzRequest(); + AuthorizeRequest authzRequest = serverConnection.getAuthzRequest(); if (authzRequest != null) { queryContext = authzRequest.queryAuthorize(queryString, regionNames); String newQueryString = queryContext.getQuery(); @@ -88,11 +88,11 @@ public class Query extends BaseCommandQuery { } } - processQuery(msg, query, queryString, regionNames, start, null, queryContext, servConn, true); + processQuery(clientMessage, query, queryString, regionNames, start, null, queryContext, serverConnection, true); } catch (QueryInvalidException e) { throw new QueryInvalidException(e.getMessage() + queryString); } catch (QueryExecutionLowMemoryException e) { - writeQueryResponseException(msg, e, false, servConn); + writeQueryResponseException(clientMessage, e, serverConnection); } } http://git-wip-us.apache.org/repos/asf/geode/blob/db81427f/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Query651.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Query651.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Query651.java index 5849431..97f5d56 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Query651.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Query651.java @@ -44,40 +44,40 @@ public class Query651 extends BaseCommandQuery { protected Query651() {} @Override - public void cmdExecute(Message msg, ServerConnection servConn, long start) + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start) throws IOException, InterruptedException { // Based on MessageType.DESTROY // Added by gregp 10/18/05 - 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 queryString = msg.getPart(0).getString(); + String queryString = clientMessage.getPart(0).getString(); long compiledQueryId = 0; Object[] queryParams = null; try { - if (msg.getMessageType() == MessageType.QUERY_WITH_PARAMETERS) { + if (clientMessage.getMessageType() == MessageType.QUERY_WITH_PARAMETERS) { // Query with parameters supported from 6.6 onwards. - int params = msg.getPart(1).getInt(); // Number of parameters. + int params = clientMessage.getPart(1).getInt(); // Number of parameters. // In case of native client there will be extra two parameters at 2 and 3 index. int paramStartIndex = 2; - if (msg.getNumberOfParts() > (1 /* type */ + 1 /* query string */ + 1 /* params length */ - + params /* number of params */)) { - int timeout = msg.getPart(3).getInt(); - servConn.setRequestSpecificTimeout(timeout); + if (clientMessage.getNumberOfParts() > (1 /* type */ + 1 /* query string */ + 1 /* params length */ + + params /* number of params */)) { + int timeout = clientMessage.getPart(3).getInt(); + serverConnection.setRequestSpecificTimeout(timeout); paramStartIndex = 4; } // Get the query execution parameters. queryParams = new Object[params]; for (int i = 0; i < queryParams.length; i++) { - queryParams[i] = msg.getPart(i + paramStartIndex).getObject(); + queryParams[i] = clientMessage.getPart(i + paramStartIndex).getObject(); } } else { // this is optional part for message specific timeout, which right now send by native client // need to take care while adding new message - if (msg.getNumberOfParts() == 3) { - int timeout = msg.getPart(2).getInt(); - servConn.setRequestSpecificTimeout(timeout); + if (clientMessage.getNumberOfParts() == 3) { + int timeout = clientMessage.getPart(2).getInt(); + serverConnection.setRequestSpecificTimeout(timeout); } } } catch (ClassNotFoundException cne) { @@ -85,19 +85,19 @@ public class Query651 extends BaseCommandQuery { } if (logger.isDebugEnabled()) { - logger.debug("{}: Received query request from {} queryString: {}{}", servConn.getName(), - servConn.getSocketString(), queryString, + logger.debug("{}: Received query request from {} queryString: {}{}", serverConnection.getName(), + serverConnection.getSocketString(), queryString, (queryParams != null ? (" with num query parameters :" + queryParams.length) : "")); } try { // Create query QueryService queryService = - servConn.getCachedRegionHelper().getCache().getLocalQueryService(); + serverConnection.getCachedRegionHelper().getCache().getLocalQueryService(); org.apache.geode.cache.query.Query query = null; if (queryParams != null) { // Its a compiled query. - CacheClientNotifier ccn = servConn.getAcceptor().getCacheClientNotifier(); + CacheClientNotifier ccn = serverConnection.getAcceptor().getCacheClientNotifier(); query = ccn.getCompiledQuery(queryString); if (query == null) { // This is first time the query is seen by this server. @@ -114,7 +114,7 @@ public class Query651 extends BaseCommandQuery { // Authorization check QueryOperationContext queryContext = null; - AuthorizeRequest authzRequest = servConn.getAuthzRequest(); + AuthorizeRequest authzRequest = serverConnection.getAuthzRequest(); if (authzRequest != null) { queryContext = authzRequest.queryAuthorize(queryString, regionNames, queryParams); String newQueryString = queryContext.getQuery(); @@ -128,8 +128,7 @@ public class Query651 extends BaseCommandQuery { } } - processQueryUsingParams(msg, query, queryString, regionNames, start, null, queryContext, - servConn, true, queryParams); + processQueryUsingParams(clientMessage, query, queryString, regionNames, start, null, queryContext, serverConnection, true, queryParams); } catch (QueryInvalidException e) { throw new QueryInvalidException(e.getMessage() + queryString); } http://git-wip-us.apache.org/repos/asf/geode/blob/db81427f/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterDataSerializers.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterDataSerializers.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterDataSerializers.java index 7d28d52..d1c101f 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterDataSerializers.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterDataSerializers.java @@ -37,22 +37,22 @@ public class RegisterDataSerializers extends BaseCommand { private RegisterDataSerializers() {} - public void cmdExecute(Message msg, ServerConnection servConn, long start) + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start) throws IOException, ClassNotFoundException { if (logger.isDebugEnabled()) { logger.debug("{}: Received register dataserializer request ({} parts) from {}", - servConn.getName(), msg.getNumberOfParts(), servConn.getSocketString()); + serverConnection.getName(), clientMessage.getNumberOfParts(), serverConnection.getSocketString()); } - int noOfParts = msg.getNumberOfParts(); + int noOfParts = clientMessage.getNumberOfParts(); // 2 parts per instantiator and one eventId part int noOfDataSerializers = (noOfParts - 1) / 2; // retrieve eventID from the last Part - ByteBuffer eventIdPartsBuffer = ByteBuffer.wrap(msg.getPart(noOfParts - 1).getSerializedForm()); + ByteBuffer eventIdPartsBuffer = ByteBuffer.wrap(clientMessage.getPart(noOfParts - 1).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); byte[][] serializedDataSerializers = new byte[noOfDataSerializers * 2][]; boolean caughtCNFE = false; @@ -60,12 +60,12 @@ public class RegisterDataSerializers extends BaseCommand { try { for (int i = 0; i < noOfParts - 1; i = i + 2) { - Part dataSerializerClassNamePart = msg.getPart(i); + Part dataSerializerClassNamePart = clientMessage.getPart(i); serializedDataSerializers[i] = dataSerializerClassNamePart.getSerializedForm(); String dataSerializerClassName = (String) CacheServerHelper.deserialize(serializedDataSerializers[i]); - Part idPart = msg.getPart(i + 1); + Part idPart = clientMessage.getPart(i + 1); serializedDataSerializers[i + 1] = idPart.getSerializedForm(); int id = idPart.getInt(); @@ -73,7 +73,7 @@ public class RegisterDataSerializers extends BaseCommand { try { dataSerializerClass = InternalDataSerializer.getCachedClass(dataSerializerClassName); InternalDataSerializer.register(dataSerializerClass, true, eventId, - servConn.getProxyID()); + serverConnection.getProxyID()); } catch (ClassNotFoundException e) { // If a ClassNotFoundException is caught, store it, but continue // processing other instantiators @@ -82,26 +82,26 @@ public class RegisterDataSerializers extends BaseCommand { } } } catch (Exception e) { - writeException(msg, e, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); } // If a ClassNotFoundException was caught while processing the // instantiators, send it back to the client. Note: This only sends // the last CNFE. if (caughtCNFE) { - writeException(msg, cnfe, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, cnfe, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); } // Send reply to client if necessary. If an exception occurs in the above // code, then the reply has already been sent. - if (!servConn.getTransientFlag(RESPONDED)) { - writeReply(msg, servConn); + if (!serverConnection.getTransientFlag(RESPONDED)) { + writeReply(clientMessage, serverConnection); } if (logger.isDebugEnabled()) { - logger.debug("Registered dataserializer for MembershipId = {}", servConn.getMembershipID()); + logger.debug("Registered dataserializer for MembershipId = {}", serverConnection.getMembershipID()); } } } http://git-wip-us.apache.org/repos/asf/geode/blob/db81427f/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInstantiators.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInstantiators.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInstantiators.java index 1e701fc..2b63337 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInstantiators.java +++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInstantiators.java @@ -49,23 +49,23 @@ public class RegisterInstantiators extends BaseCommand { private RegisterInstantiators() {} @Override - public void cmdExecute(Message msg, ServerConnection servConn, long start) + public void cmdExecute(Message clientMessage, ServerConnection serverConnection, long start) throws IOException, ClassNotFoundException { if (logger.isDebugEnabled()) { logger.debug("{}: Received register instantiator request ({} parts) from {}", - servConn.getName(), msg.getNumberOfParts(), servConn.getSocketString()); + serverConnection.getName(), clientMessage.getNumberOfParts(), serverConnection.getSocketString()); } - int noOfParts = msg.getNumberOfParts(); + int noOfParts = clientMessage.getNumberOfParts(); // Assert parts Assert.assertTrue((noOfParts - 1) % 3 == 0); // 3 parts per instantiator and one eventId part int noOfInstantiators = (noOfParts - 1) / 3; // retrieve eventID from the last Part - ByteBuffer eventIdPartsBuffer = ByteBuffer.wrap(msg.getPart(noOfParts - 1).getSerializedForm()); + ByteBuffer eventIdPartsBuffer = ByteBuffer.wrap(clientMessage.getPart(noOfParts - 1).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); byte[][] serializedInstantiators = new byte[noOfInstantiators * 3][]; boolean caughtCNFE = false; @@ -73,17 +73,17 @@ public class RegisterInstantiators extends BaseCommand { try { for (int i = 0; i < noOfParts - 1; i = i + 3) { - Part instantiatorPart = msg.getPart(i); + Part instantiatorPart = clientMessage.getPart(i); serializedInstantiators[i] = instantiatorPart.getSerializedForm(); String instantiatorClassName = (String) CacheServerHelper.deserialize(serializedInstantiators[i]); - Part instantiatedPart = msg.getPart(i + 1); + Part instantiatedPart = clientMessage.getPart(i + 1); serializedInstantiators[i + 1] = instantiatedPart.getSerializedForm(); String instantiatedClassName = (String) CacheServerHelper.deserialize(serializedInstantiators[i + 1]); - Part idPart = msg.getPart(i + 2); + Part idPart = clientMessage.getPart(i + 2); serializedInstantiators[i + 2] = idPart.getSerializedForm(); int id = idPart.getInt(); @@ -92,7 +92,7 @@ public class RegisterInstantiators extends BaseCommand { instantiatorClass = InternalDataSerializer.getCachedClass(instantiatorClassName); instantiatedClass = InternalDataSerializer.getCachedClass(instantiatedClassName); InternalInstantiator.register(instantiatorClass, instantiatedClass, id, true, eventId, - servConn.getProxyID()); + serverConnection.getProxyID()); } catch (ClassNotFoundException e) { // If a ClassNotFoundException is caught, store it, but continue // processing other instantiators @@ -102,17 +102,17 @@ public class RegisterInstantiators extends BaseCommand { } } catch (Exception e) { logger.warn(LocalizedMessage.create(LocalizedStrings.RegisterInstantiators_BAD_CLIENT, - new Object[] {servConn.getMembershipID(), e.getLocalizedMessage()})); - writeException(msg, e, false, servConn); - servConn.setAsTrue(RESPONDED); + new Object[] { serverConnection.getMembershipID(), e.getLocalizedMessage()})); + writeException(clientMessage, e, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); } // If a ClassNotFoundException was caught while processing the // instantiators, send it back to the client. Note: This only sends // the last CNFE. if (caughtCNFE) { - writeException(msg, cnfe, false, servConn); - servConn.setAsTrue(RESPONDED); + writeException(clientMessage, cnfe, false, serverConnection); + serverConnection.setAsTrue(RESPONDED); // Send the instantiators on to other clients if we hit an error // due to a missing class, because they were not distributed @@ -120,7 +120,7 @@ public class RegisterInstantiators extends BaseCommand { // been distributed if successfully registered. ClientInstantiatorMessage clientInstantiatorMessage = new ClientInstantiatorMessage(EnumListenerEvent.AFTER_REGISTER_INSTANTIATOR, - serializedInstantiators, servConn.getProxyID(), eventId); + serializedInstantiators, serverConnection.getProxyID(), eventId); // Notify other clients CacheClientNotifier.routeClientMessage(clientInstantiatorMessage); @@ -129,12 +129,12 @@ public class RegisterInstantiators extends BaseCommand { // Send reply to client if necessary. If an exception occurs in the above // code, then the reply has already been sent. - if (!servConn.getTransientFlag(RESPONDED)) { - writeReply(msg, servConn); + if (!serverConnection.getTransientFlag(RESPONDED)) { + writeReply(clientMessage, serverConnection); } if (logger.isDebugEnabled()) { - logger.debug("Registered instantiators for MembershipId = {}", servConn.getMembershipID()); + logger.debug("Registered instantiators for MembershipId = {}", serverConnection.getMembershipID()); } }