You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by jb...@apache.org on 2022/06/15 01:36:19 UTC
[geode] branch develop updated: GEODE-10326: Convert MessageType to enum. (#7688)
This is an automated email from the ASF dual-hosted git repository.
jbarrett pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode.git
The following commit(s) were added to refs/heads/develop by this push:
new 03ab34d62f GEODE-10326: Convert MessageType to enum. (#7688)
03ab34d62f is described below
commit 03ab34d62f8742b02fabea4c38bb19141736a588
Author: Jacob Barrett <jb...@pivotal.io>
AuthorDate: Tue Jun 14 18:36:13 2022 -0700
GEODE-10326: Convert MessageType to enum. (#7688)
* Migrate unit tests to JUnit 5.
* Add missing unit tests.
---
.../sockets/CqNameToOpHashMapIntegrationTest.java | 18 +-
.../apache/geode/codeAnalysis/excludedClasses.txt | 1 +
.../codeAnalysis/sanctionedDataSerializables.txt | 2 +-
.../geode/cache/client/internal/AbstractOp.java | 19 +-
.../client/internal/AbstractOpWithTimeout.java | 4 +-
.../geode/cache/client/internal/AddPDXEnumOp.java | 2 +-
.../geode/cache/client/internal/AddPDXTypeOp.java | 2 +-
.../cache/client/internal/AuthenticateUserOp.java | 6 +-
.../geode/cache/client/internal/ClearOp.java | 2 +-
.../cache/client/internal/CloseConnectionOp.java | 2 +-
.../geode/cache/client/internal/CommitOp.java | 2 +-
.../geode/cache/client/internal/ContainsKeyOp.java | 2 +-
.../geode/cache/client/internal/DestroyOp.java | 2 +-
.../cache/client/internal/DestroyRegionOp.java | 2 +-
.../client/internal/ExecuteFunctionNoAckOp.java | 6 +-
.../cache/client/internal/ExecuteFunctionOp.java | 18 +-
.../internal/ExecuteRegionFunctionNoAckOp.java | 6 +-
.../client/internal/ExecuteRegionFunctionOp.java | 20 +-
.../internal/ExecuteRegionFunctionSingleHopOp.java | 16 +-
.../geode/cache/client/internal/GetAllOp.java | 2 +-
.../client/internal/GetClientPRMetaDataOp.java | 14 +-
.../internal/GetClientPartitionAttributesOp.java | 14 +-
.../geode/cache/client/internal/GetEntryOp.java | 2 +-
.../cache/client/internal/GetEventValueOp.java | 6 +-
.../client/internal/GetFunctionAttributeOp.java | 2 +-
.../apache/geode/cache/client/internal/GetOp.java | 2 +-
.../cache/client/internal/GetPDXEnumByIdOp.java | 2 +-
.../geode/cache/client/internal/GetPDXEnumsOp.java | 6 +-
.../cache/client/internal/GetPDXIdForEnumOp.java | 6 +-
.../cache/client/internal/GetPDXIdForTypeOp.java | 6 +-
.../cache/client/internal/GetPDXTypeByIdOp.java | 2 +-
.../geode/cache/client/internal/GetPDXTypesOp.java | 6 +-
.../geode/cache/client/internal/InvalidateOp.java | 2 +-
.../geode/cache/client/internal/KeySetOp.java | 6 +-
.../geode/cache/client/internal/MakePrimaryOp.java | 2 +-
.../apache/geode/cache/client/internal/PingOp.java | 2 +-
.../geode/cache/client/internal/PrimaryAckOp.java | 2 +-
.../cache/client/internal/ProxyCacheCloseOp.java | 6 +-
.../geode/cache/client/internal/PutAllOp.java | 2 +-
.../apache/geode/cache/client/internal/PutOp.java | 6 +-
.../geode/cache/client/internal/QueryOp.java | 6 +-
.../cache/client/internal/ReadyForEventsOp.java | 2 +-
.../client/internal/RegisterDataSerializersOp.java | 2 +-
.../client/internal/RegisterInstantiatorsOp.java | 2 +-
.../cache/client/internal/RegisterInterestOp.java | 18 +-
.../geode/cache/client/internal/RemoveAllOp.java | 2 +-
.../geode/cache/client/internal/RollbackOp.java | 2 +-
.../apache/geode/cache/client/internal/SizeOp.java | 2 +-
.../geode/cache/client/internal/TXFailoverOp.java | 2 +-
.../cache/client/internal/TXSynchronizationOp.java | 23 +-
.../client/internal/UnregisterInterestListOp.java | 2 +-
.../client/internal/UnregisterInterestOp.java | 2 +-
.../cache/query/internal/CqQueryVsdStats.java | 9 +-
.../geode/cache/query/internal/cq/CqService.java | 4 +-
.../cache/query/internal/cq/MissingCqService.java | 7 +-
.../internal/cache/DistributedCacheOperation.java | 2 +-
.../internal/cache/DistributedPutAllOperation.java | 2 +-
.../cache/DistributedRemoveAllOperation.java | 2 +-
.../geode/internal/cache/FilterRoutingInfo.java | 23 +-
.../apache/geode/internal/cache/TXManagerImpl.java | 3 +-
.../ServerToClientFunctionResultSender.java | 9 +-
.../ServerToClientFunctionResultSender65.java | 3 +-
.../geode/internal/cache/tier/MessageType.java | 498 +++++++--------------
.../internal/cache/tier/sockets/BaseCommand.java | 15 +-
.../cache/tier/sockets/BaseCommandQuery.java | 21 +-
.../cache/tier/sockets/CacheClientNotifier.java | 4 +-
.../cache/tier/sockets/CacheClientUpdater.java | 70 +--
.../cache/tier/sockets/ChunkedMessage.java | 4 +-
.../tier/sockets/ClientRegistrationMetadata.java | 3 +-
.../tier/sockets/ClientUpdateMessageImpl.java | 97 ++--
.../cache/tier/sockets/CommandInitializer.java | 62 +--
.../cache/tier/sockets/CommandRegistry.java | 5 +-
.../cache/tier/sockets/HAEventWrapper.java | 12 +-
.../geode/internal/cache/tier/sockets/Message.java | 31 +-
.../cache/tier/sockets/ServerConnection.java | 20 +-
.../cache/tier/sockets/ServerResponseMatrix.java | 12 +-
.../tier/sockets/ServerSideHandshakeFactory.java | 3 +-
.../cache/tier/sockets/command/Default.java | 2 +-
.../cache/tier/sockets/command/DestroyRegion.java | 4 +-
.../tier/sockets/command/ExecuteFunction70.java | 4 +-
.../sockets/command/ExecuteRegionFunction66.java | 2 +-
.../command/ExecuteRegionFunctionSingleHop.java | 2 +-
.../sockets/command/GatewayReceiverCommand.java | 2 +-
.../client/internal/AbstractOpWithTimeoutTest.java | 11 +-
.../cache/DistributedCacheOperationTest.java | 2 +-
.../geode/internal/cache/FilterInfoTest.java | 7 +-
...verToClientFunctionResultSender65JUnitTest.java | 4 +-
...erverToClientFunctionResultSenderJUnitTest.java | 4 +-
.../geode/internal/cache/tier/MessageTypeTest.java | 71 +++
.../tier/sockets/CacheClientNotifierTest.java | 5 +-
.../tier/sockets/ClientUpdateMessageImplTest.java | 233 ++++++++--
.../cache/tier/sockets/CommandInitializerTest.java | 18 +-
.../cache/tier/sockets/ServerConnectionTest.java | 2 +
.../sockets/command/ExecuteFunction70Test.java | 9 +-
.../ClientAuthenticationPart2DUnitTest.java | 7 +-
.../cq/internal/CqServiceFactoryImplTest.java | 14 +-
.../query/cq/internal/CqServiceFactoryImpl.java | 14 +-
.../cache/query/cq/internal/CqServiceImpl.java | 88 ++--
.../cache/query/cq/internal/command/CloseCQ.java | 2 +-
.../query/cq/internal/command/ExecuteCQ61.java | 6 +-
.../query/cq/internal/command/GetCQStats.java | 2 +-
.../query/cq/internal/command/GetDurableCQs.java | 2 +-
.../cache/query/cq/internal/command/MonitorCQ.java | 4 +-
.../cache/query/cq/internal/command/StopCQ.java | 2 +-
.../cache/query/cq/internal/ops/CloseCQOp.java | 2 +-
.../cache/query/cq/internal/ops/CreateCQOp.java | 12 +-
.../query/cq/internal/ops/CreateCQWithIROp.java | 2 +-
.../query/cq/internal/ops/GetDurableCQsOp.java | 8 +-
.../cache/query/cq/internal/ops/StopCQOp.java | 2 +-
.../cq/internal/CqServiceFactoryImplTest.java | 14 +-
.../client/locator/GatewaySenderBatchOp.java | 12 +-
111 files changed, 963 insertions(+), 828 deletions(-)
diff --git a/geode-core/src/integrationTest/java/org/apache/geode/internal/cache/tier/sockets/CqNameToOpHashMapIntegrationTest.java b/geode-core/src/integrationTest/java/org/apache/geode/internal/cache/tier/sockets/CqNameToOpHashMapIntegrationTest.java
index b3d8eea1bb..e4ec0b7fc8 100644
--- a/geode-core/src/integrationTest/java/org/apache/geode/internal/cache/tier/sockets/CqNameToOpHashMapIntegrationTest.java
+++ b/geode-core/src/integrationTest/java/org/apache/geode/internal/cache/tier/sockets/CqNameToOpHashMapIntegrationTest.java
@@ -33,6 +33,7 @@ import org.junit.Test;
import org.apache.geode.DataSerializer;
import org.apache.geode.internal.InternalDataSerializer;
+import org.apache.geode.internal.cache.tier.MessageType;
public class CqNameToOpHashMapIntegrationTest {
/**
@@ -57,13 +58,13 @@ public class CqNameToOpHashMapIntegrationTest {
new ClientUpdateMessageImpl.CqNameToOpHashMap(numEntries);
for (int i = 0; i < numEntries; ++i) {
- originalCqNameToOpHashMap.add(String.valueOf(i), i);
- modifiedCqNameToOpHashMap.add(String.valueOf(i), i);
+ originalCqNameToOpHashMap.add(String.valueOf(i), MessageType.REQUEST);
+ modifiedCqNameToOpHashMap.add(String.valueOf(i), MessageType.REQUEST);
}
CompletableFuture<Void> removeFromHashMapTask = CompletableFuture.runAsync(() -> {
for (int i = 0; i < numEntries; ++i) {
- modifiedCqNameToOpHashMap.remove(String.valueOf(i), i);
+ modifiedCqNameToOpHashMap.remove(String.valueOf(i), MessageType.REQUEST);
}
});
@@ -87,18 +88,15 @@ public class CqNameToOpHashMapIntegrationTest {
for (int j = 0; j < cqNamesSize; j++) {
String cqNamesKey = DataSerializer.readObject(inputStream);
- Integer cqNamesValue = DataSerializer.<Integer>readObject(inputStream);
- reconstructedCqNameToOpHashMap.add(cqNamesKey, cqNamesValue);
+ Integer cqNamesValue = DataSerializer.readObject(inputStream);
+ reconstructedCqNameToOpHashMap.add(cqNamesKey, MessageType.valueOf(cqNamesValue));
}
// The reconstructed map should have some subset of the entries in the cqNameToOpHashMap,
// but the specific contents will depend on timing with the removeFromHashMap task.
- MapDifference<String, Integer> reconstructedVersusOriginalHashMapDifference =
+ MapDifference<String, MessageType> reconstructedVersusOriginalHashMapDifference =
Maps.difference(reconstructedCqNameToOpHashMap, originalCqNameToOpHashMap);
- assertThat(reconstructedVersusOriginalHashMapDifference.entriesInCommon().size() >= 0)
- .isTrue();
- assertThat(reconstructedVersusOriginalHashMapDifference.entriesDiffering().size() == 0)
- .isTrue();
+ assertThat(reconstructedVersusOriginalHashMapDifference.entriesDiffering()).hasSize(0);
} catch (Exception ex) {
throw new RuntimeException("Failed to serialize/deserialize the CqNameToOpHashMap", ex);
}
diff --git a/geode-core/src/integrationTest/resources/org/apache/geode/codeAnalysis/excludedClasses.txt b/geode-core/src/integrationTest/resources/org/apache/geode/codeAnalysis/excludedClasses.txt
index 3939917e49..835662d7a8 100644
--- a/geode-core/src/integrationTest/resources/org/apache/geode/codeAnalysis/excludedClasses.txt
+++ b/geode-core/src/integrationTest/resources/org/apache/geode/codeAnalysis/excludedClasses.txt
@@ -105,3 +105,4 @@ org/apache/geode/cache/query/internal/xml/ElementType$3
org/apache/geode/internal/net/ByteBufferVendor$OpenAttemptTimedOut
org/apache/geode/internal/cache/wan/GatewaySenderEventImpl$TransactionMetadataDisposition
org/apache/geode/internal/cache/tier/InterestType
+org/apache/geode/internal/cache/tier/MessageType
diff --git a/geode-core/src/integrationTest/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt b/geode-core/src/integrationTest/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt
index 90c0cd442d..84309aac4f 100644
--- a/geode-core/src/integrationTest/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt
+++ b/geode-core/src/integrationTest/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt
@@ -1742,7 +1742,7 @@ fromData,175
toData,198
org/apache/geode/internal/cache/tier/sockets/HAEventWrapper,2
-fromData,460
+fromData,467
toData,106
org/apache/geode/internal/cache/tier/sockets/InterestResultPolicyImpl,2
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/AbstractOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/AbstractOp.java
index e3d4cb6e8a..a7a46a90d8 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/AbstractOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/AbstractOp.java
@@ -53,7 +53,7 @@ public abstract class AbstractOp implements Op {
private boolean allowDuplicateMetadataRefresh;
- protected AbstractOp(int msgType, int msgParts) {
+ protected AbstractOp(MessageType msgType, int msgParts) {
msg = new Message(msgParts, KnownVersion.CURRENT);
getMessage().setMessageType(msgType);
}
@@ -134,8 +134,7 @@ public abstract class AbstractOp implements Op {
}
userId = id;
}
- secureLogger.debug("{} Using userId {}: ",
- MessageType.getString(getMessage().getMessageType()), userId);
+ secureLogger.debug("{} Using userId {}: ", getMessage().getMessageType(), userId);
try (HeapDataOutputStream hdos = new HeapDataOutputStream(KnownVersion.CURRENT)) {
hdos.writeLong(cnx.getConnectionID());
hdos.writeLong(userId);
@@ -246,7 +245,7 @@ public abstract class AbstractOp implements Op {
/**
* Return true of <code>messageType</code> indicates the operation had an error on the server.
*/
- protected abstract boolean isErrorResponse(int msgType);
+ protected abstract boolean isErrorResponse(MessageType msgType);
/**
* Process a response that contains an ack.
@@ -257,7 +256,7 @@ public abstract class AbstractOp implements Op {
* exception.
*/
protected void processAck(Message msg, String opName) throws Exception {
- final int msgType = msg.getMessageType();
+ final MessageType msgType = msg.getMessageType();
if (msgType != MessageType.REPLY) {
Part part = msg.getPart(0);
if (msgType == MessageType.EXCEPTION) {
@@ -273,7 +272,7 @@ public abstract class AbstractOp implements Op {
} else if (isErrorResponse(msgType)) {
throw new ServerOperationException(part.getString());
} else {
- throw new InternalGemFireError("Unexpected message type " + MessageType.getString(msgType));
+ throw new InternalGemFireError("Unexpected message type " + msgType);
}
}
}
@@ -289,7 +288,7 @@ public abstract class AbstractOp implements Op {
*/
protected Object processObjResponse(Message msg, String opName) throws Exception {
Part part = msg.getPart(0);
- final int msgType = msg.getMessageType();
+ final MessageType msgType = msg.getMessageType();
if (msgType == MessageType.RESPONSE) {
return part.getObject();
} else {
@@ -301,7 +300,7 @@ public abstract class AbstractOp implements Op {
} else if (isErrorResponse(msgType)) {
throw new ServerOperationException(part.getString());
} else {
- throw new InternalGemFireError("Unexpected message type " + MessageType.getString(msgType));
+ throw new InternalGemFireError("Unexpected message type " + msgType);
}
}
}
@@ -338,7 +337,7 @@ public abstract class AbstractOp implements Op {
void processChunkedResponse(ChunkedMessage msg, String opName, ChunkHandler callback)
throws Exception {
msg.readHeader();
- final int msgType = msg.getMessageType();
+ final MessageType msgType = msg.getMessageType();
if (msgType == MessageType.RESPONSE) {
do {
msg.receiveChunk();
@@ -357,7 +356,7 @@ public abstract class AbstractOp implements Op {
Part part = msg.getPart(0);
throw new ServerOperationException(part.getString());
} else {
- throw new IOException("Unexpected message type " + MessageType.getString(msgType));
+ throw new IOException("Unexpected message type " + msgType);
}
}
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/AbstractOpWithTimeout.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/AbstractOpWithTimeout.java
index dd6c9b500c..41eb8d2364 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/AbstractOpWithTimeout.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/AbstractOpWithTimeout.java
@@ -17,6 +17,8 @@ package org.apache.geode.cache.client.internal;
import java.net.Socket;
+import org.apache.geode.internal.cache.tier.MessageType;
+
/**
* Derivatives of this class can have an alternative socket timeout from the default connection
* timeout.
@@ -25,7 +27,7 @@ public abstract class AbstractOpWithTimeout extends AbstractOp {
private final int timeoutMs;
- AbstractOpWithTimeout(int msgType, int msgParts, int timeoutMs) {
+ AbstractOpWithTimeout(MessageType msgType, int msgParts, int timeoutMs) {
super(msgType, msgParts);
this.timeoutMs = timeoutMs;
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/AddPDXEnumOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/AddPDXEnumOp.java
index 5c5a94b59c..23dae31a53 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/AddPDXEnumOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/AddPDXEnumOp.java
@@ -58,7 +58,7 @@ public class AddPDXEnumOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return false;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/AddPDXTypeOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/AddPDXTypeOp.java
index b4366c6569..8a30afc092 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/AddPDXTypeOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/AddPDXTypeOp.java
@@ -58,7 +58,7 @@ public class AddPDXTypeOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return false;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/AuthenticateUserOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/AuthenticateUserOp.java
index 2cd9c22600..aa2760d8fb 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/AuthenticateUserOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/AuthenticateUserOp.java
@@ -211,7 +211,7 @@ public class AuthenticateUserOp {
final @NotNull Connection connection) throws Exception {
byte[] bytes;
Part part = msg.getPart(0);
- final int msgType = msg.getMessageType();
+ final MessageType msgType = msg.getMessageType();
long userId = -1;
if (msgType == MessageType.RESPONSE) {
bytes = (byte[]) part.getObject();
@@ -253,12 +253,12 @@ public class AuthenticateUserOp {
} else if (isErrorResponse(msgType)) {
throw new ServerOperationException(part.getString());
} else {
- throw new InternalGemFireError("Unexpected message type " + MessageType.getString(msgType));
+ throw new InternalGemFireError("Unexpected message type " + msgType);
}
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.REQUESTDATAERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ClearOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ClearOp.java
index d96698dcae..8c9cbaf5c0 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ClearOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ClearOp.java
@@ -81,7 +81,7 @@ public class ClearOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.CLEAR_REGION_DATA_ERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/CloseConnectionOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/CloseConnectionOp.java
index 3b11298657..9c4165058b 100755
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/CloseConnectionOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/CloseConnectionOp.java
@@ -75,7 +75,7 @@ public class CloseConnectionOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return false;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/CommitOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/CommitOp.java
index ae54092558..16d0cc192d 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/CommitOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/CommitOp.java
@@ -85,7 +85,7 @@ public class CommitOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.EXCEPTION;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ContainsKeyOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ContainsKeyOp.java
index fdb59a6825..3be0c03da4 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ContainsKeyOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ContainsKeyOp.java
@@ -69,7 +69,7 @@ public class ContainsKeyOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.CONTAINS_KEY_DATA_ERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/DestroyOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/DestroyOp.java
index 970d26f67b..98d3b5dcda 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/DestroyOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/DestroyOp.java
@@ -235,7 +235,7 @@ public class DestroyOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.DESTROY_DATA_ERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/DestroyRegionOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/DestroyRegionOp.java
index d20d25aaa7..cade1f9574 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/DestroyRegionOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/DestroyRegionOp.java
@@ -81,7 +81,7 @@ public class DestroyRegionOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.DESTROY_REGION_DATA_ERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteFunctionNoAckOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteFunctionNoAckOp.java
index c644f41e9f..a404694876 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteFunctionNoAckOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteFunctionNoAckOp.java
@@ -179,7 +179,7 @@ public class ExecuteFunctionNoAckOp {
@Override
protected Object processResponse(final @NotNull Message msg) throws Exception {
- final int msgType = msg.getMessageType();
+ final MessageType msgType = msg.getMessageType();
if (msgType == MessageType.REPLY) {
return null;
} else {
@@ -191,14 +191,14 @@ public class ExecuteFunctionNoAckOp {
logger.warn("Function execution without result encountered an Exception on server.");
} else {
throw new InternalGemFireError(
- "Unexpected message type " + MessageType.getString(msgType));
+ "Unexpected message type " + msgType);
}
return null;
}
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.EXECUTE_FUNCTION_ERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteFunctionOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteFunctionOp.java
index 13ae1232f3..2f9652696b 100755
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteFunctionOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteFunctionOp.java
@@ -16,6 +16,8 @@
package org.apache.geode.cache.client.internal;
import static org.apache.geode.internal.cache.execute.AbstractExecution.DEFAULT_CLIENT_FUNCTION_TIMEOUT;
+import static org.apache.geode.internal.cache.tier.MessageType.EXECUTE_FUNCTION;
+import static org.apache.geode.internal.cache.tier.MessageType.EXECUTE_FUNCTION_ERROR;
import java.util.ArrayList;
import java.util.List;
@@ -195,7 +197,7 @@ public class ExecuteFunctionOp {
byte isReexecute,
String[] groups, boolean allMembers, boolean ignoreFailedMembers,
final int timeoutMs) {
- super(MessageType.EXECUTE_FUNCTION, MSG_PARTS, timeoutMs);
+ super(EXECUTE_FUNCTION, MSG_PARTS, timeoutMs);
byte fnState = AbstractExecution.getFunctionState(function.isHA(), function.hasResult(),
function.optimizeForWrite());
@@ -227,7 +229,7 @@ public class ExecuteFunctionOp {
MemberMappedArgument memberMappedArg, byte hasResult, ResultCollector rc,
boolean isFnSerializationReqd, boolean isHA, boolean optimizeForWrite, byte isReexecute,
String[] groups, boolean allMembers, boolean ignoreFailedMembers, final int timeoutMs) {
- super(MessageType.EXECUTE_FUNCTION, MSG_PARTS, timeoutMs);
+ super(EXECUTE_FUNCTION, MSG_PARTS, timeoutMs);
byte fnState = AbstractExecution.getFunctionState(isHA, hasResult == (byte) 1,
optimizeForWrite);
@@ -251,7 +253,7 @@ public class ExecuteFunctionOp {
}
ExecuteFunctionOpImpl(ExecuteFunctionOpImpl op, byte isReexecute) {
- super(MessageType.EXECUTE_FUNCTION, MSG_PARTS, op.getTimeoutMs());
+ super(EXECUTE_FUNCTION, MSG_PARTS, op.getTimeoutMs());
resultCollector = op.resultCollector;
function = op.function;
functionId = op.functionId;
@@ -318,7 +320,7 @@ public class ExecuteFunctionOp {
// Read the header which describes the type of message following
executeFunctionResponseMsg.readHeader();
switch (executeFunctionResponseMsg.getMessageType()) {
- case MessageType.EXECUTE_FUNCTION_RESULT:
+ case EXECUTE_FUNCTION_RESULT:
if (logger.isDebugEnabled()) {
logger.debug(
"ExecuteFunctionOpImpl#processResponse: received message of type EXECUTE_FUNCTION_RESULT.");
@@ -367,7 +369,7 @@ public class ExecuteFunctionOp {
"ExecuteFunctionOpImpl#processResponse: received all the results from server successfully.");
}
return null;
- case MessageType.EXCEPTION:
+ case EXCEPTION:
if (logger.isDebugEnabled()) {
logger.debug(
"ExecuteFunctionOpImpl#processResponse: received message of type EXCEPTION");
@@ -383,7 +385,7 @@ public class ExecuteFunctionOp {
throw new ServerOperationException(
": While performing a remote execute Function" + t.getMessage(), t);
}
- case MessageType.EXECUTE_FUNCTION_ERROR:
+ case EXECUTE_FUNCTION_ERROR:
if (logger.isDebugEnabled()) {
logger.debug(
"ExecuteFunctionOpImpl#processResponse: received message of type EXECUTE_FUNCTION_ERROR");
@@ -402,8 +404,8 @@ public class ExecuteFunctionOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
- return msgType == MessageType.EXECUTE_FUNCTION_ERROR;
+ protected boolean isErrorResponse(MessageType msgType) {
+ return msgType == EXECUTE_FUNCTION_ERROR;
}
@Override
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteRegionFunctionNoAckOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteRegionFunctionNoAckOp.java
index 9e7439b206..af92e34fda 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteRegionFunctionNoAckOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteRegionFunctionNoAckOp.java
@@ -172,7 +172,7 @@ public class ExecuteRegionFunctionNoAckOp {
@Override
protected Object processResponse(final @NotNull Message msg) throws Exception {
- final int msgType = msg.getMessageType();
+ final MessageType msgType = msg.getMessageType();
if (msgType == MessageType.REPLY) {
return null;
} else {
@@ -184,7 +184,7 @@ public class ExecuteRegionFunctionNoAckOp {
logger.warn("Function execution without result encountered an Exception on server.");
} else {
throw new InternalGemFireError(
- "Unexpected message type " + MessageType.getString(msgType));
+ "Unexpected message type " + msgType);
}
return null;
}
@@ -192,7 +192,7 @@ public class ExecuteRegionFunctionNoAckOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.EXECUTE_REGION_FUNCTION_ERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteRegionFunctionOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteRegionFunctionOp.java
index b3dc3aba6f..a330a01eba 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteRegionFunctionOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteRegionFunctionOp.java
@@ -16,6 +16,8 @@
package org.apache.geode.cache.client.internal;
import static org.apache.geode.internal.cache.execute.AbstractExecution.DEFAULT_CLIENT_FUNCTION_TIMEOUT;
+import static org.apache.geode.internal.cache.tier.MessageType.EXECUTE_REGION_FUNCTION;
+import static org.apache.geode.internal.cache.tier.MessageType.EXECUTE_REGION_FUNCTION_ERROR;
import java.util.ArrayList;
import java.util.Collections;
@@ -193,7 +195,7 @@ public class ExecuteRegionFunctionOp {
ExecuteRegionFunctionOpImpl(String region, Function function,
ServerRegionFunctionExecutor serverRegionExecutor, ResultCollector rc,
final int timeoutMs) {
- super(MessageType.EXECUTE_REGION_FUNCTION,
+ super(EXECUTE_REGION_FUNCTION,
getMessagePartCount(serverRegionExecutor.getFilter().size(), 0), timeoutMs);
executeOnBucketSet = serverRegionExecutor.getExecuteOnBucketSetFlag();
byte flags = ExecuteFunctionHelper.createFlags(executeOnBucketSet, isReExecute);
@@ -214,7 +216,7 @@ public class ExecuteRegionFunctionOp {
// For testing only
ExecuteRegionFunctionOpImpl() {
- super(MessageType.EXECUTE_REGION_FUNCTION, 0, DEFAULT_CLIENT_FUNCTION_TIMEOUT);
+ super(EXECUTE_REGION_FUNCTION, 0, DEFAULT_CLIENT_FUNCTION_TIMEOUT);
resultCollector = null;
function = null;
isReExecute = (byte) 0;
@@ -231,7 +233,7 @@ public class ExecuteRegionFunctionOp {
ServerRegionFunctionExecutor serverRegionExecutor, ResultCollector rc, byte hasResult,
boolean isHA, boolean optimizeForWrite,
boolean calculateFnState, final int timeoutMs) {
- super(MessageType.EXECUTE_REGION_FUNCTION,
+ super(EXECUTE_REGION_FUNCTION,
getMessagePartCount(serverRegionExecutor.getFilter().size(), 0), timeoutMs);
byte functionState = hasResult;
@@ -263,7 +265,7 @@ public class ExecuteRegionFunctionOp {
ExecuteRegionFunctionOpImpl(ExecuteRegionFunctionOpImpl op, byte isReExecute,
Set<String> removedNodes) {
- super(MessageType.EXECUTE_REGION_FUNCTION,
+ super(EXECUTE_REGION_FUNCTION,
getMessagePartCount(op.executor.getFilter().size(), removedNodes.size()),
op.getTimeoutMs());
this.isReExecute = isReExecute;
@@ -305,7 +307,7 @@ public class ExecuteRegionFunctionOp {
try {
executeFunctionResponseMsg.readHeader();
switch (executeFunctionResponseMsg.getMessageType()) {
- case MessageType.EXECUTE_REGION_FUNCTION_RESULT:
+ case EXECUTE_REGION_FUNCTION_RESULT:
final boolean isDebugEnabled = logger.isDebugEnabled();
if (isDebugEnabled) {
logger.debug(
@@ -412,7 +414,7 @@ public class ExecuteRegionFunctionOp {
resultCollector.endResults();
return null;
- case MessageType.EXCEPTION:
+ case EXCEPTION:
if (logger.isDebugEnabled()) {
logger.debug(
"ExecuteRegionFunctionOpImpl#processResponse: received message of type EXCEPTION. The number of parts are : {}",
@@ -438,7 +440,7 @@ public class ExecuteRegionFunctionOp {
throw new ServerOperationException(s, (Throwable) obj);
}
break;
- case MessageType.EXECUTE_REGION_FUNCTION_ERROR:
+ case EXECUTE_REGION_FUNCTION_ERROR:
if (logger.isDebugEnabled()) {
logger.debug(
"ExecuteRegionFunctionOpImpl#processResponse: received message of type EXECUTE_REGION_FUNCTION_ERROR");
@@ -481,8 +483,8 @@ public class ExecuteRegionFunctionOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
- return msgType == MessageType.EXECUTE_REGION_FUNCTION_ERROR;
+ protected boolean isErrorResponse(MessageType msgType) {
+ return msgType == EXECUTE_REGION_FUNCTION_ERROR;
}
@Override
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteRegionFunctionSingleHopOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteRegionFunctionSingleHopOp.java
index 81f9d0bec5..86461a86b3 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteRegionFunctionSingleHopOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteRegionFunctionSingleHopOp.java
@@ -16,6 +16,8 @@
package org.apache.geode.cache.client.internal;
import static org.apache.geode.internal.cache.execute.AbstractExecution.DEFAULT_CLIENT_FUNCTION_TIMEOUT;
+import static org.apache.geode.internal.cache.tier.MessageType.EXECUTE_REGION_FUNCTION_ERROR;
+import static org.apache.geode.internal.cache.tier.MessageType.EXECUTE_REGION_FUNCTION_SINGLE_HOP;
import java.util.ArrayList;
import java.util.HashSet;
@@ -161,7 +163,7 @@ public class ExecuteRegionFunctionSingleHopOp {
Set<String> removedNodes, boolean allBuckets, final int timeoutMs) {
// What is this 8 that is getting added to filter and removednode sizes?
// It should have been used as a constant and documented
- super(MessageType.EXECUTE_REGION_FUNCTION_SINGLE_HOP,
+ super(EXECUTE_REGION_FUNCTION_SINGLE_HOP,
8 + serverRegionExecutor.getFilter().size() + removedNodes.size(), timeoutMs);
isHA = function.isHA();
optimizeForWrite = function.optimizeForWrite();
@@ -207,7 +209,7 @@ public class ExecuteRegionFunctionSingleHopOp {
final int timeoutMs) {
// What is this 8 that is getting added to filter and removednode sizes?
// It should have been used as a constant and documented
- super(MessageType.EXECUTE_REGION_FUNCTION_SINGLE_HOP,
+ super(EXECUTE_REGION_FUNCTION_SINGLE_HOP,
8 + serverRegionExecutor.getFilter().size() + removedNodes.size(), timeoutMs);
this.isHA = isHA;
this.optimizeForWrite = optimizeForWrite;
@@ -260,7 +262,7 @@ public class ExecuteRegionFunctionSingleHopOp {
try {
executeFunctionResponseMsg.readHeader();
switch (executeFunctionResponseMsg.getMessageType()) {
- case MessageType.EXECUTE_REGION_FUNCTION_RESULT:
+ case EXECUTE_REGION_FUNCTION_RESULT:
final boolean isDebugEnabled = logger.isDebugEnabled();
if (isDebugEnabled) {
logger.debug(
@@ -336,7 +338,7 @@ public class ExecuteRegionFunctionSingleHopOp {
}
return null;
- case MessageType.EXCEPTION:
+ case EXCEPTION:
if (logger.isDebugEnabled()) {
logger.debug(
"ExecuteRegionFunctionSingleHopOpImpl#processResponse: received message of type EXCEPTION");
@@ -362,7 +364,7 @@ public class ExecuteRegionFunctionSingleHopOp {
throw new ServerOperationException(s, (Throwable) obj);
}
break;
- case MessageType.EXECUTE_REGION_FUNCTION_ERROR:
+ case EXECUTE_REGION_FUNCTION_ERROR:
if (logger.isDebugEnabled()) {
logger.debug(
"ExecuteRegionFunctionSingleHopOpImpl#processResponse: received message of type EXECUTE_REGION_FUNCTION_ERROR");
@@ -410,8 +412,8 @@ public class ExecuteRegionFunctionSingleHopOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
- return msgType == MessageType.EXECUTE_REGION_FUNCTION_ERROR;
+ protected boolean isErrorResponse(MessageType msgType) {
+ return msgType == EXECUTE_REGION_FUNCTION_ERROR;
}
@Override
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetAllOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetAllOp.java
index 6e82a593a5..6cdfc93c99 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetAllOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetAllOp.java
@@ -219,7 +219,7 @@ public class GetAllOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.GET_ALL_DATA_ERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetClientPRMetaDataOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetClientPRMetaDataOp.java
index 55b2537dde..165216798c 100755
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetClientPRMetaDataOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetClientPRMetaDataOp.java
@@ -14,6 +14,8 @@
*/
package org.apache.geode.cache.client.internal;
+import static org.apache.geode.internal.cache.tier.MessageType.GET_CLIENT_PR_METADATA;
+
import java.util.List;
import java.util.Set;
@@ -61,7 +63,7 @@ public class GetClientPRMetaDataOp {
ClientMetadataService cms = null;
public GetClientPRMetaDataOpImpl(String regionFullPath, ClientMetadataService cms) {
- super(MessageType.GET_CLIENT_PR_METADATA, 1);
+ super(GET_CLIENT_PR_METADATA, 1);
this.regionFullPath = regionFullPath;
this.cms = cms;
getMessage().addStringPart(regionFullPath, true);
@@ -82,17 +84,17 @@ public class GetClientPRMetaDataOp {
@Override
protected Object processResponse(final @NotNull Message msg) throws Exception {
switch (msg.getMessageType()) {
- case MessageType.GET_CLIENT_PR_METADATA_ERROR:
+ case GET_CLIENT_PR_METADATA_ERROR:
String errorMsg = msg.getPart(0).getString();
if (logger.isDebugEnabled()) {
logger.debug(errorMsg);
}
throw new ServerOperationException(errorMsg);
- case MessageType.RESPONSE_CLIENT_PR_METADATA:
+ case RESPONSE_CLIENT_PR_METADATA:
final boolean isDebugEnabled = logger.isDebugEnabled();
if (isDebugEnabled) {
logger.debug("GetClientPRMetaDataOpImpl#processResponse: received message of type : {}"
- + MessageType.getString(msg.getMessageType()));
+ + msg.getMessageType());
}
int numParts = msg.getNumberOfParts();
ClientPartitionAdvisor advisor = cms.getClientPartitionAdvisor(regionFullPath);
@@ -125,7 +127,7 @@ public class GetClientPRMetaDataOp {
}
cms.setMetadataStable(true);
return null;
- case MessageType.EXCEPTION:
+ case EXCEPTION:
if (logger.isDebugEnabled()) {
logger.debug(
"GetClientPRMetaDataOpImpl#processResponse: received message of type EXCEPTION");
@@ -160,7 +162,7 @@ public class GetClientPRMetaDataOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return false;
}
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetClientPartitionAttributesOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetClientPartitionAttributesOp.java
index fecd9951b2..f804a0435b 100755
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetClientPartitionAttributesOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetClientPartitionAttributesOp.java
@@ -14,6 +14,10 @@
*/
package org.apache.geode.cache.client.internal;
+import static org.apache.geode.internal.cache.tier.MessageType.EXCEPTION;
+import static org.apache.geode.internal.cache.tier.MessageType.GET_CLIENT_PARTITION_ATTRIBUTES_ERROR;
+import static org.apache.geode.internal.cache.tier.MessageType.RESPONSE_CLIENT_PARTITION_ATTRIBUTES;
+
import java.util.Set;
import org.apache.logging.log4j.Logger;
@@ -86,18 +90,18 @@ public class GetClientPartitionAttributesOp {
@Override
protected Object processResponse(final @NotNull Message msg) throws Exception {
switch (msg.getMessageType()) {
- case MessageType.GET_CLIENT_PARTITION_ATTRIBUTES_ERROR:
+ case GET_CLIENT_PARTITION_ATTRIBUTES_ERROR:
String errorMsg = msg.getPart(0).getString();
if (logger.isDebugEnabled()) {
logger.debug(errorMsg);
}
throw new ServerOperationException(errorMsg);
- case MessageType.RESPONSE_CLIENT_PARTITION_ATTRIBUTES:
+ case RESPONSE_CLIENT_PARTITION_ATTRIBUTES:
final boolean isDebugEnabled = logger.isDebugEnabled();
if (isDebugEnabled) {
logger.debug(
"GetClientPartitionAttributesOpImpl#processResponse: received message of type : {}",
- MessageType.getString(msg.getMessageType()));
+ msg.getMessageType());
}
int bucketCount;
String colocatedWith;
@@ -126,7 +130,7 @@ public class GetClientPartitionAttributesOp {
new ClientPartitionAdvisor(bucketCount, colocatedWith, partitionResolverName, fpaSet);
return advisor;
- case MessageType.EXCEPTION:
+ case EXCEPTION:
if (logger.isDebugEnabled()) {
logger.debug(
"GetClientPartitionAttributesOpImpl#processResponse: received message of type EXCEPTION");
@@ -162,7 +166,7 @@ public class GetClientPartitionAttributesOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return false;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetEntryOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetEntryOp.java
index 6ddae50b96..fc675c5a81 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetEntryOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetEntryOp.java
@@ -59,7 +59,7 @@ public class GetEntryOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.REQUESTDATAERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetEventValueOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetEventValueOp.java
index 11d40769b5..65f419f1e0 100755
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetEventValueOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetEventValueOp.java
@@ -74,7 +74,7 @@ public class GetEventValueOp {
@Override
protected Object processResponse(final @NotNull Message msg) throws Exception {
Part part = msg.getPart(0);
- final int msgType = msg.getMessageType();
+ final MessageType msgType = msg.getMessageType();
if (msgType == MessageType.RESPONSE) {
return part;
} else {
@@ -90,13 +90,13 @@ public class GetEventValueOp {
throw new ServerOperationException(part.getString());
} else {
throw new InternalGemFireError(
- "Unexpected message type " + MessageType.getString(msgType));
+ "Unexpected message type " + msgType);
}
}
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.REQUESTDATAERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetFunctionAttributeOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetFunctionAttributeOp.java
index a294613579..551d109be8 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetFunctionAttributeOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetFunctionAttributeOp.java
@@ -46,7 +46,7 @@ public class GetFunctionAttributeOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.REQUESTDATAERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetOp.java
index 27c3b7d00e..2d441cccdc 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetOp.java
@@ -215,7 +215,7 @@ public class GetOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.REQUESTDATAERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXEnumByIdOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXEnumByIdOp.java
index 50815574ba..42e20a781c 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXEnumByIdOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXEnumByIdOp.java
@@ -55,7 +55,7 @@ public class GetPDXEnumByIdOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return false;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXEnumsOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXEnumsOp.java
index cbc2157513..d5d9b3fbb1 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXEnumsOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXEnumsOp.java
@@ -50,7 +50,7 @@ public class GetPDXEnumsOp {
@Override
protected Object processResponse(final @NotNull Message msg) throws Exception {
Part part = msg.getPart(0);
- int msgType = msg.getMessageType();
+ final MessageType msgType = msg.getMessageType();
if (msgType == MessageType.RESPONSE) {
return part.getObject();
@@ -64,13 +64,13 @@ public class GetPDXEnumsOp {
} else {
throw new InternalGemFireError(
- "Unexpected message type " + MessageType.getString(msgType));
+ "Unexpected message type " + msgType);
}
}
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return false;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXIdForEnumOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXIdForEnumOp.java
index 2189611c64..afad323cdc 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXIdForEnumOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXIdForEnumOp.java
@@ -56,7 +56,7 @@ public class GetPDXIdForEnumOp {
@Override
protected Object processResponse(final @NotNull Message msg) throws Exception {
Part part = msg.getPart(0);
- final int msgType = msg.getMessageType();
+ final MessageType msgType = msg.getMessageType();
if (msgType == MessageType.RESPONSE) {
return part.getInt();
} else {
@@ -69,13 +69,13 @@ public class GetPDXIdForEnumOp {
throw new ServerOperationException(part.getString());
} else {
throw new InternalGemFireError(
- "Unexpected message type " + MessageType.getString(msgType));
+ "Unexpected message type " + msgType);
}
}
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return false;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXIdForTypeOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXIdForTypeOp.java
index c8c9a64ee3..521f2b2039 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXIdForTypeOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXIdForTypeOp.java
@@ -56,7 +56,7 @@ public class GetPDXIdForTypeOp {
@Override
protected Object processResponse(final @NotNull Message msg) throws Exception {
Part part = msg.getPart(0);
- final int msgType = msg.getMessageType();
+ final MessageType msgType = msg.getMessageType();
if (msgType == MessageType.RESPONSE) {
return part.getInt();
} else {
@@ -69,13 +69,13 @@ public class GetPDXIdForTypeOp {
throw new ServerOperationException(part.getString());
} else {
throw new InternalGemFireError(
- "Unexpected message type " + MessageType.getString(msgType));
+ "Unexpected message type " + msgType);
}
}
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return false;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXTypeByIdOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXTypeByIdOp.java
index eda309379a..865969734d 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXTypeByIdOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXTypeByIdOp.java
@@ -55,7 +55,7 @@ public class GetPDXTypeByIdOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return false;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXTypesOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXTypesOp.java
index ba0508eafd..15eb53f121 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXTypesOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/GetPDXTypesOp.java
@@ -50,7 +50,7 @@ public class GetPDXTypesOp {
@Override
protected Object processResponse(final @NotNull Message msg) throws Exception {
Part part = msg.getPart(0);
- int msgType = msg.getMessageType();
+ MessageType msgType = msg.getMessageType();
if (msgType == MessageType.RESPONSE) {
return part.getObject();
@@ -64,13 +64,13 @@ public class GetPDXTypesOp {
} else {
throw new InternalGemFireError(
- "Unexpected message type " + MessageType.getString(msgType));
+ "Unexpected message type " + msgType);
}
}
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return false;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/InvalidateOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/InvalidateOp.java
index 62031e5453..3463afd828 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/InvalidateOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/InvalidateOp.java
@@ -166,7 +166,7 @@ public class InvalidateOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.INVALIDATE_ERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/KeySetOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/KeySetOp.java
index d2c8b6bac2..09e1879f14 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/KeySetOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/KeySetOp.java
@@ -72,7 +72,7 @@ public class KeySetOp {
final Exception[] exceptionRef = new Exception[1];
keySetResponseMessage.readHeader();
- final int msgType = keySetResponseMessage.getMessageType();
+ final MessageType msgType = keySetResponseMessage.getMessageType();
if (msgType == MessageType.RESPONSE) {
do {
keySetResponseMessage.receiveChunk();
@@ -99,7 +99,7 @@ public class KeySetOp {
throw new ServerOperationException(part.getString());
} else {
throw new InternalGemFireError(
- "Unexpected message type " + MessageType.getString(msgType));
+ "Unexpected message type " + msgType);
}
}
@@ -111,7 +111,7 @@ public class KeySetOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.KEY_SET_DATA_ERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/MakePrimaryOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/MakePrimaryOp.java
index b514de2238..b8af6e5a20 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/MakePrimaryOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/MakePrimaryOp.java
@@ -68,7 +68,7 @@ public class MakePrimaryOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return false;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/PingOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/PingOp.java
index 828da2e23f..9b69e1be4d 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/PingOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/PingOp.java
@@ -83,7 +83,7 @@ public class PingOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return false;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/PrimaryAckOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/PrimaryAckOp.java
index a85bb88dfa..34d6599ec5 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/PrimaryAckOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/PrimaryAckOp.java
@@ -73,7 +73,7 @@ public class PrimaryAckOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return false;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ProxyCacheCloseOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ProxyCacheCloseOp.java
index 4a70bf316b..1f821961f4 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ProxyCacheCloseOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ProxyCacheCloseOp.java
@@ -77,7 +77,7 @@ public class ProxyCacheCloseOp {
@Override
protected Object processResponse(final @NotNull Message msg) throws Exception {
Part part = msg.getPart(0);
- final int msgType = msg.getMessageType();
+ final MessageType msgType = msg.getMessageType();
if (msgType == MessageType.REPLY) {
return part.getObject();
} else if (msgType == MessageType.EXCEPTION) {
@@ -88,12 +88,12 @@ public class ProxyCacheCloseOp {
} else if (isErrorResponse(msgType)) {
throw new ServerOperationException(part.getString());
} else {
- throw new InternalGemFireError("Unexpected message type " + MessageType.getString(msgType));
+ throw new InternalGemFireError("Unexpected message type " + msgType);
}
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.REQUESTDATAERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/PutAllOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/PutAllOp.java
index 00141eeb87..8639cf19c1 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/PutAllOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/PutAllOp.java
@@ -398,7 +398,7 @@ public class PutAllOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.PUT_DATA_ERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/PutOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/PutOp.java
index 6d17d0ba6c..7cb0023c36 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/PutOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/PutOp.java
@@ -348,7 +348,7 @@ public class PutOp {
* @since GemFire 6.1
*/
private void processAck(Message msg, Connection con) throws Exception {
- final int msgType = msg.getMessageType();
+ final MessageType msgType = msg.getMessageType();
// Update delta stats
if (deltaSent && region != null) {
region.getCachePerfStats().incDeltasSent();
@@ -376,13 +376,13 @@ public class PutOp {
throw new ServerOperationException(part.getString());
} else {
throw new InternalGemFireError(
- "Unexpected message type " + MessageType.getString(msgType));
+ "Unexpected message type " + msgType);
}
}
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.PUT_DATA_ERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/QueryOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/QueryOp.java
index 294fed8041..f37b653e16 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/QueryOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/QueryOp.java
@@ -99,7 +99,7 @@ public class QueryOp {
*
* @throws org.apache.geode.SerializationException if serialization fails
*/
- protected QueryOpImpl(int msgType, int numParts) {
+ protected QueryOpImpl(MessageType msgType, int numParts) {
super(msgType, numParts);
}
@@ -209,8 +209,8 @@ public class QueryOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
- return msgType == MessageType.QUERY_DATA_ERROR || msgType == MessageType.CQDATAERROR_MSG_TYPE
+ protected boolean isErrorResponse(MessageType msgType) {
+ return msgType == MessageType.QUERY_DATA_ERROR || msgType == MessageType.CQDATAERROR
|| msgType == MessageType.CQ_EXCEPTION_TYPE;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ReadyForEventsOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ReadyForEventsOp.java
index 2d8f1515c0..8a71e2ca46 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ReadyForEventsOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ReadyForEventsOp.java
@@ -68,7 +68,7 @@ public class ReadyForEventsOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return false;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/RegisterDataSerializersOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/RegisterDataSerializersOp.java
index 14884cbbfe..2d05a21514 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/RegisterDataSerializersOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/RegisterDataSerializersOp.java
@@ -101,7 +101,7 @@ public class RegisterDataSerializersOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return false;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/RegisterInstantiatorsOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/RegisterInstantiatorsOp.java
index 10adf9644f..4a16089d18 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/RegisterInstantiatorsOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/RegisterInstantiatorsOp.java
@@ -132,7 +132,7 @@ public class RegisterInstantiatorsOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return false;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/RegisterInterestOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/RegisterInterestOp.java
index b5cca57f25..a5be9b7bb1 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/RegisterInterestOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/RegisterInterestOp.java
@@ -14,6 +14,8 @@
*/
package org.apache.geode.cache.client.internal;
+import static org.apache.geode.internal.cache.tier.MessageType.REGISTER_INTEREST;
+import static org.apache.geode.internal.cache.tier.MessageType.REGISTER_INTEREST_DATA_ERROR;
import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
import java.util.ArrayList;
@@ -130,7 +132,7 @@ public class RegisterInterestOp {
final boolean isDurable,
final boolean receiveUpdatesAsInvalidates,
final @NotNull DataPolicy regionDataPolicy) {
- super(MessageType.REGISTER_INTEREST, 7);
+ super(REGISTER_INTEREST, 7);
this.region = region;
getMessage().addStringPart(region, true);
getMessage().addIntPart(interestType.ordinal());
@@ -155,7 +157,7 @@ public class RegisterInterestOp {
*
* @throws org.apache.geode.SerializationException if serialization fails
*/
- protected RegisterInterestOpImpl(String region, int msgType, int numParts) {
+ protected RegisterInterestOpImpl(String region, MessageType msgType, int numParts) {
super(msgType, numParts);
this.region = region;
}
@@ -175,7 +177,7 @@ public class RegisterInterestOp {
ChunkedMessage chunkedMessage = (ChunkedMessage) m;
chunkedMessage.readHeader();
switch (chunkedMessage.getMessageType()) {
- case MessageType.RESPONSE_FROM_PRIMARY: {
+ case RESPONSE_FROM_PRIMARY: {
LocalRegion localRegion = null;
try {
@@ -233,11 +235,11 @@ public class RegisterInterestOp {
}
return serverKeys;
}
- case MessageType.RESPONSE_FROM_SECONDARY:
+ case RESPONSE_FROM_SECONDARY:
// Read the chunk
chunkedMessage.receiveChunk();
return null;
- case MessageType.EXCEPTION:
+ case EXCEPTION:
// Read the chunk
chunkedMessage.receiveChunk();
// Deserialize the result
@@ -248,7 +250,7 @@ public class RegisterInterestOp {
String s = this + ": While performing a remote " + getOpName();
throw new ServerOperationException(s, (Throwable) obj);
}
- case MessageType.REGISTER_INTEREST_DATA_ERROR:
+ case REGISTER_INTEREST_DATA_ERROR:
// Read the chunk
chunkedMessage.receiveChunk();
@@ -266,8 +268,8 @@ public class RegisterInterestOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
- return msgType == MessageType.REGISTER_INTEREST_DATA_ERROR;
+ protected boolean isErrorResponse(MessageType msgType) {
+ return msgType == REGISTER_INTEREST_DATA_ERROR;
}
@Override
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/RemoveAllOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/RemoveAllOp.java
index 4f0496212e..6611038aa3 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/RemoveAllOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/RemoveAllOp.java
@@ -361,7 +361,7 @@ public class RemoveAllOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.PUT_DATA_ERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/RollbackOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/RollbackOp.java
index 4746d4a053..798b423cf0 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/RollbackOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/RollbackOp.java
@@ -62,7 +62,7 @@ public class RollbackOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.EXCEPTION;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/SizeOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/SizeOp.java
index 8976e45ec0..03557ddc4f 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/SizeOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/SizeOp.java
@@ -57,7 +57,7 @@ public class SizeOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.SIZE_ERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/TXFailoverOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/TXFailoverOp.java
index ff632ec7ad..33d2e68400 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/TXFailoverOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/TXFailoverOp.java
@@ -56,7 +56,7 @@ public class TXFailoverOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.EXCEPTION;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/TXSynchronizationOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/TXSynchronizationOp.java
index ada72048cb..2feaa0727b 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/TXSynchronizationOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/TXSynchronizationOp.java
@@ -25,7 +25,6 @@ import org.apache.geode.internal.cache.TXCommitMessage;
import org.apache.geode.internal.cache.TXManagerImpl;
import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.sockets.Message;
-import org.apache.geode.internal.cache.tier.sockets.Part;
/**
* TXSynchronizationOp sends JTA beforeCompletion and afterCompletion messages to the server pool.
@@ -76,20 +75,20 @@ public class TXSynchronizationOp {
@Override
protected void processAck(Message msg, String opName) throws Exception {
- final int msgType = msg.getMessageType();
+ final MessageType msgType = msg.getMessageType();
if (msgType == MessageType.REPLY) {
return;
- } else {
- Part part = msg.getPart(0);
- if (msgType == MessageType.EXCEPTION) {
- Throwable t = (Throwable) part.getObject();
- if (t instanceof CommitConflictException
- || t instanceof SynchronizationCommitConflictException) {
- throw (GemFireException) t;
- }
+ }
+
+ if (msgType == MessageType.EXCEPTION) {
+ Throwable t = (Throwable) msg.getPart(0).getObject();
+ if (t instanceof CommitConflictException
+ || t instanceof SynchronizationCommitConflictException) {
+ throw (GemFireException) t;
}
- super.processAck(msg, opName);
}
+
+ super.processAck(msg, opName);
}
@@ -124,7 +123,7 @@ public class TXSynchronizationOp {
* @see org.apache.geode.cache.client.internal.AbstractOp#isErrorResponse(int)
*/
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.REQUESTDATAERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/UnregisterInterestListOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/UnregisterInterestListOp.java
index 878115cb42..69b3f7138c 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/UnregisterInterestListOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/UnregisterInterestListOp.java
@@ -76,7 +76,7 @@ public class UnregisterInterestListOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.UNREGISTER_INTEREST_DATA_ERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/UnregisterInterestOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/UnregisterInterestOp.java
index 20242474fd..3ac932e4ac 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/UnregisterInterestOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/UnregisterInterestOp.java
@@ -76,7 +76,7 @@ public class UnregisterInterestOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.UNREGISTER_INTEREST_DATA_ERROR;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/query/internal/CqQueryVsdStats.java b/geode-core/src/main/java/org/apache/geode/cache/query/internal/CqQueryVsdStats.java
index 8da361ec5b..f7e3d05478 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/query/internal/CqQueryVsdStats.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/query/internal/CqQueryVsdStats.java
@@ -300,23 +300,22 @@ public class CqQueryVsdStats {
*
* @param cqEvent object the type of CQ event
*/
- public void updateStats(Integer cqEvent) {
+ public void updateStats(MessageType cqEvent) {
if (cqEvent == null) {
return;
}
incNumEvents();
switch (cqEvent) {
- case MessageType.LOCAL_CREATE:
+ case LOCAL_CREATE:
incNumInserts();
return;
- case MessageType.LOCAL_UPDATE:
+ case LOCAL_UPDATE:
incNumUpdates();
return;
- case MessageType.LOCAL_DESTROY:
+ case LOCAL_DESTROY:
incNumDeletes();
return;
default:
- return;
}
}
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/CqService.java b/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/CqService.java
index ff29eb9a34..2356a3a699 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/CqService.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/CqService.java
@@ -39,6 +39,7 @@ import org.apache.geode.cache.query.RegionNotFoundException;
import org.apache.geode.distributed.internal.DistributionAdvisor.Profile;
import org.apache.geode.internal.cache.EventID;
import org.apache.geode.internal.cache.FilterRoutingInfo;
+import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
@@ -172,7 +173,8 @@ public interface CqService {
* @param cqs list of cqs with the cq operation from the Server.
* @param messageType base operation
*/
- void dispatchCqListeners(HashMap<String, Integer> cqs, int messageType, Object key, Object value,
+ void dispatchCqListeners(HashMap<String, MessageType> cqs, MessageType messageType, Object key,
+ Object value,
byte[] delta, QueueManager qManager, EventID eventId);
void processEvents(CacheEvent<?, ?> event, Profile localProfile, Profile[] profiles,
diff --git a/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/MissingCqService.java b/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/MissingCqService.java
index a276d58f36..891d00b6c9 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/MissingCqService.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/MissingCqService.java
@@ -40,6 +40,7 @@ import org.apache.geode.cache.query.RegionNotFoundException;
import org.apache.geode.distributed.internal.DistributionAdvisor.Profile;
import org.apache.geode.internal.cache.EventID;
import org.apache.geode.internal.cache.FilterRoutingInfo;
+import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
@@ -132,9 +133,9 @@ public class MissingCqService implements CqService {
}
@Override
- public void dispatchCqListeners(HashMap<String, Integer> cqs, int messageType, Object key,
- Object value,
- byte[] delta, QueueManager qManager, EventID eventId) {
+ public void dispatchCqListeners(HashMap<String, MessageType> cqs, MessageType messageType,
+ Object key,
+ Object value, byte[] delta, QueueManager qManager, EventID eventId) {
throw new IllegalStateException("CqService is not available.");
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java
index e32a2afebe..955cb5fd6b 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java
@@ -725,7 +725,7 @@ public abstract class DistributedCacheOperation {
}
void doRemoveDestroyTokensFromCqResultKeys(FilterInfo filterInfo, ServerCQ cq) {
- for (Map.Entry<Long, Integer> e : filterInfo.getCQs().entrySet()) {
+ for (Map.Entry<Long, MessageType> e : filterInfo.getCQs().entrySet()) {
Long cqID = e.getKey();
// For the CQs satisfying the event with destroy CQEvent, remove
// the entry form CQ cache.
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedPutAllOperation.java b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedPutAllOperation.java
index 0f043a03bb..403c07c009 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedPutAllOperation.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedPutAllOperation.java
@@ -820,7 +820,7 @@ public class DistributedPutAllOperation extends AbstractUpdateOperation {
@Override
void doRemoveDestroyTokensFromCqResultKeys(FilterInfo filterInfo, ServerCQ cq) {
- for (Map.Entry<Long, Integer> e : filterInfo.getCQs().entrySet()) {
+ for (Map.Entry<Long, MessageType> e : filterInfo.getCQs().entrySet()) {
Long cqID = e.getKey();
// For the CQs satisfying the event with destroy CQEvent, remove
// the entry from CQ cache.
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedRemoveAllOperation.java b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedRemoveAllOperation.java
index 0dcbb209d9..92c3fa827f 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedRemoveAllOperation.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedRemoveAllOperation.java
@@ -590,7 +590,7 @@ public class DistributedRemoveAllOperation extends AbstractUpdateOperation {
@Override
void doRemoveDestroyTokensFromCqResultKeys(FilterInfo filterInfo, ServerCQ cq) {
- for (Map.Entry<Long, Integer> e : filterInfo.getCQs().entrySet()) {
+ for (Map.Entry<Long, MessageType> e : filterInfo.getCQs().entrySet()) {
Long cqID = e.getKey();
// For the CQs satisfying the event with destroy CQEvent, remove
// the entry from CQ cache.
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/FilterRoutingInfo.java b/geode-core/src/main/java/org/apache/geode/internal/cache/FilterRoutingInfo.java
index f73918d7a9..63596c95a8 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/FilterRoutingInfo.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/FilterRoutingInfo.java
@@ -31,6 +31,7 @@ import org.apache.geode.distributed.internal.membership.InternalDistributedMembe
import org.apache.geode.internal.HeapDataOutputStream;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.VersionedDataSerializable;
+import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.serialization.ByteArrayDataInput;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.serialization.StaticSerialization;
@@ -62,7 +63,7 @@ public class FilterRoutingInfo implements VersionedDataSerializable {
*
* @param cqInfo map of server side CQ Name to CQ event type.
*/
- public void setLocalCqInfo(HashMap<Long, Integer> cqInfo) {
+ public void setLocalCqInfo(HashMap<Long, MessageType> cqInfo) {
if (localFilterInfo == null) {
localFilterInfo = new FilterInfo();
}
@@ -100,7 +101,8 @@ public class FilterRoutingInfo implements VersionedDataSerializable {
* @param member for which CQs are satisfied.
* @param cqInfo map of server side CQ Name to CQ event type.
*/
- public void setCqRoutingInfo(InternalDistributedMember member, HashMap<Long, Integer> cqInfo) {
+ public void setCqRoutingInfo(InternalDistributedMember member,
+ HashMap<Long, MessageType> cqInfo) {
FilterInfo fInfo = new FilterInfo();
fInfo.setCQs(cqInfo);
serverFilterInfo.put(member, fInfo);
@@ -251,7 +253,7 @@ public class FilterRoutingInfo implements VersionedDataSerializable {
private static final long serialVersionUID = 0;
/** Map holding Cq filterID and CqEvent Type */
- private HashMap<Long, Integer> cqs;
+ private HashMap<Long, MessageType> cqs;
/**
* serialized routing data. This is only deserialized when requested so that routing information
@@ -275,9 +277,7 @@ public class FilterRoutingInfo implements VersionedDataSerializable {
if (cqs == null) {
cqs = new HashMap<>();
}
- for (Map.Entry<Long, Integer> entry : other.cqs.entrySet()) {
- cqs.put(entry.getKey(), entry.getValue());
- }
+ cqs.putAll(other.cqs);
}
if (other.interestedClients != null) {
if (interestedClients == null) {
@@ -327,11 +327,11 @@ public class FilterRoutingInfo implements VersionedDataSerializable {
} else {
hdos.writeBoolean(true);
InternalDataSerializer.writeArrayLength(cqs.size(), hdos);
- for (final Map.Entry<Long, Integer> longIntegerEntry : cqs.entrySet()) {
+ for (final Map.Entry<Long, MessageType> longIntegerEntry : cqs.entrySet()) {
// most cq IDs and all event types are small ints, so we use an optimized
// write that serializes 7 bits at a time in a compact form
InternalDataSerializer.writeUnsignedVL(longIntegerEntry.getKey(), hdos);
- InternalDataSerializer.writeUnsignedVL(longIntegerEntry.getValue(), hdos);
+ InternalDataSerializer.writeUnsignedVL(longIntegerEntry.getValue().id, hdos);
}
}
InternalDataSerializer.writeSetOfLongs(interestedClients, longIDs, hdos);
@@ -340,14 +340,14 @@ public class FilterRoutingInfo implements VersionedDataSerializable {
DataSerializer.writeByteArray(myData, hdos.size(), out);
}
- public HashMap<Long, Integer> getCQs() {
+ public HashMap<Long, MessageType> getCQs() {
if (cqs == null && myData != null) {
deserialize();
}
return cqs;
}
- public void setCQs(HashMap<Long, Integer> cqs) {
+ public void setCQs(HashMap<Long, MessageType> cqs) {
this.cqs = cqs;
}
@@ -386,7 +386,8 @@ public class FilterRoutingInfo implements VersionedDataSerializable {
cqs = new HashMap<>(numEntries);
for (int i = 0; i < numEntries; i++) {
Long key = InternalDataSerializer.readUnsignedVL(dis);
- Integer value = (int) InternalDataSerializer.readUnsignedVL(dis);
+ MessageType value =
+ MessageType.valueOf((int) InternalDataSerializer.readUnsignedVL(dis));
cqs.put(key, value);
}
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/TXManagerImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/TXManagerImpl.java
index 2dffabe3c0..96b8a2a7d5 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/TXManagerImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/TXManagerImpl.java
@@ -58,7 +58,6 @@ import org.apache.geode.distributed.internal.MembershipListener;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.SystemTimer.SystemTimerTask;
import org.apache.geode.internal.cache.entries.AbstractRegionEntry;
-import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.sockets.Message;
import org.apache.geode.internal.statistics.StatisticsClock;
import org.apache.geode.internal.util.concurrent.CustomEntryConcurrentHashMap;
@@ -1011,7 +1010,7 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
}
if (logger.isDebugEnabled()) {
logger.debug("masqueradeAs tx {} for client message {}", val,
- MessageType.getString(msg.getMessageType()));
+ msg.getMessageType());
}
return val;
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/execute/ServerToClientFunctionResultSender.java b/geode-core/src/main/java/org/apache/geode/internal/cache/execute/ServerToClientFunctionResultSender.java
index d07434be7a..d471b2a856 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/execute/ServerToClientFunctionResultSender.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/execute/ServerToClientFunctionResultSender.java
@@ -29,6 +29,7 @@ import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.internal.cache.execute.metrics.FunctionStatsManager;
import org.apache.geode.internal.cache.tier.Command;
+import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.sockets.ChunkedMessage;
import org.apache.geode.internal.cache.tier.sockets.Message;
import org.apache.geode.internal.cache.tier.sockets.ServerConnection;
@@ -38,11 +39,11 @@ import org.apache.geode.logging.internal.log4j.api.LogService;
public class ServerToClientFunctionResultSender implements ResultSender {
private static final Logger logger = LogService.getLogger();
- protected ChunkedMessage msg = null;
+ protected ChunkedMessage msg;
- protected ServerConnection sc = null;
+ protected ServerConnection sc;
- protected int messageType = -1;
+ protected MessageType messageType;
protected volatile boolean headerSent = false;
@@ -68,7 +69,7 @@ public class ServerToClientFunctionResultSender implements ResultSender {
return lastResultReceived;
}
- public ServerToClientFunctionResultSender(ChunkedMessage msg, int messageType,
+ public ServerToClientFunctionResultSender(ChunkedMessage msg, MessageType messageType,
ServerConnection sc, Function function, ExecuteFunctionOperationContext authzContext) {
this.msg = msg;
this.msg.setVersion(sc.getClientVersion());
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/execute/ServerToClientFunctionResultSender65.java b/geode-core/src/main/java/org/apache/geode/internal/cache/execute/ServerToClientFunctionResultSender65.java
index 55dbe28dc1..698372e11e 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/execute/ServerToClientFunctionResultSender65.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/execute/ServerToClientFunctionResultSender65.java
@@ -28,6 +28,7 @@ import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.internal.cache.execute.metrics.FunctionStatsManager;
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;
import org.apache.geode.internal.cache.tier.sockets.ChunkedMessage;
import org.apache.geode.internal.cache.tier.sockets.ServerConnection;
@@ -41,7 +42,7 @@ import org.apache.geode.logging.internal.log4j.api.LogService;
public class ServerToClientFunctionResultSender65 extends ServerToClientFunctionResultSender {
private static final Logger logger = LogService.getLogger();
- public ServerToClientFunctionResultSender65(ChunkedMessage msg, int messageType,
+ public ServerToClientFunctionResultSender65(ChunkedMessage msg, MessageType messageType,
ServerConnection sc, Function function, ExecuteFunctionOperationContext authzContext) {
super(msg, messageType, sc, function, authzContext);
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/MessageType.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/MessageType.java
index 85f692226f..a853dfbd6d 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/MessageType.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/MessageType.java
@@ -12,604 +12,430 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
+
package org.apache.geode.internal.cache.tier;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import org.apache.geode.annotations.Immutable;
+
/**
* Pre-defined message types supported by the system
*
* @since GemFire 2.0.2
*/
-public class MessageType {
+public enum MessageType {
/** An invalid message, typically due to problems receiving a complete message */
- public static final int INVALID = -1;
+ INVALID(-1),
/** A request for a value from a region */
- public static final int REQUEST = 0;
+ REQUEST(0),
/** The value from a region */
- public static final int RESPONSE = 1;
+ RESPONSE(1),
- /** An acception occurred while accessing a region */
- public static final int EXCEPTION = 2;
+ /** An exception occurred while accessing a region */
+ EXCEPTION(2),
/** The get request data was bad (null key, for example) */
- public static final int REQUESTDATAERROR = 3;
+ REQUESTDATAERROR(3),
- // public static final int DATANOTFOUNDERROR = 4;
+ // DATANOTFOUNDERROR(4),
/** A ping request */
- public static final int PING = 5;
+ PING(5),
/** A ping reply */
- public static final int REPLY = 6;
+ REPLY(6),
/**
* Put data into the cache
*
* @since GemFire 3.5
*/
- public static final int PUT = 7;
+ PUT(7),
/**
* The put request data was bad (null key or value, for example)
*
* @since GemFire 3.5
*/
- public static final int PUT_DATA_ERROR = 8;
+ PUT_DATA_ERROR(8),
/**
* Destroy an entry from the cache
*
* @since GemFire 3.5
*/
- public static final int DESTROY = 9;
+ DESTROY(9),
/**
* The destroy request data was bad (null key, for example)
*
* @since GemFire 3.5
*/
- public static final int DESTROY_DATA_ERROR = 10;
+ DESTROY_DATA_ERROR(10),
/**
* Destroy a region from the cache
*
* @since GemFire 3.5
*/
- public static final int DESTROY_REGION = 11;
+ DESTROY_REGION(11),
/**
* The destroy region request data was bad (null region name, for example)
*
* @since GemFire 3.5
*/
- public static final int DESTROY_REGION_DATA_ERROR = 12;
+ DESTROY_REGION_DATA_ERROR(12),
/**
* A client wished to be notified of updates
*
* @since GemFire 3.5
*/
- public static final int CLIENT_NOTIFICATION = 13;
+ CLIENT_NOTIFICATION(13),
/**
* Information about a client receiving updates has changed
*
* @since GemFire 3.5
*/
- public static final int UPDATE_CLIENT_NOTIFICATION = 14;
+ UPDATE_CLIENT_NOTIFICATION(14),
/**
* The receiver (which is an edge client in this case) should locally invalidate a piece of data
*
* @since GemFire 3.5
*/
- public static final int LOCAL_INVALIDATE = 15;
+ LOCAL_INVALIDATE(15),
/**
* The receiver (which is an edge client in this case) should locally destroy a piece of data
*
* @since GemFire 3.5
*/
- public static final int LOCAL_DESTROY = 16;
+ LOCAL_DESTROY(16),
/**
* The receiver (which is an edge client in this case) should locally destroy a region
*
* @since GemFire 3.5
*/
- public static final int LOCAL_DESTROY_REGION = 17;
+ LOCAL_DESTROY_REGION(17),
/**
* A message to close the client connection
*
* @since GemFire 3.5
*/
- public static final int CLOSE_CONNECTION = 18;
+ CLOSE_CONNECTION(18),
/**
* A message to process a batch of messages
*
* @since GemFire 4.1.1
*/
- public static final int PROCESS_BATCH = 19;
+ PROCESS_BATCH(19),
/**
* A message to register interest in a specific key
*
* @since GemFire 4.1.2
*/
- public static final int REGISTER_INTEREST = 20;
+ REGISTER_INTEREST(20),
/**
* The register interest request data was bad (null region name, for example)
*
* @since GemFire 4.1.2
*/
- public static final int REGISTER_INTEREST_DATA_ERROR = 21;
+ REGISTER_INTEREST_DATA_ERROR(21),
/**
* A message to unregister interest in a specific key
*
* @since GemFire 4.1.2
*/
- public static final int UNREGISTER_INTEREST = 22;
+ UNREGISTER_INTEREST(22),
/**
* The unregister interest request data was bad (null region name, for example)
*
* @since GemFire 4.1.2
*/
- public static final int UNREGISTER_INTEREST_DATA_ERROR = 23;
+ UNREGISTER_INTEREST_DATA_ERROR(23),
/**
* A message to register interest in a specific list of keys
*
* @since GemFire 4.1.2
*/
- public static final int REGISTER_INTEREST_LIST = 24;
+ REGISTER_INTEREST_LIST(24),
/**
* A message to unregister interest in a specific list of keys
*
* @since GemFire 4.1.2
*/
- public static final int UNREGISTER_INTEREST_LIST = 25;
+ UNREGISTER_INTEREST_LIST(25),
/**
* An unknown message type. This is being used for responses.
*
* @since GemFire 4.1.2
*/
- public static final int UNKNOWN_MESSAGE_TYPE_ERROR = 26;
+ UNKNOWN_MESSAGE_TYPE_ERROR(26),
/**
* The receiver (which is an edge client in this case) should locally create a piece of data
*
* @since GemFire 4.1.2
*/
- public static final int LOCAL_CREATE = 27;
+ LOCAL_CREATE(27),
/**
* The receiver (which is an edge client in this case) should locally update a piece of data
*
* @since GemFire 4.1.2
*/
- public static final int LOCAL_UPDATE = 28;
+ LOCAL_UPDATE(28),
- public static final int CREATE_REGION = 29;
- public static final int CREATE_REGION_DATA_ERROR = 30;
+ CREATE_REGION(29),
+ CREATE_REGION_DATA_ERROR(30),
/**
* A message to make primary server
*
* @since GemFire 5.1
*/
- public static final int MAKE_PRIMARY = 31;
+ MAKE_PRIMARY(31),
/**
* Response message type from primary server
*
* @since GemFire 5.1
*/
- public static final int RESPONSE_FROM_PRIMARY = 32;
+ RESPONSE_FROM_PRIMARY(32),
/**
* Response message type from secondary server
*
* @since GemFire 5.1
*/
- public static final int RESPONSE_FROM_SECONDARY = 33;
+ RESPONSE_FROM_SECONDARY(33),
/**
* A query request message
*
- * <p>
- * author gregp
- *
* @since GemFire 4.1.1
*/
- public static final int QUERY = 34;
+ QUERY(34),
- public static final int QUERY_DATA_ERROR = 35;
+ QUERY_DATA_ERROR(35),
- public static final int CLEAR_REGION = 36;
+ CLEAR_REGION(36),
- public static final int CLEAR_REGION_DATA_ERROR = 37;
+ CLEAR_REGION_DATA_ERROR(37),
- public static final int CONTAINS_KEY = 38;
+ CONTAINS_KEY(38),
- public static final int CONTAINS_KEY_DATA_ERROR = 39;
+ CONTAINS_KEY_DATA_ERROR(39),
- public static final int KEY_SET = 40;
+ KEY_SET(40),
- public static final int KEY_SET_DATA_ERROR = 41;
+ KEY_SET_DATA_ERROR(41),
- public static final int EXECUTECQ_MSG_TYPE = 42;
+ EXECUTECQ(42),
- public static final int EXECUTECQ_WITH_IR_MSG_TYPE = 43;
+ EXECUTECQ_WITH_IR(43),
- public static final int STOPCQ_MSG_TYPE = 44;
+ STOPCQ(44),
- public static final int CLOSECQ_MSG_TYPE = 45;
+ CLOSECQ(45),
- public static final int CLOSECLIENTCQS_MSG_TYPE = 46;
+ CLOSECLIENTCQS(46),
- public static final int CQDATAERROR_MSG_TYPE = 47;
+ CQDATAERROR(47),
- public static final int GETCQSTATS_MSG_TYPE = 48;
+ GETCQSTATS(48),
- public static final int MONITORCQ_MSG_TYPE = 49;
+ MONITORCQ(49),
- public static final int CQ_EXCEPTION_TYPE = 50;
+ CQ_EXCEPTION_TYPE(50),
/**
* A message to register the Instantiators.
*
* @since GemFire 5.0
*/
- public static final int REGISTER_INSTANTIATORS = 51;
+ REGISTER_INSTANTIATORS(51),
- public static final int PERIODIC_ACK = 52;
+ PERIODIC_ACK(52),
- public static final int CLIENT_READY = 53;
+ CLIENT_READY(53),
- public static final int CLIENT_MARKER = 54;
+ CLIENT_MARKER(54),
- public static final int INVALIDATE_REGION = 55;
+ INVALIDATE_REGION(55),
- public static final int PUTALL = 56;
+ PUTALL(56),
- public static final int GET_ALL_DATA_ERROR = 58;
+ GET_ALL_DATA_ERROR(58),
- public static final int EXECUTE_REGION_FUNCTION = 59;
+ EXECUTE_REGION_FUNCTION(59),
- public static final int EXECUTE_REGION_FUNCTION_RESULT = 60;
+ EXECUTE_REGION_FUNCTION_RESULT(60),
- public static final int EXECUTE_REGION_FUNCTION_ERROR = 61;
+ EXECUTE_REGION_FUNCTION_ERROR(61),
- public static final int EXECUTE_FUNCTION = 62;
+ EXECUTE_FUNCTION(62),
- public static final int EXECUTE_FUNCTION_RESULT = 63;
+ EXECUTE_FUNCTION_RESULT(63),
- public static final int EXECUTE_FUNCTION_ERROR = 64;
+ EXECUTE_FUNCTION_ERROR(64),
- public static final int CLIENT_REGISTER_INTEREST = 65;
+ CLIENT_REGISTER_INTEREST(65),
- public static final int CLIENT_UNREGISTER_INTEREST = 66;
+ CLIENT_UNREGISTER_INTEREST(66),
- public static final int REGISTER_DATASERIALIZERS = 67;
+ REGISTER_DATASERIALIZERS(67),
- public static final int REQUEST_EVENT_VALUE = 68;
+ REQUEST_EVENT_VALUE(68),
- public static final int REQUEST_EVENT_VALUE_ERROR = 69;
+ REQUEST_EVENT_VALUE_ERROR(69),
- public static final int PUT_DELTA_ERROR = 70;
+ PUT_DELTA_ERROR(70),
- public static final int GET_CLIENT_PR_METADATA = 71;
+ GET_CLIENT_PR_METADATA(71),
- public static final int RESPONSE_CLIENT_PR_METADATA = 72;
+ RESPONSE_CLIENT_PR_METADATA(72),
- public static final int GET_CLIENT_PARTITION_ATTRIBUTES = 73;
+ GET_CLIENT_PARTITION_ATTRIBUTES(73),
- public static final int RESPONSE_CLIENT_PARTITION_ATTRIBUTES = 74;
+ RESPONSE_CLIENT_PARTITION_ATTRIBUTES(74),
- public static final int GET_CLIENT_PR_METADATA_ERROR = 75;
+ GET_CLIENT_PR_METADATA_ERROR(75),
- public static final int GET_CLIENT_PARTITION_ATTRIBUTES_ERROR = 76;
+ GET_CLIENT_PARTITION_ATTRIBUTES_ERROR(76),
- public static final int USER_CREDENTIAL_MESSAGE = 77;
+ USER_CREDENTIAL_MESSAGE(77),
- public static final int REMOVE_USER_AUTH = 78;
+ REMOVE_USER_AUTH(78),
- public static final int EXECUTE_REGION_FUNCTION_SINGLE_HOP = 79;
+ EXECUTE_REGION_FUNCTION_SINGLE_HOP(79),
- public static final int QUERY_WITH_PARAMETERS = 80;
+ QUERY_WITH_PARAMETERS(80),
- public static final int SIZE = 81;
- public static final int SIZE_ERROR = 82;
+ SIZE(81),
+ SIZE_ERROR(82),
- public static final int INVALIDATE = 83;
- public static final int INVALIDATE_ERROR = 84;
+ INVALIDATE(83),
+ INVALIDATE_ERROR(84),
- public static final int COMMIT = 85;
- public static final int COMMIT_ERROR = 86;
+ COMMIT(85),
+ COMMIT_ERROR(86),
- public static final int ROLLBACK = 87;
+ ROLLBACK(87),
- public static final int TX_FAILOVER = 88;
+ TX_FAILOVER(88),
- public static final int GET_ENTRY = 89;
+ GET_ENTRY(89),
- public static final int TX_SYNCHRONIZATION = 90;
+ TX_SYNCHRONIZATION(90),
- public static final int GET_FUNCTION_ATTRIBUTES = 91;
+ GET_FUNCTION_ATTRIBUTES(91),
- public static final int GET_PDX_TYPE_BY_ID = 92;
+ GET_PDX_TYPE_BY_ID(92),
- public static final int GET_PDX_ID_FOR_TYPE = 93;
+ GET_PDX_ID_FOR_TYPE(93),
- public static final int ADD_PDX_TYPE = 94;
+ ADD_PDX_TYPE(94),
- public static final int GET_ALL_FOR_RI = 95; // this message type is not used after 6.x
+ GET_ALL_FOR_RI(95), // this message type is not used after 6.x
- public static final int ADD_PDX_ENUM = 96;
- public static final int GET_PDX_ID_FOR_ENUM = 97;
- public static final int GET_PDX_ENUM_BY_ID = 98;
+ ADD_PDX_ENUM(96),
+ GET_PDX_ID_FOR_ENUM(97),
+ GET_PDX_ENUM_BY_ID(98),
- public static final int SERVER_TO_CLIENT_PING = 99;
+ SERVER_TO_CLIENT_PING(99),
- public static final int GET_ALL_70 = 100;
+ GET_ALL_70(100),
/** gets the pdx type definitions @since GemFire 7.0 */
- public static final int GET_PDX_TYPES = 101;
+ GET_PDX_TYPES(101),
/** gets the pdx enum definitions @since GemFire 7.0 */
- public static final int GET_PDX_ENUMS = 102;
+ GET_PDX_ENUMS(102),
- public static final int TOMBSTONE_OPERATION = 103;
+ TOMBSTONE_OPERATION(103),
- public static final int GATEWAY_RECEIVER_COMMAND = 104;
+ GATEWAY_RECEIVER_COMMAND(104),
- public static final int GETDURABLECQS_MSG_TYPE = 105;
+ GETDURABLECQS(105),
- public static final int GET_DURABLE_CQS_DATA_ERROR = 106;
+ GET_DURABLE_CQS_DATA_ERROR(106),
- public static final int GET_ALL_WITH_CALLBACK = 107;
+ GET_ALL_WITH_CALLBACK(107),
- public static final int PUT_ALL_WITH_CALLBACK = 108;
+ PUT_ALL_WITH_CALLBACK(108),
- public static final int REMOVE_ALL = 109;
+ REMOVE_ALL(109),
/**
* @since Geode 1.15
*/
- public static final int CLIENT_RE_AUTHENTICATE = 110;
- /**
- * Must be equal to last valid message id.
- */
- private static final int LAST_VALID_MESSAGE_ID = CLIENT_RE_AUTHENTICATE;
+ CLIENT_RE_AUTHENTICATE(110);
+
+
+ public final int id;
+
+ MessageType(final int id) {
+ this.id = id;
+ }
+
+ @Immutable
+ private static final MessageType[] messageTypes = new MessageType[111];
+
+ static {
+ for (final MessageType messageType : values()) {
+ if (messageType == INVALID) {
+ continue;
+ }
+ final MessageType otherMessageType = messageTypes[messageType.id];
+ if (null != (otherMessageType)) {
+ throw new IllegalStateException(
+ String.format("Duplicate MessageType id %d found in %s and %s", messageType.id,
+ messageType, otherMessageType));
+ }
- public static boolean validate(int messageType) {
- if (messageType == -1 || messageType == 4) {
- return false;
+ messageTypes[messageType.id] = messageType;
}
- boolean valid = (messageType >= 0 && messageType <= LAST_VALID_MESSAGE_ID);
- // logger.warning("Message type is valid: " + valid);
- return valid;
}
- public static String getString(int type) {
- switch (type) {
- case INVALID:
- return "INVALID";
- case REQUEST:
- return "REQUEST";
- case RESPONSE:
- return "RESPONSE";
- case EXCEPTION:
- return "EXCEPTION";
- case REQUESTDATAERROR:
- return "REQUESTDATAERROR";
- case PING:
- return "PING";
- case REPLY:
- return "REPLY";
- case PUT:
- return "PUT";
- case PUTALL:
- return "PUTALL";
- case PUT_DATA_ERROR:
- return "PUT_DATA_ERROR";
- case DESTROY:
- return "DESTROY";
- case DESTROY_DATA_ERROR:
- return "DESTROY_DATA_ERROR";
- case DESTROY_REGION:
- return "DESTROY_REGION";
- case DESTROY_REGION_DATA_ERROR:
- return "DESTROY_REGION_DATA_ERROR";
- case CLIENT_NOTIFICATION:
- return "CLIENT_NOTIFICATION";
- case UPDATE_CLIENT_NOTIFICATION:
- return "UPDATE_CLIENT_NOTIFICATION";
- case LOCAL_INVALIDATE:
- return "LOCAL_INVALIDATE";
- case LOCAL_DESTROY:
- return "LOCAL_DESTROY";
- case LOCAL_DESTROY_REGION:
- return "LOCAL_DESTROY_REGION";
- case CLOSE_CONNECTION:
- return "CLOSE_CONNECTION";
- case PROCESS_BATCH:
- return "PROCESS_BATCH";
- case REGISTER_INTEREST:
- return "REGISTER_INTEREST";
- case REGISTER_INTEREST_DATA_ERROR:
- return "REGISTER_INTEREST_DATA_ERROR";
- case UNREGISTER_INTEREST:
- return "UNREGISTER_INTEREST";
- case REGISTER_INTEREST_LIST:
- return "REGISTER_INTEREST_LIST";
- case UNREGISTER_INTEREST_LIST:
- return "UNREGISTER_INTEREST_LIST";
- case UNKNOWN_MESSAGE_TYPE_ERROR:
- return "UNKNOWN_MESSAGE_TYPE_ERROR";
- case LOCAL_CREATE:
- return "LOCAL_CREATE";
- case CREATE_REGION:
- return "CREATE_REGION";
- case CREATE_REGION_DATA_ERROR:
- return "CREATE_REGION_DATA_ERROR";
- case MAKE_PRIMARY:
- return "MAKE_PRIMARY";
- case RESPONSE_FROM_PRIMARY:
- return "RESPONSE_FROM_PRIMARY";
- case RESPONSE_FROM_SECONDARY:
- return "RESPONSE_FROM_SECONDARY";
- case QUERY:
- return "QUERY";
- case QUERY_WITH_PARAMETERS:
- return "QUERY_WITH_PARAMETERS";
- case EXECUTECQ_MSG_TYPE:
- return "EXECUTECQ";
- case EXECUTECQ_WITH_IR_MSG_TYPE:
- return "EXECUTECQ_WITH_IR";
- case STOPCQ_MSG_TYPE:
- return "STOPCQ";
- case CLOSECQ_MSG_TYPE:
- return "CLOSECQ";
- case CLOSECLIENTCQS_MSG_TYPE:
- return "CLOSECLIENTCQS";
- case CQDATAERROR_MSG_TYPE:
- return "CQDATAERROR";
- case GETCQSTATS_MSG_TYPE:
- return "GETCQSTATS";
- case MONITORCQ_MSG_TYPE:
- return "MONITORCQ";
- case CQ_EXCEPTION_TYPE:
- return "CQ_EXCEPTION_TYPE";
- case QUERY_DATA_ERROR:
- return "QUERY_DATA_ERROR";
- case CLEAR_REGION:
- return "CLEAR_REGION";
- case CLEAR_REGION_DATA_ERROR:
- return "CLEAR_REGION_DATA_ERROR";
- case CONTAINS_KEY:
- return "CONTAINS_KEY";
- case CONTAINS_KEY_DATA_ERROR:
- return "CONTAINS_KEY_DATA_ERROR";
- case KEY_SET:
- return "KEY_SET";
- case KEY_SET_DATA_ERROR:
- return "KEY_SET_DATA_ERROR";
- case REGISTER_INSTANTIATORS:
- return "REGISTER_INSTANTIATORS";
- case PERIODIC_ACK:
- return "PERIODIC_ACK";
- case CLIENT_READY:
- return "CLIENT_READY";
- case CLIENT_MARKER:
- return "CLIENT_MARKER";
- case INVALIDATE_REGION:
- return "INVALIDATE_REGION";
- case EXECUTE_REGION_FUNCTION:
- return "EXECUTE_REGION_FUNCTION";
- case EXECUTE_REGION_FUNCTION_RESULT:
- return "EXECUTE_REGION_FUNCTION_RESULT";
- case EXECUTE_REGION_FUNCTION_ERROR:
- return "EXECUTE_REGION_FUNCTION_ERROR";
- case EXECUTE_FUNCTION:
- return "EXECUTE_FUNCTION";
- case EXECUTE_FUNCTION_RESULT:
- return "EXECUTE_FUNCTION_RESULT";
- case EXECUTE_FUNCTION_ERROR:
- return "EXECUTE_FUNCTION_ERROR";
- case CLIENT_REGISTER_INTEREST:
- return "CLIENT_REGISTER_INTEREST";
- case CLIENT_UNREGISTER_INTEREST:
- return "CLIENT_UNREGISTER_INTEREST";
- case REGISTER_DATASERIALIZERS:
- return "REGISTER_DATASERIALIZERS";
- case GET_CLIENT_PR_METADATA:
- return "GET_CLIENT_PR_METADATA";
- case RESPONSE_CLIENT_PR_METADATA:
- return "RESPONSE_CLIENT_PR_METADATA";
- case GET_CLIENT_PARTITION_ATTRIBUTES:
- return "GET_CLIENT_PARTITION_ATTRIBUTES";
- case RESPONSE_CLIENT_PARTITION_ATTRIBUTES:
- return "RESPONSE_CLIENT_PARTITION_ATTRIBUTES";
- case GET_CLIENT_PR_METADATA_ERROR:
- return "GET_CLIENT_PR_METADATA_ERROR";
- case GET_CLIENT_PARTITION_ATTRIBUTES_ERROR:
- return "GET_CLIENT_PARTITION_ATTRIBUTES_ERROR";
- case USER_CREDENTIAL_MESSAGE:
- return "USER_CREDENTIAL_MESSAGE";
- case REMOVE_USER_AUTH:
- return "REMOVE_USER_AUTH";
- case EXECUTE_REGION_FUNCTION_SINGLE_HOP:
- return "EXECUTE_REGION_FUNCTION_SINGLE_HOP";
- case GET_PDX_ID_FOR_TYPE:
- return "GET_PDX_ID_FOR_TYPE";
- case GET_PDX_TYPE_BY_ID:
- return "GET_PDX_TYPE_BY_ID";
- case ADD_PDX_TYPE:
- return "ADD_PDX_TYPE";
- case ADD_PDX_ENUM:
- return "ADD_PDX_ENUM";
- case GET_PDX_ID_FOR_ENUM:
- return "GET_PDX_ID_FOR_ENUM";
- case GET_PDX_ENUM_BY_ID:
- return "GET_PDX_ENUM_BY_ID";
- case SIZE:
- return "SIZE";
- case SIZE_ERROR:
- return "SIZE_ERROR";
- case INVALIDATE:
- return "INVALIDATE";
- case INVALIDATE_ERROR:
- return "INVALIDATE_ERROR";
- case COMMIT:
- return "COMMIT";
- case COMMIT_ERROR:
- return "COMMIT_ERROR";
- case ROLLBACK:
- return "ROLLBACK";
- case TX_FAILOVER:
- return "TX_FAILOVER";
- case GET_ENTRY:
- return "GET_ENTRY";
- case TX_SYNCHRONIZATION:
- return "TX_SYNCHRONIZATION";
- case GET_FUNCTION_ATTRIBUTES:
- return "GET_FUNCTION_ATTRIBUTES";
- case GET_ALL_FOR_RI:
- return "GET_ALL_FOR_RI";
- case SERVER_TO_CLIENT_PING:
- return "SERVER_TO_CLIENT_PING";
- case GATEWAY_RECEIVER_COMMAND:
- return "GATEWAY_RECEIVER_COMMAND";
- case TOMBSTONE_OPERATION:
- return "TOMBSTONE_OPERATION";
- case GET_PDX_TYPES:
- return "GET_PDX_TYPES";
- case GET_PDX_ENUMS:
- return "GET_PDX_ENUMS";
- case GET_ALL_70:
- return "GET_ALL_70";
- case GETDURABLECQS_MSG_TYPE:
- return "GETDURABLECQS_MSG_TYPE";
- case GET_DURABLE_CQS_DATA_ERROR:
- return "GET_DURABLE_CQS_DATA_ERROR";
- case GET_ALL_WITH_CALLBACK:
- return "GET_ALL_WITH_CALLBACK";
- case PUT_ALL_WITH_CALLBACK:
- return "PUT_ALL_WITH_CALLBACK";
- case REMOVE_ALL:
- return "REMOVE_ALL";
- default:
- return Integer.toString(type);
+ private static @Nullable MessageType getMessageType(final int id) {
+ if (id == INVALID.id) {
+ return INVALID;
+ } else if (id >= 0 && id < messageTypes.length) {
+ return messageTypes[id];
}
+ return null;
}
+
+ public static @NotNull MessageType valueOf(final int id) {
+ final MessageType messageType = getMessageType(id);
+ if (messageType == null) {
+ throw new EnumConstantNotPresentException(MessageType.class, String.valueOf(id));
+ }
+ return messageType;
+ }
+
+ public static boolean validate(final int id) {
+ final MessageType messageType = getMessageType(id);
+ return null != messageType && INVALID != messageType;
+ }
+
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/BaseCommand.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/BaseCommand.java
index 9566723f74..9000a5503c 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/BaseCommand.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/BaseCommand.java
@@ -573,7 +573,7 @@ public abstract class BaseCommand implements Command {
return e;
}
- protected static void writeException(final @NotNull Message origMsg, final int msgType,
+ protected static void writeException(final @NotNull Message origMsg, final MessageType msgType,
final @NotNull Throwable e, final boolean isSevere,
final @NotNull ServerConnection serverConnection) throws IOException {
Throwable theException = getClientException(serverConnection, e);
@@ -599,7 +599,7 @@ public abstract class BaseCommand implements Command {
}
}
- protected static void writeErrorResponse(final Message origMsg, final int messageType,
+ protected static void writeErrorResponse(final Message origMsg, final MessageType messageType,
final @NotNull ServerConnection serverConnection) throws IOException {
Message errorMsg = serverConnection.getErrorResponseMessage();
errorMsg.setMessageType(messageType);
@@ -610,7 +610,8 @@ public abstract class BaseCommand implements Command {
errorMsg.send(serverConnection);
}
- protected static void writeErrorResponse(final @NotNull Message origMsg, final int messageType,
+ protected static void writeErrorResponse(final @NotNull Message origMsg,
+ final MessageType messageType,
final @NotNull String msg,
final @NotNull ServerConnection serverConnection) throws IOException {
Message errorMsg = serverConnection.getErrorResponseMessage();
@@ -734,7 +735,7 @@ public abstract class BaseCommand implements Command {
}
protected static void writeChunkedErrorResponse(final @NotNull Message origMsg,
- final int messageType, final @NotNull String message,
+ final @NotNull MessageType messageType, final @NotNull String message,
final @NotNull ServerConnection serverConnection) throws IOException {
// Send chunked response header identifying error message
ChunkedMessage chunkedResponseMsg = serverConnection.getChunkedResponseMessage();
@@ -757,7 +758,7 @@ public abstract class BaseCommand implements Command {
}
protected static void writeFunctionResponseException(final @NotNull Message origMsg,
- final int messageType,
+ final @NotNull MessageType messageType,
final @NotNull ServerConnection serverConnection, final @NotNull Throwable exception)
throws IOException {
Throwable e = getClientException(serverConnection, exception);
@@ -792,7 +793,7 @@ public abstract class BaseCommand implements Command {
}
protected static void writeFunctionResponseError(final @NotNull Message origMsg,
- final int messageType, final @NotNull String message,
+ final @NotNull MessageType messageType, final @NotNull String message,
final @NotNull ServerConnection servConn) throws IOException {
ChunkedMessage functionResponseMsg = servConn.getFunctionResponseMessage();
ChunkedMessage chunkedResponseMsg = servConn.getChunkedResponseMessage();
@@ -820,7 +821,7 @@ public abstract class BaseCommand implements Command {
}
protected static void writeKeySetErrorResponse(final @NotNull Message origMsg,
- final int messageType, final @NotNull String message,
+ final @NotNull MessageType messageType, final @NotNull String message,
final @NotNull ServerConnection servConn) throws IOException {
// Send chunked response header identifying error message
ChunkedMessage chunkedResponseMsg = servConn.getKeySetResponseMessage();
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/BaseCommandQuery.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/BaseCommandQuery.java
index 21dcbef39b..b67c3a7a85 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/BaseCommandQuery.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/BaseCommandQuery.java
@@ -14,6 +14,9 @@
*/
package org.apache.geode.internal.cache.tier.sockets;
+import static org.apache.geode.internal.cache.tier.MessageType.CQ_EXCEPTION_TYPE;
+import static org.apache.geode.internal.cache.tier.MessageType.RESPONSE;
+
import java.io.IOException;
import java.util.Collection;
import java.util.List;
@@ -188,7 +191,7 @@ public abstract class BaseCommandQuery extends BaseCommand {
stats.incProcessQueryTime(start - oldStart);
if (sendResults) {
- queryResponseMsg.setMessageType(MessageType.RESPONSE);
+ queryResponseMsg.setMessageType(RESPONSE);
queryResponseMsg.setTransactionId(msg.getTransactionId());
queryResponseMsg.sendHeader();
}
@@ -224,7 +227,7 @@ public abstract class BaseCommandQuery extends BaseCommand {
} else if (result instanceof Integer) {
if (sendResults) {
- queryResponseMsg.setMessageType(MessageType.RESPONSE);
+ queryResponseMsg.setMessageType(RESPONSE);
queryResponseMsg.setTransactionId(msg.getTransactionId());
queryResponseMsg.sendHeader();
writeQueryResponseChunk(result, null, true, servConn);
@@ -251,7 +254,7 @@ public abstract class BaseCommandQuery extends BaseCommand {
if (msg != null && logger.isDebugEnabled()) {
logger.debug(
"{}: ignoring message of type {} from client {} because shutdown occurred during message processing.",
- servConn.getName(), MessageType.getString(msg.getMessageType()), servConn.getProxyID());
+ servConn.getName(), msg.getMessageType(), servConn.getProxyID());
}
servConn.setFlagProcessMessagesAsFalse();
servConn.setClientDisconnectedException(se);
@@ -283,7 +286,7 @@ public abstract class BaseCommandQuery extends BaseCommand {
return results.getCollectionType();
}
- protected void sendCqResponse(int msgType, String msgStr, int txId, Throwable e,
+ protected void sendCqResponse(MessageType msgType, String msgStr, int txId, Throwable e,
ServerConnection servConn) throws IOException {
ChunkedMessage cqMsg = servConn.getChunkedResponseMessage();
if (logger.isDebugEnabled()) {
@@ -291,16 +294,16 @@ public abstract class BaseCommandQuery extends BaseCommand {
}
switch (msgType) {
- case MessageType.REPLY:
+ case REPLY:
cqMsg.setNumberOfParts(1);
break;
- case MessageType.CQDATAERROR_MSG_TYPE:
+ case CQDATAERROR:
logger.warn(msgStr);
cqMsg.setNumberOfParts(1);
break;
- case MessageType.CQ_EXCEPTION_TYPE:
+ case CQ_EXCEPTION_TYPE:
String exMsg = "";
if (e != null) {
exMsg = e.getLocalizedMessage();
@@ -313,9 +316,9 @@ public abstract class BaseCommandQuery extends BaseCommand {
break;
default:
- msgType = MessageType.CQ_EXCEPTION_TYPE;
+ msgType = CQ_EXCEPTION_TYPE;
cqMsg.setNumberOfParts(1);
- msgStr += "Uknown query Exception.";
+ msgStr += "Unknown query Exception.";
break;
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifier.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifier.java
index a6197e024f..2098a8e24f 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifier.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifier.java
@@ -754,7 +754,7 @@ public class CacheClientNotifier {
// Add CQ info.
if (filterInfo.getCQs() != null) {
- for (Map.Entry<Long, Integer> e : filterInfo.getCQs().entrySet()) {
+ for (Map.Entry<Long, MessageType> e : filterInfo.getCQs().entrySet()) {
Long cqID = e.getKey();
String cqName = regionProfile.getRealCqID(cqID);
if (cqName == null) {
@@ -834,7 +834,7 @@ public class CacheClientNotifier {
FilterProfile regionProfile = ((InternalRegion) event.getRegion()).getFilterProfile();
if (event.getOperation().isEntry() && filterInfo.getCQs() != null) {
EntryEvent<?, ?> entryEvent = (EntryEvent<?, ?>) event;
- for (Map.Entry<Long, Integer> e : filterInfo.getCQs().entrySet()) {
+ for (Map.Entry<Long, MessageType> e : filterInfo.getCQs().entrySet()) {
Long cqID = e.getKey();
String cqName = regionProfile.getRealCqID(cqID);
if (cqName != null) {
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientUpdater.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientUpdater.java
index 9b1728e094..edc8aba4c8 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientUpdater.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientUpdater.java
@@ -14,6 +14,14 @@
*/
package org.apache.geode.internal.cache.tier.sockets;
+;
+
+import static org.apache.geode.internal.cache.tier.MessageType.LOCAL_CREATE;
+import static org.apache.geode.internal.cache.tier.MessageType.LOCAL_UPDATE;
+import static org.apache.geode.internal.cache.tier.MessageType.REGISTER_DATASERIALIZERS;
+import static org.apache.geode.internal.cache.tier.MessageType.REGISTER_INSTANTIATORS;
+import static org.apache.geode.internal.cache.tier.MessageType.SERVER_TO_CLIENT_PING;
+
import java.io.IOException;
import java.io.InputStream;
import java.io.InterruptedIOException;
@@ -648,7 +656,7 @@ public class CacheClientUpdater extends LoggingThread implements ClientUpdater,
// object, it will be stored as a CachedDeserializable and
// deserialized only when requested.
- boolean isCreate = clientMessage.getMessageType() == MessageType.LOCAL_CREATE;
+ boolean isCreate = clientMessage.getMessageType() == LOCAL_CREATE;
if (isDebugEnabled) {
logger.debug(
@@ -697,7 +705,7 @@ public class CacheClientUpdater extends LoggingThread implements ClientUpdater,
try {
// Create an event and put the entry
newEvent = EntryEventImpl.create(region,
- clientMessage.getMessageType() == MessageType.LOCAL_CREATE ? Operation.CREATE
+ clientMessage.getMessageType() == LOCAL_CREATE ? Operation.CREATE
: Operation.UPDATE,
key, null /* newValue */, callbackArgument /* callbackArg */, true /* originRemote */,
eventId.getDistributedMember());
@@ -708,7 +716,7 @@ public class CacheClientUpdater extends LoggingThread implements ClientUpdater,
region.basicBridgeClientUpdate(eventId.getDistributedMember(), key, objectValue,
deltaBytes,
isValueObject, callbackArgument,
- clientMessage.getMessageType() == MessageType.LOCAL_CREATE,
+ clientMessage.getMessageType() == LOCAL_CREATE,
qManager.getState().getProcessedMarker() || !isDurableClient, newEvent,
eventId);
@@ -727,7 +735,7 @@ public class CacheClientUpdater extends LoggingThread implements ClientUpdater,
region.basicBridgeClientUpdate(eventId.getDistributedMember(), key, objectValue, null,
isValueObject, callbackArgument,
- clientMessage.getMessageType() == MessageType.LOCAL_CREATE,
+ clientMessage.getMessageType() == LOCAL_CREATE,
qManager.getState().getProcessedMarker() || !isDurableClient, newEvent,
eventId);
@@ -879,7 +887,7 @@ public class CacheClientUpdater extends LoggingThread implements ClientUpdater,
logger.debug("Received message has CQ Event. Number of cqs interested in the event : {}",
numCqsPart.getInt() / 2);
}
- processCqs(clientMessage, partCnt, numCqsPart.getInt(), regionOpType.getInt(),
+ processCqs(clientMessage, partCnt, numCqsPart.getInt(), MessageType.valueOf(regionOpType.getInt()),
key, null);
isOpCompleted = true;
}
@@ -1273,14 +1281,14 @@ public class CacheClientUpdater extends LoggingThread implements ClientUpdater,
* Processes message to invoke CQ listeners.
*/
private int processCqs(Message clientMessage, int startMessagePart, int numCqParts,
- int messageType, Object key, Object value) {
+ MessageType messageType, Object key, Object value) {
return processCqs(clientMessage, startMessagePart, numCqParts, messageType, key, value, null,
null);
}
private int processCqs(Message clientMessage, int startMessagePart, int numCqParts,
- int messageType, Object key, Object value, byte[] delta, EventID eventId) {
- HashMap cqs = new HashMap();
+ MessageType messageType, Object key, Object value, byte[] delta, EventID eventId) {
+ HashMap<String, MessageType> cqs = new HashMap<>();
final boolean isDebugEnabled = logger.isDebugEnabled();
for (int cqCnt = 0; cqCnt < numCqParts;) {
@@ -1294,10 +1302,12 @@ public class CacheClientUpdater extends LoggingThread implements ClientUpdater,
Part cqNamePart = clientMessage.getPart(startMessagePart + cqCnt++);
// Get CQ Op.
Part cqOpPart = clientMessage.getPart(startMessagePart + cqCnt++);
- cqs.put(cqNamePart.getString(), cqOpPart.getInt());
+ final String cqName = cqNamePart.getString();
+ final MessageType cqOp = MessageType.valueOf(cqOpPart.getInt());
+ cqs.put(cqName, cqOp);
if (sb != null) {
- sb.append(cqNamePart.getString()).append(" op=").append(cqOpPart.getInt()).append(" ");
+ sb.append(cqName).append(" op=").append(cqOp).append(" ");
}
} catch (Exception ignore) {
logger.warn(
@@ -1606,7 +1616,7 @@ public class CacheClientUpdater extends LoggingThread implements ClientUpdater,
}
// If the message is a ping, ignore it
- if (clientMessage.getMessageType() == MessageType.SERVER_TO_CLIENT_PING) {
+ if (clientMessage.getMessageType() == SERVER_TO_CLIENT_PING) {
if (isDebugEnabled) {
logger.debug("{}: Received ping", this);
}
@@ -1614,8 +1624,8 @@ public class CacheClientUpdater extends LoggingThread implements ClientUpdater,
}
boolean isDeltaSent = false;
- boolean isCreateOrUpdate = clientMessage.getMessageType() == MessageType.LOCAL_CREATE
- || clientMessage.getMessageType() == MessageType.LOCAL_UPDATE;
+ boolean isCreateOrUpdate = clientMessage.getMessageType() == LOCAL_CREATE
+ || clientMessage.getMessageType() == LOCAL_UPDATE;
if (isCreateOrUpdate) {
isDeltaSent = (Boolean) clientMessage.getPart(2).getObject();
}
@@ -1630,8 +1640,8 @@ public class CacheClientUpdater extends LoggingThread implements ClientUpdater,
EventID eventId = (EventID) eid.getObject();
// no need to verify if the instantiator msg is duplicate or not
- if (clientMessage.getMessageType() != MessageType.REGISTER_INSTANTIATORS
- && clientMessage.getMessageType() != MessageType.REGISTER_DATASERIALIZERS) {
+ if (clientMessage.getMessageType() != REGISTER_INSTANTIATORS
+ && clientMessage.getMessageType() != REGISTER_DATASERIALIZERS) {
if (qManager.getState().verifyIfDuplicate(eventId,
!(isDurableClient || isDeltaSent))) {
continue;
@@ -1647,49 +1657,49 @@ public class CacheClientUpdater extends LoggingThread implements ClientUpdater,
// Process the message
switch (clientMessage.getMessageType()) {
- case MessageType.LOCAL_CREATE:
- case MessageType.LOCAL_UPDATE:
+ case LOCAL_CREATE:
+ case LOCAL_UPDATE:
handleUpdate(clientMessage);
break;
- case MessageType.LOCAL_INVALIDATE:
+ case LOCAL_INVALIDATE:
handleInvalidate(clientMessage);
break;
- case MessageType.LOCAL_DESTROY:
+ case LOCAL_DESTROY:
handleDestroy(clientMessage);
break;
- case MessageType.CLIENT_RE_AUTHENTICATE:
+ case CLIENT_RE_AUTHENTICATE:
handleAuthenticate();
break;
- case MessageType.LOCAL_DESTROY_REGION:
+ case LOCAL_DESTROY_REGION:
handleDestroyRegion(clientMessage);
break;
- case MessageType.CLEAR_REGION:
+ case CLEAR_REGION:
handleClearRegion(clientMessage);
break;
- case MessageType.REGISTER_INSTANTIATORS:
+ case REGISTER_INSTANTIATORS:
handleRegisterInstantiator(clientMessage, eventId);
break;
- case MessageType.REGISTER_DATASERIALIZERS:
+ case REGISTER_DATASERIALIZERS:
handleRegisterDataSerializer(clientMessage, eventId);
break;
- case MessageType.CLIENT_MARKER:
+ case CLIENT_MARKER:
handleMarker(clientMessage);
break;
- case MessageType.INVALIDATE_REGION:
+ case INVALIDATE_REGION:
handleInvalidateRegion(clientMessage);
break;
- case MessageType.CLIENT_REGISTER_INTEREST:
+ case CLIENT_REGISTER_INTEREST:
handleRegisterInterest(clientMessage);
break;
- case MessageType.CLIENT_UNREGISTER_INTEREST:
+ case CLIENT_UNREGISTER_INTEREST:
handleUnregisterInterest(clientMessage);
break;
- case MessageType.TOMBSTONE_OPERATION:
+ case TOMBSTONE_OPERATION:
handleTombstoneOperation(clientMessage);
break;
default:
logger.warn("{}: Received an unsupported message (type={})",
- new Object[] {this, MessageType.getString(clientMessage.getMessageType())});
+ new Object[] {this, clientMessage.getMessageType()});
break;
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ChunkedMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ChunkedMessage.java
index 9f8a0309f8..4199f62aaf 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ChunkedMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ChunkedMessage.java
@@ -190,7 +190,7 @@ public class ChunkedMessage extends Message {
// Set the header and payload fields only after receiving all the
// socket data, providing better message consistency in the face
// of exceptional conditions (e.g. IO problems, timeouts etc.)
- messageType = type;
+ messageType = MessageType.valueOf(type);
numberOfParts = numParts; // Already set in setPayloadFields via setNumberOfParts
transactionId = txid;
}
@@ -338,7 +338,7 @@ public class ChunkedMessage extends Message {
protected void getDSCODEsForWrite() {
final ByteBuffer cb = getCommBuffer();
cb.clear();
- cb.putInt(messageType);
+ cb.putInt(messageType.id);
cb.putInt(numberOfParts);
cb.putInt(transactionId);
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientRegistrationMetadata.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientRegistrationMetadata.java
index 5d1217d3d2..777b25b361 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientRegistrationMetadata.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientRegistrationMetadata.java
@@ -29,6 +29,7 @@ import org.apache.geode.cache.UnsupportedVersionException;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.tier.Command;
import org.apache.geode.internal.cache.tier.CommunicationMode;
+import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.serialization.VersionedDataInputStream;
import org.apache.geode.internal.serialization.VersionedDataOutputStream;
@@ -126,7 +127,7 @@ class ClientRegistrationMetadata {
if (clientVersion == null) {
message = KnownVersion.unsupportedVersionMessage(clientVersionOrdinal);
} else {
- final Map<Integer, Command> commands =
+ final Map<MessageType, Command> commands =
CommandInitializer.getDefaultInstance().get(clientVersion);
if (commands == null) {
message = "No commands registered for version " + clientVersion + ".";
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientUpdateMessageImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientUpdateMessageImpl.java
index 0707af212d..e4b65155fb 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientUpdateMessageImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientUpdateMessageImpl.java
@@ -17,6 +17,8 @@ package org.apache.geode.internal.cache.tier.sockets;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
@@ -24,6 +26,7 @@ import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.logging.log4j.Logger;
+import org.jetbrains.annotations.NotNull;
import org.apache.geode.DataSerializer;
import org.apache.geode.GemFireIOException;
@@ -445,7 +448,8 @@ public class ClientUpdateMessageImpl implements ClientUpdateMessage, Sizeable, N
if (clientHasCq) {
if (message.getMessageType() == MessageType.LOCAL_INVALIDATE) {
// in case of invalidate, set the region operation type.
- message.addIntPart(isCreate() ? MessageType.LOCAL_CREATE : MessageType.LOCAL_UPDATE);
+ message
+ .addIntPart(isCreate() ? MessageType.LOCAL_CREATE.id : MessageType.LOCAL_UPDATE.id);
}
addCqsToMessage(proxyId, message);
}
@@ -471,7 +475,7 @@ public class ClientUpdateMessageImpl implements ClientUpdateMessage, Sizeable, N
if (isInvalidate()) {
// This is to take care when invalidate message is getting sent
// to the Client. See the code for create/update operation.
- message.addIntPart(MessageType.LOCAL_INVALIDATE);
+ message.addIntPart(MessageType.LOCAL_INVALIDATE.id);
}
addCqsToMessage(proxyId, message);
}
@@ -597,7 +601,8 @@ public class ClientUpdateMessageImpl implements ClientUpdateMessage, Sizeable, N
}
}
- synchronized void addClientCq(ClientProxyMembershipID clientId, String cqName, Integer cqEvent) {
+ synchronized void addClientCq(ClientProxyMembershipID clientId, String cqName,
+ MessageType cqEvent) {
if (_clientCqs == null) {
_clientCqs = new ClientCqConcurrentMap();
_hasCqs = true;
@@ -973,76 +978,78 @@ public class ClientUpdateMessageImpl implements ClientUpdateMessage, Sizeable, N
*/
boolean isFull();
- void addToMessage(Message message);
+ void addToMessage(@NotNull Message message);
int size();
+ @NotNull
String[] getNames();
- void add(String name, Integer op);
+ void add(@NotNull String name, @NotNull MessageType op);
- void delete(String name);
+ void delete(@NotNull String name);
}
+
/**
* Contains either zero or one String to int tuples. This is a common case and this impl has a
* much smaller memory footprint than a HashMap with one entry.
*/
public static class CqNameToOpSingleEntry implements CqNameToOp {
private String[] name;
- private int op;
+ private MessageType op;
private static final String[] EMPTY_NAMES_ARRAY = new String[0];
-
- CqNameToOpSingleEntry(String name, Integer op) {
- initializeName(name);
- this.op = op;
+ /**
+ * Construct empty tuple.
+ */
+ CqNameToOpSingleEntry() {
+ setEmpty();
}
- private void initializeName(String name) {
- this.name = new String[] {name};
+ CqNameToOpSingleEntry(final @NotNull String name, final @NotNull MessageType op) {
+ setOne(name, op);
}
@Override
public void sendTo(DataOutput out) throws IOException {
// When serialized it needs to look just as if writeObject was called on a HASH_MAP
out.writeByte(DSCODE.HASH_MAP.toByte());
- int size = size();
+ final int size = size();
InternalDataSerializer.writeArrayLength(size, out);
if (size > 0) {
DataSerializer.writeObject(name[0], out);
- DataSerializer.writeObject(op, out);
+ DataSerializer.writeObject(op.id, out);
}
}
@Override
public boolean isEmpty() {
- return name == null;
+ return size() == 0;
}
@Override
- public void addToMessage(Message message) {
+ public void addToMessage(@NotNull Message message) {
if (!isEmpty()) {
message.addStringPart(name[0], true);
- message.addIntPart(op);
+ message.addIntPart(op.id);
}
}
@Override
public int size() {
- return isEmpty() ? 0 : 1;
+ return name.length;
}
@Override
- public String[] getNames() {
- return (isEmpty()) ? EMPTY_NAMES_ARRAY : name;
+ public @NotNull String @NotNull [] getNames() {
+ return name;
}
@Override
- public void add(String name, Integer op) {
+ public void add(final @NotNull String name, final @NotNull MessageType op) {
if (isEmpty()) {
- this.name = new String[] {name};
- this.op = op;
+ setOne(name, op);
} else if (this.name[0].equals(name)) {
this.op = op;
} else {
@@ -1051,9 +1058,9 @@ public class ClientUpdateMessageImpl implements ClientUpdateMessage, Sizeable, N
}
@Override
- public void delete(String name) {
- if (name.equals(this.name[0])) {
- this.name = null;
+ public void delete(@NotNull String name) {
+ if (!isEmpty() && name.equals(this.name[0])) {
+ setEmpty();
}
}
@@ -1061,12 +1068,24 @@ public class ClientUpdateMessageImpl implements ClientUpdateMessage, Sizeable, N
public boolean isFull() {
return !isEmpty();
}
+
+ private void setEmpty() {
+ name = EMPTY_NAMES_ARRAY;
+ this.op = null;
+ }
+
+ private void setOne(final @NotNull String name, final @NotNull MessageType op) {
+ this.name = new String[] {name};
+ this.op = op;
+ }
+
}
+
/**
* Basically just a ConcurrentHashMap<String, Integer> but limits itself to the CqNameToOp
* interface.
*/
- public static class CqNameToOpHashMap extends ConcurrentHashMap<String, Integer>
+ public static class CqNameToOpHashMap extends ConcurrentHashMap<String, MessageType>
implements CqNameToOp {
public CqNameToOpHashMap(int initialCapacity) {
super(initialCapacity, 1.0f);
@@ -1081,35 +1100,41 @@ public class ClientUpdateMessageImpl implements ClientUpdateMessage, Sizeable, N
public void sendTo(DataOutput out) throws IOException {
// When serialized it needs to look just as if writeObject was called on a HASH_MAP
out.writeByte(DSCODE.HASH_MAP.toByte());
- DataSerializer.writeConcurrentHashMap(this, out);
+ Collection<Entry<String, MessageType>> entrySnapshot = new ArrayList<>(this.entrySet());
+ InternalDataSerializer.writeArrayLength(entrySnapshot.size(), out);
+ for (Map.Entry<String, MessageType> entry : entrySnapshot) {
+ DataSerializer.writeObject(entry.getKey(), out);
+ DataSerializer.writeObject(entry.getValue().id, out);
+ }
+
}
@Override
- public String[] getNames() {
+ public @NotNull String @NotNull [] getNames() {
String[] cqNames = new String[size()];
cqNames = keySet().toArray(cqNames);
return cqNames;
}
@Override
- public void addToMessage(Message message) {
- for (Entry<String, Integer> entry : entrySet()) {
+ public void addToMessage(@NotNull Message message) {
+ for (Entry<String, MessageType> entry : entrySet()) {
// Add CQ Name.
String cq = entry.getKey();
message.addStringPart(cq, true);
// Add CQ Op.
- int op = entry.getValue();
- message.addIntPart(op);
+ MessageType op = entry.getValue();
+ message.addIntPart(op.id);
}
}
@Override
- public void add(String name, Integer op) {
+ public void add(@NotNull String name, @NotNull MessageType op) {
put(name, op);
}
@Override
- public void delete(String name) {
+ public void delete(@NotNull String name) {
remove(name);
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CommandInitializer.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CommandInitializer.java
index dfa2097712..dab9bdf16e 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CommandInitializer.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CommandInitializer.java
@@ -108,8 +108,8 @@ public class CommandInitializer implements CommandRegistry {
return instance;
}
- final Map<KnownVersion, Map<Integer, Command>> unmodifiableRegisteredCommands;
- final LinkedHashMap<KnownVersion, ConcurrentMap<Integer, Command>> modifiableRegisteredCommands;
+ final Map<KnownVersion, Map<MessageType, Command>> unmodifiableRegisteredCommands;
+ final LinkedHashMap<KnownVersion, ConcurrentMap<MessageType, Command>> modifiableRegisteredCommands;
public CommandInitializer() {
modifiableRegisteredCommands = initializeAllCommands();
@@ -117,10 +117,10 @@ public class CommandInitializer implements CommandRegistry {
}
@Override
- public void register(int messageType,
+ public void register(MessageType messageType,
Map<KnownVersion, Command> versionToNewCommand) {
if (!registerCommand(messageType, versionToNewCommand, modifiableRegisteredCommands)) {
- throw new InternalGemFireError(String.format("Message %d was not registered.", messageType));
+ throw new InternalGemFireError(String.format("Message %s was not registered.", messageType));
}
}
@@ -132,7 +132,7 @@ public class CommandInitializer implements CommandRegistry {
* @return immutable {@link Map} for {@link MessageType} to {@link Command}.
*/
@Override
- public Map<Integer, Command> get(final KnownVersion version) {
+ public Map<MessageType, Command> get(final KnownVersion version) {
return unmodifiableRegisteredCommands.get(version);
}
@@ -143,17 +143,18 @@ public class CommandInitializer implements CommandRegistry {
* otherwise false.
* @throws InternalGemFireError if a different command was already registered.
*/
- boolean registerCommand(final int messageType,
+ boolean registerCommand(final MessageType messageType,
final Map<KnownVersion, Command> versionToNewCommand,
- final LinkedHashMap<KnownVersion, ConcurrentMap<Integer, Command>> allCommands) {
+ final LinkedHashMap<KnownVersion, ConcurrentMap<MessageType, Command>> allCommands) {
boolean modified = false;
Command command = null;
- for (Map.Entry<KnownVersion, ConcurrentMap<Integer, Command>> entry : allCommands.entrySet()) {
+ for (Map.Entry<KnownVersion, ConcurrentMap<MessageType, Command>> entry : allCommands
+ .entrySet()) {
KnownVersion version = entry.getKey();
// Get the current set of commands for this version.
- Map<Integer, Command> commandMap = entry.getValue();
+ Map<MessageType, Command> commandMap = entry.getValue();
// See if we have a new command to insert into this map. Otherwise, keep using the command we
// have already read
@@ -165,7 +166,7 @@ public class CommandInitializer implements CommandRegistry {
Command oldCommand = commandMap.get(messageType);
if (oldCommand != null && oldCommand != command) {
throw new InternalGemFireError("Command is already defined int the map for message Type "
- + MessageType.getString(messageType) + ". Old Value=" + commandMap.get(messageType)
+ + messageType + ". Old Value=" + commandMap.get(messageType)
+ ", newValue=" + command + ", version=" + version);
}
commandMap.put(messageType, command);
@@ -175,14 +176,14 @@ public class CommandInitializer implements CommandRegistry {
return modified;
}
- private static LinkedHashMap<KnownVersion, ConcurrentMap<Integer, Command>> initializeAllCommands() {
- final LinkedHashMap<KnownVersion, ConcurrentMap<Integer, Command>> allCommands =
+ private static LinkedHashMap<KnownVersion, ConcurrentMap<MessageType, Command>> initializeAllCommands() {
+ final LinkedHashMap<KnownVersion, ConcurrentMap<MessageType, Command>> allCommands =
new LinkedHashMap<>();
- final ConcurrentMap<Integer, Command> gfe81Commands = buildGfe81Commands();
+ final ConcurrentMap<MessageType, Command> gfe81Commands = buildGfe81Commands();
allCommands.put(KnownVersion.GFE_81, gfe81Commands);
- final ConcurrentMap<Integer, Command> gfe90Commands =
+ final ConcurrentMap<MessageType, Command> gfe90Commands =
buildGfe90Commands(allCommands.get(KnownVersion.GFE_81));
allCommands.put(KnownVersion.GFE_90, gfe90Commands);
allCommands.put(KnownVersion.GEODE_1_1_0, gfe90Commands);
@@ -194,7 +195,7 @@ public class CommandInitializer implements CommandRegistry {
allCommands.put(KnownVersion.GEODE_1_6_0, gfe90Commands);
allCommands.put(KnownVersion.GEODE_1_7_0, gfe90Commands);
- final ConcurrentMap<Integer, Command> geode18Commands =
+ final ConcurrentMap<MessageType, Command> geode18Commands =
buildGeode18Commands(allCommands.get(KnownVersion.GEODE_1_7_0));
allCommands.put(KnownVersion.GEODE_1_8_0, geode18Commands);
allCommands.put(KnownVersion.GEODE_1_9_0, geode18Commands);
@@ -213,36 +214,36 @@ public class CommandInitializer implements CommandRegistry {
return allCommands;
}
- private static ConcurrentMap<Integer, Command> buildGeode18Commands(
- final ConcurrentMap<Integer, Command> baseCommands) {
- final ConcurrentMap<Integer, Command> commands = new ConcurrentHashMap<>(baseCommands);
+ private static ConcurrentMap<MessageType, Command> buildGeode18Commands(
+ final ConcurrentMap<MessageType, Command> baseCommands) {
+ final ConcurrentMap<MessageType, Command> commands = new ConcurrentHashMap<>(baseCommands);
initializeGeode18Commands(commands);
return commands;
}
- private static ConcurrentMap<Integer, Command> buildGfe90Commands(
- final ConcurrentMap<Integer, Command> baseCommands) {
- final ConcurrentMap<Integer, Command> commands = new ConcurrentHashMap<>(baseCommands);
+ private static ConcurrentMap<MessageType, Command> buildGfe90Commands(
+ final ConcurrentMap<MessageType, Command> baseCommands) {
+ final ConcurrentMap<MessageType, Command> commands = new ConcurrentHashMap<>(baseCommands);
initializeGfe90Commands(commands);
return commands;
}
- private static ConcurrentMap<Integer, Command> buildGfe81Commands() {
- final ConcurrentMap<Integer, Command> commands = new ConcurrentHashMap<>();
+ private static ConcurrentMap<MessageType, Command> buildGfe81Commands() {
+ final ConcurrentMap<MessageType, Command> commands = new ConcurrentHashMap<>();
initializeGfe81Commands(commands);
return commands;
}
- static void initializeGeode18Commands(final Map<Integer, Command> commands) {
+ static void initializeGeode18Commands(final Map<MessageType, Command> commands) {
commands.put(MessageType.EXECUTE_REGION_FUNCTION, ExecuteRegionFunctionGeode18.getCommand());
}
- static void initializeGfe90Commands(final Map<Integer, Command> commands) {
+ static void initializeGfe90Commands(final Map<MessageType, Command> commands) {
commands.put(MessageType.QUERY_WITH_PARAMETERS, QueryWithParametersGeode10.getCommand());
commands.put(MessageType.QUERY, QueryGeode10.getCommand());
}
- static void initializeGfe81Commands(final Map<Integer, Command> commands) {
+ static void initializeGfe81Commands(final Map<MessageType, Command> commands) {
commands.put(MessageType.PING, Ping.getCommand());
commands.put(MessageType.QUERY, Query.getCommand());
commands.put(MessageType.CLEAR_REGION, ClearRegion.getCommand());
@@ -309,11 +310,12 @@ public class CommandInitializer implements CommandRegistry {
commands.put(MessageType.REMOVE_ALL, RemoveAll.getCommand());
}
- static Map<KnownVersion, Map<Integer, Command>> makeUnmodifiable(
- final Map<KnownVersion, ConcurrentMap<Integer, Command>> modifiableMap) {
- final Map<KnownVersion, Map<Integer, Command>> unmodifiableMap =
+ static Map<KnownVersion, Map<MessageType, Command>> makeUnmodifiable(
+ final Map<KnownVersion, ConcurrentMap<MessageType, Command>> modifiableMap) {
+ final Map<KnownVersion, Map<MessageType, Command>> unmodifiableMap =
new LinkedHashMap<>(modifiableMap.size());
- for (Map.Entry<KnownVersion, ConcurrentMap<Integer, Command>> e : modifiableMap.entrySet()) {
+ for (Map.Entry<KnownVersion, ConcurrentMap<MessageType, Command>> e : modifiableMap
+ .entrySet()) {
unmodifiableMap.put(e.getKey(), unmodifiableMap(e.getValue()));
}
return unmodifiableMap(unmodifiableMap);
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CommandRegistry.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CommandRegistry.java
index b0b948cd7d..ed51b37a82 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CommandRegistry.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CommandRegistry.java
@@ -19,6 +19,7 @@ import java.util.Map;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.internal.cache.tier.Command;
+import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.serialization.KnownVersion;
/**
@@ -38,7 +39,7 @@ public interface CommandRegistry {
* @throws InternalGemFireError if a different command is already registered for given version or
* if no command was added to the registry.
*/
- void register(int messageType, Map<KnownVersion, Command> versionToNewCommand);
+ void register(MessageType messageType, Map<KnownVersion, Command> versionToNewCommand);
- Map<Integer, Command> get(KnownVersion version);
+ Map<MessageType, Command> get(KnownVersion version);
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/HAEventWrapper.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/HAEventWrapper.java
index ab8ec87988..5cce6ed734 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/HAEventWrapper.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/HAEventWrapper.java
@@ -28,6 +28,7 @@ import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.cache.Conflatable;
import org.apache.geode.internal.cache.EnumListenerEvent;
import org.apache.geode.internal.cache.EventID;
+import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.sockets.ClientUpdateMessageImpl.ClientCqConcurrentMap;
import org.apache.geode.internal.cache.tier.sockets.ClientUpdateMessageImpl.CqNameToOp;
import org.apache.geode.internal.cache.tier.sockets.ClientUpdateMessageImpl.CqNameToOpHashMap;
@@ -330,16 +331,17 @@ public class HAEventWrapper implements Conflatable, DataSerializableFixedID, Siz
"The value of a ConcurrentHashMap is not allowed to be null.");
} else if (cqNamesSize == 1) {
String cqNamesKey = DataSerializer.readObject(in);
- Integer cqNamesValue = DataSerializer.<Integer>readObject(in);
- value = new CqNameToOpSingleEntry(cqNamesKey, cqNamesValue);
+ Integer cqNamesValue = DataSerializer.readObject(in);
+ value =
+ new CqNameToOpSingleEntry(cqNamesKey, MessageType.valueOf(cqNamesValue));
} else if (cqNamesSize == 0) {
- value = new CqNameToOpSingleEntry(null, 0);
+ value = new CqNameToOpSingleEntry();
} else {
value = new CqNameToOpHashMap(cqNamesSize);
for (int j = 0; j < cqNamesSize; j++) {
String cqNamesKey = DataSerializer.readObject(in);
- Integer cqNamesValue = DataSerializer.<Integer>readObject(in);
- value.add(cqNamesKey, cqNamesValue);
+ Integer cqNamesValue = DataSerializer.readObject(in);
+ value.add(cqNamesKey, MessageType.valueOf(cqNamesValue));
}
}
} else if (typeByte == DSCODE.NULL.toByte()) {
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/Message.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/Message.java
index 055924832c..d5e1f7ca5e 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/Message.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/Message.java
@@ -27,6 +27,7 @@ import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import org.apache.logging.log4j.Logger;
+import org.jetbrains.annotations.NotNull;
import org.apache.geode.SerializationException;
import org.apache.geode.annotations.Immutable;
@@ -81,7 +82,7 @@ import org.apache.geode.util.internal.GeodeGlossary;
public class Message {
// Tentative workaround to avoid OOM stated in #46754.
- public static final ThreadLocal<Integer> MESSAGE_TYPE = new ThreadLocal<>();
+ public static final ThreadLocal<MessageType> MESSAGE_TYPE = new ThreadLocal<>();
public static final String MAX_MESSAGE_SIZE_PROPERTY =
GeodeGlossary.GEMFIRE_PREFIX + "client.max-message-size";
@@ -141,12 +142,12 @@ public class Message {
private static final int maxMessageSize =
Integer.getInteger(MAX_MESSAGE_SIZE_PROPERTY, DEFAULT_MAX_MESSAGE_SIZE);
- protected int messageType;
+ protected MessageType messageType;
private int payloadLength = 0;
- int numberOfParts = 0;
+ int numberOfParts;
protected int transactionId = TXManagerImpl.NOTX;
int currentPart = 0;
- private Part[] partsList = null;
+ private Part[] partsList;
private ByteBuffer cachedCommBuffer;
protected Socket socket = null;
private SocketChannel socketChannel = null;
@@ -193,13 +194,9 @@ public class Message {
return (byte[]) securePart.getObject();
}
- public void setMessageType(int msgType) {
+ public void setMessageType(final @NotNull MessageType messageType) {
messageModified = true;
- if (!MessageType.validate(msgType)) {
- throw new IllegalArgumentException(
- "Invalid MessageType");
- }
- messageType = msgType;
+ this.messageType = messageType;
}
public void setVersion(KnownVersion clientVersion) {
@@ -444,7 +441,7 @@ public class Message {
currentPart++;
}
- public int getMessageType() {
+ public MessageType getMessageType() {
return messageType;
}
@@ -530,7 +527,7 @@ public class Message {
if (isRetry) {
flagsByte |= MESSAGE_IS_RETRY;
}
- getCommBuffer().putInt(messageType).putInt(msgLen).putInt(numberOfParts)
+ getCommBuffer().putInt(messageType.id).putInt(msgLen).putInt(numberOfParts)
.putInt(transactionId).put(flagsByte);
}
@@ -781,7 +778,7 @@ public class Message {
isRetry = (bits & MESSAGE_IS_RETRY) != 0;
bits &= MESSAGE_IS_RETRY_MASK;
flags = bits;
- messageType = type;
+ messageType = MessageType.valueOf(type);
readPayloadFields(numParts, len);
@@ -853,15 +850,15 @@ public class Message {
len, numParts));
}
- Integer msgType = MESSAGE_TYPE.get();
- if (msgType != null && msgType == MessageType.PING) {
+ MessageType msgType = MESSAGE_TYPE.get();
+ if (msgType == MessageType.PING) {
// set it to null right away.
MESSAGE_TYPE.set(null);
// Some number which will not throw OOM but still be acceptable for a ping operation.
int pingParts = 10;
if (numParts > pingParts) {
throw new IOException("Part length ( " + numParts + " ) is inconsistent for "
- + MessageType.getString(msgType) + " operation.");
+ + msgType + " operation.");
}
}
@@ -1049,7 +1046,7 @@ public class Message {
@Override
public String toString() {
StringBuilder sb = new StringBuilder();
- sb.append("type=").append(MessageType.getString(messageType));
+ sb.append("type=").append(messageType);
sb.append("; payloadLength=").append(payloadLength);
sb.append("; numberOfParts=").append(numberOfParts);
sb.append("; hasSecurePart=").append(isSecureMode());
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnection.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnection.java
index 004ebd4730..debcbb7646 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnection.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnection.java
@@ -117,7 +117,7 @@ public class ServerConnection implements Runnable {
public static boolean allowInternalMessagesWithoutCredentials =
!Boolean.getBoolean(DISALLOW_INTERNAL_MESSAGES_WITHOUT_CREDENTIALS_NAME);
- private Map<Integer, Command> commands;
+ private Map<MessageType, Command> commands;
protected final SecurityService securityService;
@@ -477,7 +477,7 @@ public class ServerConnection implements Runnable {
}
}
- protected Map<Integer, Command> getCommands() {
+ protected Map<MessageType, Command> getCommands() {
return commands;
}
@@ -839,7 +839,7 @@ public class ServerConnection implements Runnable {
if (!processMessages || crHelper.isShutdown()) {
if (logger.isDebugEnabled()) {
logger.debug("{} ignoring message of type {} from client {} due to shutdown.",
- getName(), MessageType.getString(message.getMessageType()), proxyId);
+ getName(), message.getMessageType(), proxyId);
}
return;
}
@@ -858,7 +858,7 @@ public class ServerConnection implements Runnable {
if (logger.isTraceEnabled()) {
logger.trace("{} received {} with txid {}", getName(),
- MessageType.getString(message.getMessageType()), message.getTransactionId());
+ message.getMessageType(), message.getTransactionId());
if (message.getTransactionId() < -1) {
message.setTransactionId(-1);
}
@@ -910,7 +910,7 @@ public class ServerConnection implements Runnable {
}
long uniqueId = getUniqueId();
- String messageType = MessageType.getString(requestMessage.getMessageType());
+ String messageType = requestMessage.getMessageType().toString();
if (uniqueId == 0 || uniqueId == -1) {
logger.debug("No unique ID yet. {}, {}", messageType, getName());
return null;
@@ -1105,7 +1105,7 @@ public class ServerConnection implements Runnable {
commands = CommandInitializer.getDefaultInstance().get(clientVersion);
}
- private Command getCommand(Integer messageType) {
+ private Command getCommand(MessageType messageType) {
return commands.get(messageType);
}
@@ -1275,13 +1275,13 @@ public class ServerConnection implements Runnable {
if (AcceptorImpl.isAuthenticationRequired() && logger.isDebugEnabled()) {
logger.debug(
"ServerConnection.updateAndGetSecurityPart() not adding security part for message type {}",
- MessageType.getString(requestMessage.messageType));
+ requestMessage.messageType);
}
return null;
}
public boolean isInternalMessage(Message message, boolean allowOldInternalMessages) {
- int messageType = message.getMessageType();
+ final MessageType messageType = message.getMessageType();
boolean isInternalMessage = messageType == MessageType.PING
|| messageType == MessageType.REQUEST_EVENT_VALUE || messageType == MessageType.MAKE_PRIMARY
|| messageType == MessageType.REMOVE_USER_AUTH || messageType == MessageType.CLIENT_READY
@@ -1295,8 +1295,8 @@ public class ServerConnection implements Runnable {
// we allow older clients to not send credentials for a handful of messages if and only if a
// system property is set. This allows a rolling upgrade to be performed.
if (!isInternalMessage && allowOldInternalMessages) {
- isInternalMessage = messageType == MessageType.GETCQSTATS_MSG_TYPE
- || messageType == MessageType.MONITORCQ_MSG_TYPE
+ isInternalMessage = messageType == MessageType.GETCQSTATS
+ || messageType == MessageType.MONITORCQ
|| messageType == MessageType.REGISTER_DATASERIALIZERS
|| messageType == MessageType.REGISTER_INSTANTIATORS
|| messageType == MessageType.ADD_PDX_TYPE
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerResponseMatrix.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerResponseMatrix.java
index 8e347b6462..18ed37f129 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerResponseMatrix.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerResponseMatrix.java
@@ -14,6 +14,8 @@
*/
package org.apache.geode.internal.cache.tier.sockets;
+;
+
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.RegionEntry;
import org.apache.geode.internal.cache.Token;
@@ -75,7 +77,7 @@ public class ServerResponseMatrix {
}
public static boolean checkForValidStateAfterNotification(LocalRegion region, Object key,
- int operation) {
+ MessageType operation) {
// 4 nonexistent/destroyed , valid , invalid,local-invalid ,
// 4 create , update , invalidate, destroy
@@ -107,16 +109,16 @@ public class ServerResponseMatrix {
int notificationState = -1;
switch (operation) {
- case MessageType.LOCAL_CREATE:
+ case LOCAL_CREATE:
notificationState = 0;
break;
- case MessageType.LOCAL_UPDATE:
+ case LOCAL_UPDATE:
notificationState = 1;
break;
- case MessageType.LOCAL_INVALIDATE:
+ case LOCAL_INVALIDATE:
notificationState = 2;
break;
- case MessageType.LOCAL_DESTROY:
+ case LOCAL_DESTROY:
notificationState = 3;
break;
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerSideHandshakeFactory.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerSideHandshakeFactory.java
index dca14be5e1..0db449f40f 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerSideHandshakeFactory.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerSideHandshakeFactory.java
@@ -29,6 +29,7 @@ import org.apache.geode.cache.VersionException;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.internal.cache.tier.Command;
import org.apache.geode.internal.cache.tier.CommunicationMode;
+import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.ServerSideHandshake;
import org.apache.geode.internal.security.SecurityService;
import org.apache.geode.internal.serialization.KnownVersion;
@@ -79,7 +80,7 @@ class ServerSideHandshakeFactory {
final KnownVersion clientVersion = Versioning.getKnownVersionOrDefault(
Versioning.getVersion(clientVersionOrdinal), KnownVersion.TEST_VERSION);
if (clientVersion != KnownVersion.TEST_VERSION) {
- final Map<Integer, Command> commands =
+ final Map<MessageType, Command> commands =
CommandInitializer.getDefaultInstance().get(clientVersion);
if (commands != null) {
return clientVersion;
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Default.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Default.java
index 2e102ee313..db7dc7838b 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Default.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Default.java
@@ -46,7 +46,7 @@ public class Default extends BaseCommand {
logger.fatal("{}: Unknown message type ({}) with tx: {} from {}",
new Object[] {serverConnection.getName(),
- MessageType.getString(clientMessage.getMessageType()),
+ clientMessage.getMessageType(),
clientMessage.getTransactionId(),
serverConnection.getSocketString()});
writeErrorResponse(clientMessage, MessageType.UNKNOWN_MESSAGE_TYPE_ERROR, serverConnection);
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/DestroyRegion.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/DestroyRegion.java
index c5e55c4c13..7e864f016c 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/DestroyRegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/DestroyRegion.java
@@ -78,7 +78,7 @@ public class DestroyRegion extends BaseCommand {
if (logger.isDebugEnabled()) {
logger.debug(
"{} ignoring message of type {} from client {} because shutdown occurred during message processing.",
- serverConnection.getName(), MessageType.getString(clientMessage.getMessageType()),
+ serverConnection.getName(), clientMessage.getMessageType(),
serverConnection.getProxyID());
}
@@ -147,7 +147,7 @@ public class DestroyRegion extends BaseCommand {
if (logger.isDebugEnabled()) {
logger.debug(
"{} ignoring message of type {} from client {} because shutdown occurred during message processing.",
- serverConnection.getName(), MessageType.getString(clientMessage.getMessageType()),
+ serverConnection.getName(), clientMessage.getMessageType(),
serverConnection.getProxyID());
}
serverConnection.setFlagProcessMessagesAsFalse();
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteFunction70.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteFunction70.java
index 53795fb189..f529eed5d9 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteFunction70.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteFunction70.java
@@ -476,7 +476,7 @@ public class ExecuteFunction70 extends BaseCommand {
}
interface ServerToClientFunctionResultSender65Factory {
- ServerToClientFunctionResultSender65 create(ChunkedMessage msg, int messageType,
+ ServerToClientFunctionResultSender65 create(ChunkedMessage msg, MessageType messageType,
ServerConnection sc, Function function, ExecuteFunctionOperationContext authzContext);
}
@@ -488,7 +488,7 @@ public class ExecuteFunction70 extends BaseCommand {
private static class DefaultServerToClientFunctionResultSender65Factory
implements ServerToClientFunctionResultSender65Factory {
@Override
- public ServerToClientFunctionResultSender65 create(ChunkedMessage msg, int messageType,
+ public ServerToClientFunctionResultSender65 create(ChunkedMessage msg, MessageType messageType,
ServerConnection sc, Function function, ExecuteFunctionOperationContext authzContext) {
return new ServerToClientFunctionResultSender65(msg, messageType, sc, function, authzContext);
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunction66.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunction66.java
index 80f6cfb636..722904ee7c 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunction66.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunction66.java
@@ -443,7 +443,7 @@ public class ExecuteRegionFunction66 extends BaseCommand {
}
}
- protected static void writeFunctionResponseException(Message origMsg, int messageType,
+ protected static void writeFunctionResponseException(Message origMsg, MessageType messageType,
String message, ServerConnection serverConnection, Throwable e) throws IOException {
ChunkedMessage functionResponseMsg = serverConnection.getFunctionResponseMessage();
ChunkedMessage chunkedResponseMsg = serverConnection.getChunkedResponseMessage();
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunctionSingleHop.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunctionSingleHop.java
index 08152696f1..c13a08808d 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunctionSingleHop.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunctionSingleHop.java
@@ -348,7 +348,7 @@ public class ExecuteRegionFunctionSingleHop extends BaseCommand {
}
}
- protected static void writeFunctionResponseException(Message origMsg, int messageType,
+ protected static void writeFunctionResponseException(Message origMsg, MessageType messageType,
String message, ServerConnection serverConnection, Throwable e) throws IOException {
ChunkedMessage functionResponseMsg = serverConnection.getFunctionResponseMessage();
ChunkedMessage chunkedResponseMsg = serverConnection.getChunkedResponseMessage();
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GatewayReceiverCommand.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GatewayReceiverCommand.java
index 5ebd08f747..14e3ccf061 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GatewayReceiverCommand.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GatewayReceiverCommand.java
@@ -614,7 +614,7 @@ public class GatewayReceiverCommand extends BaseCommand {
if (logger.isDebugEnabled()) {
logger.debug(
"{} ignoring message of type {} from client {} because shutdown occurred during message processing.",
- serverConnection.getName(), MessageType.getString(clientMessage.getMessageType()),
+ serverConnection.getName(), clientMessage.getMessageType(),
serverConnection.getProxyID());
}
serverConnection.setFlagProcessMessagesAsFalse();
diff --git a/geode-core/src/test/java/org/apache/geode/cache/client/internal/AbstractOpWithTimeoutTest.java b/geode-core/src/test/java/org/apache/geode/cache/client/internal/AbstractOpWithTimeoutTest.java
index 1ae144065f..6ec6cc5d70 100644
--- a/geode-core/src/test/java/org/apache/geode/cache/client/internal/AbstractOpWithTimeoutTest.java
+++ b/geode-core/src/test/java/org/apache/geode/cache/client/internal/AbstractOpWithTimeoutTest.java
@@ -15,6 +15,7 @@
package org.apache.geode.cache.client.internal;
+import static org.apache.geode.internal.cache.tier.MessageType.REQUEST;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.CALLS_REAL_METHODS;
import static org.mockito.Mockito.mock;
@@ -27,11 +28,13 @@ import java.net.Socket;
import org.junit.Test;
+import org.apache.geode.internal.cache.tier.MessageType;
+
public class AbstractOpWithTimeoutTest {
private abstract static class AbstractTestOpWithTimeout extends AbstractOpWithTimeout {
- public AbstractTestOpWithTimeout(int msgType, int msgParts, int timeoutMs) {
+ public AbstractTestOpWithTimeout(MessageType msgType, int msgParts, int timeoutMs) {
super(msgType, msgParts, timeoutMs);
}
@@ -47,7 +50,7 @@ public class AbstractOpWithTimeoutTest {
@Test
public void attemptWillSetAndResetSoTimeout() throws Exception {
final AbstractOpWithTimeout mockOp = mock(AbstractTestOpWithTimeout.class,
- withSettings().useConstructor(0, 0, 123).defaultAnswer(CALLS_REAL_METHODS));
+ withSettings().useConstructor(REQUEST, 0, 123).defaultAnswer(CALLS_REAL_METHODS));
final Socket socket = mock(Socket.class);
when(socket.getSoTimeout()).thenReturn(456);
@@ -66,7 +69,7 @@ public class AbstractOpWithTimeoutTest {
@Test
public void attemptWontSetAndResetSoTimeout() throws Exception {
final AbstractOpWithTimeout mockOp = mock(AbstractTestOpWithTimeout.class,
- withSettings().useConstructor(0, 0, 123).defaultAnswer(CALLS_REAL_METHODS));
+ withSettings().useConstructor(REQUEST, 0, 123).defaultAnswer(CALLS_REAL_METHODS));
final Socket socket = mock(Socket.class);
when(socket.getSoTimeout()).thenReturn(123);
@@ -83,7 +86,7 @@ public class AbstractOpWithTimeoutTest {
@Test
public void getTimeoutMs() {
final AbstractOpWithTimeout mockOp = mock(AbstractOpWithTimeout.class,
- withSettings().useConstructor(0, 0, 123).defaultAnswer(CALLS_REAL_METHODS));
+ withSettings().useConstructor(REQUEST, 0, 123).defaultAnswer(CALLS_REAL_METHODS));
assertThat(mockOp.getTimeoutMs()).isEqualTo(123);
}
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/DistributedCacheOperationTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/DistributedCacheOperationTest.java
index ff4bb2a7c9..15f3ace22b 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/DistributedCacheOperationTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/DistributedCacheOperationTest.java
@@ -115,7 +115,7 @@ public class DistributedCacheOperationTest {
@Test
public void testDoRemoveDestroyTokensFromCqResultKeys() {
Object key = new Object();
- HashMap<Long, Integer> hashMap = new HashMap<>();
+ HashMap<Long, MessageType> hashMap = new HashMap<>();
hashMap.put(1L, MessageType.LOCAL_DESTROY);
EntryEventImpl baseEvent = mock(EntryEventImpl.class);
ServerCQ serverCQ = mock(ServerCQ.class);
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/FilterInfoTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/FilterInfoTest.java
index 090bd99cbe..34b883d60c 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/FilterInfoTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/FilterInfoTest.java
@@ -25,6 +25,7 @@ import org.junit.Test;
import org.apache.geode.internal.HeapDataOutputStream;
import org.apache.geode.internal.cache.FilterRoutingInfo.FilterInfo;
+import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.serialization.ByteArrayDataInput;
import org.apache.geode.internal.serialization.KnownVersion;
@@ -35,9 +36,9 @@ public class FilterInfoTest {
@Test
public void validateSerialization() throws IOException, ClassNotFoundException {
FilterInfo serialized = new FilterInfo();
- HashMap<Long, Integer> cqs = new HashMap<>();
- cqs.put(1L, 1);
- cqs.put(2L, 2);
+ HashMap<Long, MessageType> cqs = new HashMap<>();
+ cqs.put(1L, MessageType.RESPONSE);
+ cqs.put(2L, MessageType.EXCEPTION);
serialized.setCQs(cqs);
Set<Long> clients = new HashSet<>();
clients.add(1L);
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/execute/ServerToClientFunctionResultSender65JUnitTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/execute/ServerToClientFunctionResultSender65JUnitTest.java
index 7f2e216d64..9ff2240711 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/execute/ServerToClientFunctionResultSender65JUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/execute/ServerToClientFunctionResultSender65JUnitTest.java
@@ -19,6 +19,7 @@ import static org.mockito.Mockito.when;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.operations.ExecuteFunctionOperationContext;
+import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.sockets.AcceptorImpl;
import org.apache.geode.internal.cache.tier.sockets.ChunkedMessage;
import org.apache.geode.internal.cache.tier.sockets.ServerConnection;
@@ -37,7 +38,8 @@ public class ServerToClientFunctionResultSender65JUnitTest
when(serverConnection.getAcceptor()).thenReturn(acceptor);
when(acceptor.isSelector()).thenReturn(true);
when(acceptor.isRunning()).thenReturn(true);
- return new ServerToClientFunctionResultSender65(msg, 1, serverConnection, function,
+ return new ServerToClientFunctionResultSender65(msg, MessageType.RESPONSE, serverConnection,
+ function,
executeFunctionOperationContext);
}
}
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/execute/ServerToClientFunctionResultSenderJUnitTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/execute/ServerToClientFunctionResultSenderJUnitTest.java
index c3701731f9..d44adcc329 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/execute/ServerToClientFunctionResultSenderJUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/execute/ServerToClientFunctionResultSenderJUnitTest.java
@@ -32,6 +32,7 @@ import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.tier.CachedRegionHelper;
+import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.sockets.AcceptorImpl;
import org.apache.geode.internal.cache.tier.sockets.ChunkedMessage;
import org.apache.geode.internal.cache.tier.sockets.ServerConnection;
@@ -56,7 +57,8 @@ public class ServerToClientFunctionResultSenderJUnitTest {
when(acceptor.isSelector()).thenReturn(true);
when(acceptor.isRunning()).thenReturn(true);
- return new ServerToClientFunctionResultSender(msg, 1, serverConnection, function,
+ return new ServerToClientFunctionResultSender(msg, MessageType.RESPONSE, serverConnection,
+ function,
executeFunctionOperationContext);
}
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/MessageTypeTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/MessageTypeTest.java
new file mode 100644
index 0000000000..d37430a566
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/MessageTypeTest.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.geode.internal.cache.tier;
+
+import static org.apache.geode.internal.cache.tier.MessageType.validate;
+import static org.apache.geode.internal.cache.tier.MessageType.valueOf;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import org.junit.jupiter.api.Test;
+
+class MessageTypeTest {
+
+ @Test
+ void valueOfInvalidIdReturnsInvalidEnumValue() {
+ assertThat(valueOf(-1)).isSameAs(MessageType.INVALID);
+ }
+
+ @Test
+ void valueOfValidIdReturnsEnumValue() {
+ assertThat(valueOf(0)).isSameAs(MessageType.REQUEST);
+ assertThat(valueOf(42)).isSameAs(MessageType.EXECUTECQ);
+ }
+
+ @Test
+ void valueOfGapIdWillThrowException() {
+ assertThatThrownBy(() -> valueOf(57)).isInstanceOf(EnumConstantNotPresentException.class);
+ }
+
+ @Test
+ void valueOfOutOfRangeIdThrowsException() {
+ assertThatThrownBy(() -> valueOf(200)).isInstanceOf(EnumConstantNotPresentException.class);
+ assertThatThrownBy(() -> valueOf(-200)).isInstanceOf(EnumConstantNotPresentException.class);
+ }
+
+ @Test
+ void validateOutOfRangeIdReturnsFalse() {
+ assertThat(validate(-200)).isFalse();
+ assertThat(validate(200)).isFalse();
+ }
+
+ @Test
+ void validateGapIdReturnsFalse() {
+ assertThat(validate(57)).isFalse();
+ }
+
+ @Test
+ void validateInvalidIdReturnsFalse() {
+ assertThat(validate(-1)).isFalse();
+ }
+
+ @Test
+ void validateValidIdReturnsTrue() {
+ assertThat(validate(0)).isTrue();
+ assertThat(validate(42)).isTrue();
+ }
+
+}
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifierTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifierTest.java
index c0cde17ad7..88da44f4fb 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifierTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifierTest.java
@@ -66,6 +66,7 @@ import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.InternalCacheEvent;
import org.apache.geode.internal.cache.InternalRegion;
import org.apache.geode.internal.cache.RegionQueueException;
+import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.sockets.ClientRegistrationEventQueueManager.ClientRegistrationEventQueue;
import org.apache.geode.internal.statistics.StatisticsClock;
import org.apache.geode.internal.statistics.StatisticsManager;
@@ -340,8 +341,8 @@ public class CacheClientNotifierTest {
InternalCacheEvent internalCacheEvent = mock(InternalCacheEvent.class);
ServerCQ serverCQ = mock(ServerCQ.class);
- HashMap<Long, Integer> cqs = new HashMap<>();
- cqs.put(CQ_ID, 123);
+ HashMap<Long, MessageType> cqs = new HashMap<>();
+ cqs.put(CQ_ID, MessageType.RESPONSE);
when(filterInfo.getCQs())
.thenReturn(cqs);
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/ClientUpdateMessageImplTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/ClientUpdateMessageImplTest.java
index 6d6095dd7d..0b4a8f00a9 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/ClientUpdateMessageImplTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/ClientUpdateMessageImplTest.java
@@ -16,28 +16,38 @@
package org.apache.geode.internal.cache.tier.sockets;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.Assert.assertNotNull;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
import static org.mockito.Mockito.withSettings;
+import java.io.DataOutput;
+import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.Future;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
import org.apache.geode.CopyHelper;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.DurableClientAttributes;
import org.apache.geode.internal.cache.EnumListenerEvent;
import org.apache.geode.internal.cache.LocalRegion;
+import org.apache.geode.internal.cache.tier.MessageType;
+import org.apache.geode.internal.cache.tier.sockets.ClientUpdateMessageImpl.CqNameToOpSingleEntry;
+import org.apache.geode.internal.serialization.DSCODE;
import org.apache.geode.internal.statistics.StatisticsClock;
import org.apache.geode.test.fake.Fakes;
-import org.apache.geode.test.junit.rules.ExecutorServiceRule;
public class ClientUpdateMessageImplTest implements Serializable {
private final ClientProxyMembershipID client1 = mock(ClientProxyMembershipID.class);
@@ -45,23 +55,27 @@ public class ClientUpdateMessageImplTest implements Serializable {
private final ClientUpdateMessageImpl.ClientCqConcurrentMap clientCqs =
new ClientUpdateMessageImpl.ClientCqConcurrentMap();
- @Rule
- public ExecutorServiceRule executorService = new ExecutorServiceRule();
+ public ExecutorService executorService = Executors.newCachedThreadPool();
- @Before
- public void setUp() {
+ @BeforeEach
+ void beforeEach() {
ClientUpdateMessageImpl.CqNameToOpHashMap cqs1 =
new ClientUpdateMessageImpl.CqNameToOpHashMap(2);
- cqs1.add("cqName1", 1);
- cqs1.add("cqName2", 2);
+ cqs1.add("cqName1", MessageType.RESPONSE);
+ cqs1.add("cqName2", MessageType.EXCEPTION);
clientCqs.put(client1, cqs1);
- ClientUpdateMessageImpl.CqNameToOpSingleEntry cqs2 =
- new ClientUpdateMessageImpl.CqNameToOpSingleEntry("cqName3", 3);
+ CqNameToOpSingleEntry cqs2 =
+ new CqNameToOpSingleEntry("cqName3", MessageType.REQUESTDATAERROR);
clientCqs.put(client2, cqs2);
}
+ @AfterEach
+ void afterEach() {
+ assertThat(executorService.shutdownNow()).isEmpty();
+ }
+
@Test
- public void addInterestedClientTest() {
+ void addInterestedClientTest() {
ClientUpdateMessageImpl clientUpdateMessageImpl = new ClientUpdateMessageImpl();
ClientProxyMembershipID clientProxyMembershipID = mock(ClientProxyMembershipID.class);
@@ -79,18 +93,18 @@ public class ClientUpdateMessageImplTest implements Serializable {
}
@Test
- public void serializeClientUpdateMessageNullInterestLists() {
+ void serializeClientUpdateMessageNullInterestLists() {
ClientUpdateMessageImpl clientUpdateMessageImpl = getTestClientUpdateMessage();
ClientUpdateMessageImpl clientUpdateMessageCopy = CopyHelper.copy(clientUpdateMessageImpl);
- assertNotNull(clientUpdateMessageCopy);
+ assertThat(clientUpdateMessageCopy).isNotNull();
assertThat(clientUpdateMessageCopy.hasClientsInterestedInUpdates()).isFalse();
assertThat(clientUpdateMessageCopy.hasClientsInterestedInInvalidates()).isFalse();
}
@Test
- public void serializeClientUpdateMessageWithInterestLists() {
+ void serializeClientUpdateMessageWithInterestLists() {
ClientUpdateMessageImpl clientUpdateMessageImpl = getTestClientUpdateMessage();
DistributedMember distributedMember =
@@ -123,7 +137,7 @@ public class ClientUpdateMessageImplTest implements Serializable {
ClientUpdateMessageImpl clientUpdateMessageCopy = CopyHelper.copy(clientUpdateMessageImpl);
- assertNotNull(clientUpdateMessageCopy);
+ assertThat(clientUpdateMessageCopy).isNotNull();
assertThat(clientUpdateMessageCopy.isClientInterestedInUpdates(interestedClientID)).isTrue();
assertThat(clientUpdateMessageCopy.isClientInterestedInInvalidates(interestedClientID))
.isTrue();
@@ -137,11 +151,11 @@ public class ClientUpdateMessageImplTest implements Serializable {
}
@Test
- public void addClientCqCanBeExecutedConcurrently() throws Exception {
+ void addClientCqCanBeExecutedConcurrently() throws Exception {
ClientUpdateMessageImpl clientUpdateMessageImpl = new ClientUpdateMessageImpl();
int numOfEvents = 4;
- int[] cqEvents = new int[numOfEvents];
+ MessageType[] cqEvents = new MessageType[numOfEvents];
String[] cqNames = new String[numOfEvents];
ClientProxyMembershipID[] clients = new ClientProxyMembershipID[numOfEvents];
prepareCqInfo(numOfEvents, cqEvents, cqNames, clients);
@@ -155,21 +169,21 @@ public class ClientUpdateMessageImplTest implements Serializable {
(ClientUpdateMessageImpl.CqNameToOpHashMap) clientUpdateMessageImpl.getClientCqs()
.get(client1);
for (int i = 0; i < 3; i++) {
- assertThat(client1Cqs.get("cqName" + i)).isEqualTo(i);
+ assertThat(client1Cqs.get("cqName" + i)).isEqualTo(MessageType.valueOf(i));
}
assertThat(clientUpdateMessageImpl.getClientCqs().get(client2)).isInstanceOf(
- ClientUpdateMessageImpl.CqNameToOpSingleEntry.class);
- ClientUpdateMessageImpl.CqNameToOpSingleEntry client2Cqs =
- (ClientUpdateMessageImpl.CqNameToOpSingleEntry) clientUpdateMessageImpl.getClientCqs()
+ CqNameToOpSingleEntry.class);
+ CqNameToOpSingleEntry client2Cqs =
+ (CqNameToOpSingleEntry) clientUpdateMessageImpl.getClientCqs()
.get(client2);
assertThat(client2Cqs.isEmpty()).isFalse();
}
- private void prepareCqInfo(int numOfEvents, int[] cqEvents, String[] cqNames,
+ private void prepareCqInfo(int numOfEvents, MessageType[] cqEvents, String[] cqNames,
ClientProxyMembershipID[] clients) {
for (int i = 0; i < numOfEvents; i++) {
- cqEvents[i] = i;
+ cqEvents[i] = MessageType.valueOf(i);
cqNames[i] = "cqName" + i;
if (i < 3) {
clients[i] = client1;
@@ -180,23 +194,23 @@ public class ClientUpdateMessageImplTest implements Serializable {
}
private void addClientCqConcurrently(ClientUpdateMessageImpl clientUpdateMessageImpl,
- int numOfEvents, int[] cqEvents, String[] cqNames, ClientProxyMembershipID[] clients)
+ int numOfEvents, MessageType[] cqEvents, String[] cqNames, ClientProxyMembershipID[] clients)
throws InterruptedException, java.util.concurrent.ExecutionException {
- List<Future<Void>> futures = new ArrayList<>();
+ List<Future<?>> futures = new ArrayList<>(numOfEvents);
for (int i = 0; i < numOfEvents; i++) {
ClientProxyMembershipID client = clients[i];
String cqName = cqNames[i];
- int cqEvent = cqEvents[i];
+ MessageType cqEvent = cqEvents[i];
futures.add(executorService
.submit(() -> clientUpdateMessageImpl.addClientCq(client, cqName, cqEvent)));
}
- for (Future<Void> future : futures) {
+ for (Future<?> future : futures) {
future.get();
}
}
@Test
- public void addOrSetClientCqsCanSetIfCqsMapIsNull() {
+ void addOrSetClientCqsCanSetIfCqsMapIsNull() {
ClientUpdateMessageImpl clientUpdateMessageImpl = new ClientUpdateMessageImpl();
clientUpdateMessageImpl.addOrSetClientCqs(client1, clientCqs);
@@ -205,10 +219,11 @@ public class ClientUpdateMessageImplTest implements Serializable {
}
@Test
- public void addOrSetClientCqsCanAddCqsIfCqsMapNotNull() {
+ void addOrSetClientCqsCanAddCqsIfCqsMapNotNull() {
ClientUpdateMessageImpl clientUpdateMessageImpl = new ClientUpdateMessageImpl();
ClientProxyMembershipID clientProxyMembershipID = mock(ClientProxyMembershipID.class);
- clientUpdateMessageImpl.addClientCq(clientProxyMembershipID, "cqName", 10);
+ clientUpdateMessageImpl.addClientCq(clientProxyMembershipID, "cqName",
+ MessageType.DESTROY_DATA_ERROR);
clientUpdateMessageImpl.addOrSetClientCqs(client1, clientCqs);
@@ -218,13 +233,159 @@ public class ClientUpdateMessageImplTest implements Serializable {
ClientUpdateMessageImpl.CqNameToOpHashMap client1Cqs =
(ClientUpdateMessageImpl.CqNameToOpHashMap) clientUpdateMessageImpl.getClientCqs()
.get(client1);
- assertThat(client1Cqs.get("cqName1")).isEqualTo(1);
- assertThat(client1Cqs.get("cqName2")).isEqualTo(2);
+ assertThat(client1Cqs.get("cqName1")).isEqualTo(MessageType.RESPONSE);
+ assertThat(client1Cqs.get("cqName2")).isEqualTo(MessageType.EXCEPTION);
assertThat(clientUpdateMessageImpl.getClientCqs().get(clientProxyMembershipID)).isInstanceOf(
- ClientUpdateMessageImpl.CqNameToOpSingleEntry.class);
+ CqNameToOpSingleEntry.class);
assertThat(clientUpdateMessageImpl.getClientCqs().get(client2)).isNull();
}
+ public static class CqNameToOpSingleEntryTest {
+ @Test
+ void isEmptyIsTrueWhenConstructedEmpty() {
+ assertThat(new CqNameToOpSingleEntry().isEmpty()).isTrue();
+ }
+
+ @Test
+ void sizeIsZeroWhenConstructedEmpty() {
+ assertThat(new CqNameToOpSingleEntry().size()).isZero();
+ }
+
+ @Test
+ void isFullIsFalseWhenConstructedEmpty() {
+ assertThat(new CqNameToOpSingleEntry().isFull()).isFalse();
+ }
+
+ @Test
+ void getNamesIsEmptyWhenConstructedEmpty() {
+ assertThat(new CqNameToOpSingleEntry().getNames()).isEmpty();
+ }
+
+ @Test
+ void canAddWhenConstructedEmpty() {
+ final CqNameToOpSingleEntry cqNameToOpSingleEntry = new CqNameToOpSingleEntry();
+ cqNameToOpSingleEntry.add("something", MessageType.UPDATE_CLIENT_NOTIFICATION);
+ assertThat(cqNameToOpSingleEntry.isEmpty()).isFalse();
+ assertThat(cqNameToOpSingleEntry.size()).isOne();
+ assertThat(cqNameToOpSingleEntry.isFull()).isTrue();
+ assertThat(cqNameToOpSingleEntry.getNames()).containsExactly("something");
+ }
+
+ @Test
+ void canDeleteWhenConstructedEmpty() {
+ final CqNameToOpSingleEntry cqNameToOpSingleEntry = new CqNameToOpSingleEntry();
+ cqNameToOpSingleEntry.delete("something");
+ assertThat(cqNameToOpSingleEntry.isEmpty()).isTrue();
+ assertThat(cqNameToOpSingleEntry.size()).isZero();
+ assertThat(cqNameToOpSingleEntry.isFull()).isFalse();
+ assertThat(cqNameToOpSingleEntry.getNames()).isEmpty();
+ }
+
+ @Test
+ void sendToWritesEmptyHashMapWhenConstructedEmpty() throws IOException {
+ final DataOutput dataOutput = mock(DataOutput.class);
+
+ final CqNameToOpSingleEntry cqNameToOpSingleEntry = new CqNameToOpSingleEntry();
+ cqNameToOpSingleEntry.sendTo(dataOutput);
+
+ final InOrder order = inOrder(dataOutput);
+ order.verify(dataOutput).writeByte(DSCODE.HASH_MAP.toByte());
+ order.verify(dataOutput).writeByte(0);
+ order.verifyNoMoreInteractions();
+ }
+
+ @Test
+ void addToMessageAddsNothingWhenConstructedEmpty() {
+ final Message message = mock(Message.class);
+
+ final CqNameToOpSingleEntry cqNameToOpSingleEntry = new CqNameToOpSingleEntry();
+ cqNameToOpSingleEntry.addToMessage(message);
+
+ verifyNoInteractions(message);
+ }
+
+ @Test
+ void isEmptyIsFalseWhenConstructedWithName() {
+ assertThat(
+ new CqNameToOpSingleEntry("something", MessageType.UPDATE_CLIENT_NOTIFICATION).isEmpty())
+ .isFalse();
+ }
+
+ @Test
+ void sizeIsOneWhenConstructedWithName() {
+ assertThat(
+ new CqNameToOpSingleEntry("something", MessageType.UPDATE_CLIENT_NOTIFICATION).size())
+ .isOne();
+ }
+
+ @Test
+ void isFullIsTrueWhenConstructedWithName() {
+ assertThat(
+ new CqNameToOpSingleEntry("something", MessageType.UPDATE_CLIENT_NOTIFICATION).isFull())
+ .isTrue();
+ }
+
+ @Test
+ void canAddWithSameNameWhenConstructedWithName() {
+ final CqNameToOpSingleEntry cqNameToOpSingleEntry =
+ new CqNameToOpSingleEntry("something", MessageType.UPDATE_CLIENT_NOTIFICATION);
+ cqNameToOpSingleEntry.add("something", MessageType.RESPONSE);
+ assertThat(cqNameToOpSingleEntry.isEmpty()).isFalse();
+ assertThat(cqNameToOpSingleEntry.size()).isOne();
+ assertThat(cqNameToOpSingleEntry.isFull()).isTrue();
+ assertThat(cqNameToOpSingleEntry.getNames()).containsExactly("something");
+ }
+
+ @Test
+ void canNotAddWithDifferentNameWhenConstructedWithName() {
+ final CqNameToOpSingleEntry cqNameToOpSingleEntry =
+ new CqNameToOpSingleEntry("something", MessageType.UPDATE_CLIENT_NOTIFICATION);
+ assertThatThrownBy(() -> cqNameToOpSingleEntry.add("other", MessageType.RESPONSE))
+ .isInstanceOf(IllegalStateException.class);
+ }
+
+ @Test
+ void canDeleteSameNameWhenConstructedWithName() {
+ final CqNameToOpSingleEntry cqNameToOpSingleEntry =
+ new CqNameToOpSingleEntry("something", MessageType.UPDATE_CLIENT_NOTIFICATION);
+ cqNameToOpSingleEntry.delete("something");
+ assertThat(cqNameToOpSingleEntry.isEmpty()).isTrue();
+ assertThat(cqNameToOpSingleEntry.size()).isZero();
+ assertThat(cqNameToOpSingleEntry.isFull()).isFalse();
+ assertThat(cqNameToOpSingleEntry.getNames()).isEmpty();
+ }
+
+ @Test
+ void sendToWritesSingleEntryHashMapWhenConstructedWithName() throws IOException {
+ final DataOutput dataOutput = mock(DataOutput.class);
+
+ final CqNameToOpSingleEntry cqNameToOpSingleEntry =
+ new CqNameToOpSingleEntry("something", MessageType.UPDATE_CLIENT_NOTIFICATION);
+ cqNameToOpSingleEntry.sendTo(dataOutput);
+
+ final InOrder order = inOrder(dataOutput);
+ order.verify(dataOutput).writeByte(DSCODE.HASH_MAP.toByte());
+ order.verify(dataOutput).writeByte(1);
+ order.verify(dataOutput).writeBytes(eq("something"));
+ order.verify(dataOutput).writeInt(eq(14));
+ // won't verify all remaining interactions
+ }
+
+ @Test
+ void addToMessageAddsSingleEntryWhenConstructedWithName() {
+ final Message message = mock(Message.class);
+
+ final CqNameToOpSingleEntry cqNameToOpSingleEntry =
+ new CqNameToOpSingleEntry("something", MessageType.UPDATE_CLIENT_NOTIFICATION);
+ cqNameToOpSingleEntry.addToMessage(message);
+
+ final InOrder order = inOrder(message);
+ order.verify(message).addStringPart(eq("something"), eq(true));
+ order.verify(message).addIntPart(eq(MessageType.UPDATE_CLIENT_NOTIFICATION.id));
+ order.verifyNoMoreInteractions();
+ }
+
+ }
}
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/CommandInitializerTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/CommandInitializerTest.java
index 766abdedd4..cdfd3e816e 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/CommandInitializerTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/CommandInitializerTest.java
@@ -102,7 +102,7 @@ public class CommandInitializerTest {
@Test
public void initializeGeode18Commands() {
@SuppressWarnings("unchecked")
- final Map<Integer, Command> commands = mock(Map.class);
+ final Map<MessageType, Command> commands = mock(Map.class);
CommandInitializer.initializeGeode18Commands(commands);
@@ -115,7 +115,7 @@ public class CommandInitializerTest {
@Test
public void initializeGfe90Commands() {
@SuppressWarnings("unchecked")
- final Map<Integer, Command> commands = mock(Map.class);
+ final Map<MessageType, Command> commands = mock(Map.class);
CommandInitializer.initializeGfe90Commands(commands);
@@ -129,7 +129,7 @@ public class CommandInitializerTest {
@Test
public void initializeGfe82Commands() {
@SuppressWarnings("unchecked")
- final Map<Integer, Command> commands = mock(Map.class);
+ final Map<MessageType, Command> commands = mock(Map.class);
CommandInitializer.initializeGfe81Commands(commands);
@@ -207,9 +207,9 @@ public class CommandInitializerTest {
@Test
public void commandMapUnmodifiable() {
final CommandInitializer commandInitializer = new CommandInitializer();
- final Map<Integer, Command> commands =
+ final Map<MessageType, Command> commands =
commandInitializer.get(KnownVersion.CURRENT.getClientServerProtocolVersion());
- assertThatThrownBy(() -> commands.put(1, Put70.getCommand()))
+ assertThatThrownBy(() -> commands.put(MessageType.RESPONSE, Put70.getCommand()))
.isInstanceOf(UnsupportedOperationException.class);
}
@@ -220,11 +220,11 @@ public class CommandInitializerTest {
newCommandMap.put(KnownVersion.CURRENT.getClientServerProtocolVersion(), command);
final CommandInitializer commandInitializer = new CommandInitializer();
- final Map<Integer, Command> commands =
+ final Map<MessageType, Command> commands =
commandInitializer.get(KnownVersion.CURRENT.getClientServerProtocolVersion());
- assertThat(commands).doesNotContainKeys(-2);
- commandInitializer.register(-2, newCommandMap);
- assertThat(commands).containsEntry(-2, command);
+ assertThat(commands).doesNotContainKeys(MessageType.RESPONSE);
+ commandInitializer.register(MessageType.RESPONSE, newCommandMap);
+ assertThat(commands).containsEntry(MessageType.RESPONSE, command);
}
}
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/ServerConnectionTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/ServerConnectionTest.java
index e32a031a3c..4481158cb5 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/ServerConnectionTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/ServerConnectionTest.java
@@ -55,6 +55,7 @@ import org.apache.geode.internal.cache.tier.CachedRegionHelper;
import org.apache.geode.internal.cache.tier.Command;
import org.apache.geode.internal.cache.tier.CommunicationMode;
import org.apache.geode.internal.cache.tier.Encryptor;
+import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.ServerSideHandshake;
import org.apache.geode.internal.cache.tier.sockets.command.PutUserCredentials;
import org.apache.geode.internal.monitoring.ThreadsMonitoring;
@@ -195,6 +196,7 @@ public class ServerConnectionTest {
@Test
public void bindSubject() {
ServerConnection connection = spy(serverConnection);
+ connection.getRequestMessage().setMessageType(MessageType.REQUEST);
initializeClientUserAuth(connection);
when(acceptor.getCacheClientNotifier()).thenReturn(notifier);
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteFunction70Test.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteFunction70Test.java
index 4b0b43972f..f34ec7d434 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteFunction70Test.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteFunction70Test.java
@@ -16,7 +16,6 @@ package org.apache.geode.internal.cache.tier.sockets.command;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doCallRealMethod;
import static org.mockito.Mockito.doThrow;
@@ -162,7 +161,7 @@ public class ExecuteFunction70Test {
when(serverConnection.getHandshake()).thenReturn(mock(ServerSideHandshake.class));
when(serverConnection.getReplyMessage()).thenReturn(replyMessage);
- when(serverToClientFunctionResultSender65Factory.create(any(), anyInt(), any(), any(), any()))
+ when(serverToClientFunctionResultSender65Factory.create(any(), any(), any(), any(), any()))
.thenReturn(
serverToClientFunctionResultSender65);
@@ -178,7 +177,7 @@ public class ExecuteFunction70Test {
executeFunction.cmdExecute(message, serverConnection, securityService, 0);
verify(serverToClientFunctionResultSender65Factory).create(eq(functionResponseMessage),
- anyInt(), any(), any(), any());
+ any(), any(), any(), any());
}
@Test
@@ -191,7 +190,7 @@ public class ExecuteFunction70Test {
verify(securityService).authorize(ResourcePermissions.DATA_WRITE);
verify(serverToClientFunctionResultSender65Factory).create(eq(functionResponseMessage),
- anyInt(), any(), any(), any());
+ any(), any(), any(), any());
}
@Test
@@ -220,7 +219,7 @@ public class ExecuteFunction70Test {
verify(authzRequest).executeFunctionAuthorize(eq(FUNCTION_ID), any(), any(), any(),
eq(false));
verify(serverToClientFunctionResultSender65Factory).create(eq(functionResponseMessage),
- anyInt(), any(), any(), any());
+ any(), any(), any(), any());
}
@Test
diff --git a/geode-core/src/upgradeTest/java/org/apache/geode/security/ClientAuthenticationPart2DUnitTest.java b/geode-core/src/upgradeTest/java/org/apache/geode/security/ClientAuthenticationPart2DUnitTest.java
index 3490b559b5..2711d15761 100644
--- a/geode-core/src/upgradeTest/java/org/apache/geode/security/ClientAuthenticationPart2DUnitTest.java
+++ b/geode-core/src/upgradeTest/java/org/apache/geode/security/ClientAuthenticationPart2DUnitTest.java
@@ -78,10 +78,11 @@ public class ClientAuthenticationPart2DUnitTest extends ClientAuthenticationTest
when(serverConnection.isInternalMessage(any(Message.class), any(Boolean.class)))
.thenCallRealMethod();
- int[] oldInternalMessages = new int[] {MessageType.ADD_PDX_TYPE, MessageType.ADD_PDX_ENUM,
- MessageType.REGISTER_INSTANTIATORS, MessageType.REGISTER_DATASERIALIZERS};
+ MessageType[] oldInternalMessages =
+ new MessageType[] {MessageType.ADD_PDX_TYPE, MessageType.ADD_PDX_ENUM,
+ MessageType.REGISTER_INSTANTIATORS, MessageType.REGISTER_DATASERIALIZERS};
- for (final int oldInternalMessage : oldInternalMessages) {
+ for (final MessageType oldInternalMessage : oldInternalMessages) {
Message message = mock(Message.class);
when(message.getMessageType()).thenReturn(oldInternalMessage);
diff --git a/geode-cq/src/integrationTest/java/org/apache/geode/cache/query/cq/internal/CqServiceFactoryImplTest.java b/geode-cq/src/integrationTest/java/org/apache/geode/cache/query/cq/internal/CqServiceFactoryImplTest.java
index 1e92fea631..c0172843f7 100644
--- a/geode-cq/src/integrationTest/java/org/apache/geode/cache/query/cq/internal/CqServiceFactoryImplTest.java
+++ b/geode-cq/src/integrationTest/java/org/apache/geode/cache/query/cq/internal/CqServiceFactoryImplTest.java
@@ -46,20 +46,20 @@ public class CqServiceFactoryImplTest {
CommandRegistry commandRegistry = new CommandInitializer();
- final Integer[] messageTypes = {
- MessageType.EXECUTECQ_MSG_TYPE, MessageType.EXECUTECQ_WITH_IR_MSG_TYPE,
- MessageType.GETCQSTATS_MSG_TYPE, MessageType.MONITORCQ_MSG_TYPE,
- MessageType.STOPCQ_MSG_TYPE, MessageType.CLOSECQ_MSG_TYPE,
- MessageType.GETDURABLECQS_MSG_TYPE};
+ final MessageType[] messageTypes = {
+ MessageType.EXECUTECQ, MessageType.EXECUTECQ_WITH_IR,
+ MessageType.GETCQSTATS, MessageType.MONITORCQ,
+ MessageType.STOPCQ, MessageType.CLOSECQ,
+ MessageType.GETDURABLECQS};
- final Set<Integer> initialKeys = commandRegistry.get(KnownVersion.OLDEST).keySet();
+ final Set<MessageType> initialKeys = commandRegistry.get(KnownVersion.OLDEST).keySet();
assertThat(initialKeys).doesNotContain(messageTypes);
final CqServiceFactoryImpl cqServiceFactory = new CqServiceFactoryImpl();
cqServiceFactory.initialize();
cqServiceFactory.create(internalCache, commandRegistry);
- final Set<Integer> expectedKeys = new HashSet<>(initialKeys);
+ final Set<MessageType> expectedKeys = new HashSet<>(initialKeys);
expectedKeys.addAll(asList(messageTypes));
assertThat(commandRegistry.get(KnownVersion.OLDEST)).containsOnlyKeys(expectedKeys);
}
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/CqServiceFactoryImpl.java b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/CqServiceFactoryImpl.java
index 60cee606b9..f95af547a4 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/CqServiceFactoryImpl.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/CqServiceFactoryImpl.java
@@ -41,19 +41,19 @@ public class CqServiceFactoryImpl implements CqServiceFactory {
@Override
public CqService create(InternalCache cache, CommandRegistry commandRegistry) {
- commandRegistry.register(MessageType.EXECUTECQ_MSG_TYPE,
+ commandRegistry.register(MessageType.EXECUTECQ,
singletonMap(KnownVersion.OLDEST, ExecuteCQ61.getCommand()));
- commandRegistry.register(MessageType.EXECUTECQ_WITH_IR_MSG_TYPE,
+ commandRegistry.register(MessageType.EXECUTECQ_WITH_IR,
singletonMap(KnownVersion.OLDEST, ExecuteCQ61.getCommand()));
- commandRegistry.register(MessageType.GETCQSTATS_MSG_TYPE,
+ commandRegistry.register(MessageType.GETCQSTATS,
singletonMap(KnownVersion.OLDEST, GetCQStats.getCommand()));
- commandRegistry.register(MessageType.MONITORCQ_MSG_TYPE,
+ commandRegistry.register(MessageType.MONITORCQ,
singletonMap(KnownVersion.OLDEST, MonitorCQ.getCommand()));
- commandRegistry.register(MessageType.STOPCQ_MSG_TYPE,
+ commandRegistry.register(MessageType.STOPCQ,
singletonMap(KnownVersion.OLDEST, StopCQ.getCommand()));
- commandRegistry.register(MessageType.CLOSECQ_MSG_TYPE,
+ commandRegistry.register(MessageType.CLOSECQ,
singletonMap(KnownVersion.OLDEST, CloseCQ.getCommand()));
- commandRegistry.register(MessageType.GETDURABLECQS_MSG_TYPE,
+ commandRegistry.register(MessageType.GETDURABLECQS,
singletonMap(KnownVersion.OLDEST, GetDurableCQs.getCommand()));
return new CqServiceImpl(cache);
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/CqServiceImpl.java b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/CqServiceImpl.java
index 3f3578778a..0775494d3b 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/CqServiceImpl.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/CqServiceImpl.java
@@ -14,6 +14,14 @@
*/
package org.apache.geode.cache.query.cq.internal;
+import static org.apache.geode.internal.cache.tier.MessageType.CLEAR_REGION;
+import static org.apache.geode.internal.cache.tier.MessageType.DESTROY_REGION;
+import static org.apache.geode.internal.cache.tier.MessageType.EXCEPTION;
+import static org.apache.geode.internal.cache.tier.MessageType.INVALIDATE_REGION;
+import static org.apache.geode.internal.cache.tier.MessageType.LOCAL_CREATE;
+import static org.apache.geode.internal.cache.tier.MessageType.LOCAL_DESTROY;
+import static org.apache.geode.internal.cache.tier.MessageType.LOCAL_UPDATE;
+
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -93,11 +101,6 @@ import org.apache.geode.util.internal.GeodeGlossary;
public class CqServiceImpl implements CqService {
private static final Logger logger = LogService.getLogger();
- private static final Integer MESSAGE_TYPE_LOCAL_CREATE = MessageType.LOCAL_CREATE;
- private static final Integer MESSAGE_TYPE_LOCAL_UPDATE = MessageType.LOCAL_UPDATE;
- private static final Integer MESSAGE_TYPE_LOCAL_DESTROY = MessageType.LOCAL_DESTROY;
- private static final Integer MESSAGE_TYPE_EXCEPTION = MessageType.EXCEPTION;
-
/**
* System property to evaluate the query even though the initial results are not required when cq
* is executed using the execute() method.
@@ -927,15 +930,16 @@ public class CqServiceImpl implements CqService {
}
@Override
- public void dispatchCqListeners(HashMap<String, Integer> cqs, int messageType, Object key,
+ public void dispatchCqListeners(HashMap<String, MessageType> cqs, MessageType messageType,
+ Object key,
Object value, byte[] delta, QueueManager qManager, EventID eventId) {
Object[] fullValue = new Object[1];
- Iterator<Map.Entry<String, Integer>> iter = cqs.entrySet().iterator();
+ Iterator<Map.Entry<String, MessageType>> iter = cqs.entrySet().iterator();
String cqName = null;
final boolean isDebugEnabled = logger.isDebugEnabled();
while (iter.hasNext()) {
try {
- Map.Entry<String, Integer> entry = iter.next();
+ Map.Entry<String, MessageType> entry = iter.next();
cqName = entry.getKey();
ClientCQImpl cQuery = (ClientCQImpl) getCq(cqName);
@@ -947,10 +951,10 @@ public class CqServiceImpl implements CqService {
continue;
}
- Integer cqOp = entry.getValue();
+ MessageType cqOp = entry.getValue();
// If Region destroy event, close the cq.
- if (cqOp == MessageType.DESTROY_REGION) {
+ if (cqOp == DESTROY_REGION) {
// The close will also invoke the listeners close().
try {
cQuery.close(false);
@@ -1156,30 +1160,30 @@ public class CqServiceImpl implements CqService {
/**
* Returns the Operation for the given EnumListenerEvent type.
*/
- private Operation getOperation(int eventType) {
+ private Operation getOperation(MessageType eventType) {
Operation op = null;
switch (eventType) {
- case MessageType.LOCAL_CREATE:
+ case LOCAL_CREATE:
op = Operation.CREATE;
break;
- case MessageType.LOCAL_UPDATE:
+ case LOCAL_UPDATE:
op = Operation.UPDATE;
break;
- case MessageType.LOCAL_DESTROY:
+ case LOCAL_DESTROY:
op = Operation.DESTROY;
break;
- case MessageType.LOCAL_INVALIDATE:
+ case LOCAL_INVALIDATE:
op = Operation.INVALIDATE;
break;
- case MessageType.CLEAR_REGION:
+ case CLEAR_REGION:
op = Operation.REGION_CLEAR;
break;
- case MessageType.INVALIDATE_REGION:
+ case INVALIDATE_REGION:
op = Operation.REGION_INVALIDATE;
break;
}
@@ -1211,7 +1215,7 @@ public class CqServiceImpl implements CqService {
if (isDebugEnabled) {
logger.debug("CQ service processing region event {}", event);
}
- Integer cqRegionEvent = generateCqRegionEvent(event);
+ MessageType cqRegionEvent = generateCqRegionEvent(event);
for (int i = -1; i < profiles.length; i++) {
CacheProfile cf;
@@ -1227,11 +1231,10 @@ public class CqServiceImpl implements CqService {
if (pf == null || pf.getCqMap().isEmpty()) {
continue;
}
- Map cqs = pf.getCqMap();
- HashMap<Long, Integer> cqInfo = new HashMap<>();
- for (final Object o : cqs.entrySet()) {
- Map.Entry cqEntry = (Map.Entry) o;
- ServerCQImpl cQuery = (ServerCQImpl) cqEntry.getValue();
+ Map<?, ServerCQImpl> cqs = pf.getCqMap();
+ HashMap<Long, MessageType> cqInfo = new HashMap<>();
+ for (final Map.Entry<?, ServerCQImpl> cqEntry : cqs.entrySet()) {
+ ServerCQImpl cQuery = cqEntry.getValue();
if (!event.isOriginRemote() && event.getOperation().isRegionDestroy()
&& !((LocalRegion) event.getRegion()).isUsedForPartitionedRegionBucket()) {
try {
@@ -1278,7 +1281,7 @@ public class CqServiceImpl implements CqService {
|| event.getOperation().isDestroy() || event.getOperation().isInvalidate()
|| (event.getOperation().isCreate() && isDupEvent));
- HashMap<String, Integer> matchedCqs = new HashMap<>();
+ HashMap<String, MessageType> matchedCqs = new HashMap<>();
long executionStartTime;
for (int i = -1; i < profiles.length; i++) {
CacheProfile cf;
@@ -1295,7 +1298,7 @@ public class CqServiceImpl implements CqService {
continue;
}
- Map cqs = pf.getCqMap();
+ Map<?, ServerCQImpl> cqs = pf.getCqMap();
if (isDebugEnabled) {
logger.debug("Profile for {} processing {} CQs", cf.peerMemberId, cqs.size());
@@ -1315,11 +1318,10 @@ public class CqServiceImpl implements CqService {
}
}
- HashMap<Long, Integer> cqInfo = new HashMap<>();
+ HashMap<Long, MessageType> cqInfo = new HashMap<>();
- for (Object o : cqs.entrySet()) {
- Map.Entry cqEntry = (Map.Entry) o;
- ServerCQImpl cQuery = (ServerCQImpl) cqEntry.getValue();
+ for (Map.Entry<?, ServerCQImpl> cqEntry : cqs.entrySet()) {
+ ServerCQImpl cQuery = cqEntry.getValue();
b_cqResults_newValue = false;
b_cqResults_oldValue = false;
queryOldValue = false;
@@ -1333,7 +1335,7 @@ public class CqServiceImpl implements CqService {
logger.debug("Processing CQ : {} Key: {}", cqName, eventKey);
}
- Integer cqEvent = null;
+ MessageType cqEvent = null;
if (matchedCqs.containsKey(cqName)) {
cqEvent = matchedCqs.get(cqName);
if (isDebugEnabled) {
@@ -1343,10 +1345,10 @@ public class CqServiceImpl implements CqService {
continue;
}
// Update the Cache Results for this CQ.
- if (cqEvent == MessageType.LOCAL_CREATE
- || cqEvent == MessageType.LOCAL_UPDATE) {
+ if (cqEvent == LOCAL_CREATE
+ || cqEvent == LOCAL_UPDATE) {
cQuery.addToCqResultKeys(eventKey);
- } else if (cqEvent == MessageType.LOCAL_DESTROY) {
+ } else if (cqEvent == LOCAL_DESTROY) {
cQuery.markAsDestroyedInCqResultKeys(eventKey);
}
} else {
@@ -1429,13 +1431,13 @@ public class CqServiceImpl implements CqService {
}
if (error) {
- cqEvent = MESSAGE_TYPE_EXCEPTION;
+ cqEvent = EXCEPTION;
} else {
if (b_cqResults_newValue) {
if (b_cqResults_oldValue) {
- cqEvent = MESSAGE_TYPE_LOCAL_UPDATE;
+ cqEvent = LOCAL_UPDATE;
} else {
- cqEvent = MESSAGE_TYPE_LOCAL_CREATE;
+ cqEvent = LOCAL_CREATE;
}
// If its create and caching is enabled, cache the key
// for this CQ.
@@ -1444,7 +1446,7 @@ public class CqServiceImpl implements CqService {
// Base invalidate operation is treated as destroy.
// When the invalidate comes through, the entry will no longer
// satisfy the query and will need to be deleted.
- cqEvent = MESSAGE_TYPE_LOCAL_DESTROY;
+ cqEvent = LOCAL_DESTROY;
// If caching is enabled, mark this event's key as removed
// from the CQ cache.
cQuery.markAsDestroyedInCqResultKeys(eventKey);
@@ -1498,16 +1500,16 @@ public class CqServiceImpl implements CqService {
} // iteration over Profiles.
}
- private Integer generateCqRegionEvent(CacheEvent event) {
- Integer cqEvent = null;
+ private MessageType generateCqRegionEvent(CacheEvent event) {
if (event.getOperation().isRegionDestroy()) {
- cqEvent = MessageType.DESTROY_REGION;
+ return DESTROY_REGION;
} else if (event.getOperation().isRegionInvalidate()) {
- cqEvent = MessageType.INVALIDATE_REGION;
+ return INVALIDATE_REGION;
} else if (event.getOperation().isClear()) {
- cqEvent = MessageType.CLEAR_REGION;
+ return CLEAR_REGION;
}
- return cqEvent;
+
+ return null;
}
/**
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/CloseCQ.java b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/CloseCQ.java
index c8aeab4864..bfb4ed9acb 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/CloseCQ.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/CloseCQ.java
@@ -73,7 +73,7 @@ public class CloseCQ extends BaseCQCommand {
if (cqName == null) {
String err =
"The cqName for the cq close request is null";
- sendCqResponse(MessageType.CQDATAERROR_MSG_TYPE, err, clientMessage.getTransactionId(), null,
+ sendCqResponse(MessageType.CQDATAERROR, err, clientMessage.getTransactionId(), null,
serverConnection);
return;
}
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/ExecuteCQ61.java b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/ExecuteCQ61.java
index 8451a067ea..98764ec98c 100755
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/ExecuteCQ61.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/ExecuteCQ61.java
@@ -96,7 +96,7 @@ public class ExecuteCQ61 extends BaseCQCommand {
if (logger.isDebugEnabled()) {
logger.debug("{}: Received {} request from {} CqName: {} queryString: {}",
- serverConnection.getName(), MessageType.getString(clientMessage.getMessageType()),
+ serverConnection.getName(), clientMessage.getMessageType(),
serverConnection.getSocketString(), cqName, cqQueryString);
}
@@ -108,7 +108,7 @@ public class ExecuteCQ61 extends BaseCQCommand {
// This should have been taken care at the client.
String err =
"Server notifyBySubscription mode is set to false. CQ execution is not supported in this mode.";
- sendCqResponse(MessageType.CQDATAERROR_MSG_TYPE, err, clientMessage.getTransactionId(),
+ sendCqResponse(MessageType.CQDATAERROR, err, clientMessage.getTransactionId(),
null, serverConnection);
return;
}
@@ -171,7 +171,7 @@ public class ExecuteCQ61 extends BaseCQCommand {
}
final boolean sendResults =
- clientMessage.getMessageType() == MessageType.EXECUTECQ_WITH_IR_MSG_TYPE;
+ clientMessage.getMessageType() == MessageType.EXECUTECQ_WITH_IR;
// Execute the query only if it is execute with initial results or
// if it is a non PR query with execute query and maintain keys flags set
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/GetCQStats.java b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/GetCQStats.java
index 5b9d587a3c..81dfc57ed9 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/GetCQStats.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/GetCQStats.java
@@ -70,7 +70,7 @@ public class GetCQStats extends BaseCQCommand {
// Process the query request
if (cqName == null) {
String err = "The cqName for the cq stats request is null";
- sendCqResponse(MessageType.CQDATAERROR_MSG_TYPE, err, clientMessage.getTransactionId(), null,
+ sendCqResponse(MessageType.CQDATAERROR, err, clientMessage.getTransactionId(), null,
serverConnection);
return;
}
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/GetDurableCQs.java b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/GetDurableCQs.java
index b14ee36bb0..6c564baed5 100755
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/GetDurableCQs.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/GetDurableCQs.java
@@ -64,7 +64,7 @@ public class GetDurableCQs extends BaseCQCommand {
if (logger.isDebugEnabled()) {
logger.debug("{}: Received {} request from {}", serverConnection.getName(),
- MessageType.getString(clientMessage.getMessageType()),
+ clientMessage.getMessageType(),
serverConnection.getSocketString());
}
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/MonitorCQ.java b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/MonitorCQ.java
index 7056277f99..8d0472c3c6 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/MonitorCQ.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/MonitorCQ.java
@@ -55,7 +55,7 @@ public class MonitorCQ extends BaseCQCommand {
// This should have been taken care at the client - remove?
String err = String.format("%s: The MonitorCq operation is invalid.",
serverConnection.getName());
- sendCqResponse(MessageType.CQDATAERROR_MSG_TYPE, err, clientMessage.getTransactionId(), null,
+ sendCqResponse(MessageType.CQDATAERROR, err, clientMessage.getTransactionId(), null,
serverConnection);
return;
}
@@ -69,7 +69,7 @@ public class MonitorCQ extends BaseCQCommand {
String err =
String.format("%s: A null Region name was passed for MonitorCq operation.",
serverConnection.getName());
- sendCqResponse(MessageType.CQDATAERROR_MSG_TYPE, err, clientMessage.getTransactionId(),
+ sendCqResponse(MessageType.CQDATAERROR, err, clientMessage.getTransactionId(),
null, serverConnection);
return;
}
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/StopCQ.java b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/StopCQ.java
index 707154d25c..de336a3041 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/StopCQ.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/command/StopCQ.java
@@ -74,7 +74,7 @@ public class StopCQ extends BaseCQCommand {
if (cqName == null) {
String err =
"The cqName for the cq stop request is null";
- sendCqResponse(MessageType.CQDATAERROR_MSG_TYPE, err, clientMessage.getTransactionId(), null,
+ sendCqResponse(MessageType.CQDATAERROR, err, clientMessage.getTransactionId(), null,
serverConnection);
return;
}
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ops/CloseCQOp.java b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ops/CloseCQOp.java
index 9c18b55ff9..d5a9a911a6 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ops/CloseCQOp.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ops/CloseCQOp.java
@@ -47,7 +47,7 @@ public class CloseCQOp {
* @throws org.apache.geode.SerializationException if serialization fails
*/
public CloseCQOpImpl(String cqName) {
- super(MessageType.CLOSECQ_MSG_TYPE, 1);
+ super(MessageType.CLOSECQ, 1);
getMessage().addStringPart(cqName);
}
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ops/CreateCQOp.java b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ops/CreateCQOp.java
index 9f6315a4ce..dfa04fc6c7 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ops/CreateCQOp.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ops/CreateCQOp.java
@@ -89,7 +89,7 @@ public class CreateCQOp {
public CreateCQOpImpl(final @NotNull String cqName, final @NotNull String queryStr,
final int cqState, final boolean isDurable,
final @NotNull DataPolicy regionDataPolicy) {
- super(MessageType.EXECUTECQ_MSG_TYPE, 5);
+ super(MessageType.EXECUTECQ, 5);
getMessage().addStringPart(cqName);
getMessage().addStringPart(queryStr);
getMessage().addIntPart(cqState);
@@ -109,7 +109,7 @@ public class CreateCQOp {
protected Object processResponse(final @NotNull Message m) throws Exception {
ChunkedMessage msg = (ChunkedMessage) m;
msg.readHeader();
- int msgType = msg.getMessageType();
+ MessageType msgType = msg.getMessageType();
msg.receiveChunk();
if (msgType == MessageType.REPLY) {
return Boolean.TRUE;
@@ -132,7 +132,7 @@ public class CreateCQOp {
throw new ServerOperationException(errorMessage);
} else {
throw new InternalGemFireError(
- "Unexpected message type " + MessageType.getString(msgType));
+ "Unexpected message type " + msgType);
}
}
}
@@ -142,7 +142,7 @@ public class CreateCQOp {
*
* @throws org.apache.geode.SerializationException if serialization fails
*/
- protected CreateCQOpImpl(int msgType, int numParts) {
+ protected CreateCQOpImpl(MessageType msgType, int numParts) {
super(msgType, numParts);
}
@@ -151,8 +151,8 @@ public class CreateCQOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
- return msgType == MessageType.CQDATAERROR_MSG_TYPE
+ protected boolean isErrorResponse(MessageType msgType) {
+ return msgType == MessageType.CQDATAERROR
|| msgType == MessageType.CQ_EXCEPTION_TYPE;
}
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ops/CreateCQWithIROp.java b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ops/CreateCQWithIROp.java
index 50e9e53a98..ea345b8c32 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ops/CreateCQWithIROp.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ops/CreateCQWithIROp.java
@@ -63,7 +63,7 @@ public class CreateCQWithIROp {
public CreateCQWithIROpImpl(final @NotNull String cqName, final @NotNull String queryStr,
final int cqState, final boolean isDurable,
final @NotNull DataPolicy regionDataPolicy) {
- super(MessageType.EXECUTECQ_WITH_IR_MSG_TYPE, 5);
+ super(MessageType.EXECUTECQ_WITH_IR, 5);
getMessage().addStringPart(cqName);
getMessage().addStringPart(queryStr);
getMessage().addIntPart(cqState);
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ops/GetDurableCQsOp.java b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ops/GetDurableCQsOp.java
index 2f00b60273..1e485c247d 100755
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ops/GetDurableCQsOp.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ops/GetDurableCQsOp.java
@@ -57,7 +57,7 @@ public class GetDurableCQsOp {
* @throws org.apache.geode.SerializationException if serialization fails
*/
public GetDurableCQsOpImpl() {
- super(MessageType.GETDURABLECQS_MSG_TYPE, 1 /* numparts */);
+ super(MessageType.GETDURABLECQS, 1 /* numparts */);
}
@Override
@@ -93,7 +93,7 @@ public class GetDurableCQsOp {
final Exception[] exceptionRef = new Exception[1];
getDurableCQsResponseMsg.readHeader();
- final int msgType = getDurableCQsResponseMsg.getMessageType();
+ final MessageType msgType = getDurableCQsResponseMsg.getMessageType();
if (msgType == MessageType.RESPONSE) {
do {
getDurableCQsResponseMsg.receiveChunk();
@@ -122,7 +122,7 @@ public class GetDurableCQsOp {
throw new ServerOperationException(part.getString());
} else {
throw new InternalGemFireError(
- "Unexpected message type " + MessageType.getString(msgType));
+ "Unexpected message type " + msgType);
}
}
@@ -134,7 +134,7 @@ public class GetDurableCQsOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return msgType == MessageType.GET_DURABLE_CQS_DATA_ERROR;
}
}
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ops/StopCQOp.java b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ops/StopCQOp.java
index d285d0d687..e19e35548e 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ops/StopCQOp.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ops/StopCQOp.java
@@ -47,7 +47,7 @@ public class StopCQOp {
* @throws org.apache.geode.SerializationException if serialization fails
*/
public StopCQOpImpl(String cqName) {
- super(MessageType.STOPCQ_MSG_TYPE, 1);
+ super(MessageType.STOPCQ, 1);
getMessage().addStringPart(cqName);
}
diff --git a/geode-cq/src/test/java/org/apache/geode/cache/query/cq/internal/CqServiceFactoryImplTest.java b/geode-cq/src/test/java/org/apache/geode/cache/query/cq/internal/CqServiceFactoryImplTest.java
index 69b7e6a7cc..85bc57a936 100644
--- a/geode-cq/src/test/java/org/apache/geode/cache/query/cq/internal/CqServiceFactoryImplTest.java
+++ b/geode-cq/src/test/java/org/apache/geode/cache/query/cq/internal/CqServiceFactoryImplTest.java
@@ -54,19 +54,19 @@ public class CqServiceFactoryImplTest {
cqServiceFactory.initialize();
cqServiceFactory.create(internalCache, commandRegistry);
- verify(commandRegistry).register(MessageType.EXECUTECQ_MSG_TYPE,
+ verify(commandRegistry).register(MessageType.EXECUTECQ,
singletonMap(KnownVersion.OLDEST, ExecuteCQ61.getCommand()));
- verify(commandRegistry).register(MessageType.EXECUTECQ_WITH_IR_MSG_TYPE,
+ verify(commandRegistry).register(MessageType.EXECUTECQ_WITH_IR,
singletonMap(KnownVersion.OLDEST, ExecuteCQ61.getCommand()));
- verify(commandRegistry).register(MessageType.GETCQSTATS_MSG_TYPE,
+ verify(commandRegistry).register(MessageType.GETCQSTATS,
singletonMap(KnownVersion.OLDEST, GetCQStats.getCommand()));
- verify(commandRegistry).register(MessageType.MONITORCQ_MSG_TYPE,
+ verify(commandRegistry).register(MessageType.MONITORCQ,
singletonMap(KnownVersion.OLDEST, MonitorCQ.getCommand()));
- verify(commandRegistry).register(MessageType.STOPCQ_MSG_TYPE,
+ verify(commandRegistry).register(MessageType.STOPCQ,
singletonMap(KnownVersion.OLDEST, StopCQ.getCommand()));
- verify(commandRegistry).register(MessageType.CLOSECQ_MSG_TYPE,
+ verify(commandRegistry).register(MessageType.CLOSECQ,
singletonMap(KnownVersion.OLDEST, CloseCQ.getCommand()));
- verify(commandRegistry).register(MessageType.GETDURABLECQS_MSG_TYPE,
+ verify(commandRegistry).register(MessageType.GETDURABLECQS,
singletonMap(KnownVersion.OLDEST, GetDurableCQs.getCommand()));
verifyNoMoreInteractions(commandRegistry);
diff --git a/geode-wan/src/main/java/org/apache/geode/cache/wan/internal/client/locator/GatewaySenderBatchOp.java b/geode-wan/src/main/java/org/apache/geode/cache/wan/internal/client/locator/GatewaySenderBatchOp.java
index 2a8182b242..e0ee40f550 100755
--- a/geode-wan/src/main/java/org/apache/geode/cache/wan/internal/client/locator/GatewaySenderBatchOp.java
+++ b/geode-wan/src/main/java/org/apache/geode/cache/wan/internal/client/locator/GatewaySenderBatchOp.java
@@ -14,6 +14,8 @@
*/
package org.apache.geode.cache.wan.internal.client.locator;
+import static org.apache.geode.internal.cache.tier.MessageType.GATEWAY_RECEIVER_COMMAND;
+
import java.net.SocketTimeoutException;
import java.util.List;
@@ -77,7 +79,7 @@ public class GatewaySenderBatchOp {
*/
public GatewaySenderGFEBatchOpImpl(List events, int batchId, boolean removeFromQueueOnException,
int dsId, boolean isRetry) {
- super(MessageType.GATEWAY_RECEIVER_COMMAND, calcPartCount(events));
+ super(GATEWAY_RECEIVER_COMMAND, calcPartCount(events));
if (isRetry) {
getMessage().setIsRetry();
}
@@ -131,7 +133,7 @@ public class GatewaySenderBatchOp {
}
public GatewaySenderGFEBatchOpImpl() {
- super(MessageType.GATEWAY_RECEIVER_COMMAND, 0);
+ super(GATEWAY_RECEIVER_COMMAND, 0);
}
@Override
@@ -230,7 +232,7 @@ public class GatewaySenderBatchOp {
try {
// Read the header which describes the type of message following
switch (msg.getMessageType()) {
- case MessageType.REPLY:
+ case REPLY:
// Read the chunk
Part part0 = msg.getPart(0);
if (part0.isBytes() && part0.getLength() == 1 && part0.getSerializedForm()[0] == 0) {
@@ -241,7 +243,7 @@ public class GatewaySenderBatchOp {
int numEvents = msg.getPart(1).getInt();
ack = new GatewayAck(batchId, numEvents);
break;
- case MessageType.EXCEPTION:
+ case EXCEPTION:
part0 = msg.getPart(0);
Object obj = part0.getObject();
@@ -273,7 +275,7 @@ public class GatewaySenderBatchOp {
}
@Override
- protected boolean isErrorResponse(int msgType) {
+ protected boolean isErrorResponse(MessageType msgType) {
return false;
}