You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by bs...@apache.org on 2016/02/19 18:28:37 UTC
incubator-geode git commit: GEODE-981: NPE in Message.clear()
Repository: incubator-geode
Updated Branches:
refs/heads/develop 75d6a8b68 -> ab9cedf8f
GEODE-981: NPE in Message.clear()
This fixes the NPE and adds a unit test for it. I also did some cleanup
of Message, removing unused methods and old commented out code and
changing the earlyAck byte to be "flags" and not allow external
manipulation of the byte's bits.
Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/ab9cedf8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/ab9cedf8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/ab9cedf8
Branch: refs/heads/develop
Commit: ab9cedf8f253e3c049138df282352dd28c0b252a
Parents: 75d6a8b
Author: Bruce Schuchardt <bs...@pivotal.io>
Authored: Fri Feb 19 09:27:23 2016 -0800
Committer: Bruce Schuchardt <bs...@pivotal.io>
Committed: Fri Feb 19 09:27:23 2016 -0800
----------------------------------------------------------------------
.../cache/client/internal/AbstractOp.java | 2 +-
.../cache/client/internal/AddPDXEnumOp.java | 2 +-
.../cache/client/internal/AddPDXTypeOp.java | 4 +-
.../client/internal/AuthenticateUserOp.java | 4 +-
.../client/internal/CloseConnectionOp.java | 2 +-
.../gemfire/cache/client/internal/CommitOp.java | 2 +-
.../client/internal/GetClientPRMetaDataOp.java | 2 +-
.../GetClientPartitionAttributesOp.java | 2 +-
.../cache/client/internal/GetEventValueOp.java | 2 +-
.../client/internal/GetFunctionAttributeOp.java | 2 +-
.../cache/client/internal/GetPDXEnumByIdOp.java | 4 +-
.../cache/client/internal/GetPDXEnumsOp.java | 2 +-
.../client/internal/GetPDXIdForEnumOp.java | 4 +-
.../client/internal/GetPDXIdForTypeOp.java | 4 +-
.../cache/client/internal/GetPDXTypeByIdOp.java | 4 +-
.../cache/client/internal/GetPDXTypesOp.java | 2 +-
.../cache/client/internal/MakePrimaryOp.java | 2 +-
.../gemfire/cache/client/internal/PingOp.java | 2 +-
.../cache/client/internal/PrimaryAckOp.java | 2 +-
.../client/internal/ProxyCacheCloseOp.java | 2 +-
.../cache/client/internal/ReadyForEventsOp.java | 2 +-
.../internal/RegisterDataSerializersOp.java | 2 +-
.../internal/RegisterInstantiatorsOp.java | 2 +-
.../cache/client/internal/RollbackOp.java | 2 +-
.../gemfire/cache/client/internal/SizeOp.java | 2 +-
.../cache/client/internal/TXFailoverOp.java | 2 +-
.../client/internal/TXSynchronizationOp.java | 2 +-
.../gemfire/internal/cache/TXManagerImpl.java | 2 +-
.../cache/tier/sockets/BaseCommand.java | 2 +-
.../internal/cache/tier/sockets/Message.java | 236 ++++++-------------
.../internal/cache/tier/sockets/Part.java | 9 +-
.../gemfire/distributed/LocatorDUnitTest.java | 2 +
.../cache/tier/sockets/MessageJUnitTest.java | 75 ++++++
.../client/internal/GatewaySenderBatchOp.java | 2 +-
34 files changed, 183 insertions(+), 211 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/AbstractOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/AbstractOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/AbstractOp.java
index 0ae0437..28c1e0f 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/AbstractOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/AbstractOp.java
@@ -114,7 +114,7 @@ public abstract class AbstractOp implements Op {
protected void sendMessage(Connection cnx) throws Exception {
if (cnx.getServer().getRequiresCredentials()) {
// Security is enabled on client as well as on server
- getMessage().setEarlyAck(Message.MESSAGE_HAS_SECURE_PART);
+ getMessage().setMessageHasSecurePartFlag();
HeapDataOutputStream hdos = new HeapDataOutputStream(Version.CURRENT);
long userId = -1;
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/AddPDXEnumOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/AddPDXEnumOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/AddPDXEnumOp.java
index b88948d..ff2bf1c 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/AddPDXEnumOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/AddPDXEnumOp.java
@@ -91,7 +91,7 @@ public class AddPDXEnumOp {
//most of the other messages like this.
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/AddPDXTypeOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/AddPDXTypeOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/AddPDXTypeOp.java
index 92dd246..9fbc674 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/AddPDXTypeOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/AddPDXTypeOp.java
@@ -87,11 +87,9 @@ public class AddPDXTypeOp {
return false;
}
- //TODO - no idea what this mumbo jumbo means, but it's on
- //most of the other messages like this.
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/AuthenticateUserOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/AuthenticateUserOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/AuthenticateUserOp.java
index b0f0cec..b03c7b9 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/AuthenticateUserOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/AuthenticateUserOp.java
@@ -121,7 +121,7 @@ public class AuthenticateUserOp {
tmpSecurityProperties, server, false, (InternalLogWriter)sys.getLogWriter(), (InternalLogWriter)sys
.getSecurityLogWriter());
- getMessage().setEarlyAck(Message.MESSAGE_HAS_SECURE_PART);
+ getMessage().setMessageHasSecurePartFlag();
HeapDataOutputStream heapdos = new HeapDataOutputStream(Version.CURRENT);
try {
DataSerializer.writeProperties(credentials, heapdos);
@@ -144,7 +144,7 @@ public class AuthenticateUserOp {
this.securityProperties = securityProps;
this.needsServerLocation = needsServer;
- getMessage().setEarlyAck(Message.MESSAGE_HAS_SECURE_PART);
+ getMessage().setMessageHasSecurePartFlag();
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/CloseConnectionOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/CloseConnectionOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/CloseConnectionOp.java
index 943f0a3..7de38fe 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/CloseConnectionOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/CloseConnectionOp.java
@@ -67,7 +67,7 @@ public class CloseConnectionOp {
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/CommitOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/CommitOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/CommitOp.java
index e04a466..c9c6dd7 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/CommitOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/CommitOp.java
@@ -85,7 +85,7 @@ public class CommitOp {
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetClientPRMetaDataOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetClientPRMetaDataOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetClientPRMetaDataOp.java
index f88d1e9..8bae6ff 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetClientPRMetaDataOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetClientPRMetaDataOp.java
@@ -81,7 +81,7 @@ public class GetClientPRMetaDataOp {
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetClientPartitionAttributesOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetClientPartitionAttributesOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetClientPartitionAttributesOp.java
index 004872c..e1a8870 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetClientPartitionAttributesOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetClientPartitionAttributesOp.java
@@ -86,7 +86,7 @@ public class GetClientPartitionAttributesOp {
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetEventValueOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetEventValueOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetEventValueOp.java
index 7dfe9af..1038ede 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetEventValueOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetEventValueOp.java
@@ -72,7 +72,7 @@ public class GetEventValueOp {
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetFunctionAttributeOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetFunctionAttributeOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetFunctionAttributeOp.java
index 9e63fba..177ea26 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetFunctionAttributeOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetFunctionAttributeOp.java
@@ -77,7 +77,7 @@ public class GetFunctionAttributeOp {
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXEnumByIdOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXEnumByIdOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXEnumByIdOp.java
index d3e9efb..1e22d81 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXEnumByIdOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXEnumByIdOp.java
@@ -83,11 +83,9 @@ public class GetPDXEnumByIdOp {
return false;
}
- //TODO - no idea what this mumbo jumbo means, but it's on
- //most of the other messages like this.
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXEnumsOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXEnumsOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXEnumsOp.java
index 0cb5d33..0590caf 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXEnumsOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXEnumsOp.java
@@ -105,7 +105,7 @@ public class GetPDXEnumsOp {
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXIdForEnumOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXIdForEnumOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXIdForEnumOp.java
index 81a2b1b..bac2e80 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXIdForEnumOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXIdForEnumOp.java
@@ -104,11 +104,9 @@ public class GetPDXIdForEnumOp {
protected boolean participateInTransaction() {
return false;
}
- //TODO - no idea what this mumbo jumbo means, but it's on
- //most of the other messages like this.
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXIdForTypeOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXIdForTypeOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXIdForTypeOp.java
index d771cb6..1b71f71 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXIdForTypeOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXIdForTypeOp.java
@@ -104,11 +104,9 @@ public class GetPDXIdForTypeOp {
protected boolean participateInTransaction() {
return false;
}
- //TODO - no idea what this mumbo jumbo means, but it's on
- //most of the other messages like this.
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXTypeByIdOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXTypeByIdOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXTypeByIdOp.java
index e0b2810..1d9bd2d 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXTypeByIdOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXTypeByIdOp.java
@@ -83,11 +83,9 @@ public class GetPDXTypeByIdOp {
return false;
}
- //TODO - no idea what this mumbo jumbo means, but it's on
- //most of the other messages like this.
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXTypesOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXTypesOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXTypesOp.java
index 2990192..262cb9a 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXTypesOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetPDXTypesOp.java
@@ -105,7 +105,7 @@ public class GetPDXTypesOp {
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/MakePrimaryOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/MakePrimaryOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/MakePrimaryOp.java
index 59b99f0..27d80b1 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/MakePrimaryOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/MakePrimaryOp.java
@@ -62,7 +62,7 @@ public class MakePrimaryOp {
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PingOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PingOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PingOp.java
index e0bc81b..e70d50a 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PingOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PingOp.java
@@ -66,7 +66,7 @@ public class PingOp {
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
startTime = System.currentTimeMillis();
getMessage().send(false);
Message.messageType.set(MessageType.PING);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PrimaryAckOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PrimaryAckOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PrimaryAckOp.java
index 0b65c56..4ee680a 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PrimaryAckOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PrimaryAckOp.java
@@ -72,7 +72,7 @@ public class PrimaryAckOp {
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ProxyCacheCloseOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ProxyCacheCloseOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ProxyCacheCloseOp.java
index 53b8fa9..2747fa8 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ProxyCacheCloseOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ProxyCacheCloseOp.java
@@ -45,7 +45,7 @@ public class ProxyCacheCloseOp {
public ProxyCacheCloseOpImpl(ExecutablePool pool, Properties securityProps,
boolean keepAlive) {
super(MessageType.REMOVE_USER_AUTH, 1);
- getMessage().setEarlyAck(Message.MESSAGE_HAS_SECURE_PART);
+ getMessage().setMessageHasSecurePartFlag();
getMessage().addBytesPart(keepAlive ? new byte[] {1} : new byte[] {0});
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ReadyForEventsOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ReadyForEventsOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ReadyForEventsOp.java
index a003538..d2631fc 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ReadyForEventsOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ReadyForEventsOp.java
@@ -62,7 +62,7 @@ public class ReadyForEventsOp {
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/RegisterDataSerializersOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/RegisterDataSerializersOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/RegisterDataSerializersOp.java
index b2b975f..869ad64 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/RegisterDataSerializersOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/RegisterDataSerializersOp.java
@@ -131,7 +131,7 @@ public class RegisterDataSerializersOp {
}
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/RegisterInstantiatorsOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/RegisterInstantiatorsOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/RegisterInstantiatorsOp.java
index 93d3756..0d5a137 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/RegisterInstantiatorsOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/RegisterInstantiatorsOp.java
@@ -173,7 +173,7 @@ public class RegisterInstantiatorsOp {
}
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/RollbackOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/RollbackOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/RollbackOp.java
index 2793f32..6f01b96 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/RollbackOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/RollbackOp.java
@@ -92,7 +92,7 @@ public class RollbackOp {
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/SizeOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/SizeOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/SizeOp.java
index 6d69083..42cc225 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/SizeOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/SizeOp.java
@@ -85,7 +85,7 @@ public class SizeOp {
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/TXFailoverOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/TXFailoverOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/TXFailoverOp.java
index 1fecc7d..64ee66e 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/TXFailoverOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/TXFailoverOp.java
@@ -86,7 +86,7 @@ public class TXFailoverOp {
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/TXSynchronizationOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/TXSynchronizationOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/TXSynchronizationOp.java
index 48d66f2..34ecf4d 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/TXSynchronizationOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/TXSynchronizationOp.java
@@ -155,7 +155,7 @@ public class TXSynchronizationOp {
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/TXManagerImpl.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/TXManagerImpl.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/TXManagerImpl.java
index f4216ac..de49fea 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/TXManagerImpl.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/TXManagerImpl.java
@@ -777,7 +777,7 @@ public final class TXManagerImpl implements CacheTransactionManager,
if (val == null) {
synchronized(this.hostedTXStates) {
val = this.hostedTXStates.get(key);
- if (val == null && msg.canStartRemoteTransaction()) {
+ if (val == null) {
// [sjigyasu] TODO: Conditionally create object based on distributed or non-distributed tx mode
if (msg instanceof TransactionMessage && ((TransactionMessage)msg).isTransactionDistributed()) {
val = new DistTXStateProxyImplOnDatanode(this, key, memberId);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/BaseCommand.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/BaseCommand.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/BaseCommand.java
index 52ea6f5..dd13f19 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/BaseCommand.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/BaseCommand.java
@@ -931,7 +931,7 @@ public abstract class BaseCommand implements Command {
try {
requestMsg = servConn.getRequestMessage();
requestMsg.recv(servConn, MAX_INCOMING_DATA, incomingDataLimiter,
- MAX_INCOMING_MSGS, incomingMsgLimiter);
+ incomingMsgLimiter);
return requestMsg;
}
catch (EOFException eof) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/Message.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/Message.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/Message.java
index f6866bf..4bfd44b 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/Message.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/Message.java
@@ -34,14 +34,12 @@ import com.gemstone.gemfire.internal.Assert;
import com.gemstone.gemfire.internal.HeapDataOutputStream;
import com.gemstone.gemfire.internal.SocketUtils;
import com.gemstone.gemfire.internal.Version;
-import com.gemstone.gemfire.internal.cache.CachedDeserializable;
import com.gemstone.gemfire.internal.cache.TXManagerImpl;
import com.gemstone.gemfire.internal.cache.tier.MessageType;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
import com.gemstone.gemfire.internal.logging.LogService;
import com.gemstone.gemfire.internal.logging.log4j.LocalizedMessage;
import com.gemstone.gemfire.internal.offheap.StoredObject;
-import com.gemstone.gemfire.internal.offheap.annotations.Retained;
import com.gemstone.gemfire.internal.offheap.annotations.Unretained;
import com.gemstone.gemfire.internal.util.BlobHelper;
@@ -61,7 +59,7 @@ import com.gemstone.gemfire.internal.util.BlobHelper;
* transId - int - 4 bytes filled in by the requestor, copied back into
* the response
*
- * earlyAck - byte- 1 byte filled in by the requestor
+ * flags - byte- 1 byte filled in by the requestor
* len1
* part1
* .
@@ -81,33 +79,17 @@ import com.gemstone.gemfire.internal.util.BlobHelper;
*
* @see com.gemstone.gemfire.internal.cache.tier.MessageType
*
- * @author Sudhir Menon
- * @since 2.0.2
*/
public class Message {
private static final Logger logger = LogService.getLogger();
- @Override
- public String toString() {
- StringBuffer sb = new StringBuffer();
- sb.append("type=" + MessageType.getString(msgType));
- sb.append("; payloadLength=" + payloadLength);
- sb.append("; numberOfParts=" + numberOfParts);
- sb.append("; transactionId=" + transactionId);
- //sb.append("; bufferLength=" + bufferLength);
- sb.append("; currentPart=" + currentPart);
- sb.append("; messageModified=" + messageModified);
- sb.append("; earlyAck=" + earlyAck);
- for (int i = 0; i < numberOfParts; i ++) {
- sb.append("; part[" + i + "]={");
- sb.append(this.partsList[i].toString());
- sb.append("}");
- }
- return sb.toString();
- }
+ private static final int PART_HEADER_SIZE = 5; // 4 bytes for length, 1 byte for isObject
+
+ private static final int FIXED_LENGTH = 17;
+
+ private static final ThreadLocal<ByteBuffer> tlCommBuffer = new ThreadLocal<>();
- protected final static int FIXED_LENGTH = 17;
protected int msgType;
protected int payloadLength=0;
protected int numberOfParts =0;
@@ -122,10 +104,10 @@ public class Message {
protected boolean messageModified = true;
/** is this message a retry of a previously sent message? */
protected boolean isRetry;
- private byte earlyAck = 0x00;
+ private byte flags = 0x00;
protected MessageStats msgStats = null;
protected ServerConnection sc = null;
- private int MAX_DATA = -1;
+ private int maxIncomingMessageLength = -1;
private Semaphore dataLimiter = null;
// private int MAX_MSGS = -1;
private Semaphore msgLimiter = null;
@@ -133,8 +115,10 @@ public class Message {
private int chunkSize = 1024;//Default Chunk Size.
protected Part securePart = null;
+ private boolean isMetaRegion = false;
+
- // These two statics are fields shoved into the earlyAck byte for transmission.
+ // These two statics are fields shoved into the flags byte for transmission.
// The MESSAGE_IS_RETRY bit is stripped out during deserialization but the other
// is left in place
public static final byte MESSAGE_HAS_SECURE_PART = (byte)0x02;
@@ -181,31 +165,14 @@ public class Message {
this.version = clientVersion;
}
- /**
- * Sets whether this message is early-ack
- * @param earlyAck whether this message is early-ack
- */
- public void setEarlyAck(boolean earlyAck) {
- if (earlyAck) {
- this.earlyAck = 0x01;
- } else {
- this.earlyAck = 0x00;
- }
+ public void setMessageHasSecurePartFlag() {
+ this.flags = (byte)(this.flags | MESSAGE_HAS_SECURE_PART);
}
-
- // TODO (ashetkar) To be removed later.
- public void setEarlyAck(byte earlyAck) {
- // Check that the passed in value is within the acceptable range.
- if (0x00 <= earlyAck && earlyAck <= 0x02) {
- this.earlyAck |= earlyAck;
- }
+
+ public void clearMessageHasSecurePartFlag() {
+ this.flags = (byte)(this.flags & MESSAGE_HAS_SECURE_PART);
}
- /*
- * public void setPayloadLength(int payloadLength) {
- this.payloadLength = payloadLength;
- }*/
-
/**
* Sets and builds the {@link Part}s that are sent
* in the payload of the Message
@@ -252,6 +219,14 @@ public class Message {
public void setChunkSize(int chunkSize) {
this.chunkSize = chunkSize;
}
+
+ /**
+ * When building a Message this will return the number of the
+ * next Part to be added to the message
+ */
+ public int getNextPartNumber() {
+ return this.currentPart;
+ }
public void addStringPart(String str) {
if (str==null) {
@@ -266,15 +241,6 @@ public class Message {
}
}
- /**
- * Sets whether or not a
- * <code>DataOutputStream</code>/<code>DataOutputStream</code>
- * should be used to send/receive data.
- public void setUseDataStream (boolean useDataStream) {
- this.useDataStream = useDataStream;
- }
- */
-
/*
* Adds a new part to this message that contains a <code>byte</code>
* array (as opposed to a serialized object).
@@ -295,7 +261,7 @@ public class Message {
}
}
- public void addDeltaPart(HeapDataOutputStream hdos) { // TODO: Amogh- Should it be just DataOutput?
+ public void addDeltaPart(HeapDataOutputStream hdos) {
this.messageModified = true;
Part part = partsList[this.currentPart];
part.setPartState(hdos, false);
@@ -364,12 +330,6 @@ public class Message {
// the heap bb to the existing direct bb without needing to allocate extra direct bbs.
// Delaying the flush uses more direct memory but reduces the number of system calls.
try {
-// logger.fine("hitesh before serializatino: " );
-//
-// if (o != null ){
-// logger.fine("hitesh before serializatino: " + o.toString());
-// logger.fine("hitesh before serializatino: " + o.getClass().getName());
-// }
BlobHelper.serializeTo(o, hdos);
} catch (IOException ex) {
throw new SerializationException("failed serializing object", ex);
@@ -385,8 +345,6 @@ public class Message {
if (zipValues) {
throw new UnsupportedOperationException("zipValues no longer supported");
-// byte[] b = CacheServerHelper.serialize(o, zipValues);
-// addRawPart(b, true);
} else {
HeapDataOutputStream hdos;
Version v = version;
@@ -395,12 +353,6 @@ public class Message {
}
hdos = new HeapDataOutputStream(chunkSize, v);
try {
-// logger.fine("hitesh before serializatino: " );
-//
-// if (o != null ){
-// logger.fine("hitesh before serializatino: " + o.toString());
-// logger.fine("hitesh before serializatino: " + o.getClass().getName());
-// }
BlobHelper.serializeTo(o, hdos);
} catch (IOException ex) {
throw new SerializationException("failed serializing object", ex);
@@ -468,19 +420,8 @@ public class Message {
return null;
}
- public boolean getEarlyAck() {
- return this.earlyAck == 0x01 ? true: false;
- }
-
- // TODO (ashetkar) To be removed
- public byte getEarlyAckByte() {
- return this.earlyAck;
- }
-
- private static ThreadLocal tlCommBuffer = new ThreadLocal();
-
public static ByteBuffer setTLCommBuffer(ByteBuffer bb) {
- ByteBuffer result = (ByteBuffer)tlCommBuffer.get();
+ ByteBuffer result = tlCommBuffer.get();
tlCommBuffer.set(bb);
return result;
}
@@ -490,7 +431,7 @@ public class Message {
return this.cachedCommBuffer;
}
else {
- return (ByteBuffer)tlCommBuffer.get();
+ return tlCommBuffer.get();
}
}
@@ -505,14 +446,15 @@ public class Message {
this.msgStats.decMessagesBeingReceived(len);
}
}
- if (this.socket != null) {
- getCommBuffer().clear();
+ ByteBuffer buffer = getCommBuffer();
+ if (buffer != null) {
+ buffer.clear();
}
clearParts();
if (len != 0 && this.dataLimiter != null) {
this.dataLimiter.release(len);
this.dataLimiter = null;
- this.MAX_DATA = 0;
+ this.maxIncomingMessageLength = 0;
}
if (this.hdrRead) {
if (this.msgLimiter != null) {
@@ -521,29 +463,28 @@ public class Message {
}
this.hdrRead = false;
}
+ this.flags = 0;
}
protected void packHeaderInfoForSending(int msgLen, boolean isSecurityHeader) {
- //TODO:hitesh setting second bit of early ack for client
+ //TODO:hitesh setting second bit of flags byte for client
//this is not require but this makes all changes easily at client side right now
//just see this bit and process security header
- byte eAck = this.earlyAck;
+ byte flagsByte = this.flags;
if (isSecurityHeader) {
- eAck |= MESSAGE_HAS_SECURE_PART;
+ flagsByte |= MESSAGE_HAS_SECURE_PART;
}
if (this.isRetry) {
- eAck |= MESSAGE_IS_RETRY;
+ flagsByte |= MESSAGE_IS_RETRY;
}
getCommBuffer()
.putInt(this.msgType)
.putInt(msgLen)
.putInt(this.numberOfParts)
.putInt(this.transactionId)
- .put(eAck);
+ .put(flagsByte);
}
- private static final int PART_HEADER_SIZE = 5; // 4 bytes for length, 1 byte for isObject
-
protected Part getSecurityPart() {
if (this.sc != null ) {
//look types right put get etc
@@ -557,15 +498,13 @@ public class Message {
this.securePart.setPartState(bytes, false);
}
- private boolean m_isMetaRegion = false;
-
public void setMetaRegion(boolean isMetaRegion) {
- this.m_isMetaRegion = isMetaRegion;
+ this.isMetaRegion = isMetaRegion;
}
public boolean getAndResetIsMetaRegion() {
- boolean isMetaRegion = this.m_isMetaRegion;
- this.m_isMetaRegion = false;
+ boolean isMetaRegion = this.isMetaRegion;
+ this.isMetaRegion = false;
return isMetaRegion;
}
@@ -598,7 +537,6 @@ public class Message {
numOfSecureParts = 1;
}
- //this.logger.fine("hitesh sendbytes forServer_SecurityPart " + numOfSecureParts);
int totalPartLen = 0;
for (int i=0;i<this.numberOfParts;i++){
Part part = this.partsList[i];
@@ -740,7 +678,7 @@ public class Message {
final int len = cb.getInt();
final int numParts = cb.getInt();
final int txid = cb.getInt();
- byte early = cb.get();
+ byte bits = cb.get();
cb.clear();
if (!MessageType.validate(type)) {
@@ -783,8 +721,8 @@ public class Message {
} // for
}
if (len > 0) {
- if (this.MAX_DATA > 0 && len > this.MAX_DATA) {
- throw new IOException(LocalizedStrings.Message_MESSAGE_SIZE_0_EXCEEDED_MAX_LIMIT_OF_1.toLocalizedString(new Object[] {Integer.valueOf(len), Integer.valueOf(this.MAX_DATA)}));
+ if (this.maxIncomingMessageLength > 0 && len > this.maxIncomingMessageLength) {
+ throw new IOException(LocalizedStrings.Message_MESSAGE_SIZE_0_EXCEEDED_MAX_LIMIT_OF_1.toLocalizedString(new Object[] {Integer.valueOf(len), Integer.valueOf(this.maxIncomingMessageLength)}));
}
if (this.dataLimiter != null) {
for (;;) {
@@ -825,13 +763,12 @@ public class Message {
this.payloadLength = len; // makes sure payloadLength gets set now so we will dec on clear
}
- this.isRetry = (early & MESSAGE_IS_RETRY) != 0;
- early = (byte)(early & MESSAGE_IS_RETRY_MASK);
-
- //TODO:hitesh it was below ??
- this.earlyAck = early;
+ this.isRetry = (bits & MESSAGE_IS_RETRY) != 0;
+ bits = (byte)(bits & MESSAGE_IS_RETRY_MASK);
+ this.flags = bits;
+ // TODO why is the msgType set twice, here and after reading the payload fields?
this.msgType = type;
- //this.logger.fine("Before reading message parts, earlyAck already read as " + this.earlyAck);
+
readPayloadFields(numParts, len);
// Set the header and payload fields only after receiving all the
@@ -841,43 +778,13 @@ public class Message {
this.payloadLength = len;
// this.numberOfParts = numParts; Already set in setPayloadFields via setNumberOfParts
this.transactionId = txid;
- this.earlyAck = early;
+ this.flags = bits;
if (this.sc != null) {
// Keep track of the fact that a message is being processed.
this.sc.updateProcessingMessage();
}
}
-// static final int MAX_PART_BUFFERS = 2;
-// static final int MIN_PART_BUFFER_SIZE = 999;
-// static final int MAX_PART_BUFFER_SIZE = 1024*1024*11;
-// static ArrayList partBuffers = new ArrayList(2);
-// static int partBufferIdx = 0;
-// static {
-// for (int i=0; i < MAX_PART_BUFFERS; i++) {
-// partBuffers.add(i, null);
-// }
-// }
-
-// private static synchronized byte[] getPartBuffer(int size) {
-// byte[] result;
-// synchronized (partBuffers) {
-// result = (byte[])partBuffers.get(partBufferIdx);
-// if (result == null) {
-// result = new byte[size];
-// partBuffers.add(partBufferIdx, result);
-// } else if (result.length != size) {
-// // can't use a cached one
-// return null;
-// }
-// partBufferIdx++;
-// if (partBufferIdx >= MAX_PART_BUFFERS) {
-// partBufferIdx = 0;
-// }
-// }
-// return result;
-// }
-
protected void readPayloadFields(final int numParts, final int len)
throws IOException {
//TODO:Hitesh
@@ -912,11 +819,9 @@ public class Message {
int readSecurePart = 0;
//TODO:Hitesh look if securePart can be cached here
- //this.logger.fine("readPayloadFields() early ack = " + this.earlyAck);
readSecurePart = checkAndSetSecurityPart();
int bytesRemaining = len;
- //this.logger.fine("readPayloadFields() : numParts=" + numParts + " len=" + len);
for (int i = 0; ((i < numParts + readSecurePart) || ((readSecurePart == 1) && (cb
.remaining() > 0))); i++) {
int bytesReadThisTime = readPartChunk(bytesRemaining);
@@ -934,14 +839,8 @@ public class Message {
int partLen = cb.getInt();
byte partType = cb.get();
byte[] partBytes = null;
-// this.logger.fine("readPayloadFields(): partLen=" + partLen + " partType=" + partType);
if (partLen > 0) {
-// if (partLen >= MIN_PART_BUFFER_SIZE && partLen <= MAX_PART_BUFFER_SIZE) {
-// partBytes = getPartBuffer(partLen);
-// }
-// if (partBytes == null) {
- partBytes = new byte[partLen];
-// }
+ partBytes = new byte[partLen];
int alreadyReadBytes = cb.remaining();
if (alreadyReadBytes > 0) {
if (partLen < alreadyReadBytes) {
@@ -961,7 +860,6 @@ public class Message {
}
cb.limit(bytesThisTime);
int res = this.sockCh.read(cb);
- //System.out.println("DEBUG: part read " + res + " bytes commBuffer=" + cb);
if (res != -1) {
cb.flip();
bytesRemaining -= res;
@@ -980,7 +878,6 @@ public class Message {
res = this.is.read(partBytes, off, remaining);
}
catch (SocketTimeoutException e) {
-// res = 0;
// TODO: add cancellation check
throw e;
}
@@ -1002,7 +899,7 @@ public class Message {
}
protected int checkAndSetSecurityPart() {
- if ((this.earlyAck | MESSAGE_HAS_SECURE_PART) == this.earlyAck) {
+ if ((this.flags | MESSAGE_HAS_SECURE_PART) == this.flags) {
this.securePart = new Part();
return 1;
}
@@ -1018,7 +915,6 @@ public class Message {
*/
private int readPartChunk(int bytesRemaining) throws IOException {
final ByteBuffer cb = getCommBuffer();
- //this.logger.info("DEBUG: commBuffer.remaining=" + cb.remaining());
if (cb.remaining() >= PART_HEADER_SIZE) {
// we already have the next part header in commBuffer so just return
return 0;
@@ -1042,7 +938,6 @@ public class Message {
}
while (remaining > 0) {
int res = this.sockCh.read(cb);
- //System.out.println("DEBUG: partChunk read " + res + " bytes commBuffer=" + cb);
if (res != -1) {
remaining -= res;
bytesRead += res;
@@ -1067,7 +962,6 @@ public class Message {
res = this.is.read(cb.array(), pos, bytesToRead);
}
catch (SocketTimeoutException e) {
-// res = 0;
// TODO add a cancellation check
throw e;
}
@@ -1097,6 +991,26 @@ public class Message {
}
this.currentPart=0;
}
+
+ @Override
+ public String toString() {
+ StringBuffer sb = new StringBuffer();
+ sb.append("type=").append(MessageType.getString(msgType));
+ sb.append("; payloadLength=").append(payloadLength);
+ sb.append("; numberOfParts=").append(numberOfParts);
+ sb.append("; transactionId=").append(transactionId);
+ sb.append("; currentPart=").append(currentPart);
+ sb.append("; messageModified=").append(messageModified);
+ sb.append("; flags=").append(Integer.toHexString(flags));
+ for (int i = 0; i < numberOfParts; i ++) {
+ sb.append("; part[").append(i).append("]={");
+ sb.append(this.partsList[i].toString());
+ sb.append("}");
+ }
+ return sb.toString();
+ }
+
+
public void setComms(ServerConnection sc, Socket socket, ByteBuffer bb, MessageStats msgStats) throws IOException {
this.sc = sc;
setComms(socket, bb, msgStats);
@@ -1174,17 +1088,13 @@ public class Message {
throw new IOException(LocalizedStrings.Message_DEAD_CONNECTION.toLocalizedString());
}
}
- public void recv(ServerConnection sc, int MAX_DATA, Semaphore dataLimiter, int MAX_MSGS, Semaphore msgLimiter)
+ public void recv(ServerConnection sc, int maxMessageLength, Semaphore dataLimiter, Semaphore msgLimiter)
throws IOException {
this.sc = sc;
- this.MAX_DATA = MAX_DATA;
+ this.maxIncomingMessageLength = maxMessageLength;
this.dataLimiter = dataLimiter;
-// this.MAX_MSGS = MAX_MSGS;
this.msgLimiter = msgLimiter;
recv();
}
- public boolean canStartRemoteTransaction() {
- return true;
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/Part.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/Part.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/Part.java
index 5418c68..f5f6326 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/Part.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/Part.java
@@ -69,10 +69,6 @@ public class Part {
this.typeCode = tc;
}
-// public void init(HeapDataOutputStream os, byte typeCode) {
-// this.part = os;
-// this.typeCode = typeCode;
-// }
public void clear() {
this.part = null;
@@ -97,9 +93,7 @@ public class Part {
public boolean isBytes() {
return this.typeCode == BYTE_CODE || this.typeCode == EMPTY_BYTEARRAY_CODE;
}
-// public boolean isString() {
-// return this.typeCode == STRING_CODE;
-// }
+
public void setPartState(byte[] b, boolean isObject) {
if (isObject) {
this.typeCode = OBJECT_CODE;
@@ -111,6 +105,7 @@ public class Part {
}
this.part = b;
}
+
public void setPartState(HeapDataOutputStream os, boolean isObject) {
if (isObject) {
this.typeCode = OBJECT_CODE;
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/LocatorDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/LocatorDUnitTest.java b/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/LocatorDUnitTest.java
index f74e0de..3c6a90a 100644
--- a/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/LocatorDUnitTest.java
+++ b/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/LocatorDUnitTest.java
@@ -1091,6 +1091,8 @@ public class LocatorDUnitTest extends DistributedTestCase {
* members of the distributed system join it. This ensures that
* members start up okay, and that handling of a stopped locator
* is correct.
+ * <p>The locator is then restarted and is shown to take over the
+ * role of membership coordinator.
*/
public void testOneLocator() throws Exception {
disconnectAllFromDS();
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/MessageJUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/MessageJUnitTest.java b/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/MessageJUnitTest.java
new file mode 100755
index 0000000..3dc5a7d
--- /dev/null
+++ b/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/MessageJUnitTest.java
@@ -0,0 +1,75 @@
+/*
+ * 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 com.gemstone.gemfire.internal.cache.tier.sockets;
+
+import static org.junit.Assert.*;
+import static org.mockito.Matchers.*;
+import static org.mockito.Mockito.*;
+
+import java.net.Socket;
+import java.nio.ByteBuffer;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import com.gemstone.gemfire.internal.Version;
+import com.gemstone.gemfire.internal.offheap.HeapByteBufferMemoryChunkJUnitTest;
+import com.gemstone.gemfire.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class MessageJUnitTest {
+
+ Message message;
+ Socket mockSocket;
+ MessageStats mockStats;
+ ByteBuffer msgBuffer;
+
+ @Before
+ public void setUp() throws Exception {
+ mockSocket = mock(Socket.class);
+ message = new Message(5, Version.CURRENT);
+ assertEquals(5, message.getNumberOfParts());
+ mockStats = mock(MessageStats.class);
+ msgBuffer = ByteBuffer.allocate(1000);
+ message.setComms(mockSocket, msgBuffer, mockStats);
+ }
+
+ @Test
+ public void clearDoesNotThrowNPE() throws Exception{
+ // unsetComms clears the message's ByteBuffer, which was causing an NPE during shutdown
+ // when clear() was invoked
+ message.unsetComms();
+ message.clear();
+ }
+
+ @Test
+ public void numberOfPartsIsAdjusted() {
+ int numParts = message.getNumberOfParts();
+ message.setNumberOfParts(2*numParts);
+ assertEquals(2*numParts, message.getNumberOfParts());
+ message.addBytesPart(new byte[1]);
+ message.addIntPart(2);
+ message.addLongPart(3);
+ message.addObjPart("4");
+ message.addStringPart("5");
+ assertEquals(5, message.getNextPartNumber());
+ }
+
+ // TODO many more tests are needed
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ab9cedf8/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/GatewaySenderBatchOp.java
----------------------------------------------------------------------
diff --git a/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/GatewaySenderBatchOp.java b/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/GatewaySenderBatchOp.java
index 2eb0896..564591d 100644
--- a/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/GatewaySenderBatchOp.java
+++ b/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/GatewaySenderBatchOp.java
@@ -241,7 +241,7 @@ public class GatewaySenderBatchOp {
@Override
protected void sendMessage(Connection cnx) throws Exception {
- getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().clearMessageHasSecurePartFlag();
getMessage().send(false);
}