You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by hi...@apache.org on 2017/08/22 18:59:44 UTC
[1/7] geode git commit: GEODE-3406: Locator accepts Protobuf requests
Repository: geode
Updated Branches:
refs/heads/develop be4551191 -> 31e82d6d6
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRegionNamesRequestOperationHandlerJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRegionNamesRequestOperationHandlerJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRegionNamesRequestOperationHandlerJUnitTest.java
index 29902e6..fd84d41 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRegionNamesRequestOperationHandlerJUnitTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRegionNamesRequestOperationHandlerJUnitTest.java
@@ -17,7 +17,6 @@ package org.apache.geode.protocol.protobuf.operations;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
-import java.nio.charset.Charset;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
@@ -28,7 +27,8 @@ import org.junit.experimental.categories.Category;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.Region;
-import org.apache.geode.protocol.protobuf.BasicTypes;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.protobuf.RegionAPI;
import org.apache.geode.protocol.protobuf.Result;
import org.apache.geode.protocol.protobuf.Success;
@@ -62,11 +62,12 @@ public class GetRegionNamesRequestOperationHandlerJUnitTest extends OperationHan
}
@Test
- public void processReturnsCacheRegions() throws CodecAlreadyRegisteredForTypeException,
- UnsupportedEncodingTypeException, CodecNotRegisteredForTypeException {
- Result<RegionAPI.GetRegionNamesResponse> result =
- operationHandler.process(serializationServiceStub,
- ProtobufRequestUtilities.createGetRegionNamesRequest(), cacheStub);
+ public void processReturnsCacheRegions()
+ throws CodecAlreadyRegisteredForTypeException, UnsupportedEncodingTypeException,
+ CodecNotRegisteredForTypeException, InvalidExecutionContextException {
+ Result<RegionAPI.GetRegionNamesResponse> result = operationHandler.process(
+ serializationServiceStub, ProtobufRequestUtilities.createGetRegionNamesRequest(),
+ new ExecutionContext(cacheStub));
Assert.assertTrue(result instanceof Success);
RegionAPI.GetRegionNamesResponse getRegionsResponse = result.getMessage();
@@ -84,14 +85,15 @@ public class GetRegionNamesRequestOperationHandlerJUnitTest extends OperationHan
}
@Test
- public void processReturnsNoCacheRegions() throws CodecAlreadyRegisteredForTypeException,
- UnsupportedEncodingTypeException, CodecNotRegisteredForTypeException {
+ public void processReturnsNoCacheRegions()
+ throws CodecAlreadyRegisteredForTypeException, UnsupportedEncodingTypeException,
+ CodecNotRegisteredForTypeException, InvalidExecutionContextException {
Cache emptyCache = mock(Cache.class);;
when(emptyCache.rootRegions())
.thenReturn(Collections.unmodifiableSet(new HashSet<Region<String, String>>()));
- Result<RegionAPI.GetRegionNamesResponse> result =
- operationHandler.process(serializationServiceStub,
- ProtobufRequestUtilities.createGetRegionNamesRequest(), emptyCache);
+ Result<RegionAPI.GetRegionNamesResponse> result = operationHandler.process(
+ serializationServiceStub, ProtobufRequestUtilities.createGetRegionNamesRequest(),
+ new ExecutionContext(emptyCache));
Assert.assertTrue(result instanceof Success);
RegionAPI.GetRegionNamesResponse getRegionsResponse = result.getMessage();
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRegionRequestOperationHandlerJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRegionRequestOperationHandlerJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRegionRequestOperationHandlerJUnitTest.java
index 5cfa6b3..6762f66 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRegionRequestOperationHandlerJUnitTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRegionRequestOperationHandlerJUnitTest.java
@@ -19,6 +19,8 @@ import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.Scope;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.MessageUtil;
import org.apache.geode.protocol.protobuf.BasicTypes;
import org.apache.geode.protocol.protobuf.ClientProtocol;
@@ -57,8 +59,9 @@ public class GetRegionRequestOperationHandlerJUnitTest extends OperationHandlerJ
}
@Test
- public void processReturnsCacheRegions() throws CodecAlreadyRegisteredForTypeException,
- UnsupportedEncodingTypeException, CodecNotRegisteredForTypeException {
+ public void processReturnsCacheRegions()
+ throws CodecAlreadyRegisteredForTypeException, UnsupportedEncodingTypeException,
+ CodecNotRegisteredForTypeException, InvalidExecutionContextException {
RegionAttributes regionAttributesStub = mock(RegionAttributes.class);
when(cacheStub.getRegion(TEST_REGION1)).thenReturn(region1Stub);
@@ -72,7 +75,7 @@ public class GetRegionRequestOperationHandlerJUnitTest extends OperationHandlerJ
Result<RegionAPI.GetRegionResponse> result = operationHandler.process(serializationServiceStub,
- MessageUtil.makeGetRegionRequest(TEST_REGION1), cacheStub);
+ MessageUtil.makeGetRegionRequest(TEST_REGION1), new ExecutionContext(cacheStub));
RegionAPI.GetRegionResponse response = result.getMessage();
BasicTypes.Region region = response.getRegion();
Assert.assertEquals(TEST_REGION1, region.getName());
@@ -89,14 +92,15 @@ public class GetRegionRequestOperationHandlerJUnitTest extends OperationHandlerJ
}
@Test
- public void processReturnsNoCacheRegions() throws CodecAlreadyRegisteredForTypeException,
- UnsupportedEncodingTypeException, CodecNotRegisteredForTypeException {
+ public void processReturnsNoCacheRegions()
+ throws CodecAlreadyRegisteredForTypeException, UnsupportedEncodingTypeException,
+ CodecNotRegisteredForTypeException, InvalidExecutionContextException {
Cache emptyCache = mock(Cache.class);
when(emptyCache.rootRegions())
.thenReturn(Collections.unmodifiableSet(new HashSet<Region<String, String>>()));
String unknownRegionName = "UNKNOWN_REGION";
Result<RegionAPI.GetRegionResponse> result = operationHandler.process(serializationServiceStub,
- MessageUtil.makeGetRegionRequest(unknownRegionName), emptyCache);
+ MessageUtil.makeGetRegionRequest(unknownRegionName), new ExecutionContext(emptyCache));
Assert.assertTrue(result instanceof Failure);
Assert.assertEquals(ProtocolErrorCode.REGION_NOT_FOUND.codeValue,
result.getErrorMessage().getError().getErrorCode());
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRequestOperationHandlerJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRequestOperationHandlerJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRequestOperationHandlerJUnitTest.java
index 0213bf7..af35f6b 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRequestOperationHandlerJUnitTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRequestOperationHandlerJUnitTest.java
@@ -16,6 +16,8 @@ package org.apache.geode.protocol.protobuf.operations;
import com.google.protobuf.ByteString;
import org.apache.geode.cache.Region;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.protobuf.BasicTypes;
import org.apache.geode.protocol.protobuf.Failure;
import org.apache.geode.protocol.protobuf.ProtocolErrorCode;
@@ -70,10 +72,10 @@ public class GetRequestOperationHandlerJUnitTest extends OperationHandlerJUnitTe
@Test
public void processReturnsTheEncodedValueFromTheRegion()
throws CodecAlreadyRegisteredForTypeException, UnsupportedEncodingTypeException,
- CodecNotRegisteredForTypeException {
+ CodecNotRegisteredForTypeException, InvalidExecutionContextException {
RegionAPI.GetRequest getRequest = generateTestRequest(false, false, false);
- Result<RegionAPI.GetResponse> result =
- operationHandler.process(serializationServiceStub, getRequest, cacheStub);
+ Result<RegionAPI.GetResponse> result = operationHandler.process(serializationServiceStub,
+ getRequest, new ExecutionContext(cacheStub));
Assert.assertTrue(result instanceof Success);
Assert.assertEquals(BasicTypes.EncodedValue.ValueCase.STRINGRESULT,
@@ -85,10 +87,10 @@ public class GetRequestOperationHandlerJUnitTest extends OperationHandlerJUnitTe
@Test
public void processReturnsUnsucessfulResponseForInvalidRegion()
throws CodecAlreadyRegisteredForTypeException, UnsupportedEncodingTypeException,
- CodecNotRegisteredForTypeException {
+ CodecNotRegisteredForTypeException, InvalidExecutionContextException {
RegionAPI.GetRequest getRequest = generateTestRequest(true, false, false);
- Result<RegionAPI.GetResponse> response =
- operationHandler.process(serializationServiceStub, getRequest, cacheStub);
+ Result<RegionAPI.GetResponse> response = operationHandler.process(serializationServiceStub,
+ getRequest, new ExecutionContext(cacheStub));
Assert.assertTrue(response instanceof Failure);
Assert.assertEquals(ProtocolErrorCode.REGION_NOT_FOUND.codeValue,
@@ -98,10 +100,10 @@ public class GetRequestOperationHandlerJUnitTest extends OperationHandlerJUnitTe
@Test
public void processReturnsKeyNotFoundWhenKeyIsNotFound()
throws CodecAlreadyRegisteredForTypeException, UnsupportedEncodingTypeException,
- CodecNotRegisteredForTypeException {
+ CodecNotRegisteredForTypeException, InvalidExecutionContextException {
RegionAPI.GetRequest getRequest = generateTestRequest(false, true, false);
- Result<RegionAPI.GetResponse> response =
- operationHandler.process(serializationServiceStub, getRequest, cacheStub);
+ Result<RegionAPI.GetResponse> response = operationHandler.process(serializationServiceStub,
+ getRequest, new ExecutionContext(cacheStub));
Assert.assertTrue(response instanceof Success);
}
@@ -109,10 +111,10 @@ public class GetRequestOperationHandlerJUnitTest extends OperationHandlerJUnitTe
@Test
public void processReturnsLookupFailureWhenKeyFoundWithNoValue()
throws CodecAlreadyRegisteredForTypeException, UnsupportedEncodingTypeException,
- CodecNotRegisteredForTypeException {
+ CodecNotRegisteredForTypeException, InvalidExecutionContextException {
RegionAPI.GetRequest getRequest = generateTestRequest(false, false, true);
- Result<RegionAPI.GetResponse> response =
- operationHandler.process(serializationServiceStub, getRequest, cacheStub);
+ Result<RegionAPI.GetResponse> response = operationHandler.process(serializationServiceStub,
+ getRequest, new ExecutionContext(cacheStub));
Assert.assertTrue(response instanceof Success);
}
@@ -120,7 +122,8 @@ public class GetRequestOperationHandlerJUnitTest extends OperationHandlerJUnitTe
@Test
public void processReturnsErrorWhenUnableToDecodeRequest()
throws CodecAlreadyRegisteredForTypeException, UnsupportedEncodingTypeException,
- CodecNotRegisteredForTypeException, UnsupportedEncodingException {
+ CodecNotRegisteredForTypeException, UnsupportedEncodingException,
+ InvalidExecutionContextException {
CodecNotRegisteredForTypeException exception =
new CodecNotRegisteredForTypeException("error finding codec for type");
when(serializationServiceStub.decode(any(), any())).thenThrow(exception);
@@ -132,8 +135,8 @@ public class GetRequestOperationHandlerJUnitTest extends OperationHandlerJUnitTe
.setCustomEncodedValue(customEncodedValueBuilder).build();
RegionAPI.GetRequest getRequest =
ProtobufRequestUtilities.createGetRequest(TEST_REGION, encodedKey).getGetRequest();
- Result<RegionAPI.GetResponse> response =
- operationHandler.process(serializationServiceStub, getRequest, cacheStub);
+ Result<RegionAPI.GetResponse> response = operationHandler.process(serializationServiceStub,
+ getRequest, new ExecutionContext(cacheStub));
Assert.assertTrue(response instanceof Failure);
Assert.assertEquals(ProtocolErrorCode.VALUE_ENCODING_ERROR.codeValue,
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/PutAllRequestOperationHandlerJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/PutAllRequestOperationHandlerJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/PutAllRequestOperationHandlerJUnitTest.java
index c13d542..d3fff49 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/PutAllRequestOperationHandlerJUnitTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/PutAllRequestOperationHandlerJUnitTest.java
@@ -15,6 +15,8 @@
package org.apache.geode.protocol.protobuf.operations;
import org.apache.geode.cache.Region;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.protobuf.BasicTypes;
import org.apache.geode.protocol.protobuf.RegionAPI;
import org.apache.geode.protocol.protobuf.Result;
@@ -66,12 +68,13 @@ public class PutAllRequestOperationHandlerJUnitTest extends OperationHandlerJUni
}
@Test
- public void processInsertsMultipleValidEntriesInCache() throws UnsupportedEncodingTypeException,
- CodecNotRegisteredForTypeException, CodecAlreadyRegisteredForTypeException {
+ public void processInsertsMultipleValidEntriesInCache()
+ throws UnsupportedEncodingTypeException, CodecNotRegisteredForTypeException,
+ CodecAlreadyRegisteredForTypeException, InvalidExecutionContextException {
PutAllRequestOperationHandler operationHandler = new PutAllRequestOperationHandler();
Result<RegionAPI.PutAllResponse> result = operationHandler.process(serializationServiceStub,
- generateTestRequest(false, true), cacheStub);
+ generateTestRequest(false, true), new ExecutionContext(cacheStub));
Assert.assertTrue(result instanceof Success);
@@ -85,7 +88,7 @@ public class PutAllRequestOperationHandlerJUnitTest extends OperationHandlerJUni
PutAllRequestOperationHandler operationHandler = new PutAllRequestOperationHandler();
Result<RegionAPI.PutAllResponse> result = operationHandler.process(serializationServiceStub,
- generateTestRequest(true, true), cacheStub);
+ generateTestRequest(true, true), new ExecutionContext(cacheStub));
assertTrue(result instanceof Success);
verify(regionMock).put(TEST_KEY1, TEST_VALUE1);
@@ -104,7 +107,7 @@ public class PutAllRequestOperationHandlerJUnitTest extends OperationHandlerJUni
PutAllRequestOperationHandler operationHandler = new PutAllRequestOperationHandler();
Result<RegionAPI.PutAllResponse> result = operationHandler.process(serializationServiceStub,
- generateTestRequest(false, false), cacheStub);
+ generateTestRequest(false, false), new ExecutionContext(cacheStub));
assertTrue(result instanceof Success);
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/PutRequestOperationHandlerJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/PutRequestOperationHandlerJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/PutRequestOperationHandlerJUnitTest.java
index fc697e4..64d9f67 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/PutRequestOperationHandlerJUnitTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/PutRequestOperationHandlerJUnitTest.java
@@ -16,6 +16,8 @@ package org.apache.geode.protocol.protobuf.operations;
import com.google.protobuf.ByteString;
import org.apache.geode.cache.Region;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.protobuf.BasicTypes;
import org.apache.geode.protocol.protobuf.Failure;
import org.apache.geode.protocol.protobuf.ProtocolErrorCode;
@@ -61,11 +63,12 @@ public class PutRequestOperationHandlerJUnitTest extends OperationHandlerJUnitTe
}
@Test
- public void test_puttingTheEncodedEntryIntoRegion() throws UnsupportedEncodingTypeException,
- CodecNotRegisteredForTypeException, CodecAlreadyRegisteredForTypeException {
+ public void test_puttingTheEncodedEntryIntoRegion()
+ throws UnsupportedEncodingTypeException, CodecNotRegisteredForTypeException,
+ CodecAlreadyRegisteredForTypeException, InvalidExecutionContextException {
PutRequestOperationHandler operationHandler = new PutRequestOperationHandler();
- Result<RegionAPI.PutResponse> result =
- operationHandler.process(serializationServiceStub, generateTestRequest(), cacheStub);
+ Result<RegionAPI.PutResponse> result = operationHandler.process(serializationServiceStub,
+ generateTestRequest(), new ExecutionContext(cacheStub));
assertTrue(result instanceof Success);
@@ -74,9 +77,9 @@ public class PutRequestOperationHandlerJUnitTest extends OperationHandlerJUnitTe
}
@Test
- public void test_invalidEncodingType()
- throws CodecAlreadyRegisteredForTypeException, UnsupportedEncodingTypeException,
- CodecNotRegisteredForTypeException, UnsupportedEncodingException {
+ public void test_invalidEncodingType() throws CodecAlreadyRegisteredForTypeException,
+ UnsupportedEncodingTypeException, CodecNotRegisteredForTypeException,
+ UnsupportedEncodingException, InvalidExecutionContextException {
String exceptionText = "unsupported type!";
UnsupportedEncodingTypeException exception =
new UnsupportedEncodingTypeException(exceptionText);
@@ -95,8 +98,8 @@ public class PutRequestOperationHandlerJUnitTest extends OperationHandlerJUnitTe
BasicTypes.Entry testEntry = ProtobufUtilities.createEntry(encodedKey, testValue);
RegionAPI.PutRequest putRequest =
ProtobufRequestUtilities.createPutRequest(TEST_REGION, testEntry).getPutRequest();
- Result<RegionAPI.PutResponse> result =
- operationHandler.process(serializationServiceStub, putRequest, cacheStub);
+ Result<RegionAPI.PutResponse> result = operationHandler.process(serializationServiceStub,
+ putRequest, new ExecutionContext(cacheStub));
assertTrue(result instanceof Failure);
assertEquals(ProtocolErrorCode.VALUE_ENCODING_ERROR.codeValue,
@@ -104,12 +107,13 @@ public class PutRequestOperationHandlerJUnitTest extends OperationHandlerJUnitTe
}
@Test
- public void test_RegionNotFound() throws CodecAlreadyRegisteredForTypeException,
- UnsupportedEncodingTypeException, CodecNotRegisteredForTypeException {
+ public void test_RegionNotFound()
+ throws CodecAlreadyRegisteredForTypeException, UnsupportedEncodingTypeException,
+ CodecNotRegisteredForTypeException, InvalidExecutionContextException {
when(cacheStub.getRegion(TEST_REGION)).thenReturn(null);
PutRequestOperationHandler operationHandler = new PutRequestOperationHandler();
- Result<RegionAPI.PutResponse> result =
- operationHandler.process(serializationServiceStub, generateTestRequest(), cacheStub);
+ Result<RegionAPI.PutResponse> result = operationHandler.process(serializationServiceStub,
+ generateTestRequest(), new ExecutionContext(cacheStub));
assertTrue(result instanceof Failure);
assertEquals(ProtocolErrorCode.REGION_NOT_FOUND.codeValue,
@@ -117,13 +121,14 @@ public class PutRequestOperationHandlerJUnitTest extends OperationHandlerJUnitTe
}
@Test
- public void test_RegionThrowsClasscastException() throws CodecAlreadyRegisteredForTypeException,
- UnsupportedEncodingTypeException, CodecNotRegisteredForTypeException {
+ public void test_RegionThrowsClasscastException()
+ throws CodecAlreadyRegisteredForTypeException, UnsupportedEncodingTypeException,
+ CodecNotRegisteredForTypeException, InvalidExecutionContextException {
when(regionMock.put(any(), any())).thenThrow(ClassCastException.class);
PutRequestOperationHandler operationHandler = new PutRequestOperationHandler();
- Result<RegionAPI.PutResponse> result =
- operationHandler.process(serializationServiceStub, generateTestRequest(), cacheStub);
+ Result<RegionAPI.PutResponse> result = operationHandler.process(serializationServiceStub,
+ generateTestRequest(), new ExecutionContext(cacheStub));
assertTrue(result instanceof Failure);
assertEquals(ProtocolErrorCode.CONSTRAINT_VIOLATION.codeValue,
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/RemoveRequestOperationHandlerJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/RemoveRequestOperationHandlerJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/RemoveRequestOperationHandlerJUnitTest.java
index 3b917b7..47d6231 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/RemoveRequestOperationHandlerJUnitTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/RemoveRequestOperationHandlerJUnitTest.java
@@ -16,6 +16,8 @@ package org.apache.geode.protocol.protobuf.operations;
import com.google.protobuf.ByteString;
import org.apache.geode.cache.Region;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.protobuf.BasicTypes;
import org.apache.geode.protocol.protobuf.ClientProtocol;
import org.apache.geode.protocol.protobuf.Failure;
@@ -71,10 +73,10 @@ public class RemoveRequestOperationHandlerJUnitTest extends OperationHandlerJUni
@Test
public void processValidKeyRemovesTheEntryAndReturnSuccess()
throws CodecAlreadyRegisteredForTypeException, UnsupportedEncodingTypeException,
- CodecNotRegisteredForTypeException {
+ CodecNotRegisteredForTypeException, InvalidExecutionContextException {
RegionAPI.RemoveRequest removeRequest = generateTestRequest(false, false).getRemoveRequest();
- Result<RegionAPI.RemoveResponse> result =
- operationHandler.process(serializationServiceStub, removeRequest, cacheStub);
+ Result<RegionAPI.RemoveResponse> result = operationHandler.process(serializationServiceStub,
+ removeRequest, new ExecutionContext(cacheStub));
assertTrue(result instanceof Success);
verify(regionStub).remove(TEST_KEY);
@@ -83,10 +85,10 @@ public class RemoveRequestOperationHandlerJUnitTest extends OperationHandlerJUni
@Test
public void processReturnsUnsucessfulResponseForInvalidRegion()
throws CodecAlreadyRegisteredForTypeException, UnsupportedEncodingTypeException,
- CodecNotRegisteredForTypeException {
+ CodecNotRegisteredForTypeException, InvalidExecutionContextException {
RegionAPI.RemoveRequest removeRequest = generateTestRequest(true, false).getRemoveRequest();
- Result<RegionAPI.RemoveResponse> result =
- operationHandler.process(serializationServiceStub, removeRequest, cacheStub);
+ Result<RegionAPI.RemoveResponse> result = operationHandler.process(serializationServiceStub,
+ removeRequest, new ExecutionContext(cacheStub));
assertTrue(result instanceof Failure);
assertEquals(ProtocolErrorCode.REGION_NOT_FOUND.codeValue,
@@ -96,10 +98,10 @@ public class RemoveRequestOperationHandlerJUnitTest extends OperationHandlerJUni
@Test
public void processReturnsSuccessWhenKeyIsNotFound()
throws CodecAlreadyRegisteredForTypeException, UnsupportedEncodingTypeException,
- CodecNotRegisteredForTypeException {
+ CodecNotRegisteredForTypeException, InvalidExecutionContextException {
RegionAPI.RemoveRequest removeRequest = generateTestRequest(false, true).getRemoveRequest();
- Result<RegionAPI.RemoveResponse> result =
- operationHandler.process(serializationServiceStub, removeRequest, cacheStub);
+ Result<RegionAPI.RemoveResponse> result = operationHandler.process(serializationServiceStub,
+ removeRequest, new ExecutionContext(cacheStub));
assertTrue(result instanceof Success);
}
@@ -107,7 +109,8 @@ public class RemoveRequestOperationHandlerJUnitTest extends OperationHandlerJUni
@Test
public void processReturnsErrorWhenUnableToDecodeRequest()
throws CodecAlreadyRegisteredForTypeException, UnsupportedEncodingTypeException,
- CodecNotRegisteredForTypeException, UnsupportedEncodingException {
+ CodecNotRegisteredForTypeException, UnsupportedEncodingException,
+ InvalidExecutionContextException {
CodecNotRegisteredForTypeException exception =
new CodecNotRegisteredForTypeException("error finding codec for type");
when(serializationServiceStub.decode(any(), any())).thenThrow(exception);
@@ -120,8 +123,8 @@ public class RemoveRequestOperationHandlerJUnitTest extends OperationHandlerJUni
RegionAPI.RemoveRequest removeRequest =
ProtobufRequestUtilities.createRemoveRequest(TEST_REGION, encodedKey).getRemoveRequest();;
- Result<RegionAPI.RemoveResponse> result =
- operationHandler.process(serializationServiceStub, removeRequest, cacheStub);
+ Result<RegionAPI.RemoveResponse> result = operationHandler.process(serializationServiceStub,
+ removeRequest, new ExecutionContext(cacheStub));
assertTrue(result instanceof Failure);
assertEquals(ProtocolErrorCode.VALUE_ENCODING_ERROR.codeValue,
[5/7] geode git commit: GEODE-3406: Exclude new exception class from
analyze serialiable test
Posted by hi...@apache.org.
GEODE-3406: Exclude new exception class from analyze serialiable test
Signed-off-by: Hitesh Khamesra <hk...@pivotal.io>
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/a4fc1ddf
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/a4fc1ddf
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/a4fc1ddf
Branch: refs/heads/develop
Commit: a4fc1ddf628766cd5f3e7fb9e9469f523d20dbf2
Parents: 0eb320f
Author: Alexander Murmann <am...@pivotal.io>
Authored: Fri Aug 18 11:01:26 2017 -0700
Committer: Hitesh Khamesra <hk...@pivotal.io>
Committed: Tue Aug 22 10:55:27 2017 -0700
----------------------------------------------------------------------
.../resources/org/apache/geode/codeAnalysis/excludedClasses.txt | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/a4fc1ddf/geode-core/src/test/resources/org/apache/geode/codeAnalysis/excludedClasses.txt
----------------------------------------------------------------------
diff --git a/geode-core/src/test/resources/org/apache/geode/codeAnalysis/excludedClasses.txt b/geode-core/src/test/resources/org/apache/geode/codeAnalysis/excludedClasses.txt
index 43220fa..fbd582a 100644
--- a/geode-core/src/test/resources/org/apache/geode/codeAnalysis/excludedClasses.txt
+++ b/geode-core/src/test/resources/org/apache/geode/codeAnalysis/excludedClasses.txt
@@ -10,6 +10,7 @@ org/apache/geode/distributed/internal/DistributionManager
org/apache/geode/internal/ExitCode
org/apache/geode/internal/JarDeployer
org/apache/geode/internal/cache/BackupLock
+org/apache/geode/internal/cache/tier/sockets/InvalidExecutionContextException
org/apache/geode/internal/logging/GemFireLevel
org/apache/geode/internal/logging/LoggingThreadGroup
org/apache/geode/internal/logging/LogWriterImpl
[2/7] geode git commit: GEODE-3406: Locator accepts Protobuf requests
Posted by hi...@apache.org.
GEODE-3406: Locator accepts Protobuf requests
Also addresses GEODE-3400, GEODE-3399
This allows the locator to respond to Protobuf requests. Currently it
will only be able to respond to getAvailableServers.
To enable this we are introducing a new value of "0" that will be sent
in place of the Gossip version. After it we expect the same magic byte
("110") as in AcceptorImpl.
This also is gated by the `geode.feature-protobuf-protocol` system
property.
The getAvailableServers request handler now uses the locator directly,
since we are on the locator.
Signed-off-by: Brian Rowe <br...@pivotal.io>
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/530f48f3
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/530f48f3
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/530f48f3
Branch: refs/heads/develop
Commit: 530f48f35a96c4f8af7e51ed03b1ee2e5e150ebd
Parents: be45511
Author: Alexander Murmann <am...@pivotal.io>
Authored: Mon Aug 14 15:08:14 2017 -0700
Committer: Hitesh Khamesra <hk...@pivotal.io>
Committed: Tue Aug 22 10:52:50 2017 -0700
----------------------------------------------------------------------
.../distributed/internal/InternalLocator.java | 4 +-
.../distributed/internal/ServerLocator.java | 4 +
.../internal/tcpserver/TcpServer.java | 77 +++++++----
.../geode/internal/cache/InternalCache.java | 4 +-
.../ClientProtoclMessageHandlerLoader.java | 64 +++++++++
.../sockets/ClientProtocolMessageHandler.java | 7 +-
.../ClientProtocolMessageHandlerLoader.java | 64 +++++++++
.../cache/tier/sockets/ExecutionContext.java | 54 ++++++++
.../GenericProtocolServerConnection.java | 3 +-
.../InvalidExecutionContextException.java | 33 +++++
.../AutoConnectionSourceImplJUnitTest.java | 8 +-
.../tcpserver/TCPServerSSLJUnitTest.java | 2 +-
.../internal/tcpserver/TcpServerJUnitTest.java | 2 +-
.../protocol/operations/OperationHandler.java | 6 +-
.../protocol/protobuf/ProtobufOpsProcessor.java | 17 ++-
.../protobuf/ProtobufStreamProcessor.java | 29 ++--
.../protocol/protobuf/ProtocolErrorCode.java | 1 +
.../GetAllRequestOperationHandler.java | 8 +-
.../GetAvailableServersOperationHandler.java | 65 ++-------
.../GetRegionNamesRequestOperationHandler.java | 8 +-
.../GetRegionRequestOperationHandler.java | 8 +-
.../operations/GetRequestOperationHandler.java | 8 +-
.../PutAllRequestOperationHandler.java | 8 +-
.../operations/PutRequestOperationHandler.java | 8 +-
.../RemoveRequestOperationHandler.java | 9 +-
.../protocol/GetAvailableServersDUnitTest.java | 108 ---------------
.../RoundTripLocatorConnectionJUnitTest.java | 132 +++++++++++++++++++
.../protobuf/ProtobufStreamProcessorTest.java | 4 +-
.../GetAllRequestOperationHandlerJUnitTest.java | 18 +--
...ailableServersOperationHandlerJUnitTest.java | 97 ++++----------
...onNamesRequestOperationHandlerJUnitTest.java | 26 ++--
...tRegionRequestOperationHandlerJUnitTest.java | 16 ++-
.../GetRequestOperationHandlerJUnitTest.java | 33 ++---
.../PutAllRequestOperationHandlerJUnitTest.java | 13 +-
.../PutRequestOperationHandlerJUnitTest.java | 39 +++---
.../RemoveRequestOperationHandlerJUnitTest.java | 27 ++--
36 files changed, 618 insertions(+), 396 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalLocator.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalLocator.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalLocator.java
index 4725518..8d2daf6 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalLocator.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalLocator.java
@@ -315,7 +315,6 @@ public class InternalLocator extends Locator implements ConnectListener {
// TODO:GEODE-1243: this.server is now a TcpServer and it should store or return its non-zero
// port in a variable to use here
-
try {
newLocator.startPeerLocation(startDistributedSystem);
if (startDistributedSystem) {
@@ -500,7 +499,7 @@ public class InternalLocator extends Locator implements ConnectListener {
this.stats = new LocatorStats();
this.server = new TcpServer(port, this.bindAddress, null, this.config, this.handler,
- new DelayedPoolStatHelper(), group, this.toString());
+ new DelayedPoolStatHelper(), group, this.toString(), this);
}
// Reset the file names with the correct port number if startLocatorAndDS was called with port
@@ -636,7 +635,6 @@ public class InternalLocator extends Locator implements ConnectListener {
*/
private void startDistributedSystem() throws UnknownHostException {
InternalDistributedSystem existing = InternalDistributedSystem.getConnectedInstance();
-
if (existing != null) {
// LOG: changed from config to info
logger.info(LocalizedMessage
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-core/src/main/java/org/apache/geode/distributed/internal/ServerLocator.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/ServerLocator.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/ServerLocator.java
index fb66b4c..27c557c 100755
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/ServerLocator.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/ServerLocator.java
@@ -103,6 +103,10 @@ public class ServerLocator implements TcpHandler, DistributionAdvisee {
this.stats = null;
}
+ public LocatorLoadSnapshot getLoadSnapshot() {
+ return loadSnapshot;
+ }
+
public ServerLocator(int port, InetAddress bindAddress, String hostNameForClients, File logFile,
ProductUseLog productUseLogWriter, String memberName, InternalDistributedSystem ds,
LocatorStats stats) throws IOException {
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/TcpServer.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/TcpServer.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/TcpServer.java
index 976f504..c3d51c1 100755
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/TcpServer.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/TcpServer.java
@@ -22,6 +22,7 @@ import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.distributed.internal.DistributionConfigImpl;
import org.apache.geode.distributed.internal.DistributionStats;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
+import org.apache.geode.distributed.internal.InternalLocator;
import org.apache.geode.distributed.internal.PoolStatHelper;
import org.apache.geode.distributed.internal.PooledExecutorWithDMStats;
import org.apache.geode.internal.DSFIDFactory;
@@ -31,6 +32,10 @@ import org.apache.geode.internal.VersionedDataInputStream;
import org.apache.geode.internal.VersionedDataOutputStream;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.tier.Acceptor;
+import org.apache.geode.internal.cache.tier.sockets.AcceptorImpl;
+import org.apache.geode.internal.cache.tier.sockets.ClientProtocolMessageHandlerLoader;
+import org.apache.geode.internal.cache.tier.sockets.ClientProtocolMessageHandler;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
import org.apache.geode.internal.cache.tier.sockets.HandShake;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.net.SocketCreator;
@@ -49,7 +54,6 @@ import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;
import java.net.URL;
-import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
@@ -78,6 +82,7 @@ public class TcpServer {
* <p>
* This should be incremented if the gossip message structures change
* <p>
+ * 0 - special indicator of a non-gossip message from a client<br>
* 1000 - gemfire 5.5 - using java serialization<br>
* 1001 - 5.7 - using DataSerializable and supporting server locator messages.<br>
* 1002 - 7.1 - sending GemFire version along with GOSSIP_VERSION in each request.
@@ -86,6 +91,7 @@ public class TcpServer {
* version number
*/
public final static int GOSSIPVERSION = 1002;
+ public final static int NON_GOSSIP_REQUEST_VERSION = 0;
// Don't change it ever. We did NOT send GemFire version in a Gossip request till 1001 version.
// This GOSSIPVERSION is used in _getVersionForAddress request for getting GemFire version of a
// GossipServer.
@@ -120,6 +126,7 @@ public class TcpServer {
private InetAddress bind_address;
private volatile boolean shuttingDown = false; // GemStoneAddition
private final PoolStatHelper poolHelper;
+ private InternalLocator internalLocator;
private final TcpHandler handler;
private PooledExecutorWithDMStats executor;
@@ -143,11 +150,12 @@ public class TcpServer {
public TcpServer(int port, InetAddress bind_address, Properties sslConfig,
DistributionConfigImpl cfg, TcpHandler handler, PoolStatHelper poolHelper,
- ThreadGroup threadGroup, String threadName) {
+ ThreadGroup threadGroup, String threadName, InternalLocator internalLocator) {
this.port = port;
this.bind_address = bind_address;
this.handler = handler;
this.poolHelper = poolHelper;
+ this.internalLocator = internalLocator;
// register DSFID types first; invoked explicitly so that all message type
// initializations do not happen in first deserialization on a possibly
// "precious" thread
@@ -334,42 +342,46 @@ public class TcpServer {
* fix for bug 33711 - client requests are spun off to another thread for processing. Requests are
* synchronized in processGossip.
*/
- private void processRequest(final Socket sock) {
+ private void processRequest(final Socket socket) {
executor.execute(() -> {
long startTime = DistributionStats.getStatTime();
DataInputStream input = null;
Object request, response;
try {
- sock.setSoTimeout(READ_TIMEOUT);
- getSocketCreator().configureServerSSLSocket(sock);
+ socket.setSoTimeout(READ_TIMEOUT);
+ getSocketCreator().configureServerSSLSocket(socket);
try {
- input = new DataInputStream(sock.getInputStream());
+ input = new DataInputStream(socket.getInputStream());
} catch (StreamCorruptedException e) {
// Some garbage can be left on the socket stream
// if a peer disappears at exactly the wrong moment.
log.debug("Discarding illegal request from "
- + (sock.getInetAddress().getHostAddress() + ":" + sock.getPort()), e);
+ + (socket.getInetAddress().getHostAddress() + ":" + socket.getPort()), e);
return;
}
- int gossipVersion = readGossipVersion(sock, input);
+ int gossipVersion = readGossipVersion(socket, input);
short versionOrdinal;
+ if (gossipVersion == NON_GOSSIP_REQUEST_VERSION) {
+ if (input.readUnsignedByte() == AcceptorImpl.PROTOBUF_CLIENT_SERVER_PROTOCOL
+ && Boolean.getBoolean("geode.feature-protobuf-protocol")) {
+ ClientProtocolMessageHandler messageHandler = ClientProtocolMessageHandlerLoader.load();
+ messageHandler.receiveMessage(input, socket.getOutputStream(),
+ new ExecutionContext(internalLocator));
+ } else {
+ rejectUnknownProtocolConnection(socket, gossipVersion);
+ return;
+ }
+ }
if (gossipVersion <= getCurrentGossipVersion()
&& GOSSIP_TO_GEMFIRE_VERSION_MAP.containsKey(gossipVersion)) {
// Create a versioned stream to remember sender's GemFire version
versionOrdinal = (short) GOSSIP_TO_GEMFIRE_VERSION_MAP.get(gossipVersion);
} else {
// Close the socket. We can not accept requests from a newer version
- try {
- sock.getOutputStream().write("unknown protocol version".getBytes());
- sock.getOutputStream().flush();
- } catch (IOException e) {
- log.debug(
- "exception in sending reply to process using unknown protocol " + gossipVersion, e);
- }
- sock.close();
+ rejectUnknownProtocolConnection(socket, gossipVersion);
return;
}
if (Version.GFE_71.compareTo(versionOrdinal) <= 0) {
@@ -378,13 +390,13 @@ public class TcpServer {
}
if (log.isDebugEnabled() && versionOrdinal != Version.CURRENT_ORDINAL) {
- log.debug("Locator reading request from " + sock.getInetAddress() + " with version "
+ log.debug("Locator reading request from " + socket.getInetAddress() + " with version "
+ Version.fromOrdinal(versionOrdinal, false));
}
input = new VersionedDataInputStream(input, Version.fromOrdinal(versionOrdinal, false));
request = DataSerializer.readObject(input);
if (log.isDebugEnabled()) {
- log.debug("Locator received request " + request + " from " + sock.getInetAddress());
+ log.debug("Locator received request " + request + " from " + socket.getInetAddress());
}
if (request instanceof ShutdownRequest) {
shuttingDown = true;
@@ -405,7 +417,7 @@ public class TcpServer {
startTime = DistributionStats.getStatTime();
if (response != null) {
- DataOutputStream output = new DataOutputStream(sock.getOutputStream());
+ DataOutputStream output = new DataOutputStream(socket.getOutputStream());
if (versionOrdinal != Version.CURRENT_ORDINAL) {
output =
new VersionedDataOutputStream(output, Version.fromOrdinal(versionOrdinal, false));
@@ -422,19 +434,19 @@ public class TcpServer {
// ignore
} catch (ClassNotFoundException ex) {
String sender = null;
- if (sock != null) {
- sender = sock.getInetAddress().getHostAddress();
+ if (socket != null) {
+ sender = socket.getInetAddress().getHostAddress();
}
log.info("Unable to process request from " + sender + " exception=" + ex.getMessage());
} catch (Exception ex) {
String sender = null;
- if (sock != null) {
- sender = sock.getInetAddress().getHostAddress();
+ if (socket != null) {
+ sender = socket.getInetAddress().getHostAddress();
}
if (ex instanceof IOException) {
// IOException could be caused by a client failure. Don't
// log with severe.
- if (!sock.isClosed()) {
+ if (!socket.isClosed()) {
log.info("Exception in processing request from " + sender, ex);
}
} else {
@@ -447,8 +459,8 @@ public class TcpServer {
} catch (Throwable ex) {
SystemFailure.checkFailure();
String sender = null;
- if (sock != null) {
- sender = sock.getInetAddress().getHostAddress();
+ if (socket != null) {
+ sender = socket.getInetAddress().getHostAddress();
}
try {
log.fatal("Exception in processing request from " + sender, ex);
@@ -461,7 +473,7 @@ public class TcpServer {
}
} finally {
try {
- sock.close();
+ socket.close();
} catch (IOException ignore) {
// ignore
}
@@ -469,6 +481,17 @@ public class TcpServer {
});
}
+ private void rejectUnknownProtocolConnection(Socket socket, int gossipVersion)
+ throws IOException {
+ try {
+ socket.getOutputStream().write("unknown protocol version".getBytes());
+ socket.getOutputStream().flush();
+ } catch (IOException e) {
+ log.debug("exception in sending reply to process using unknown protocol " + gossipVersion, e);
+ }
+ socket.close();
+ }
+
private int readGossipVersion(Socket sock, DataInputStream input) throws Exception {
// read the first byte & check for an improperly configured client pool trying
// to contact a cache server
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java b/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java
index 84aa66e..4c7a6ef 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java
@@ -75,7 +75,9 @@ import org.apache.geode.pdx.internal.TypeRegistry;
*/
public interface InternalCache extends Cache, Extensible<Cache>, CacheTime {
- InternalDistributedMember getMyId();
+ default InternalDistributedMember getMyId() {
+ return null;
+ }
Collection<DiskStore> listDiskStores();
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtoclMessageHandlerLoader.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtoclMessageHandlerLoader.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtoclMessageHandlerLoader.java
new file mode 100644
index 0000000..6654757
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtoclMessageHandlerLoader.java
@@ -0,0 +1,64 @@
+/*
+ * 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.sockets;
+
+import java.io.IOException;
+import java.net.Socket;
+import java.util.Iterator;
+import java.util.ServiceLoader;
+
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.tier.Acceptor;
+import org.apache.geode.internal.cache.tier.CachedRegionHelper;
+import org.apache.geode.internal.security.SecurityService;
+
+/**
+ * Creates instances of ServerConnection based on the connection mode provided.
+ */
+public class ClientProtoclMessageHandlerLoader {
+ private static ClientProtocolMessageHandler protobufProtocolHandler;
+ private static final Object protocolLoadLock = new Object();
+
+ public static ClientProtocolMessageHandler load() {
+ if (protobufProtocolHandler != null) {
+ return protobufProtocolHandler;
+ }
+
+ synchronized (protocolLoadLock) {
+ if (protobufProtocolHandler != null) {
+ return protobufProtocolHandler;
+ }
+
+ ServiceLoader<ClientProtocolMessageHandler> loader =
+ ServiceLoader.load(ClientProtocolMessageHandler.class);
+ Iterator<ClientProtocolMessageHandler> iterator = loader.iterator();
+
+ if (!iterator.hasNext()) {
+ throw new ServiceLoadingFailureException(
+ "ClientProtocolMessageHandler implementation not found in JVM");
+ }
+
+ ClientProtocolMessageHandler returnValue = iterator.next();
+
+ if (iterator.hasNext()) {
+ throw new ServiceLoadingFailureException(
+ "Multiple service implementations found for ClientProtocolMessageHandler");
+ }
+
+ return returnValue;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtocolMessageHandler.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtocolMessageHandler.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtocolMessageHandler.java
index 32e9e4b..38ab73e 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtocolMessageHandler.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtocolMessageHandler.java
@@ -15,12 +15,11 @@
package org.apache.geode.internal.cache.tier.sockets;
-import org.apache.geode.internal.cache.InternalCache;
-
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+
/**
* This is an interface that other modules can implement to hook into
* {@link GenericProtocolServerConnection} to handle messages sent to Geode.
@@ -30,6 +29,6 @@ import java.io.OutputStream;
* {@link GenericProtocolServerConnection}.
*/
public interface ClientProtocolMessageHandler {
- void receiveMessage(InputStream inputStream, OutputStream outputStream, InternalCache cache)
- throws IOException;
+ void receiveMessage(InputStream inputStream, OutputStream outputStream,
+ ExecutionContext executionContext) throws IOException;
}
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtocolMessageHandlerLoader.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtocolMessageHandlerLoader.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtocolMessageHandlerLoader.java
new file mode 100644
index 0000000..1dc6129
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtocolMessageHandlerLoader.java
@@ -0,0 +1,64 @@
+/*
+ * 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.sockets;
+
+import java.io.IOException;
+import java.net.Socket;
+import java.util.Iterator;
+import java.util.ServiceLoader;
+
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.tier.Acceptor;
+import org.apache.geode.internal.cache.tier.CachedRegionHelper;
+import org.apache.geode.internal.security.SecurityService;
+
+/**
+ * Creates instances of ServerConnection based on the connection mode provided.
+ */
+public class ClientProtocolMessageHandlerLoader {
+ private static ClientProtocolMessageHandler protobufProtocolHandler;
+ private static final Object protocolLoadLock = new Object();
+
+ public static ClientProtocolMessageHandler load() {
+ if (protobufProtocolHandler != null) {
+ return protobufProtocolHandler;
+ }
+
+ synchronized (protocolLoadLock) {
+ if (protobufProtocolHandler != null) {
+ return protobufProtocolHandler;
+ }
+
+ ServiceLoader<ClientProtocolMessageHandler> loader =
+ ServiceLoader.load(ClientProtocolMessageHandler.class);
+ Iterator<ClientProtocolMessageHandler> iterator = loader.iterator();
+
+ if (!iterator.hasNext()) {
+ throw new ServiceLoadingFailureException(
+ "ClientProtocolMessageHandler implementation not found in JVM");
+ }
+
+ ClientProtocolMessageHandler returnValue = iterator.next();
+
+ if (iterator.hasNext()) {
+ throw new ServiceLoadingFailureException(
+ "Multiple service implementations found for ClientProtocolMessageHandler");
+ }
+
+ return returnValue;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ExecutionContext.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ExecutionContext.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ExecutionContext.java
new file mode 100644
index 0000000..27da205
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ExecutionContext.java
@@ -0,0 +1,54 @@
+/*
+ * 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.sockets;
+
+import org.apache.geode.cache.Cache;
+import org.apache.geode.distributed.internal.InternalLocator;
+
+public class ExecutionContext {
+ private Cache cache;
+ private InternalLocator locator;
+
+ public ExecutionContext(Cache cache) {
+ this.cache = cache;
+ }
+
+ public ExecutionContext(InternalLocator locator) {
+ this.locator = locator;
+ }
+
+ // This throws if the cache isn't present because we know that non of the callers can take any
+ // reasonable action if the cache is not present
+ public Cache getCache() throws InvalidExecutionContextException {
+ if (cache != null) {
+ return cache;
+ } else {
+ throw new InvalidExecutionContextException(
+ "Execution context's cache was accessed but isn't present. Did this happen on a locator? Operations on the locator should not try to operate on a cache");
+ }
+ }
+
+ // This throws if the locator isn't present because we know that non of the callers can take any
+ // reasonable action if the locator is not present
+ public InternalLocator getLocator() throws InvalidExecutionContextException {
+ if (locator != null) {
+ return locator;
+ } else {
+ throw new InvalidExecutionContextException(
+ "Execution context's locator was accessed but isn't present. Did this happen on a server? Operations on the locator should not try to operate on a cache");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/GenericProtocolServerConnection.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/GenericProtocolServerConnection.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/GenericProtocolServerConnection.java
index 93a7f6f..8f6720e 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/GenericProtocolServerConnection.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/GenericProtocolServerConnection.java
@@ -61,7 +61,8 @@ public class GenericProtocolServerConnection extends ServerConnection {
if (!authenticator.isAuthenticated()) {
authenticator.receiveMessage(inputStream, outputStream, securityManager);
} else {
- messageHandler.receiveMessage(inputStream, outputStream, this.getCache());
+ messageHandler.receiveMessage(inputStream, outputStream,
+ new ExecutionContext(this.getCache()));
}
} catch (IOException e) {
logger.warn(e);
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/InvalidExecutionContextException.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/InvalidExecutionContextException.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/InvalidExecutionContextException.java
new file mode 100644
index 0000000..919e301
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/InvalidExecutionContextException.java
@@ -0,0 +1,33 @@
+/*
+ * 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.sockets;
+
+import org.apache.geode.annotations.Experimental;
+
+/*
+ * Indicates that OperationContext was missing required data. This will typically happen if a
+ * operation that is supposed to run on a server runs on a locator and receives a locator in its
+ * context instead of a cache. The reverse case applies as well.
+ */
+@Experimental
+public class InvalidExecutionContextException extends Exception {
+ public InvalidExecutionContextException(String message) {
+ super(message);
+ }
+
+ public InvalidExecutionContextException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-core/src/test/java/org/apache/geode/cache/client/internal/AutoConnectionSourceImplJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/cache/client/internal/AutoConnectionSourceImplJUnitTest.java b/geode-core/src/test/java/org/apache/geode/cache/client/internal/AutoConnectionSourceImplJUnitTest.java
index 5c33468..802620c 100644
--- a/geode-core/src/test/java/org/apache/geode/cache/client/internal/AutoConnectionSourceImplJUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/cache/client/internal/AutoConnectionSourceImplJUnitTest.java
@@ -18,11 +18,8 @@ import org.apache.geode.CancelCriterion;
import org.apache.geode.cache.*;
import org.apache.geode.cache.client.NoAvailableLocatorsException;
import org.apache.geode.cache.client.SubscriptionNotEnabledException;
-import org.apache.geode.cache.client.internal.AutoConnectionSourceImpl.UpdateLocatorListTask;
-import org.apache.geode.cache.client.internal.PoolImpl.PoolTask;
import org.apache.geode.cache.client.internal.locator.ClientConnectionRequest;
import org.apache.geode.cache.client.internal.locator.ClientConnectionResponse;
-import org.apache.geode.cache.client.internal.locator.LocatorListRequest;
import org.apache.geode.cache.client.internal.locator.LocatorListResponse;
import org.apache.geode.cache.query.QueryService;
import org.apache.geode.distributed.DistributedSystem;
@@ -63,7 +60,6 @@ import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
-import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -308,7 +304,7 @@ public class AutoConnectionSourceImplJUnitTest {
startFakeLocator();
int secondPort = AvailablePortHelper.getRandomAvailableTCPPort();
TcpServer server2 = new TcpServer(secondPort, InetAddress.getLocalHost(), null, null, handler,
- new FakeHelper(), Thread.currentThread().getThreadGroup(), "tcp server");
+ new FakeHelper(), Thread.currentThread().getThreadGroup(), "tcp server", null);
server2.start();
try {
@@ -392,7 +388,7 @@ public class AutoConnectionSourceImplJUnitTest {
private void startFakeLocator() throws UnknownHostException, IOException, InterruptedException {
server = new TcpServer(port, InetAddress.getLocalHost(), null, null, handler, new FakeHelper(),
- Thread.currentThread().getThreadGroup(), "Tcp Server");
+ Thread.currentThread().getThreadGroup(), "Tcp Server", null);
server.start();
Thread.sleep(500);
}
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-core/src/test/java/org/apache/geode/distributed/internal/tcpserver/TCPServerSSLJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/distributed/internal/tcpserver/TCPServerSSLJUnitTest.java b/geode-core/src/test/java/org/apache/geode/distributed/internal/tcpserver/TCPServerSSLJUnitTest.java
index 8a25aaf..229fbb9 100644
--- a/geode-core/src/test/java/org/apache/geode/distributed/internal/tcpserver/TCPServerSSLJUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/distributed/internal/tcpserver/TCPServerSSLJUnitTest.java
@@ -138,7 +138,7 @@ public class TCPServerSSLJUnitTest {
public DummyTcpServer(int port, InetAddress bind_address, Properties sslConfig,
DistributionConfigImpl cfg, TcpHandler handler, PoolStatHelper poolHelper,
ThreadGroup threadGroup, String threadName) {
- super(port, bind_address, sslConfig, cfg, handler, poolHelper, threadGroup, threadName);
+ super(port, bind_address, sslConfig, cfg, handler, poolHelper, threadGroup, threadName, null);
if (cfg == null) {
cfg = new DistributionConfigImpl(sslConfig);
}
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-core/src/test/java/org/apache/geode/distributed/internal/tcpserver/TcpServerJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/distributed/internal/tcpserver/TcpServerJUnitTest.java b/geode-core/src/test/java/org/apache/geode/distributed/internal/tcpserver/TcpServerJUnitTest.java
index eda0641..9d20e8c 100644
--- a/geode-core/src/test/java/org/apache/geode/distributed/internal/tcpserver/TcpServerJUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/distributed/internal/tcpserver/TcpServerJUnitTest.java
@@ -69,7 +69,7 @@ public class TcpServerJUnitTest {
stats = new SimpleStats();
server = new TcpServer(port, localhost, new Properties(), null, handler, stats,
- Thread.currentThread().getThreadGroup(), "server thread");
+ Thread.currentThread().getThreadGroup(), "server thread", null);
server.start();
}
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/main/java/org/apache/geode/protocol/operations/OperationHandler.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/operations/OperationHandler.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/operations/OperationHandler.java
index aa6d79e..5d9012f 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/operations/OperationHandler.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/operations/OperationHandler.java
@@ -15,7 +15,8 @@
package org.apache.geode.protocol.operations;
import org.apache.geode.annotations.Experimental;
-import org.apache.geode.cache.Cache;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.protobuf.ProtobufOpsProcessor;
import org.apache.geode.protocol.protobuf.Result;
import org.apache.geode.serialization.SerializationService;
@@ -32,6 +33,7 @@ public interface OperationHandler<Req, Resp> {
* Decode the message, deserialize contained values using the serialization service, do the work
* indicated on the provided cache, and return a response.
*/
- Result<Resp> process(SerializationService serializationService, Req request, Cache cache);
+ Result<Resp> process(SerializationService serializationService, Req request,
+ ExecutionContext executionContext) throws InvalidExecutionContextException;
}
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtobufOpsProcessor.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtobufOpsProcessor.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtobufOpsProcessor.java
index 7d75b4a..76f81e7 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtobufOpsProcessor.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtobufOpsProcessor.java
@@ -15,8 +15,10 @@
package org.apache.geode.protocol.protobuf;
import org.apache.geode.annotations.Experimental;
-import org.apache.geode.cache.Cache;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.protobuf.registry.OperationContextRegistry;
+import org.apache.geode.protocol.protobuf.utilities.ProtobufResponseUtilities;
import org.apache.geode.serialization.SerializationService;
/**
@@ -35,12 +37,19 @@ public class ProtobufOpsProcessor {
this.operationContextRegistry = operationContextRegistry;
}
- public ClientProtocol.Response process(ClientProtocol.Request request, Cache cache) {
+ public ClientProtocol.Response process(ClientProtocol.Request request, ExecutionContext context) {
ClientProtocol.Request.RequestAPICase requestType = request.getRequestAPICase();
OperationContext operationContext = operationContextRegistry.getOperationContext(requestType);
ClientProtocol.Response.Builder builder;
- Result result = operationContext.getOperationHandler().process(serializationService,
- operationContext.getFromRequest().apply(request), cache);
+ Result result;
+ try {
+ result = operationContext.getOperationHandler().process(serializationService,
+ operationContext.getFromRequest().apply(request), context);
+ } catch (InvalidExecutionContextException e) {
+ result = Failure.of(ProtobufResponseUtilities.makeErrorResponse(
+ ProtocolErrorCode.UNSUPPORTED_OPERATION.codeValue,
+ "Invalid execution context found for operation."));
+ }
builder = (ClientProtocol.Response.Builder) result.map(operationContext.getToResponse(),
operationContext.getToErrorResponse());
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtobufStreamProcessor.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtobufStreamProcessor.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtobufStreamProcessor.java
index 648ab3c..d04e49e 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtobufStreamProcessor.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtobufStreamProcessor.java
@@ -20,9 +20,8 @@ import java.io.InputStream;
import java.io.OutputStream;
import org.apache.geode.annotations.Experimental;
-import org.apache.geode.cache.Cache;
-import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.tier.sockets.ClientProtocolMessageHandler;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
import org.apache.geode.protocol.exception.InvalidProtocolMessageException;
import org.apache.geode.protocol.protobuf.registry.OperationContextRegistry;
import org.apache.geode.protocol.protobuf.serializer.ProtobufProtocolSerializer;
@@ -45,29 +44,29 @@ public class ProtobufStreamProcessor implements ClientProtocolMessageHandler {
new OperationContextRegistry());
}
- public void processOneMessage(InputStream inputStream, OutputStream outputStream, Cache cache)
- throws InvalidProtocolMessageException, IOException {
+ @Override
+ public void receiveMessage(InputStream inputStream, OutputStream outputStream,
+ ExecutionContext executionContext) throws IOException {
+ try {
+ processOneMessage(inputStream, outputStream, executionContext);
+ } catch (InvalidProtocolMessageException e) {
+ throw new IOException(e);
+ }
+ }
+
+ private void processOneMessage(InputStream inputStream, OutputStream outputStream,
+ ExecutionContext executionContext) throws InvalidProtocolMessageException, IOException {
ClientProtocol.Message message = protobufProtocolSerializer.deserialize(inputStream);
if (message == null) {
throw new EOFException("Tried to deserialize protobuf message at EOF");
}
ClientProtocol.Request request = message.getRequest();
- ClientProtocol.Response response = protobufOpsProcessor.process(request, cache);
+ ClientProtocol.Response response = protobufOpsProcessor.process(request, executionContext);
ClientProtocol.MessageHeader responseHeader =
ProtobufUtilities.createMessageHeaderForRequest(message);
ClientProtocol.Message responseMessage =
ProtobufUtilities.createProtobufResponse(responseHeader, response);
protobufProtocolSerializer.serialize(responseMessage, outputStream);
}
-
- @Override
- public void receiveMessage(InputStream inputStream, OutputStream outputStream,
- InternalCache cache) throws IOException {
- try {
- processOneMessage(inputStream, outputStream, cache);
- } catch (InvalidProtocolMessageException e) {
- throw new IOException(e);
- }
- }
}
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtocolErrorCode.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtocolErrorCode.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtocolErrorCode.java
index e3b262d..6a6f605 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtocolErrorCode.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtocolErrorCode.java
@@ -18,6 +18,7 @@ public enum ProtocolErrorCode {
GENERIC_FAILURE(1000),
VALUE_ENCODING_ERROR(1100),
UNSUPPORTED_VERSION(1101),
+ UNSUPPORTED_OPERATION(1102),
AUTHENTICATION_FAILED(1200),
AUTHORIZATION_FAILED(1201),
UNAUTHORIZED_REQUEST(1202),
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetAllRequestOperationHandler.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetAllRequestOperationHandler.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetAllRequestOperationHandler.java
index 607d1d2..75274c1 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetAllRequestOperationHandler.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetAllRequestOperationHandler.java
@@ -19,8 +19,9 @@ import java.util.Map;
import java.util.Set;
import org.apache.geode.annotations.Experimental;
-import org.apache.geode.cache.Cache;
import org.apache.geode.cache.Region;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.operations.OperationHandler;
import org.apache.geode.protocol.protobuf.BasicTypes;
import org.apache.geode.protocol.protobuf.Failure;
@@ -40,9 +41,10 @@ public class GetAllRequestOperationHandler
@Override
public Result<RegionAPI.GetAllResponse> process(SerializationService serializationService,
- RegionAPI.GetAllRequest request, Cache cache) {
+ RegionAPI.GetAllRequest request, ExecutionContext executionContext)
+ throws InvalidExecutionContextException {
String regionName = request.getRegionName();
- Region region = cache.getRegion(regionName);
+ Region region = executionContext.getCache().getRegion(regionName);
if (region == null) {
return Failure.of(ProtobufResponseUtilities
.makeErrorResponse(ProtocolErrorCode.REGION_NOT_FOUND.codeValue, "Region not found"));
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetAvailableServersOperationHandler.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetAvailableServersOperationHandler.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetAvailableServersOperationHandler.java
index 239d9f7..e7c18cd 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetAvailableServersOperationHandler.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetAvailableServersOperationHandler.java
@@ -14,33 +14,20 @@
*/
package org.apache.geode.protocol.protobuf.operations;
-import java.io.IOException;
-import java.net.InetSocketAddress;
+import java.util.ArrayList;
import java.util.Collection;
-import java.util.HashSet;
-import java.util.Properties;
-import java.util.StringTokenizer;
import java.util.stream.Collectors;
-import org.apache.commons.lang.StringUtils;
-
import org.apache.geode.annotations.Experimental;
-import org.apache.geode.cache.Cache;
-import org.apache.geode.cache.client.internal.locator.GetAllServersRequest;
-import org.apache.geode.cache.client.internal.locator.GetAllServersResponse;
-import org.apache.geode.distributed.ConfigurationProperties;
-import org.apache.geode.distributed.internal.InternalDistributedSystem;
+import org.apache.geode.distributed.internal.InternalLocator;
import org.apache.geode.distributed.internal.ServerLocation;
-import org.apache.geode.distributed.internal.tcpserver.TcpClient;
-import org.apache.geode.internal.admin.remote.DistributionLocatorId;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.operations.OperationHandler;
import org.apache.geode.protocol.protobuf.BasicTypes;
-import org.apache.geode.protocol.protobuf.Failure;
-import org.apache.geode.protocol.protobuf.ProtocolErrorCode;
import org.apache.geode.protocol.protobuf.Result;
import org.apache.geode.protocol.protobuf.ServerAPI;
import org.apache.geode.protocol.protobuf.Success;
-import org.apache.geode.protocol.protobuf.utilities.ProtobufResponseUtilities;
import org.apache.geode.serialization.SerializationService;
@Experimental
@@ -50,51 +37,19 @@ public class GetAvailableServersOperationHandler implements
@Override
public Result<ServerAPI.GetAvailableServersResponse> process(
SerializationService serializationService, ServerAPI.GetAvailableServersRequest request,
- Cache cache) {
-
- InternalDistributedSystem distributedSystem =
- (InternalDistributedSystem) cache.getDistributedSystem();
- Properties properties = distributedSystem.getProperties();
- String locatorsString = properties.getProperty(ConfigurationProperties.LOCATORS);
-
- HashSet<DistributionLocatorId> locators = new HashSet();
- StringTokenizer stringTokenizer = new StringTokenizer(locatorsString, ",");
- while (stringTokenizer.hasMoreTokens()) {
- String locator = stringTokenizer.nextToken();
- if (StringUtils.isNotEmpty(locator)) {
- locators.add(new DistributionLocatorId(locator));
- }
- }
+ ExecutionContext executionContext) throws InvalidExecutionContextException {
- TcpClient tcpClient = getTcpClient();
- for (DistributionLocatorId locator : locators) {
- try {
- return getGetAvailableServersFromLocator(tcpClient, locator.getHost());
- } catch (IOException | ClassNotFoundException e) {
- // try the next locator
- }
- }
- return Failure.of(ProtobufResponseUtilities.makeErrorResponse(
- ProtocolErrorCode.DATA_UNREACHABLE.codeValue, "Unable to find a locator"));
- }
+ InternalLocator locator = executionContext.getLocator();
+ ArrayList servers2 = locator.getServerLocatorAdvisee().getLoadSnapshot().getServers(null);
- private Result<ServerAPI.GetAvailableServersResponse> getGetAvailableServersFromLocator(
- TcpClient tcpClient, InetSocketAddress address) throws IOException, ClassNotFoundException {
- GetAllServersResponse getAllServersResponse = (GetAllServersResponse) tcpClient
- .requestToServer(address, new GetAllServersRequest(), 1000, true);
- Collection<BasicTypes.Server> servers =
- (Collection<BasicTypes.Server>) getAllServersResponse.getServers().stream()
- .map(serverLocation -> getServerProtobufMessage((ServerLocation) serverLocation))
- .collect(Collectors.toList());
+ Collection<BasicTypes.Server> servers = (Collection<BasicTypes.Server>) servers2.stream()
+ .map(serverLocation -> getServerProtobufMessage((ServerLocation) serverLocation))
+ .collect(Collectors.toList());
ServerAPI.GetAvailableServersResponse.Builder builder =
ServerAPI.GetAvailableServersResponse.newBuilder().addAllServers(servers);
return Success.of(builder.build());
}
- protected TcpClient getTcpClient() {
- return new TcpClient();
- }
-
private BasicTypes.Server getServerProtobufMessage(ServerLocation serverLocation) {
BasicTypes.Server.Builder serverBuilder = BasicTypes.Server.newBuilder();
serverBuilder.setHostname(serverLocation.getHostName()).setPort(serverLocation.getPort());
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRegionNamesRequestOperationHandler.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRegionNamesRequestOperationHandler.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRegionNamesRequestOperationHandler.java
index e5d216a..53898ed 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRegionNamesRequestOperationHandler.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRegionNamesRequestOperationHandler.java
@@ -17,8 +17,9 @@ package org.apache.geode.protocol.protobuf.operations;
import java.util.Set;
import org.apache.geode.annotations.Experimental;
-import org.apache.geode.cache.Cache;
import org.apache.geode.cache.Region;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.operations.OperationHandler;
import org.apache.geode.protocol.protobuf.RegionAPI;
import org.apache.geode.protocol.protobuf.Result;
@@ -32,8 +33,9 @@ public class GetRegionNamesRequestOperationHandler
@Override
public Result<RegionAPI.GetRegionNamesResponse> process(SerializationService serializationService,
- RegionAPI.GetRegionNamesRequest request, Cache cache) {
- Set<Region<?, ?>> regions = cache.rootRegions();
+ RegionAPI.GetRegionNamesRequest request, ExecutionContext executionContext)
+ throws InvalidExecutionContextException {
+ Set<Region<?, ?>> regions = executionContext.getCache().rootRegions();
return Success.of(ProtobufResponseUtilities.createGetRegionNamesResponse(regions));
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRegionRequestOperationHandler.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRegionRequestOperationHandler.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRegionRequestOperationHandler.java
index b563a5d..007f96b 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRegionRequestOperationHandler.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRegionRequestOperationHandler.java
@@ -15,8 +15,9 @@
package org.apache.geode.protocol.protobuf.operations;
import org.apache.geode.annotations.Experimental;
-import org.apache.geode.cache.Cache;
import org.apache.geode.cache.Region;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.operations.OperationHandler;
import org.apache.geode.protocol.protobuf.BasicTypes;
import org.apache.geode.protocol.protobuf.Failure;
@@ -34,10 +35,11 @@ public class GetRegionRequestOperationHandler
@Override
public Result<RegionAPI.GetRegionResponse> process(SerializationService serializationService,
- RegionAPI.GetRegionRequest request, Cache cache) {
+ RegionAPI.GetRegionRequest request, ExecutionContext executionContext)
+ throws InvalidExecutionContextException {
String regionName = request.getRegionName();
- Region region = cache.getRegion(regionName);
+ Region region = executionContext.getCache().getRegion(regionName);
if (region == null) {
return Failure.of(
ProtobufResponseUtilities.makeErrorResponse(ProtocolErrorCode.REGION_NOT_FOUND.codeValue,
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRequestOperationHandler.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRequestOperationHandler.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRequestOperationHandler.java
index 96c0282..8f0fef7 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRequestOperationHandler.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRequestOperationHandler.java
@@ -15,8 +15,9 @@
package org.apache.geode.protocol.protobuf.operations;
import org.apache.geode.annotations.Experimental;
-import org.apache.geode.cache.Cache;
import org.apache.geode.cache.Region;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.operations.OperationHandler;
import org.apache.geode.protocol.protobuf.BasicTypes;
import org.apache.geode.protocol.protobuf.Failure;
@@ -36,9 +37,10 @@ public class GetRequestOperationHandler
@Override
public Result<RegionAPI.GetResponse> process(SerializationService serializationService,
- RegionAPI.GetRequest request, Cache cache) {
+ RegionAPI.GetRequest request, ExecutionContext executionContext)
+ throws InvalidExecutionContextException {
String regionName = request.getRegionName();
- Region region = cache.getRegion(regionName);
+ Region region = executionContext.getCache().getRegion(regionName);
if (region == null) {
return Failure.of(ProtobufResponseUtilities
.makeErrorResponse(ProtocolErrorCode.REGION_NOT_FOUND.codeValue, "Region not found"));
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/PutAllRequestOperationHandler.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/PutAllRequestOperationHandler.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/PutAllRequestOperationHandler.java
index 253a95d..e0ebc41 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/PutAllRequestOperationHandler.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/PutAllRequestOperationHandler.java
@@ -21,8 +21,9 @@ import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.geode.annotations.Experimental;
-import org.apache.geode.cache.Cache;
import org.apache.geode.cache.Region;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.operations.OperationHandler;
import org.apache.geode.protocol.protobuf.BasicTypes;
import org.apache.geode.protocol.protobuf.Failure;
@@ -43,8 +44,9 @@ public class PutAllRequestOperationHandler
@Override
public Result<RegionAPI.PutAllResponse> process(SerializationService serializationService,
- RegionAPI.PutAllRequest putAllRequest, Cache cache) {
- Region region = cache.getRegion(putAllRequest.getRegionName());
+ RegionAPI.PutAllRequest putAllRequest, ExecutionContext executionContext)
+ throws InvalidExecutionContextException {
+ Region region = executionContext.getCache().getRegion(putAllRequest.getRegionName());
if (region == null) {
return Failure.of(ProtobufResponseUtilities.createAndLogErrorResponse(
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/PutRequestOperationHandler.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/PutRequestOperationHandler.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/PutRequestOperationHandler.java
index c24fb29..cf5afb4 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/PutRequestOperationHandler.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/PutRequestOperationHandler.java
@@ -15,8 +15,9 @@
package org.apache.geode.protocol.protobuf.operations;
import org.apache.geode.annotations.Experimental;
-import org.apache.geode.cache.Cache;
import org.apache.geode.cache.Region;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.operations.OperationHandler;
import org.apache.geode.protocol.protobuf.BasicTypes;
import org.apache.geode.protocol.protobuf.Failure;
@@ -36,9 +37,10 @@ public class PutRequestOperationHandler
@Override
public Result<RegionAPI.PutResponse> process(SerializationService serializationService,
- RegionAPI.PutRequest request, Cache cache) {
+ RegionAPI.PutRequest request, ExecutionContext executionContext)
+ throws InvalidExecutionContextException {
String regionName = request.getRegionName();
- Region region = cache.getRegion(regionName);
+ Region region = executionContext.getCache().getRegion(regionName);
if (region == null) {
return Failure.of(
ProtobufResponseUtilities.makeErrorResponse(ProtocolErrorCode.REGION_NOT_FOUND.codeValue,
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/RemoveRequestOperationHandler.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/RemoveRequestOperationHandler.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/RemoveRequestOperationHandler.java
index 59236be..052efcf 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/RemoveRequestOperationHandler.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/RemoveRequestOperationHandler.java
@@ -18,10 +18,10 @@ import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.geode.annotations.Experimental;
-import org.apache.geode.cache.Cache;
import org.apache.geode.cache.Region;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.operations.OperationHandler;
-import org.apache.geode.protocol.protobuf.BasicTypes;
import org.apache.geode.protocol.protobuf.Failure;
import org.apache.geode.protocol.protobuf.ProtocolErrorCode;
import org.apache.geode.protocol.protobuf.RegionAPI;
@@ -40,10 +40,11 @@ public class RemoveRequestOperationHandler
@Override
public Result<RegionAPI.RemoveResponse> process(SerializationService serializationService,
- RegionAPI.RemoveRequest request, Cache cache) {
+ RegionAPI.RemoveRequest request, ExecutionContext executionContext)
+ throws InvalidExecutionContextException {
String regionName = request.getRegionName();
- Region region = cache.getRegion(regionName);
+ Region region = executionContext.getCache().getRegion(regionName);
if (region == null) {
return Failure.of(ProtobufResponseUtilities
.makeErrorResponse(ProtocolErrorCode.REGION_NOT_FOUND.codeValue, "Region not found"));
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/test/java/org/apache/geode/protocol/GetAvailableServersDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/GetAvailableServersDUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/GetAvailableServersDUnitTest.java
deleted file mode 100644
index 4d6390b..0000000
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/GetAvailableServersDUnitTest.java
+++ /dev/null
@@ -1,108 +0,0 @@
-/*
- * 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.protocol;
-
-import org.apache.geode.cache.server.CacheServer;
-import org.apache.geode.internal.cache.InternalCache;
-import org.apache.geode.protocol.exception.InvalidProtocolMessageException;
-import org.apache.geode.protocol.protobuf.ClientProtocol;
-import org.apache.geode.protocol.protobuf.ServerAPI;
-import org.apache.geode.protocol.protobuf.serializer.ProtobufProtocolSerializer;
-import org.apache.geode.protocol.protobuf.utilities.ProtobufRequestUtilities;
-import org.apache.geode.protocol.protobuf.utilities.ProtobufUtilities;
-import org.apache.geode.test.dunit.DistributedTestUtils;
-import org.apache.geode.test.dunit.Host;
-import org.apache.geode.test.dunit.VM;
-import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
-import org.apache.geode.test.dunit.rules.DistributedRestoreSystemProperties;
-import org.apache.geode.test.junit.categories.DistributedTest;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-
-import java.io.IOException;
-import java.net.Socket;
-
-import static org.junit.Assert.assertEquals;
-
-@Category(DistributedTest.class)
-public class GetAvailableServersDUnitTest extends JUnit4CacheTestCase {
-
- @Rule
- public DistributedRestoreSystemProperties distributedRestoreSystemProperties =
- new DistributedRestoreSystemProperties();
-
- @Before
- public void setup() {
-
- }
-
- @Test
- public void testGetAllAvailableServersRequest()
- throws IOException, InvalidProtocolMessageException {
- Host host = Host.getHost(0);
- VM vm0 = host.getVM(0);
- VM vm1 = host.getVM(1);
- VM vm2 = host.getVM(2);
-
- int locatorPort = DistributedTestUtils.getDUnitLocatorPort();
-
- // int cacheServer1Port = vm0.invoke("Start Cache1", () -> startCacheWithCacheServer());
- int cacheServer1Port = startCacheWithCacheServer();
- int cacheServer2Port = vm1.invoke("Start Cache2", () -> startCacheWithCacheServer());
- int cacheServer3Port = vm2.invoke("Start Cache3", () -> startCacheWithCacheServer());
-
- vm0.invoke(() -> {
- Socket socket = new Socket(host.getHostName(), cacheServer1Port);
- socket.getOutputStream().write(110);
-
- ClientProtocol.Request.Builder protobufRequestBuilder =
- ProtobufUtilities.createProtobufRequestBuilder();
- ClientProtocol.Message getAvailableServersRequestMessage =
- ProtobufUtilities.createProtobufMessage(ProtobufUtilities.createMessageHeader(1233445),
- protobufRequestBuilder.setGetAvailableServersRequest(
- ProtobufRequestUtilities.createGetAvailableServersRequest()).build());
-
- ProtobufProtocolSerializer protobufProtocolSerializer = new ProtobufProtocolSerializer();
- protobufProtocolSerializer.serialize(getAvailableServersRequestMessage,
- socket.getOutputStream());
-
- ClientProtocol.Message getAvailableServersResponseMessage =
- protobufProtocolSerializer.deserialize(socket.getInputStream());
- assertEquals(1233445,
- getAvailableServersResponseMessage.getMessageHeader().getCorrelationId());
- assertEquals(ClientProtocol.Message.MessageTypeCase.RESPONSE,
- getAvailableServersResponseMessage.getMessageTypeCase());
- ClientProtocol.Response messageResponse = getAvailableServersResponseMessage.getResponse();
- assertEquals(ClientProtocol.Response.ResponseAPICase.GETAVAILABLESERVERSRESPONSE,
- messageResponse.getResponseAPICase());
- ServerAPI.GetAvailableServersResponse getAvailableServersResponse =
- messageResponse.getGetAvailableServersResponse();
- assertEquals(3, getAvailableServersResponse.getServersCount());
- });
- }
-
- private Integer startCacheWithCacheServer() throws IOException {
- System.setProperty("geode.feature-protobuf-protocol", "true");
-
- InternalCache cache = getCache();
- CacheServer cacheServer = cache.addCacheServer();
- cacheServer.setPort(0);
- cacheServer.start();
- return cacheServer.getPort();
- }
-
-}
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/test/java/org/apache/geode/protocol/RoundTripLocatorConnectionJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/RoundTripLocatorConnectionJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/RoundTripLocatorConnectionJUnitTest.java
new file mode 100644
index 0000000..799c55c
--- /dev/null
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/RoundTripLocatorConnectionJUnitTest.java
@@ -0,0 +1,132 @@
+/*
+ * 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.protocol;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.net.Socket;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.contrib.java.lang.system.RestoreSystemProperties;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.server.CacheServer;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.protocol.exception.InvalidProtocolMessageException;
+import org.apache.geode.protocol.protobuf.ClientProtocol;
+import org.apache.geode.protocol.protobuf.ProtocolErrorCode;
+import org.apache.geode.protocol.protobuf.ServerAPI;
+import org.apache.geode.protocol.protobuf.serializer.ProtobufProtocolSerializer;
+import org.apache.geode.protocol.protobuf.utilities.ProtobufRequestUtilities;
+import org.apache.geode.protocol.protobuf.utilities.ProtobufUtilities;
+import org.apache.geode.test.dunit.DistributedTestUtils;
+import org.apache.geode.test.dunit.Host;
+import org.apache.geode.test.dunit.VM;
+import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
+import org.apache.geode.test.junit.categories.DistributedTest;
+
+@Category(DistributedTest.class)
+public class RoundTripLocatorConnectionJUnitTest extends JUnit4CacheTestCase {
+
+ private Socket socket;
+ private DataOutputStream dataOutputStream;
+
+ @Rule
+ public final RestoreSystemProperties restoreSystemProperties = new RestoreSystemProperties();
+
+ @Before
+ public void setup() throws IOException {
+ Host host = Host.getHost(0);
+ int locatorPort = DistributedTestUtils.getDUnitLocatorPort();
+ int cacheServer1Port = startCacheWithCacheServer();
+
+ Host.getLocator().invoke(() -> System.setProperty("geode.feature-protobuf-protocol", "true"));
+
+ socket = new Socket(host.getHostName(), locatorPort);
+ dataOutputStream = new DataOutputStream(socket.getOutputStream());
+ dataOutputStream.writeInt(0);
+ dataOutputStream.writeByte(110);
+ }
+
+ @Test
+ public void testEchoProtobufMessageFromLocator()
+ throws IOException, InvalidProtocolMessageException {
+ ClientProtocol.Request.Builder protobufRequestBuilder =
+ ProtobufUtilities.createProtobufRequestBuilder();
+ ClientProtocol.Message getAvailableServersRequestMessage =
+ ProtobufUtilities.createProtobufMessage(ProtobufUtilities.createMessageHeader(1233445),
+ protobufRequestBuilder.setGetAvailableServersRequest(
+ ProtobufRequestUtilities.createGetAvailableServersRequest()).build());
+
+ ProtobufProtocolSerializer protobufProtocolSerializer = new ProtobufProtocolSerializer();
+ protobufProtocolSerializer.serialize(getAvailableServersRequestMessage,
+ socket.getOutputStream());
+
+ ClientProtocol.Message getAvailableServersResponseMessage =
+ protobufProtocolSerializer.deserialize(socket.getInputStream());
+ assertEquals(1233445, getAvailableServersResponseMessage.getMessageHeader().getCorrelationId());
+ assertEquals(ClientProtocol.Message.MessageTypeCase.RESPONSE,
+ getAvailableServersResponseMessage.getMessageTypeCase());
+ ClientProtocol.Response messageResponse = getAvailableServersResponseMessage.getResponse();
+ assertEquals(ClientProtocol.Response.ResponseAPICase.GETAVAILABLESERVERSRESPONSE,
+ messageResponse.getResponseAPICase());
+ ServerAPI.GetAvailableServersResponse getAvailableServersResponse =
+ messageResponse.getGetAvailableServersResponse();
+ assertEquals(1, getAvailableServersResponse.getServersCount());
+ }
+
+ @Test
+ public void testInvalidOperationReturnsFailure()
+ throws IOException, InvalidProtocolMessageException {
+ ClientProtocol.Request.Builder protobufRequestBuilder =
+ ProtobufUtilities.createProtobufRequestBuilder();
+ ClientProtocol.Message getAvailableServersRequestMessage =
+ ProtobufUtilities.createProtobufMessage(ProtobufUtilities.createMessageHeader(1233445),
+ protobufRequestBuilder
+ .setGetRegionNamesRequest(ProtobufRequestUtilities.createGetRegionNamesRequest())
+ .build());
+
+ ProtobufProtocolSerializer protobufProtocolSerializer = new ProtobufProtocolSerializer();
+ protobufProtocolSerializer.serialize(getAvailableServersRequestMessage,
+ socket.getOutputStream());
+
+ ClientProtocol.Message getAvailableServersResponseMessage =
+ protobufProtocolSerializer.deserialize(socket.getInputStream());
+ assertEquals(1233445, getAvailableServersResponseMessage.getMessageHeader().getCorrelationId());
+ assertEquals(ClientProtocol.Message.MessageTypeCase.RESPONSE,
+ getAvailableServersResponseMessage.getMessageTypeCase());
+ ClientProtocol.Response messageResponse = getAvailableServersResponseMessage.getResponse();
+ assertEquals(ClientProtocol.Response.ResponseAPICase.ERRORRESPONSE,
+ messageResponse.getResponseAPICase());
+ assertEquals(ProtocolErrorCode.UNSUPPORTED_OPERATION.codeValue,
+ messageResponse.getErrorResponse().getError().getErrorCode());
+ }
+
+ private Integer startCacheWithCacheServer() throws IOException {
+ System.setProperty("geode.feature-protobuf-protocol", "true");
+
+ InternalCache cache = getCache();
+ CacheServer cacheServer = cache.addCacheServer();
+ cacheServer.setPort(0);
+ cacheServer.start();
+ return cacheServer.getPort();
+ }
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/ProtobufStreamProcessorTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/ProtobufStreamProcessorTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/ProtobufStreamProcessorTest.java
index 87bfd52..2185b15 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/ProtobufStreamProcessorTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/ProtobufStreamProcessorTest.java
@@ -26,6 +26,7 @@ import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
import org.apache.geode.test.junit.categories.UnitTest;
@Category(UnitTest.class)
@@ -37,6 +38,7 @@ public class ProtobufStreamProcessorTest {
ProtobufStreamProcessor protobufStreamProcessor = new ProtobufStreamProcessor();
InternalCache mockInternalCache = mock(InternalCache.class);
- protobufStreamProcessor.receiveMessage(inputStream, outputStream, mockInternalCache);
+ protobufStreamProcessor.receiveMessage(inputStream, outputStream,
+ new ExecutionContext(mockInternalCache));
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetAllRequestOperationHandlerJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetAllRequestOperationHandlerJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetAllRequestOperationHandlerJUnitTest.java
index f2e3199..f4d098c 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetAllRequestOperationHandlerJUnitTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetAllRequestOperationHandlerJUnitTest.java
@@ -17,7 +17,6 @@ package org.apache.geode.protocol.protobuf.operations;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
-import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -28,13 +27,14 @@ import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.apache.geode.cache.Region;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.protobuf.BasicTypes;
import org.apache.geode.protocol.protobuf.RegionAPI;
import org.apache.geode.protocol.protobuf.Result;
import org.apache.geode.protocol.protobuf.Success;
import org.apache.geode.protocol.protobuf.utilities.ProtobufRequestUtilities;
import org.apache.geode.protocol.protobuf.utilities.ProtobufUtilities;
-import org.apache.geode.serialization.SerializationService;
import org.apache.geode.serialization.codec.StringCodec;
import org.apache.geode.serialization.exception.UnsupportedEncodingTypeException;
import org.apache.geode.serialization.registry.exception.CodecAlreadyRegisteredForTypeException;
@@ -80,9 +80,9 @@ public class GetAllRequestOperationHandlerJUnitTest extends OperationHandlerJUni
@Test
public void processReturnsExpectedValuesForValidKeys()
throws CodecAlreadyRegisteredForTypeException, UnsupportedEncodingTypeException,
- CodecNotRegisteredForTypeException {
- Result<RegionAPI.GetAllResponse> result =
- operationHandler.process(serializationServiceStub, generateTestRequest(true), cacheStub);
+ CodecNotRegisteredForTypeException, InvalidExecutionContextException {
+ Result<RegionAPI.GetAllResponse> result = operationHandler.process(serializationServiceStub,
+ generateTestRequest(true), new ExecutionContext(cacheStub));
Assert.assertTrue(result instanceof Success);
@@ -99,10 +99,10 @@ public class GetAllRequestOperationHandlerJUnitTest extends OperationHandlerJUni
}
@Test
- public void processReturnsNoEntriesForNoKeysRequested()
- throws UnsupportedEncodingTypeException, CodecNotRegisteredForTypeException {
- Result<RegionAPI.GetAllResponse> result =
- operationHandler.process(serializationServiceStub, generateTestRequest(false), cacheStub);
+ public void processReturnsNoEntriesForNoKeysRequested() throws UnsupportedEncodingTypeException,
+ CodecNotRegisteredForTypeException, InvalidExecutionContextException {
+ Result<RegionAPI.GetAllResponse> result = operationHandler.process(serializationServiceStub,
+ generateTestRequest(false), new ExecutionContext(cacheStub));
Assert.assertTrue(result instanceof Success);
http://git-wip-us.apache.org/repos/asf/geode/blob/530f48f3/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetAvailableServersOperationHandlerJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetAvailableServersOperationHandlerJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetAvailableServersOperationHandlerJUnitTest.java
index 77b088d..cff6ddc 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetAvailableServersOperationHandlerJUnitTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetAvailableServersOperationHandlerJUnitTest.java
@@ -14,14 +14,12 @@
*/
package org.apache.geode.protocol.protobuf.operations;
-import org.apache.geode.cache.client.internal.locator.GetAllServersResponse;
-import org.apache.geode.distributed.ConfigurationProperties;
-import org.apache.geode.distributed.internal.InternalDistributedSystem;
+import org.apache.geode.distributed.internal.InternalLocator;
+import org.apache.geode.distributed.internal.LocatorLoadSnapshot;
import org.apache.geode.distributed.internal.ServerLocation;
-import org.apache.geode.distributed.internal.tcpserver.TcpClient;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
+import org.apache.geode.distributed.internal.ServerLocator;
+import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
import org.apache.geode.protocol.protobuf.BasicTypes;
-import org.apache.geode.protocol.protobuf.Failure;
import org.apache.geode.protocol.protobuf.Result;
import org.apache.geode.protocol.protobuf.ServerAPI;
import org.apache.geode.protocol.protobuf.ServerAPI.GetAvailableServersResponse;
@@ -32,75 +30,48 @@ import org.junit.Before;
import org.junit.Test;
import org.junit.experimental.categories.Category;
-import java.io.IOException;
import java.util.ArrayList;
-import java.util.Properties;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyBoolean;
-import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@Category(UnitTest.class)
public class GetAvailableServersOperationHandlerJUnitTest extends OperationHandlerJUnitTest {
- private TcpClient mockTCPClient;
+ public static final String HOSTNAME_1 = "hostname1";
+ public static final int PORT_1 = 12345;
+
+ public static final String HOSTNAME_2 = "hostname2";
+ public static final int PORT_2 = 23456;
+
+ private InternalLocator internalLocatorMock;
@Before
public void setUp() throws Exception {
super.setUp();
- operationHandler = mock(GetAvailableServersOperationHandler.class);
- cacheStub = mock(GemFireCacheImpl.class);
- when(operationHandler.process(any(), any(), any())).thenCallRealMethod();
- InternalDistributedSystem mockDistributedSystem = mock(InternalDistributedSystem.class);
- when(cacheStub.getDistributedSystem()).thenReturn(mockDistributedSystem);
- Properties mockProperties = mock(Properties.class);
- when(mockDistributedSystem.getProperties()).thenReturn(mockProperties);
- String locatorString = "testLocator1Host[12345],testLocator2Host[23456]";
- when(mockProperties.getProperty(ConfigurationProperties.LOCATORS)).thenReturn(locatorString);
- mockTCPClient = mock(TcpClient.class);
- when(((GetAvailableServersOperationHandler) operationHandler).getTcpClient())
- .thenReturn(mockTCPClient);
- }
-
- @Test
- public void testServerReturnedFromHandler() throws Exception {
- when(mockTCPClient.requestToServer(any(), any(), anyInt(), anyBoolean()))
- .thenReturn(new GetAllServersResponse(new ArrayList<ServerLocation>() {
- {
- add(new ServerLocation("hostname1", 12345));
- add(new ServerLocation("hostname2", 23456));
- }
- }));
+ operationHandler = new GetAvailableServersOperationHandler();
+ internalLocatorMock = mock(InternalLocator.class);
+ ServerLocator serverLocatorAdviseeMock = mock(ServerLocator.class);
+ LocatorLoadSnapshot locatorLoadSnapshot = mock(LocatorLoadSnapshot.class);
+ ArrayList<Object> serverList = new ArrayList<>();
+ serverList.add(new ServerLocation(HOSTNAME_1, PORT_1));
+ serverList.add(new ServerLocation(HOSTNAME_2, PORT_2));
- ServerAPI.GetAvailableServersRequest getAvailableServersRequest =
- ProtobufRequestUtilities.createGetAvailableServersRequest();
- Result operationHandlerResult =
- operationHandler.process(serializationServiceStub, getAvailableServersRequest, cacheStub);
- assertTrue(operationHandlerResult instanceof Success);
- ValidateGetAvailableServersResponse(
- (GetAvailableServersResponse) operationHandlerResult.getMessage());
+ when(internalLocatorMock.getServerLocatorAdvisee()).thenReturn(serverLocatorAdviseeMock);
+ when(serverLocatorAdviseeMock.getLoadSnapshot()).thenReturn(locatorLoadSnapshot);
+ when(locatorLoadSnapshot.getServers(null)).thenReturn(serverList);
}
@Test
- public void testServerReturnedFromSecondLocatorIfFirstDown() throws Exception {
- when(mockTCPClient.requestToServer(any(), any(), anyInt(), anyBoolean()))
- .thenThrow(new IOException("BOOM!!!"))
- .thenReturn(new GetAllServersResponse(new ArrayList<ServerLocation>() {
- {
- add(new ServerLocation("hostname1", 12345));
- add(new ServerLocation("hostname2", 23456));
- }
- }));
-
+ public void testServerReturnedFromHandler() throws Exception {
ServerAPI.GetAvailableServersRequest getAvailableServersRequest =
ProtobufRequestUtilities.createGetAvailableServersRequest();
- Result operationHandlerResult =
- operationHandler.process(serializationServiceStub, getAvailableServersRequest, cacheStub);
+ Result operationHandlerResult = operationHandler.process(serializationServiceStub,
+ getAvailableServersRequest, new ExecutionContext(internalLocatorMock));
assertTrue(operationHandlerResult instanceof Success);
ValidateGetAvailableServersResponse(
(GetAvailableServersResponse) operationHandlerResult.getMessage());
@@ -110,22 +81,10 @@ public class GetAvailableServersOperationHandlerJUnitTest extends OperationHandl
GetAvailableServersResponse getAvailableServersResponse) {
assertEquals(2, getAvailableServersResponse.getServersCount());
BasicTypes.Server server = getAvailableServersResponse.getServers(0);
- assertEquals("hostname1", server.getHostname());
- assertEquals(12345, server.getPort());
+ assertEquals(HOSTNAME_1, server.getHostname());
+ assertEquals(PORT_1, server.getPort());
server = getAvailableServersResponse.getServers(1);
- assertEquals("hostname2", server.getHostname());
- assertEquals(23456, server.getPort());
- }
-
- @Test
- public void testProcessFailsIfNoLocatorsAvailable() throws Exception {
- when(mockTCPClient.requestToServer(any(), any(), anyInt(), anyBoolean()))
- .thenThrow(new IOException("BOOM!!!"));
-
- ServerAPI.GetAvailableServersRequest getAvailableServersRequest =
- ProtobufRequestUtilities.createGetAvailableServersRequest();
- Result operationHandlerResult =
- operationHandler.process(serializationServiceStub, getAvailableServersRequest, cacheStub);
- assertTrue(operationHandlerResult instanceof Failure);
+ assertEquals(HOSTNAME_2, server.getHostname());
+ assertEquals(PORT_2, server.getPort());
}
}
[4/7] geode git commit: GEODE-3406: Address PR feedback
Posted by hi...@apache.org.
GEODE-3406: Address PR feedback
* Rename ExecutionContext -> MessageExecutionContext
* Properly close socket when processing ProtoBuf request in TcpServer
* GetAvailableServersRequestHandler guards against `null` servers
* minor style changes
Signed-off-by: Alexander Murmann <am...@pivotal.io>
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/0eb320fa
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/0eb320fa
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/0eb320fa
Branch: refs/heads/develop
Commit: 0eb320fad2071c0ebb8cb1e74576ed4a231a9e3b
Parents: 530f48f
Author: Hitesh Khamesra <hk...@pivotal.io>
Authored: Fri Aug 18 10:44:57 2017 -0700
Committer: Hitesh Khamesra <hk...@pivotal.io>
Committed: Tue Aug 22 10:55:12 2017 -0700
----------------------------------------------------------------------
.../distributed/internal/InternalLocator.java | 5 +-
.../internal/tcpserver/TcpServer.java | 171 ++++++++++---------
.../geode/internal/cache/InternalCache.java | 4 +-
.../sockets/ClientProtocolMessageHandler.java | 2 +-
.../ClientProtocolMessageHandlerLoader.java | 64 -------
.../cache/tier/sockets/ExecutionContext.java | 54 ------
.../GenericProtocolServerConnection.java | 2 +-
.../tier/sockets/MessageExecutionContext.java | 56 ++++++
.../tier/sockets/MessageHandlerFactory.java | 19 +++
.../tier/sockets/ServerConnectionFactory.java | 22 +--
.../cache/tier/sockets/TcpServerFactory.java | 39 +++++
.../AutoConnectionSourceImplJUnitTest.java | 10 +-
.../tcpserver/TCPServerSSLJUnitTest.java | 3 +-
.../internal/tcpserver/TcpServerJUnitTest.java | 5 +-
.../tier/sockets/TcpServerFactoryTest.java | 19 +++
.../test/dunit/standalone/DUnitLauncher.java | 2 +
.../protocol/operations/OperationHandler.java | 4 +-
.../protocol/protobuf/ProtobufOpsProcessor.java | 5 +-
.../protobuf/ProtobufStreamProcessor.java | 7 +-
.../GetAllRequestOperationHandler.java | 4 +-
.../GetAvailableServersOperationHandler.java | 14 +-
.../GetRegionNamesRequestOperationHandler.java | 4 +-
.../GetRegionRequestOperationHandler.java | 4 +-
.../operations/GetRequestOperationHandler.java | 4 +-
.../PutAllRequestOperationHandler.java | 4 +-
.../operations/PutRequestOperationHandler.java | 4 +-
.../RemoveRequestOperationHandler.java | 4 +-
.../RoundTripLocatorConnectionJUnitTest.java | 19 ++-
.../protobuf/ProtobufStreamProcessorTest.java | 4 +-
.../GetAllRequestOperationHandlerJUnitTest.java | 6 +-
...ailableServersOperationHandlerJUnitTest.java | 42 +++--
...onNamesRequestOperationHandlerJUnitTest.java | 6 +-
...tRegionRequestOperationHandlerJUnitTest.java | 7 +-
.../GetRequestOperationHandlerJUnitTest.java | 12 +-
.../PutAllRequestOperationHandlerJUnitTest.java | 8 +-
.../PutRequestOperationHandlerJUnitTest.java | 10 +-
.../RemoveRequestOperationHandlerJUnitTest.java | 10 +-
37 files changed, 361 insertions(+), 298 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalLocator.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalLocator.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalLocator.java
index 8d2daf6..06603cc 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalLocator.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalLocator.java
@@ -62,6 +62,7 @@ import org.apache.geode.distributed.internal.tcpserver.TcpServer;
import org.apache.geode.internal.admin.remote.DistributionLocatorId;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.tier.sockets.TcpServerFactory;
import org.apache.geode.internal.cache.wan.WANServiceProvider;
import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.InternalLogWriter;
@@ -498,8 +499,8 @@ public class InternalLocator extends Locator implements ConnectListener {
ThreadGroup group = LoggingThreadGroup.createThreadGroup("Distribution locators", logger);
this.stats = new LocatorStats();
- this.server = new TcpServer(port, this.bindAddress, null, this.config, this.handler,
- new DelayedPoolStatHelper(), group, this.toString(), this);
+ this.server = new TcpServerFactory().makeTcpServer(port, this.bindAddress, null, this.config,
+ this.handler, new DelayedPoolStatHelper(), group, this.toString(), this);
}
// Reset the file names with the correct port number if startLocatorAndDS was called with port
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/TcpServer.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/TcpServer.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/TcpServer.java
index c3d51c1..d471062 100755
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/TcpServer.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/TcpServer.java
@@ -14,6 +14,32 @@
*/
package org.apache.geode.distributed.internal.tcpserver;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.EOFException;
+import java.io.File;
+import java.io.IOException;
+import java.io.StreamCorruptedException;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketAddress;
+import java.net.URL;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.net.ssl.SSLException;
+
+import org.apache.logging.log4j.Logger;
+
import org.apache.geode.CancelException;
import org.apache.geode.DataSerializer;
import org.apache.geode.SystemFailure;
@@ -32,39 +58,13 @@ import org.apache.geode.internal.VersionedDataInputStream;
import org.apache.geode.internal.VersionedDataOutputStream;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.tier.Acceptor;
-import org.apache.geode.internal.cache.tier.sockets.AcceptorImpl;
-import org.apache.geode.internal.cache.tier.sockets.ClientProtocolMessageHandlerLoader;
import org.apache.geode.internal.cache.tier.sockets.ClientProtocolMessageHandler;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
import org.apache.geode.internal.cache.tier.sockets.HandShake;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.net.SocketCreator;
import org.apache.geode.internal.net.SocketCreatorFactory;
import org.apache.geode.internal.security.SecurableCommunicationChannel;
-import org.apache.logging.log4j.Logger;
-
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.EOFException;
-import java.io.File;
-import java.io.IOException;
-import java.io.StreamCorruptedException;
-import java.net.InetAddress;
-import java.net.ServerSocket;
-import java.net.Socket;
-import java.net.SocketAddress;
-import java.net.URL;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.SynchronousQueue;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import javax.net.ssl.SSLException;
/**
* TCP server which listens on a port and delegates requests to a request handler. The server uses
@@ -99,6 +99,11 @@ public class TcpServer {
private static/* GemStoneAddition */ final Map GOSSIP_TO_GEMFIRE_VERSION_MAP = new HashMap();
+ /**
+ * For the new client-server protocol, which ignores the usual handshake mechanism.
+ */
+ public static final byte PROTOBUF_CLIENT_SERVER_PROTOCOL = (byte) 110;
+
// For test purpose only
public static boolean isTesting = false;
// Non-final field for testing to avoid any security holes in system.
@@ -126,8 +131,10 @@ public class TcpServer {
private InetAddress bind_address;
private volatile boolean shuttingDown = false; // GemStoneAddition
private final PoolStatHelper poolHelper;
- private InternalLocator internalLocator;
+ private final InternalLocator internalLocator;
private final TcpHandler handler;
+ private ClientProtocolMessageHandler messageHandler;
+
private PooledExecutorWithDMStats executor;
private final ThreadGroup threadGroup;
@@ -150,12 +157,14 @@ public class TcpServer {
public TcpServer(int port, InetAddress bind_address, Properties sslConfig,
DistributionConfigImpl cfg, TcpHandler handler, PoolStatHelper poolHelper,
- ThreadGroup threadGroup, String threadName, InternalLocator internalLocator) {
+ ThreadGroup threadGroup, String threadName, InternalLocator internalLocator,
+ ClientProtocolMessageHandler messageHandler) {
this.port = port;
this.bind_address = bind_address;
this.handler = handler;
this.poolHelper = poolHelper;
this.internalLocator = internalLocator;
+ this.messageHandler = messageHandler;
// register DSFID types first; invoked explicitly so that all message type
// initializations do not happen in first deserialization on a possibly
// "precious" thread
@@ -365,69 +374,67 @@ public class TcpServer {
short versionOrdinal;
if (gossipVersion == NON_GOSSIP_REQUEST_VERSION) {
- if (input.readUnsignedByte() == AcceptorImpl.PROTOBUF_CLIENT_SERVER_PROTOCOL
+ if (input.readUnsignedByte() == PROTOBUF_CLIENT_SERVER_PROTOCOL
&& Boolean.getBoolean("geode.feature-protobuf-protocol")) {
- ClientProtocolMessageHandler messageHandler = ClientProtocolMessageHandlerLoader.load();
messageHandler.receiveMessage(input, socket.getOutputStream(),
- new ExecutionContext(internalLocator));
+ new MessageExecutionContext(internalLocator));
} else {
rejectUnknownProtocolConnection(socket, gossipVersion);
- return;
}
- }
- if (gossipVersion <= getCurrentGossipVersion()
- && GOSSIP_TO_GEMFIRE_VERSION_MAP.containsKey(gossipVersion)) {
- // Create a versioned stream to remember sender's GemFire version
- versionOrdinal = (short) GOSSIP_TO_GEMFIRE_VERSION_MAP.get(gossipVersion);
} else {
- // Close the socket. We can not accept requests from a newer version
- rejectUnknownProtocolConnection(socket, gossipVersion);
- return;
- }
- if (Version.GFE_71.compareTo(versionOrdinal) <= 0) {
- // Recent versions of TcpClient will send the version ordinal
- versionOrdinal = input.readShort();
- }
-
- if (log.isDebugEnabled() && versionOrdinal != Version.CURRENT_ORDINAL) {
- log.debug("Locator reading request from " + socket.getInetAddress() + " with version "
- + Version.fromOrdinal(versionOrdinal, false));
- }
- input = new VersionedDataInputStream(input, Version.fromOrdinal(versionOrdinal, false));
- request = DataSerializer.readObject(input);
- if (log.isDebugEnabled()) {
- log.debug("Locator received request " + request + " from " + socket.getInetAddress());
- }
- if (request instanceof ShutdownRequest) {
- shuttingDown = true;
- // Don't call shutdown from within the worker thread, see java bug #6576792.
- // Closing the socket will cause our acceptor thread to shutdown the executor
- this.serverSocketPortAtClose = srv_sock.getLocalPort();
- srv_sock.close();
- response = new ShutdownResponse();
- } else if (request instanceof InfoRequest) {
- response = handleInfoRequest(request);
- } else if (request instanceof VersionRequest) {
- response = handleVersionRequest(request);
- } else {
- response = handler.processRequest(request);
- }
-
- handler.endRequest(request, startTime);
+ if (gossipVersion <= getCurrentGossipVersion()
+ && GOSSIP_TO_GEMFIRE_VERSION_MAP.containsKey(gossipVersion)) {
+ // Create a versioned stream to remember sender's GemFire version
+ versionOrdinal = (short) GOSSIP_TO_GEMFIRE_VERSION_MAP.get(gossipVersion);
+ } else {
+ // Close the socket. We can not accept requests from a newer version
+ rejectUnknownProtocolConnection(socket, gossipVersion);
+ return;
+ }
+ if (Version.GFE_71.compareTo(versionOrdinal) <= 0) {
+ // Recent versions of TcpClient will send the version ordinal
+ versionOrdinal = input.readShort();
+ }
- startTime = DistributionStats.getStatTime();
- if (response != null) {
- DataOutputStream output = new DataOutputStream(socket.getOutputStream());
- if (versionOrdinal != Version.CURRENT_ORDINAL) {
- output =
- new VersionedDataOutputStream(output, Version.fromOrdinal(versionOrdinal, false));
+ if (log.isDebugEnabled() && versionOrdinal != Version.CURRENT_ORDINAL) {
+ log.debug("Locator reading request from " + socket.getInetAddress() + " with version "
+ + Version.fromOrdinal(versionOrdinal, false));
+ }
+ input = new VersionedDataInputStream(input, Version.fromOrdinal(versionOrdinal, false));
+ request = DataSerializer.readObject(input);
+ if (log.isDebugEnabled()) {
+ log.debug("Locator received request " + request + " from " + socket.getInetAddress());
+ }
+ if (request instanceof ShutdownRequest) {
+ shuttingDown = true;
+ // Don't call shutdown from within the worker thread, see java bug #6576792.
+ // Closing the socket will cause our acceptor thread to shutdown the executor
+ this.serverSocketPortAtClose = srv_sock.getLocalPort();
+ srv_sock.close();
+ response = new ShutdownResponse();
+ } else if (request instanceof InfoRequest) {
+ response = handleInfoRequest(request);
+ } else if (request instanceof VersionRequest) {
+ response = handleVersionRequest(request);
+ } else {
+ response = handler.processRequest(request);
}
- DataSerializer.writeObject(response, output);
- output.flush();
- }
- handler.endResponse(request, startTime);
+ handler.endRequest(request, startTime);
+
+ startTime = DistributionStats.getStatTime();
+ if (response != null) {
+ DataOutputStream output = new DataOutputStream(socket.getOutputStream());
+ if (versionOrdinal != Version.CURRENT_ORDINAL) {
+ output =
+ new VersionedDataOutputStream(output, Version.fromOrdinal(versionOrdinal, false));
+ }
+ DataSerializer.writeObject(response, output);
+ output.flush();
+ }
+ handler.endResponse(request, startTime);
+ }
} catch (EOFException ignore) {
// client went away - ignore
} catch (CancelException ignore) {
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java b/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java
index 4c7a6ef..84aa66e 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java
@@ -75,9 +75,7 @@ import org.apache.geode.pdx.internal.TypeRegistry;
*/
public interface InternalCache extends Cache, Extensible<Cache>, CacheTime {
- default InternalDistributedMember getMyId() {
- return null;
- }
+ InternalDistributedMember getMyId();
Collection<DiskStore> listDiskStores();
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtocolMessageHandler.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtocolMessageHandler.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtocolMessageHandler.java
index 38ab73e..0ced3aa 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtocolMessageHandler.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtocolMessageHandler.java
@@ -30,5 +30,5 @@ import java.io.OutputStream;
*/
public interface ClientProtocolMessageHandler {
void receiveMessage(InputStream inputStream, OutputStream outputStream,
- ExecutionContext executionContext) throws IOException;
+ MessageExecutionContext executionContext) throws IOException;
}
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtocolMessageHandlerLoader.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtocolMessageHandlerLoader.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtocolMessageHandlerLoader.java
deleted file mode 100644
index 1dc6129..0000000
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtocolMessageHandlerLoader.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * 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.sockets;
-
-import java.io.IOException;
-import java.net.Socket;
-import java.util.Iterator;
-import java.util.ServiceLoader;
-
-import org.apache.geode.internal.cache.InternalCache;
-import org.apache.geode.internal.cache.tier.Acceptor;
-import org.apache.geode.internal.cache.tier.CachedRegionHelper;
-import org.apache.geode.internal.security.SecurityService;
-
-/**
- * Creates instances of ServerConnection based on the connection mode provided.
- */
-public class ClientProtocolMessageHandlerLoader {
- private static ClientProtocolMessageHandler protobufProtocolHandler;
- private static final Object protocolLoadLock = new Object();
-
- public static ClientProtocolMessageHandler load() {
- if (protobufProtocolHandler != null) {
- return protobufProtocolHandler;
- }
-
- synchronized (protocolLoadLock) {
- if (protobufProtocolHandler != null) {
- return protobufProtocolHandler;
- }
-
- ServiceLoader<ClientProtocolMessageHandler> loader =
- ServiceLoader.load(ClientProtocolMessageHandler.class);
- Iterator<ClientProtocolMessageHandler> iterator = loader.iterator();
-
- if (!iterator.hasNext()) {
- throw new ServiceLoadingFailureException(
- "ClientProtocolMessageHandler implementation not found in JVM");
- }
-
- ClientProtocolMessageHandler returnValue = iterator.next();
-
- if (iterator.hasNext()) {
- throw new ServiceLoadingFailureException(
- "Multiple service implementations found for ClientProtocolMessageHandler");
- }
-
- return returnValue;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ExecutionContext.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ExecutionContext.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ExecutionContext.java
deleted file mode 100644
index 27da205..0000000
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ExecutionContext.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * 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.sockets;
-
-import org.apache.geode.cache.Cache;
-import org.apache.geode.distributed.internal.InternalLocator;
-
-public class ExecutionContext {
- private Cache cache;
- private InternalLocator locator;
-
- public ExecutionContext(Cache cache) {
- this.cache = cache;
- }
-
- public ExecutionContext(InternalLocator locator) {
- this.locator = locator;
- }
-
- // This throws if the cache isn't present because we know that non of the callers can take any
- // reasonable action if the cache is not present
- public Cache getCache() throws InvalidExecutionContextException {
- if (cache != null) {
- return cache;
- } else {
- throw new InvalidExecutionContextException(
- "Execution context's cache was accessed but isn't present. Did this happen on a locator? Operations on the locator should not try to operate on a cache");
- }
- }
-
- // This throws if the locator isn't present because we know that non of the callers can take any
- // reasonable action if the locator is not present
- public InternalLocator getLocator() throws InvalidExecutionContextException {
- if (locator != null) {
- return locator;
- } else {
- throw new InvalidExecutionContextException(
- "Execution context's locator was accessed but isn't present. Did this happen on a server? Operations on the locator should not try to operate on a cache");
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/GenericProtocolServerConnection.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/GenericProtocolServerConnection.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/GenericProtocolServerConnection.java
index 8f6720e..cd1647a 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/GenericProtocolServerConnection.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/GenericProtocolServerConnection.java
@@ -62,7 +62,7 @@ public class GenericProtocolServerConnection extends ServerConnection {
authenticator.receiveMessage(inputStream, outputStream, securityManager);
} else {
messageHandler.receiveMessage(inputStream, outputStream,
- new ExecutionContext(this.getCache()));
+ new MessageExecutionContext(this.getCache()));
}
} catch (IOException e) {
logger.warn(e);
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/MessageExecutionContext.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/MessageExecutionContext.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/MessageExecutionContext.java
new file mode 100644
index 0000000..1cb8c9d
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/MessageExecutionContext.java
@@ -0,0 +1,56 @@
+/*
+ * 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.sockets;
+
+import org.apache.geode.annotations.Experimental;
+import org.apache.geode.cache.Cache;
+import org.apache.geode.distributed.internal.InternalLocator;
+
+@Experimental
+public class MessageExecutionContext {
+ private Cache cache;
+ private InternalLocator locator;
+
+ public MessageExecutionContext(Cache cache) {
+ this.cache = cache;
+ }
+
+ public MessageExecutionContext(InternalLocator locator) {
+ this.locator = locator;
+ }
+
+ // This throws if the cache isn't present because we know that non of the callers can take any
+ // reasonable action if the cache is not present
+ public Cache getCache() throws InvalidExecutionContextException {
+ if (cache != null) {
+ return cache;
+ } else {
+ throw new InvalidExecutionContextException(
+ "Operations on the locator should not to try to operate on a cache");
+ }
+ }
+
+ // This throws if the locator isn't present because we know that non of the callers can take any
+ // reasonable action if the locator is not present
+ public InternalLocator getLocator() throws InvalidExecutionContextException {
+ if (locator != null) {
+ return locator;
+ } else {
+ throw new InvalidExecutionContextException(
+ "Operations on the server should not to try to operate on a locator");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/MessageHandlerFactory.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/MessageHandlerFactory.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/MessageHandlerFactory.java
new file mode 100644
index 0000000..fd261d7
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/MessageHandlerFactory.java
@@ -0,0 +1,19 @@
+package org.apache.geode.internal.cache.tier.sockets;
+
+import java.util.Iterator;
+import java.util.ServiceLoader;
+
+public class MessageHandlerFactory {
+ public ClientProtocolMessageHandler makeMessageHandler() {
+ ServiceLoader<ClientProtocolMessageHandler> loader =
+ ServiceLoader.load(ClientProtocolMessageHandler.class);
+ Iterator<ClientProtocolMessageHandler> iterator = loader.iterator();
+
+ if (!iterator.hasNext()) {
+ throw new ServiceLoadingFailureException(
+ "There is no ClientProtocolMessageHandler implementation found in JVM");
+ }
+
+ return iterator.next();
+ }
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnectionFactory.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnectionFactory.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnectionFactory.java
index 9173f6a..d2d85f6 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnectionFactory.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnectionFactory.java
@@ -24,7 +24,6 @@ import org.apache.geode.security.StreamAuthenticator;
import java.io.IOException;
import java.net.Socket;
import java.util.HashMap;
-import java.util.Iterator;
import java.util.Map;
import java.util.ServiceLoader;
@@ -32,7 +31,7 @@ import java.util.ServiceLoader;
* Creates instances of ServerConnection based on the connection mode provided.
*/
public class ServerConnectionFactory {
- private ClientProtocolMessageHandler protobufProtocolHandler;
+ private ClientProtocolMessageHandler protocolHandler;
private Map<String, Class<? extends StreamAuthenticator>> authenticators = null;
public ServerConnectionFactory() {}
@@ -49,20 +48,13 @@ public class ServerConnectionFactory {
}
private synchronized ClientProtocolMessageHandler initializeMessageHandler() {
- if (protobufProtocolHandler != null) {
- return protobufProtocolHandler;
+ if (protocolHandler != null) {
+ return protocolHandler;
}
- ServiceLoader<ClientProtocolMessageHandler> loader =
- ServiceLoader.load(ClientProtocolMessageHandler.class);
- Iterator<ClientProtocolMessageHandler> iterator = loader.iterator();
- if (!iterator.hasNext()) {
- throw new ServiceLoadingFailureException(
- "There is no ClientProtocolMessageHandler implementation found in JVM");
- }
+ protocolHandler = new MessageHandlerFactory().makeMessageHandler();
- protobufProtocolHandler = iterator.next();
- return protobufProtocolHandler;
+ return protocolHandler;
}
private StreamAuthenticator findStreamAuthenticator(String implementationID) {
@@ -86,10 +78,10 @@ public class ServerConnectionFactory {
}
private ClientProtocolMessageHandler getClientProtocolMessageHandler() {
- if (protobufProtocolHandler == null) {
+ if (protocolHandler == null) {
initializeMessageHandler();
}
- return protobufProtocolHandler;
+ return protocolHandler;
}
public ServerConnection makeServerConnection(Socket s, InternalCache c, CachedRegionHelper helper,
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactory.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactory.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactory.java
new file mode 100644
index 0000000..991ed75
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactory.java
@@ -0,0 +1,39 @@
+package org.apache.geode.internal.cache.tier.sockets;
+
+import java.net.InetAddress;
+import java.util.Properties;
+
+import org.apache.geode.distributed.internal.DistributionConfigImpl;
+import org.apache.geode.distributed.internal.InternalLocator;
+import org.apache.geode.distributed.internal.PoolStatHelper;
+import org.apache.geode.distributed.internal.tcpserver.TcpHandler;
+import org.apache.geode.distributed.internal.tcpserver.TcpServer;
+
+public class TcpServerFactory {
+ private ClientProtocolMessageHandler protocolHandler;
+
+ public TcpServerFactory() {
+ initializeMessageHandler();
+ }
+
+ public TcpServer makeTcpServer(int port, InetAddress bind_address, Properties sslConfig,
+ DistributionConfigImpl cfg, TcpHandler handler, PoolStatHelper poolHelper,
+ ThreadGroup threadGroup, String threadName, InternalLocator internalLocator) {
+
+ return new TcpServer(port, bind_address, sslConfig, cfg, handler, poolHelper, threadGroup,
+ threadName, internalLocator, protocolHandler);
+ }
+
+ public synchronized ClientProtocolMessageHandler initializeMessageHandler() {
+ if (!Boolean.getBoolean("geode.feature-protobuf-protocol")) {
+ return null;
+ }
+ if (protocolHandler != null) {
+ return protocolHandler;
+ }
+
+ protocolHandler = new MessageHandlerFactory().makeMessageHandler();
+
+ return protocolHandler;
+ }
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-core/src/test/java/org/apache/geode/cache/client/internal/AutoConnectionSourceImplJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/cache/client/internal/AutoConnectionSourceImplJUnitTest.java b/geode-core/src/test/java/org/apache/geode/cache/client/internal/AutoConnectionSourceImplJUnitTest.java
index 802620c..e57ca83 100644
--- a/geode-core/src/test/java/org/apache/geode/cache/client/internal/AutoConnectionSourceImplJUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/cache/client/internal/AutoConnectionSourceImplJUnitTest.java
@@ -35,6 +35,7 @@ import org.apache.geode.distributed.internal.tcpserver.TcpServer;
import org.apache.geode.internal.AvailablePortHelper;
import org.apache.geode.internal.cache.PoolStats;
import org.apache.geode.internal.cache.tier.InternalClientMembership;
+import org.apache.geode.internal.cache.tier.sockets.TcpServerFactory;
import org.apache.geode.management.membership.ClientMembershipEvent;
import org.apache.geode.management.membership.ClientMembershipListener;
import org.apache.geode.test.junit.categories.ClientServerTest;
@@ -303,8 +304,9 @@ public class AutoConnectionSourceImplJUnitTest {
public void test_DiscoverLocators_whenOneLocatorWasShutdown() throws Exception {
startFakeLocator();
int secondPort = AvailablePortHelper.getRandomAvailableTCPPort();
- TcpServer server2 = new TcpServer(secondPort, InetAddress.getLocalHost(), null, null, handler,
- new FakeHelper(), Thread.currentThread().getThreadGroup(), "tcp server", null);
+ TcpServer server2 =
+ new TcpServerFactory().makeTcpServer(secondPort, InetAddress.getLocalHost(), null, null,
+ handler, new FakeHelper(), Thread.currentThread().getThreadGroup(), "tcp server", null);
server2.start();
try {
@@ -387,8 +389,8 @@ public class AutoConnectionSourceImplJUnitTest {
}
private void startFakeLocator() throws UnknownHostException, IOException, InterruptedException {
- server = new TcpServer(port, InetAddress.getLocalHost(), null, null, handler, new FakeHelper(),
- Thread.currentThread().getThreadGroup(), "Tcp Server", null);
+ server = new TcpServerFactory().makeTcpServer(port, InetAddress.getLocalHost(), null, null,
+ handler, new FakeHelper(), Thread.currentThread().getThreadGroup(), "Tcp Server", null);
server.start();
Thread.sleep(500);
}
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-core/src/test/java/org/apache/geode/distributed/internal/tcpserver/TCPServerSSLJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/distributed/internal/tcpserver/TCPServerSSLJUnitTest.java b/geode-core/src/test/java/org/apache/geode/distributed/internal/tcpserver/TCPServerSSLJUnitTest.java
index 229fbb9..c58eb31 100644
--- a/geode-core/src/test/java/org/apache/geode/distributed/internal/tcpserver/TCPServerSSLJUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/distributed/internal/tcpserver/TCPServerSSLJUnitTest.java
@@ -138,7 +138,8 @@ public class TCPServerSSLJUnitTest {
public DummyTcpServer(int port, InetAddress bind_address, Properties sslConfig,
DistributionConfigImpl cfg, TcpHandler handler, PoolStatHelper poolHelper,
ThreadGroup threadGroup, String threadName) {
- super(port, bind_address, sslConfig, cfg, handler, poolHelper, threadGroup, threadName, null);
+ super(port, bind_address, sslConfig, cfg, handler, poolHelper, threadGroup, threadName, null,
+ null);
if (cfg == null) {
cfg = new DistributionConfigImpl(sslConfig);
}
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-core/src/test/java/org/apache/geode/distributed/internal/tcpserver/TcpServerJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/distributed/internal/tcpserver/TcpServerJUnitTest.java b/geode-core/src/test/java/org/apache/geode/distributed/internal/tcpserver/TcpServerJUnitTest.java
index 9d20e8c..d02051f 100644
--- a/geode-core/src/test/java/org/apache/geode/distributed/internal/tcpserver/TcpServerJUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/distributed/internal/tcpserver/TcpServerJUnitTest.java
@@ -26,6 +26,7 @@ import org.apache.geode.distributed.internal.ClusterConfigurationService;
import org.apache.geode.distributed.internal.DistributionConfigImpl;
import org.apache.geode.distributed.internal.PoolStatHelper;
import org.apache.geode.internal.AvailablePort;
+import org.apache.geode.internal.cache.tier.sockets.TcpServerFactory;
import org.apache.geode.internal.net.SocketCreatorFactory;
import org.apache.geode.test.junit.categories.IntegrationTest;
import org.apache.geode.test.junit.categories.MembershipTest;
@@ -68,8 +69,8 @@ public class TcpServerJUnitTest {
port = AvailablePort.getRandomAvailablePort(AvailablePort.SOCKET);
stats = new SimpleStats();
- server = new TcpServer(port, localhost, new Properties(), null, handler, stats,
- Thread.currentThread().getThreadGroup(), "server thread", null);
+ server = new TcpServerFactory().makeTcpServer(port, localhost, new Properties(), null, handler,
+ stats, Thread.currentThread().getThreadGroup(), "server thread", null);
server.start();
}
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactoryTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactoryTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactoryTest.java
new file mode 100644
index 0000000..7d40d01
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactoryTest.java
@@ -0,0 +1,19 @@
+package org.apache.geode.internal.cache.tier.sockets;
+
+import static org.junit.Assert.*;
+
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.distributed.internal.tcpserver.TcpServer;
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class TcpServerFactoryTest {
+ @Test
+ public void createsATcpServer() {
+ TcpServerFactory factory = new TcpServerFactory();
+ TcpServer server = factory.makeTcpServer(80, null, null, null, null, null, null, null, null);
+ assertTrue(server != null);
+ }
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-core/src/test/java/org/apache/geode/test/dunit/standalone/DUnitLauncher.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/test/dunit/standalone/DUnitLauncher.java b/geode-core/src/test/java/org/apache/geode/test/dunit/standalone/DUnitLauncher.java
index b35270e..fd88abf 100644
--- a/geode-core/src/test/java/org/apache/geode/test/dunit/standalone/DUnitLauncher.java
+++ b/geode-core/src/test/java/org/apache/geode/test/dunit/standalone/DUnitLauncher.java
@@ -297,6 +297,8 @@ public class DUnitLauncher {
// able to do so successfully anyway
p.setProperty(DISABLE_AUTO_RECONNECT, "true");
+ System.setProperty("geode.feature-protobuf-protocol", "true");
+
try {
Locator.startLocatorAndDS(0, locatorLogFile, p);
InternalLocator internalLocator = (InternalLocator) Locator.getLocator();
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/main/java/org/apache/geode/protocol/operations/OperationHandler.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/operations/OperationHandler.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/operations/OperationHandler.java
index 5d9012f..ca3548b 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/operations/OperationHandler.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/operations/OperationHandler.java
@@ -15,7 +15,7 @@
package org.apache.geode.protocol.operations;
import org.apache.geode.annotations.Experimental;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.protobuf.ProtobufOpsProcessor;
import org.apache.geode.protocol.protobuf.Result;
@@ -34,6 +34,6 @@ public interface OperationHandler<Req, Resp> {
* indicated on the provided cache, and return a response.
*/
Result<Resp> process(SerializationService serializationService, Req request,
- ExecutionContext executionContext) throws InvalidExecutionContextException;
+ MessageExecutionContext executionContext) throws InvalidExecutionContextException;
}
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtobufOpsProcessor.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtobufOpsProcessor.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtobufOpsProcessor.java
index 76f81e7..3619e0d 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtobufOpsProcessor.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtobufOpsProcessor.java
@@ -15,7 +15,7 @@
package org.apache.geode.protocol.protobuf;
import org.apache.geode.annotations.Experimental;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.protobuf.registry.OperationContextRegistry;
import org.apache.geode.protocol.protobuf.utilities.ProtobufResponseUtilities;
@@ -37,7 +37,8 @@ public class ProtobufOpsProcessor {
this.operationContextRegistry = operationContextRegistry;
}
- public ClientProtocol.Response process(ClientProtocol.Request request, ExecutionContext context) {
+ public ClientProtocol.Response process(ClientProtocol.Request request,
+ MessageExecutionContext context) {
ClientProtocol.Request.RequestAPICase requestType = request.getRequestAPICase();
OperationContext operationContext = operationContextRegistry.getOperationContext(requestType);
ClientProtocol.Response.Builder builder;
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtobufStreamProcessor.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtobufStreamProcessor.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtobufStreamProcessor.java
index d04e49e..accb899 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtobufStreamProcessor.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/ProtobufStreamProcessor.java
@@ -21,7 +21,7 @@ import java.io.OutputStream;
import org.apache.geode.annotations.Experimental;
import org.apache.geode.internal.cache.tier.sockets.ClientProtocolMessageHandler;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.protocol.exception.InvalidProtocolMessageException;
import org.apache.geode.protocol.protobuf.registry.OperationContextRegistry;
import org.apache.geode.protocol.protobuf.serializer.ProtobufProtocolSerializer;
@@ -46,7 +46,7 @@ public class ProtobufStreamProcessor implements ClientProtocolMessageHandler {
@Override
public void receiveMessage(InputStream inputStream, OutputStream outputStream,
- ExecutionContext executionContext) throws IOException {
+ MessageExecutionContext executionContext) throws IOException {
try {
processOneMessage(inputStream, outputStream, executionContext);
} catch (InvalidProtocolMessageException e) {
@@ -55,7 +55,8 @@ public class ProtobufStreamProcessor implements ClientProtocolMessageHandler {
}
private void processOneMessage(InputStream inputStream, OutputStream outputStream,
- ExecutionContext executionContext) throws InvalidProtocolMessageException, IOException {
+ MessageExecutionContext executionContext)
+ throws InvalidProtocolMessageException, IOException {
ClientProtocol.Message message = protobufProtocolSerializer.deserialize(inputStream);
if (message == null) {
throw new EOFException("Tried to deserialize protobuf message at EOF");
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetAllRequestOperationHandler.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetAllRequestOperationHandler.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetAllRequestOperationHandler.java
index 75274c1..77cef67 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetAllRequestOperationHandler.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetAllRequestOperationHandler.java
@@ -20,7 +20,7 @@ import java.util.Set;
import org.apache.geode.annotations.Experimental;
import org.apache.geode.cache.Region;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.operations.OperationHandler;
import org.apache.geode.protocol.protobuf.BasicTypes;
@@ -41,7 +41,7 @@ public class GetAllRequestOperationHandler
@Override
public Result<RegionAPI.GetAllResponse> process(SerializationService serializationService,
- RegionAPI.GetAllRequest request, ExecutionContext executionContext)
+ RegionAPI.GetAllRequest request, MessageExecutionContext executionContext)
throws InvalidExecutionContextException {
String regionName = request.getRegionName();
Region region = executionContext.getCache().getRegion(regionName);
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetAvailableServersOperationHandler.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetAvailableServersOperationHandler.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetAvailableServersOperationHandler.java
index e7c18cd..c1c3e99 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetAvailableServersOperationHandler.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetAvailableServersOperationHandler.java
@@ -21,7 +21,7 @@ import java.util.stream.Collectors;
import org.apache.geode.annotations.Experimental;
import org.apache.geode.distributed.internal.InternalLocator;
import org.apache.geode.distributed.internal.ServerLocation;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.operations.OperationHandler;
import org.apache.geode.protocol.protobuf.BasicTypes;
@@ -37,13 +37,17 @@ public class GetAvailableServersOperationHandler implements
@Override
public Result<ServerAPI.GetAvailableServersResponse> process(
SerializationService serializationService, ServerAPI.GetAvailableServersRequest request,
- ExecutionContext executionContext) throws InvalidExecutionContextException {
+ MessageExecutionContext executionContext) throws InvalidExecutionContextException {
InternalLocator locator = executionContext.getLocator();
- ArrayList servers2 = locator.getServerLocatorAdvisee().getLoadSnapshot().getServers(null);
+ ArrayList serversFromSnapshot =
+ locator.getServerLocatorAdvisee().getLoadSnapshot().getServers(null);
+ if (serversFromSnapshot == null) {
+ serversFromSnapshot = new ArrayList();
+ }
- Collection<BasicTypes.Server> servers = (Collection<BasicTypes.Server>) servers2.stream()
- .map(serverLocation -> getServerProtobufMessage((ServerLocation) serverLocation))
+ Collection<BasicTypes.Server> servers = (Collection<BasicTypes.Server>) serversFromSnapshot
+ .stream().map(serverLocation -> getServerProtobufMessage((ServerLocation) serverLocation))
.collect(Collectors.toList());
ServerAPI.GetAvailableServersResponse.Builder builder =
ServerAPI.GetAvailableServersResponse.newBuilder().addAllServers(servers);
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRegionNamesRequestOperationHandler.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRegionNamesRequestOperationHandler.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRegionNamesRequestOperationHandler.java
index 53898ed..e2edfed 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRegionNamesRequestOperationHandler.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRegionNamesRequestOperationHandler.java
@@ -18,7 +18,7 @@ import java.util.Set;
import org.apache.geode.annotations.Experimental;
import org.apache.geode.cache.Region;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.operations.OperationHandler;
import org.apache.geode.protocol.protobuf.RegionAPI;
@@ -33,7 +33,7 @@ public class GetRegionNamesRequestOperationHandler
@Override
public Result<RegionAPI.GetRegionNamesResponse> process(SerializationService serializationService,
- RegionAPI.GetRegionNamesRequest request, ExecutionContext executionContext)
+ RegionAPI.GetRegionNamesRequest request, MessageExecutionContext executionContext)
throws InvalidExecutionContextException {
Set<Region<?, ?>> regions = executionContext.getCache().rootRegions();
return Success.of(ProtobufResponseUtilities.createGetRegionNamesResponse(regions));
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRegionRequestOperationHandler.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRegionRequestOperationHandler.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRegionRequestOperationHandler.java
index 007f96b..5a8d4d3 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRegionRequestOperationHandler.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRegionRequestOperationHandler.java
@@ -16,7 +16,7 @@ package org.apache.geode.protocol.protobuf.operations;
import org.apache.geode.annotations.Experimental;
import org.apache.geode.cache.Region;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.operations.OperationHandler;
import org.apache.geode.protocol.protobuf.BasicTypes;
@@ -35,7 +35,7 @@ public class GetRegionRequestOperationHandler
@Override
public Result<RegionAPI.GetRegionResponse> process(SerializationService serializationService,
- RegionAPI.GetRegionRequest request, ExecutionContext executionContext)
+ RegionAPI.GetRegionRequest request, MessageExecutionContext executionContext)
throws InvalidExecutionContextException {
String regionName = request.getRegionName();
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRequestOperationHandler.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRequestOperationHandler.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRequestOperationHandler.java
index 8f0fef7..504189e 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRequestOperationHandler.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/GetRequestOperationHandler.java
@@ -16,7 +16,7 @@ package org.apache.geode.protocol.protobuf.operations;
import org.apache.geode.annotations.Experimental;
import org.apache.geode.cache.Region;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.operations.OperationHandler;
import org.apache.geode.protocol.protobuf.BasicTypes;
@@ -37,7 +37,7 @@ public class GetRequestOperationHandler
@Override
public Result<RegionAPI.GetResponse> process(SerializationService serializationService,
- RegionAPI.GetRequest request, ExecutionContext executionContext)
+ RegionAPI.GetRequest request, MessageExecutionContext executionContext)
throws InvalidExecutionContextException {
String regionName = request.getRegionName();
Region region = executionContext.getCache().getRegion(regionName);
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/PutAllRequestOperationHandler.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/PutAllRequestOperationHandler.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/PutAllRequestOperationHandler.java
index e0ebc41..99c7766 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/PutAllRequestOperationHandler.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/PutAllRequestOperationHandler.java
@@ -22,7 +22,7 @@ import org.apache.logging.log4j.Logger;
import org.apache.geode.annotations.Experimental;
import org.apache.geode.cache.Region;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.operations.OperationHandler;
import org.apache.geode.protocol.protobuf.BasicTypes;
@@ -44,7 +44,7 @@ public class PutAllRequestOperationHandler
@Override
public Result<RegionAPI.PutAllResponse> process(SerializationService serializationService,
- RegionAPI.PutAllRequest putAllRequest, ExecutionContext executionContext)
+ RegionAPI.PutAllRequest putAllRequest, MessageExecutionContext executionContext)
throws InvalidExecutionContextException {
Region region = executionContext.getCache().getRegion(putAllRequest.getRegionName());
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/PutRequestOperationHandler.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/PutRequestOperationHandler.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/PutRequestOperationHandler.java
index cf5afb4..e94127b 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/PutRequestOperationHandler.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/PutRequestOperationHandler.java
@@ -16,7 +16,7 @@ package org.apache.geode.protocol.protobuf.operations;
import org.apache.geode.annotations.Experimental;
import org.apache.geode.cache.Region;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.operations.OperationHandler;
import org.apache.geode.protocol.protobuf.BasicTypes;
@@ -37,7 +37,7 @@ public class PutRequestOperationHandler
@Override
public Result<RegionAPI.PutResponse> process(SerializationService serializationService,
- RegionAPI.PutRequest request, ExecutionContext executionContext)
+ RegionAPI.PutRequest request, MessageExecutionContext executionContext)
throws InvalidExecutionContextException {
String regionName = request.getRegionName();
Region region = executionContext.getCache().getRegion(regionName);
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/RemoveRequestOperationHandler.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/RemoveRequestOperationHandler.java b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/RemoveRequestOperationHandler.java
index 052efcf..94e3504 100644
--- a/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/RemoveRequestOperationHandler.java
+++ b/geode-protobuf/src/main/java/org/apache/geode/protocol/protobuf/operations/RemoveRequestOperationHandler.java
@@ -19,7 +19,7 @@ import org.apache.logging.log4j.Logger;
import org.apache.geode.annotations.Experimental;
import org.apache.geode.cache.Region;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.operations.OperationHandler;
import org.apache.geode.protocol.protobuf.Failure;
@@ -40,7 +40,7 @@ public class RemoveRequestOperationHandler
@Override
public Result<RegionAPI.RemoveResponse> process(SerializationService serializationService,
- RegionAPI.RemoveRequest request, ExecutionContext executionContext)
+ RegionAPI.RemoveRequest request, MessageExecutionContext executionContext)
throws InvalidExecutionContextException {
String regionName = request.getRegionName();
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/test/java/org/apache/geode/protocol/RoundTripLocatorConnectionJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/RoundTripLocatorConnectionJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/RoundTripLocatorConnectionJUnitTest.java
index 799c55c..14d8c44 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/RoundTripLocatorConnectionJUnitTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/RoundTripLocatorConnectionJUnitTest.java
@@ -15,12 +15,22 @@
package org.apache.geode.protocol;
+import static org.apache.geode.distributed.ConfigurationProperties.DISABLE_AUTO_RECONNECT;
+import static org.apache.geode.distributed.ConfigurationProperties.ENABLE_CLUSTER_CONFIGURATION;
+import static org.apache.geode.distributed.ConfigurationProperties.ENABLE_NETWORK_PARTITION_DETECTION;
+import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
+import static org.apache.geode.distributed.ConfigurationProperties.LOG_LEVEL;
+import static org.apache.geode.distributed.ConfigurationProperties.MAX_WAIT_TIME_RECONNECT;
+import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
+import static org.apache.geode.distributed.ConfigurationProperties.MEMBER_TIMEOUT;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import java.io.DataOutputStream;
+import java.io.File;
import java.io.IOException;
import java.net.Socket;
+import java.util.Properties;
import org.junit.Before;
import org.junit.Rule;
@@ -29,7 +39,11 @@ import org.junit.contrib.java.lang.system.RestoreSystemProperties;
import org.junit.experimental.categories.Category;
import org.apache.geode.cache.server.CacheServer;
+import org.apache.geode.distributed.Locator;
+import org.apache.geode.distributed.internal.InternalLocator;
+import org.apache.geode.internal.AvailablePort;
import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.tier.sockets.AcceptorImpl;
import org.apache.geode.protocol.exception.InvalidProtocolMessageException;
import org.apache.geode.protocol.protobuf.ClientProtocol;
import org.apache.geode.protocol.protobuf.ProtocolErrorCode;
@@ -39,6 +53,7 @@ import org.apache.geode.protocol.protobuf.utilities.ProtobufRequestUtilities;
import org.apache.geode.protocol.protobuf.utilities.ProtobufUtilities;
import org.apache.geode.test.dunit.DistributedTestUtils;
import org.apache.geode.test.dunit.Host;
+import org.apache.geode.test.dunit.LogWriterUtils;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
import org.apache.geode.test.junit.categories.DistributedTest;
@@ -48,6 +63,7 @@ public class RoundTripLocatorConnectionJUnitTest extends JUnit4CacheTestCase {
private Socket socket;
private DataOutputStream dataOutputStream;
+ private Locator locator;
@Rule
public final RestoreSystemProperties restoreSystemProperties = new RestoreSystemProperties();
@@ -63,7 +79,8 @@ public class RoundTripLocatorConnectionJUnitTest extends JUnit4CacheTestCase {
socket = new Socket(host.getHostName(), locatorPort);
dataOutputStream = new DataOutputStream(socket.getOutputStream());
dataOutputStream.writeInt(0);
- dataOutputStream.writeByte(110);
+ // Using the constant from AcceptorImpl to ensure that magic byte is the same
+ dataOutputStream.writeByte(AcceptorImpl.PROTOBUF_CLIENT_SERVER_PROTOCOL);
}
@Test
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/ProtobufStreamProcessorTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/ProtobufStreamProcessorTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/ProtobufStreamProcessorTest.java
index 2185b15..16eb48b 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/ProtobufStreamProcessorTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/ProtobufStreamProcessorTest.java
@@ -26,7 +26,7 @@ import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.apache.geode.internal.cache.InternalCache;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.test.junit.categories.UnitTest;
@Category(UnitTest.class)
@@ -39,6 +39,6 @@ public class ProtobufStreamProcessorTest {
ProtobufStreamProcessor protobufStreamProcessor = new ProtobufStreamProcessor();
InternalCache mockInternalCache = mock(InternalCache.class);
protobufStreamProcessor.receiveMessage(inputStream, outputStream,
- new ExecutionContext(mockInternalCache));
+ new MessageExecutionContext(mockInternalCache));
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetAllRequestOperationHandlerJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetAllRequestOperationHandlerJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetAllRequestOperationHandlerJUnitTest.java
index f4d098c..64ee50b 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetAllRequestOperationHandlerJUnitTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetAllRequestOperationHandlerJUnitTest.java
@@ -27,7 +27,7 @@ import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.apache.geode.cache.Region;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.protobuf.BasicTypes;
import org.apache.geode.protocol.protobuf.RegionAPI;
@@ -82,7 +82,7 @@ public class GetAllRequestOperationHandlerJUnitTest extends OperationHandlerJUni
throws CodecAlreadyRegisteredForTypeException, UnsupportedEncodingTypeException,
CodecNotRegisteredForTypeException, InvalidExecutionContextException {
Result<RegionAPI.GetAllResponse> result = operationHandler.process(serializationServiceStub,
- generateTestRequest(true), new ExecutionContext(cacheStub));
+ generateTestRequest(true), new MessageExecutionContext(cacheStub));
Assert.assertTrue(result instanceof Success);
@@ -102,7 +102,7 @@ public class GetAllRequestOperationHandlerJUnitTest extends OperationHandlerJUni
public void processReturnsNoEntriesForNoKeysRequested() throws UnsupportedEncodingTypeException,
CodecNotRegisteredForTypeException, InvalidExecutionContextException {
Result<RegionAPI.GetAllResponse> result = operationHandler.process(serializationServiceStub,
- generateTestRequest(false), new ExecutionContext(cacheStub));
+ generateTestRequest(false), new MessageExecutionContext(cacheStub));
Assert.assertTrue(result instanceof Success);
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetAvailableServersOperationHandlerJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetAvailableServersOperationHandlerJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetAvailableServersOperationHandlerJUnitTest.java
index cff6ddc..406beea 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetAvailableServersOperationHandlerJUnitTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetAvailableServersOperationHandlerJUnitTest.java
@@ -18,7 +18,8 @@ import org.apache.geode.distributed.internal.InternalLocator;
import org.apache.geode.distributed.internal.LocatorLoadSnapshot;
import org.apache.geode.distributed.internal.ServerLocation;
import org.apache.geode.distributed.internal.ServerLocator;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.protocol.protobuf.BasicTypes;
import org.apache.geode.protocol.protobuf.Result;
import org.apache.geode.protocol.protobuf.ServerAPI;
@@ -31,6 +32,7 @@ import org.junit.Test;
import org.junit.experimental.categories.Category;
import java.util.ArrayList;
+import java.util.HashMap;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -41,13 +43,14 @@ import static org.mockito.Mockito.when;
@Category(UnitTest.class)
public class GetAvailableServersOperationHandlerJUnitTest extends OperationHandlerJUnitTest {
- public static final String HOSTNAME_1 = "hostname1";
- public static final int PORT_1 = 12345;
+ private final String HOSTNAME_1 = "hostname1";
+ private final int PORT_1 = 12345;
- public static final String HOSTNAME_2 = "hostname2";
- public static final int PORT_2 = 23456;
+ private final String HOSTNAME_2 = "hostname2";
+ private final int PORT_2 = 23456;
private InternalLocator internalLocatorMock;
+ private LocatorLoadSnapshot locatorLoadSnapshot;
@Before
public void setUp() throws Exception {
@@ -56,27 +59,44 @@ public class GetAvailableServersOperationHandlerJUnitTest extends OperationHandl
operationHandler = new GetAvailableServersOperationHandler();
internalLocatorMock = mock(InternalLocator.class);
ServerLocator serverLocatorAdviseeMock = mock(ServerLocator.class);
- LocatorLoadSnapshot locatorLoadSnapshot = mock(LocatorLoadSnapshot.class);
- ArrayList<Object> serverList = new ArrayList<>();
- serverList.add(new ServerLocation(HOSTNAME_1, PORT_1));
- serverList.add(new ServerLocation(HOSTNAME_2, PORT_2));
+ locatorLoadSnapshot = mock(LocatorLoadSnapshot.class);
+
when(internalLocatorMock.getServerLocatorAdvisee()).thenReturn(serverLocatorAdviseeMock);
when(serverLocatorAdviseeMock.getLoadSnapshot()).thenReturn(locatorLoadSnapshot);
- when(locatorLoadSnapshot.getServers(null)).thenReturn(serverList);
}
@Test
public void testServerReturnedFromHandler() throws Exception {
+ ArrayList<Object> serverList = new ArrayList<>();
+ serverList.add(new ServerLocation(HOSTNAME_1, PORT_1));
+ serverList.add(new ServerLocation(HOSTNAME_2, PORT_2));
+ when(locatorLoadSnapshot.getServers(null)).thenReturn(serverList);
+
ServerAPI.GetAvailableServersRequest getAvailableServersRequest =
ProtobufRequestUtilities.createGetAvailableServersRequest();
Result operationHandlerResult = operationHandler.process(serializationServiceStub,
- getAvailableServersRequest, new ExecutionContext(internalLocatorMock));
+ getAvailableServersRequest, new MessageExecutionContext(internalLocatorMock));
assertTrue(operationHandlerResult instanceof Success);
ValidateGetAvailableServersResponse(
(GetAvailableServersResponse) operationHandlerResult.getMessage());
}
+ @Test
+ public void testWhenServersFromSnapshotAreNullReturnsEmtpy()
+ throws InvalidExecutionContextException {
+ when(locatorLoadSnapshot.getServers(any())).thenReturn(null);
+
+ ServerAPI.GetAvailableServersRequest getAvailableServersRequest =
+ ProtobufRequestUtilities.createGetAvailableServersRequest();
+ Result operationHandlerResult = operationHandler.process(serializationServiceStub,
+ getAvailableServersRequest, new MessageExecutionContext(internalLocatorMock));
+ assertTrue(operationHandlerResult instanceof Success);
+ GetAvailableServersResponse availableServersResponse =
+ (GetAvailableServersResponse) operationHandlerResult.getMessage();
+ assertEquals(0, availableServersResponse.getServersCount());
+ }
+
private void ValidateGetAvailableServersResponse(
GetAvailableServersResponse getAvailableServersResponse) {
assertEquals(2, getAvailableServersResponse.getServersCount());
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRegionNamesRequestOperationHandlerJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRegionNamesRequestOperationHandlerJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRegionNamesRequestOperationHandlerJUnitTest.java
index fd84d41..2fcf575 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRegionNamesRequestOperationHandlerJUnitTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRegionNamesRequestOperationHandlerJUnitTest.java
@@ -27,7 +27,7 @@ import org.junit.experimental.categories.Category;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.Region;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.protobuf.RegionAPI;
import org.apache.geode.protocol.protobuf.Result;
@@ -67,7 +67,7 @@ public class GetRegionNamesRequestOperationHandlerJUnitTest extends OperationHan
CodecNotRegisteredForTypeException, InvalidExecutionContextException {
Result<RegionAPI.GetRegionNamesResponse> result = operationHandler.process(
serializationServiceStub, ProtobufRequestUtilities.createGetRegionNamesRequest(),
- new ExecutionContext(cacheStub));
+ new MessageExecutionContext(cacheStub));
Assert.assertTrue(result instanceof Success);
RegionAPI.GetRegionNamesResponse getRegionsResponse = result.getMessage();
@@ -93,7 +93,7 @@ public class GetRegionNamesRequestOperationHandlerJUnitTest extends OperationHan
.thenReturn(Collections.unmodifiableSet(new HashSet<Region<String, String>>()));
Result<RegionAPI.GetRegionNamesResponse> result = operationHandler.process(
serializationServiceStub, ProtobufRequestUtilities.createGetRegionNamesRequest(),
- new ExecutionContext(emptyCache));
+ new MessageExecutionContext(emptyCache));
Assert.assertTrue(result instanceof Success);
RegionAPI.GetRegionNamesResponse getRegionsResponse = result.getMessage();
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRegionRequestOperationHandlerJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRegionRequestOperationHandlerJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRegionRequestOperationHandlerJUnitTest.java
index 6762f66..60d4985 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRegionRequestOperationHandlerJUnitTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRegionRequestOperationHandlerJUnitTest.java
@@ -19,7 +19,7 @@ import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.Scope;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.MessageUtil;
import org.apache.geode.protocol.protobuf.BasicTypes;
@@ -75,7 +75,7 @@ public class GetRegionRequestOperationHandlerJUnitTest extends OperationHandlerJ
Result<RegionAPI.GetRegionResponse> result = operationHandler.process(serializationServiceStub,
- MessageUtil.makeGetRegionRequest(TEST_REGION1), new ExecutionContext(cacheStub));
+ MessageUtil.makeGetRegionRequest(TEST_REGION1), new MessageExecutionContext(cacheStub));
RegionAPI.GetRegionResponse response = result.getMessage();
BasicTypes.Region region = response.getRegion();
Assert.assertEquals(TEST_REGION1, region.getName());
@@ -100,7 +100,8 @@ public class GetRegionRequestOperationHandlerJUnitTest extends OperationHandlerJ
.thenReturn(Collections.unmodifiableSet(new HashSet<Region<String, String>>()));
String unknownRegionName = "UNKNOWN_REGION";
Result<RegionAPI.GetRegionResponse> result = operationHandler.process(serializationServiceStub,
- MessageUtil.makeGetRegionRequest(unknownRegionName), new ExecutionContext(emptyCache));
+ MessageUtil.makeGetRegionRequest(unknownRegionName),
+ new MessageExecutionContext(emptyCache));
Assert.assertTrue(result instanceof Failure);
Assert.assertEquals(ProtocolErrorCode.REGION_NOT_FOUND.codeValue,
result.getErrorMessage().getError().getErrorCode());
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRequestOperationHandlerJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRequestOperationHandlerJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRequestOperationHandlerJUnitTest.java
index af35f6b..6885666 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRequestOperationHandlerJUnitTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/GetRequestOperationHandlerJUnitTest.java
@@ -16,7 +16,7 @@ package org.apache.geode.protocol.protobuf.operations;
import com.google.protobuf.ByteString;
import org.apache.geode.cache.Region;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.protobuf.BasicTypes;
import org.apache.geode.protocol.protobuf.Failure;
@@ -75,7 +75,7 @@ public class GetRequestOperationHandlerJUnitTest extends OperationHandlerJUnitTe
CodecNotRegisteredForTypeException, InvalidExecutionContextException {
RegionAPI.GetRequest getRequest = generateTestRequest(false, false, false);
Result<RegionAPI.GetResponse> result = operationHandler.process(serializationServiceStub,
- getRequest, new ExecutionContext(cacheStub));
+ getRequest, new MessageExecutionContext(cacheStub));
Assert.assertTrue(result instanceof Success);
Assert.assertEquals(BasicTypes.EncodedValue.ValueCase.STRINGRESULT,
@@ -90,7 +90,7 @@ public class GetRequestOperationHandlerJUnitTest extends OperationHandlerJUnitTe
CodecNotRegisteredForTypeException, InvalidExecutionContextException {
RegionAPI.GetRequest getRequest = generateTestRequest(true, false, false);
Result<RegionAPI.GetResponse> response = operationHandler.process(serializationServiceStub,
- getRequest, new ExecutionContext(cacheStub));
+ getRequest, new MessageExecutionContext(cacheStub));
Assert.assertTrue(response instanceof Failure);
Assert.assertEquals(ProtocolErrorCode.REGION_NOT_FOUND.codeValue,
@@ -103,7 +103,7 @@ public class GetRequestOperationHandlerJUnitTest extends OperationHandlerJUnitTe
CodecNotRegisteredForTypeException, InvalidExecutionContextException {
RegionAPI.GetRequest getRequest = generateTestRequest(false, true, false);
Result<RegionAPI.GetResponse> response = operationHandler.process(serializationServiceStub,
- getRequest, new ExecutionContext(cacheStub));
+ getRequest, new MessageExecutionContext(cacheStub));
Assert.assertTrue(response instanceof Success);
}
@@ -114,7 +114,7 @@ public class GetRequestOperationHandlerJUnitTest extends OperationHandlerJUnitTe
CodecNotRegisteredForTypeException, InvalidExecutionContextException {
RegionAPI.GetRequest getRequest = generateTestRequest(false, false, true);
Result<RegionAPI.GetResponse> response = operationHandler.process(serializationServiceStub,
- getRequest, new ExecutionContext(cacheStub));
+ getRequest, new MessageExecutionContext(cacheStub));
Assert.assertTrue(response instanceof Success);
}
@@ -136,7 +136,7 @@ public class GetRequestOperationHandlerJUnitTest extends OperationHandlerJUnitTe
RegionAPI.GetRequest getRequest =
ProtobufRequestUtilities.createGetRequest(TEST_REGION, encodedKey).getGetRequest();
Result<RegionAPI.GetResponse> response = operationHandler.process(serializationServiceStub,
- getRequest, new ExecutionContext(cacheStub));
+ getRequest, new MessageExecutionContext(cacheStub));
Assert.assertTrue(response instanceof Failure);
Assert.assertEquals(ProtocolErrorCode.VALUE_ENCODING_ERROR.codeValue,
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/PutAllRequestOperationHandlerJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/PutAllRequestOperationHandlerJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/PutAllRequestOperationHandlerJUnitTest.java
index d3fff49..955013f 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/PutAllRequestOperationHandlerJUnitTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/PutAllRequestOperationHandlerJUnitTest.java
@@ -15,7 +15,7 @@
package org.apache.geode.protocol.protobuf.operations;
import org.apache.geode.cache.Region;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.protobuf.BasicTypes;
import org.apache.geode.protocol.protobuf.RegionAPI;
@@ -74,7 +74,7 @@ public class PutAllRequestOperationHandlerJUnitTest extends OperationHandlerJUni
PutAllRequestOperationHandler operationHandler = new PutAllRequestOperationHandler();
Result<RegionAPI.PutAllResponse> result = operationHandler.process(serializationServiceStub,
- generateTestRequest(false, true), new ExecutionContext(cacheStub));
+ generateTestRequest(false, true), new MessageExecutionContext(cacheStub));
Assert.assertTrue(result instanceof Success);
@@ -88,7 +88,7 @@ public class PutAllRequestOperationHandlerJUnitTest extends OperationHandlerJUni
PutAllRequestOperationHandler operationHandler = new PutAllRequestOperationHandler();
Result<RegionAPI.PutAllResponse> result = operationHandler.process(serializationServiceStub,
- generateTestRequest(true, true), new ExecutionContext(cacheStub));
+ generateTestRequest(true, true), new MessageExecutionContext(cacheStub));
assertTrue(result instanceof Success);
verify(regionMock).put(TEST_KEY1, TEST_VALUE1);
@@ -107,7 +107,7 @@ public class PutAllRequestOperationHandlerJUnitTest extends OperationHandlerJUni
PutAllRequestOperationHandler operationHandler = new PutAllRequestOperationHandler();
Result<RegionAPI.PutAllResponse> result = operationHandler.process(serializationServiceStub,
- generateTestRequest(false, false), new ExecutionContext(cacheStub));
+ generateTestRequest(false, false), new MessageExecutionContext(cacheStub));
assertTrue(result instanceof Success);
[6/7] geode git commit: GEODE-3406: Address more PR feedback
Posted by hi...@apache.org.
GEODE-3406: Address more PR feedback
Signed-off-by: Alexander Murmann <am...@pivotal.io>
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/0cc60434
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/0cc60434
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/0cc60434
Branch: refs/heads/develop
Commit: 0cc60434cd75814e755b88be39af53b32d53faeb
Parents: a4fc1dd
Author: Hitesh Khamesra <hk...@pivotal.io>
Authored: Fri Aug 18 15:10:47 2017 -0700
Committer: Hitesh Khamesra <hk...@pivotal.io>
Committed: Tue Aug 22 10:55:48 2017 -0700
----------------------------------------------------------------------
.../internal/tcpserver/TcpServer.java | 18 +++---
.../ClientProtoclMessageHandlerLoader.java | 64 --------------------
.../tier/sockets/MessageHandlerFactory.java | 15 +++++
.../cache/tier/sockets/TcpServerFactory.java | 15 +++++
.../tier/sockets/TcpServerFactoryTest.java | 15 +++++
5 files changed, 53 insertions(+), 74 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/0cc60434/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/TcpServer.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/TcpServer.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/TcpServer.java
index d471062..83f87ee 100755
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/TcpServer.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/TcpServer.java
@@ -381,16 +381,11 @@ public class TcpServer {
} else {
rejectUnknownProtocolConnection(socket, gossipVersion);
}
- } else {
- if (gossipVersion <= getCurrentGossipVersion()
- && GOSSIP_TO_GEMFIRE_VERSION_MAP.containsKey(gossipVersion)) {
- // Create a versioned stream to remember sender's GemFire version
- versionOrdinal = (short) GOSSIP_TO_GEMFIRE_VERSION_MAP.get(gossipVersion);
- } else {
- // Close the socket. We can not accept requests from a newer version
- rejectUnknownProtocolConnection(socket, gossipVersion);
- return;
- }
+ } else if (gossipVersion <= getCurrentGossipVersion()
+ && GOSSIP_TO_GEMFIRE_VERSION_MAP.containsKey(gossipVersion)) {
+ // Create a versioned stream to remember sender's GemFire version
+ versionOrdinal = (short) GOSSIP_TO_GEMFIRE_VERSION_MAP.get(gossipVersion);
+
if (Version.GFE_71.compareTo(versionOrdinal) <= 0) {
// Recent versions of TcpClient will send the version ordinal
versionOrdinal = input.readShort();
@@ -434,6 +429,9 @@ public class TcpServer {
}
handler.endResponse(request, startTime);
+ } else {
+ // Close the socket. We can not accept requests from a newer version
+ rejectUnknownProtocolConnection(socket, gossipVersion);
}
} catch (EOFException ignore) {
// client went away - ignore
http://git-wip-us.apache.org/repos/asf/geode/blob/0cc60434/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtoclMessageHandlerLoader.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtoclMessageHandlerLoader.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtoclMessageHandlerLoader.java
deleted file mode 100644
index 6654757..0000000
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProtoclMessageHandlerLoader.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * 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.sockets;
-
-import java.io.IOException;
-import java.net.Socket;
-import java.util.Iterator;
-import java.util.ServiceLoader;
-
-import org.apache.geode.internal.cache.InternalCache;
-import org.apache.geode.internal.cache.tier.Acceptor;
-import org.apache.geode.internal.cache.tier.CachedRegionHelper;
-import org.apache.geode.internal.security.SecurityService;
-
-/**
- * Creates instances of ServerConnection based on the connection mode provided.
- */
-public class ClientProtoclMessageHandlerLoader {
- private static ClientProtocolMessageHandler protobufProtocolHandler;
- private static final Object protocolLoadLock = new Object();
-
- public static ClientProtocolMessageHandler load() {
- if (protobufProtocolHandler != null) {
- return protobufProtocolHandler;
- }
-
- synchronized (protocolLoadLock) {
- if (protobufProtocolHandler != null) {
- return protobufProtocolHandler;
- }
-
- ServiceLoader<ClientProtocolMessageHandler> loader =
- ServiceLoader.load(ClientProtocolMessageHandler.class);
- Iterator<ClientProtocolMessageHandler> iterator = loader.iterator();
-
- if (!iterator.hasNext()) {
- throw new ServiceLoadingFailureException(
- "ClientProtocolMessageHandler implementation not found in JVM");
- }
-
- ClientProtocolMessageHandler returnValue = iterator.next();
-
- if (iterator.hasNext()) {
- throw new ServiceLoadingFailureException(
- "Multiple service implementations found for ClientProtocolMessageHandler");
- }
-
- return returnValue;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/geode/blob/0cc60434/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/MessageHandlerFactory.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/MessageHandlerFactory.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/MessageHandlerFactory.java
index fd261d7..2aca8c2 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/MessageHandlerFactory.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/MessageHandlerFactory.java
@@ -1,3 +1,18 @@
+/*
+ * 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.sockets;
import java.util.Iterator;
http://git-wip-us.apache.org/repos/asf/geode/blob/0cc60434/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactory.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactory.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactory.java
index 991ed75..9c6bd8c 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactory.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactory.java
@@ -1,3 +1,18 @@
+/*
+ * 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.sockets;
import java.net.InetAddress;
http://git-wip-us.apache.org/repos/asf/geode/blob/0cc60434/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactoryTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactoryTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactoryTest.java
index 7d40d01..b5c1951 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactoryTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactoryTest.java
@@ -1,3 +1,18 @@
+/*
+ * 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.sockets;
import static org.junit.Assert.*;
[7/7] geode git commit: GEODE-3406: Fixed test failures.
Posted by hi...@apache.org.
GEODE-3406: Fixed test failures.
Now TcpServerFactory doesn't look for protobuf enabled property.
If service is not available then it just ignores that
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/31e82d6d
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/31e82d6d
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/31e82d6d
Branch: refs/heads/develop
Commit: 31e82d6d64187ba7a6d71771806a0b107a811c11
Parents: 0cc6043
Author: Hitesh Khamesra <hk...@pivotal.io>
Authored: Mon Aug 21 14:46:19 2017 -0700
Committer: Hitesh Khamesra <hk...@pivotal.io>
Committed: Tue Aug 22 10:56:00 2017 -0700
----------------------------------------------------------------------
.../geode/internal/cache/tier/sockets/TcpServerFactory.java | 9 +++++----
.../apache/geode/test/dunit/standalone/DUnitLauncher.java | 2 --
2 files changed, 5 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/31e82d6d/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactory.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactory.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactory.java
index 9c6bd8c..a67d6e6 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactory.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/TcpServerFactory.java
@@ -40,14 +40,15 @@ public class TcpServerFactory {
}
public synchronized ClientProtocolMessageHandler initializeMessageHandler() {
- if (!Boolean.getBoolean("geode.feature-protobuf-protocol")) {
- return null;
- }
if (protocolHandler != null) {
return protocolHandler;
}
- protocolHandler = new MessageHandlerFactory().makeMessageHandler();
+ try {
+ protocolHandler = new MessageHandlerFactory().makeMessageHandler();
+ } catch (ServiceLoadingFailureException ex) {
+ // ignore, TcpServer will take care right now
+ }
return protocolHandler;
}
http://git-wip-us.apache.org/repos/asf/geode/blob/31e82d6d/geode-core/src/test/java/org/apache/geode/test/dunit/standalone/DUnitLauncher.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/test/dunit/standalone/DUnitLauncher.java b/geode-core/src/test/java/org/apache/geode/test/dunit/standalone/DUnitLauncher.java
index fd88abf..b35270e 100644
--- a/geode-core/src/test/java/org/apache/geode/test/dunit/standalone/DUnitLauncher.java
+++ b/geode-core/src/test/java/org/apache/geode/test/dunit/standalone/DUnitLauncher.java
@@ -297,8 +297,6 @@ public class DUnitLauncher {
// able to do so successfully anyway
p.setProperty(DISABLE_AUTO_RECONNECT, "true");
- System.setProperty("geode.feature-protobuf-protocol", "true");
-
try {
Locator.startLocatorAndDS(0, locatorLogFile, p);
InternalLocator internalLocator = (InternalLocator) Locator.getLocator();
[3/7] geode git commit: GEODE-3406: Address PR feedback
Posted by hi...@apache.org.
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/PutRequestOperationHandlerJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/PutRequestOperationHandlerJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/PutRequestOperationHandlerJUnitTest.java
index 64d9f67..d36ad41 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/PutRequestOperationHandlerJUnitTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/PutRequestOperationHandlerJUnitTest.java
@@ -16,7 +16,7 @@ package org.apache.geode.protocol.protobuf.operations;
import com.google.protobuf.ByteString;
import org.apache.geode.cache.Region;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.protobuf.BasicTypes;
import org.apache.geode.protocol.protobuf.Failure;
@@ -68,7 +68,7 @@ public class PutRequestOperationHandlerJUnitTest extends OperationHandlerJUnitTe
CodecAlreadyRegisteredForTypeException, InvalidExecutionContextException {
PutRequestOperationHandler operationHandler = new PutRequestOperationHandler();
Result<RegionAPI.PutResponse> result = operationHandler.process(serializationServiceStub,
- generateTestRequest(), new ExecutionContext(cacheStub));
+ generateTestRequest(), new MessageExecutionContext(cacheStub));
assertTrue(result instanceof Success);
@@ -99,7 +99,7 @@ public class PutRequestOperationHandlerJUnitTest extends OperationHandlerJUnitTe
RegionAPI.PutRequest putRequest =
ProtobufRequestUtilities.createPutRequest(TEST_REGION, testEntry).getPutRequest();
Result<RegionAPI.PutResponse> result = operationHandler.process(serializationServiceStub,
- putRequest, new ExecutionContext(cacheStub));
+ putRequest, new MessageExecutionContext(cacheStub));
assertTrue(result instanceof Failure);
assertEquals(ProtocolErrorCode.VALUE_ENCODING_ERROR.codeValue,
@@ -113,7 +113,7 @@ public class PutRequestOperationHandlerJUnitTest extends OperationHandlerJUnitTe
when(cacheStub.getRegion(TEST_REGION)).thenReturn(null);
PutRequestOperationHandler operationHandler = new PutRequestOperationHandler();
Result<RegionAPI.PutResponse> result = operationHandler.process(serializationServiceStub,
- generateTestRequest(), new ExecutionContext(cacheStub));
+ generateTestRequest(), new MessageExecutionContext(cacheStub));
assertTrue(result instanceof Failure);
assertEquals(ProtocolErrorCode.REGION_NOT_FOUND.codeValue,
@@ -128,7 +128,7 @@ public class PutRequestOperationHandlerJUnitTest extends OperationHandlerJUnitTe
PutRequestOperationHandler operationHandler = new PutRequestOperationHandler();
Result<RegionAPI.PutResponse> result = operationHandler.process(serializationServiceStub,
- generateTestRequest(), new ExecutionContext(cacheStub));
+ generateTestRequest(), new MessageExecutionContext(cacheStub));
assertTrue(result instanceof Failure);
assertEquals(ProtocolErrorCode.CONSTRAINT_VIOLATION.codeValue,
http://git-wip-us.apache.org/repos/asf/geode/blob/0eb320fa/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/RemoveRequestOperationHandlerJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/RemoveRequestOperationHandlerJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/RemoveRequestOperationHandlerJUnitTest.java
index 47d6231..4350ece 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/RemoveRequestOperationHandlerJUnitTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/protobuf/operations/RemoveRequestOperationHandlerJUnitTest.java
@@ -16,7 +16,7 @@ package org.apache.geode.protocol.protobuf.operations;
import com.google.protobuf.ByteString;
import org.apache.geode.cache.Region;
-import org.apache.geode.internal.cache.tier.sockets.ExecutionContext;
+import org.apache.geode.internal.cache.tier.sockets.MessageExecutionContext;
import org.apache.geode.internal.cache.tier.sockets.InvalidExecutionContextException;
import org.apache.geode.protocol.protobuf.BasicTypes;
import org.apache.geode.protocol.protobuf.ClientProtocol;
@@ -76,7 +76,7 @@ public class RemoveRequestOperationHandlerJUnitTest extends OperationHandlerJUni
CodecNotRegisteredForTypeException, InvalidExecutionContextException {
RegionAPI.RemoveRequest removeRequest = generateTestRequest(false, false).getRemoveRequest();
Result<RegionAPI.RemoveResponse> result = operationHandler.process(serializationServiceStub,
- removeRequest, new ExecutionContext(cacheStub));
+ removeRequest, new MessageExecutionContext(cacheStub));
assertTrue(result instanceof Success);
verify(regionStub).remove(TEST_KEY);
@@ -88,7 +88,7 @@ public class RemoveRequestOperationHandlerJUnitTest extends OperationHandlerJUni
CodecNotRegisteredForTypeException, InvalidExecutionContextException {
RegionAPI.RemoveRequest removeRequest = generateTestRequest(true, false).getRemoveRequest();
Result<RegionAPI.RemoveResponse> result = operationHandler.process(serializationServiceStub,
- removeRequest, new ExecutionContext(cacheStub));
+ removeRequest, new MessageExecutionContext(cacheStub));
assertTrue(result instanceof Failure);
assertEquals(ProtocolErrorCode.REGION_NOT_FOUND.codeValue,
@@ -101,7 +101,7 @@ public class RemoveRequestOperationHandlerJUnitTest extends OperationHandlerJUni
CodecNotRegisteredForTypeException, InvalidExecutionContextException {
RegionAPI.RemoveRequest removeRequest = generateTestRequest(false, true).getRemoveRequest();
Result<RegionAPI.RemoveResponse> result = operationHandler.process(serializationServiceStub,
- removeRequest, new ExecutionContext(cacheStub));
+ removeRequest, new MessageExecutionContext(cacheStub));
assertTrue(result instanceof Success);
}
@@ -124,7 +124,7 @@ public class RemoveRequestOperationHandlerJUnitTest extends OperationHandlerJUni
RegionAPI.RemoveRequest removeRequest =
ProtobufRequestUtilities.createRemoveRequest(TEST_REGION, encodedKey).getRemoveRequest();;
Result<RegionAPI.RemoveResponse> result = operationHandler.process(serializationServiceStub,
- removeRequest, new ExecutionContext(cacheStub));
+ removeRequest, new MessageExecutionContext(cacheStub));
assertTrue(result instanceof Failure);
assertEquals(ProtocolErrorCode.VALUE_ENCODING_ERROR.codeValue,