diff --git a/src/main/java/org/springframework/data/gemfire/IndexFactoryBean.java b/src/main/java/org/springframework/data/gemfire/IndexFactoryBean.java index 1075fc1f..a2eeb03d 100644 --- a/src/main/java/org/springframework/data/gemfire/IndexFactoryBean.java +++ b/src/main/java/org/springframework/data/gemfire/IndexFactoryBean.java @@ -16,12 +16,11 @@ package org.springframework.data.gemfire; -import java.util.Collection; - import org.springframework.beans.factory.BeanNameAware; import org.springframework.beans.factory.FactoryBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; import com.gemstone.gemfire.cache.RegionService; @@ -29,12 +28,7 @@ import com.gemstone.gemfire.cache.client.Pool; import com.gemstone.gemfire.cache.client.PoolManager; import com.gemstone.gemfire.cache.query.Index; import com.gemstone.gemfire.cache.query.IndexExistsException; -import com.gemstone.gemfire.cache.query.IndexInvalidException; -import com.gemstone.gemfire.cache.query.IndexNameConflictException; -import com.gemstone.gemfire.cache.query.IndexType; import com.gemstone.gemfire.cache.query.QueryService; -import com.gemstone.gemfire.cache.query.RegionNotFoundException; -import com.springsource.vfabric.licensing.log.Logger; /** * Factory bean for easy declarative creation of GemFire Indexes. @@ -48,8 +42,11 @@ public class IndexFactoryBean implements InitializingBean, BeanNameAware, Factor private String poolName; private RegionService cache; private String beanName; - private String name, expression, from, imports; - private IndexType type = IndexType.FUNCTIONAL; + private String name; + private String expression; + private String from; + private String imports; + private String type; private boolean override = true; public void afterPropertiesSet() throws Exception { @@ -67,7 +64,13 @@ public class IndexFactoryBean implements InitializingBean, BeanNameAware, Factor Assert.notNull(queryService, "Query service required for index creation"); Assert.hasText(expression, "Index expression is required"); - Assert.hasText(from, "Index from clause is required"); + Assert.hasText(from, "Index from clause (regionPath) is required"); + + if (StringUtils.hasText(type)) { + if (type.equalsIgnoreCase("KEY") || type.equalsIgnoreCase("PRIMARY_KEY")) { + Assert.isNull(imports, "The imports property is not supported for a key index"); + } + } String indexName = StringUtils.hasText(name) ? name : beanName; @@ -76,41 +79,59 @@ public class IndexFactoryBean implements InitializingBean, BeanNameAware, Factor index = createIndex(queryService, indexName); } - private Index createIndex(QueryService queryService, String indexName) throws Exception { - Collection indexes = queryService.getIndexes(); + private Index createIndex(QueryService queryService, String indexName) throws Exception { - Index old = null; + Index existingIndex = null; - for (Index index : indexes) { - if (indexName.equals(index.getName())) { - if (!override) { - return index; - } - old = index; - break; + for (Index idx : queryService.getIndexes()) { + if (idx.getName().equals(indexName)) { + existingIndex = idx; + break; } } - - if (old != null) { - // compare indices - if (from.equals(old.getFromClause()) && expression.equals(old.getIndexedExpression()) - && type.equals(old.getType())) { - return index; + if (existingIndex != null) { + if (!override) { + return existingIndex; + } else { + queryService.removeIndex(existingIndex); } } Index index = null; try { - if (StringUtils.hasText(imports)) { - index = queryService.createIndex(indexName, type, expression, from, imports); + if ("KEY".equalsIgnoreCase(type) || "PRIMARY_KEY".equalsIgnoreCase(type)) { + + index = queryService.createKeyIndex(indexName, expression, from); + + } else if ("HASH".equalsIgnoreCase(type)) { + if (StringUtils.hasText(imports)) { + index = queryService.createHashIndex(indexName, expression, from, imports); + } else { + index = queryService.createHashIndex(indexName, expression, from); + } + } else { + if (StringUtils.hasText(imports)) { + index = queryService.createIndex(indexName, expression, from, imports); + } else { + index = queryService.createIndex(indexName, expression, from); + } + } + return index; + + } catch (IndexExistsException e) { + for (Index idx : queryService.getIndexes()) { + if (idx.getName().equals(indexName)) { + return idx; + } + } + } catch (Exception e) { + if (existingIndex != null) { + if (CollectionUtils.isEmpty(queryService.getIndexes()) + || !queryService.getIndexes().contains(existingIndex)) { + queryService.getIndexes().add(existingIndex); + } + } } - else { - index = queryService.createIndex(indexName, type, expression, from); - } - - } catch (IndexExistsException e) { - // This is ok - } return index; } @@ -189,7 +210,7 @@ public class IndexFactoryBean implements InitializingBean, BeanNameAware, Factor /** * @param type the type to set */ - public void setType(IndexType type) { + public void setType(String type) { this.type = type; } diff --git a/src/main/java/org/springframework/data/gemfire/RegionFactoryBean.java b/src/main/java/org/springframework/data/gemfire/RegionFactoryBean.java index 3edcd1ee..c7676a67 100644 --- a/src/main/java/org/springframework/data/gemfire/RegionFactoryBean.java +++ b/src/main/java/org/springframework/data/gemfire/RegionFactoryBean.java @@ -24,7 +24,6 @@ import org.springframework.beans.factory.DisposableBean; import org.springframework.context.SmartLifecycle; import org.springframework.core.io.Resource; import org.springframework.data.gemfire.client.ClientRegionFactoryBean; -import org.springframework.data.gemfire.wan.SmartLifecycleGatewaySender; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; import org.springframework.util.ReflectionUtils; @@ -425,8 +424,8 @@ public class RegionFactoryBean extends RegionLookupFactoryBean imple if (!ObjectUtils.isEmpty(gatewaySenders)) { synchronized (gatewaySenders) { for (Object obj : gatewaySenders) { - SmartLifecycleGatewaySender gws = (SmartLifecycleGatewaySender) obj; - if (gws.isAutoStartup() && !gws.isRunning()) { + GatewaySender gws = (GatewaySender) obj; + if (!gws.isManualStart() && !gws.isRunning()) { gws.start(); } } @@ -443,7 +442,7 @@ public class RegionFactoryBean extends RegionLookupFactoryBean imple if (!ObjectUtils.isEmpty(gatewaySenders)) { synchronized (gatewaySenders) { for (Object obj : gatewaySenders) { - SmartLifecycleGatewaySender gws = (SmartLifecycleGatewaySender) obj; + GatewaySender gws = (GatewaySender) obj; gws.stop(); } } @@ -483,5 +482,4 @@ public class RegionFactoryBean extends RegionLookupFactoryBean imple stop(); callback.run(); } - } diff --git a/src/main/java/org/springframework/data/gemfire/config/AsyncEventQueueParser.java b/src/main/java/org/springframework/data/gemfire/config/AsyncEventQueueParser.java index 87601297..216b2bd0 100644 --- a/src/main/java/org/springframework/data/gemfire/config/AsyncEventQueueParser.java +++ b/src/main/java/org/springframework/data/gemfire/config/AsyncEventQueueParser.java @@ -49,6 +49,12 @@ public class AsyncEventQueueParser extends AbstractSingleBeanDefinitionParser { ParsingUtils.setPropertyValue(element, builder, "batch-size"); ParsingUtils.setPropertyValue(element, builder, "maximum-queue-memory"); ParsingUtils.setPropertyValue(element, builder, "disk-store-ref"); + + String diskStoreRef = element.getAttribute("disk-store-ref"); + if (StringUtils.hasText(diskStoreRef)) { + builder.addDependsOn(diskStoreRef); + } + ParsingUtils.setPropertyValue(element, builder, "persistent"); ParsingUtils.setPropertyValue(element, builder, "parallel"); ParsingUtils.setPropertyValue(element, builder, NAME_ATTRIBUTE); diff --git a/src/main/java/org/springframework/data/gemfire/function/BatchingResultSender.java b/src/main/java/org/springframework/data/gemfire/function/BatchingResultSender.java new file mode 100644 index 00000000..b0ef73bf --- /dev/null +++ b/src/main/java/org/springframework/data/gemfire/function/BatchingResultSender.java @@ -0,0 +1,171 @@ +/* + * Copyright 2002-2013 the original author or authors. + * + * Licensed 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.springframework.data.gemfire.function; + +import java.lang.reflect.Array; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Iterator; +import java.util.List; + +import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; + +import com.gemstone.gemfire.cache.execute.ResultSender; + +/** + * Sends collection results using a {@link ResultSender} in chunks determined by batchSize + * + * @author David Turanski + * + */ +public class BatchingResultSender { + private final int batchSize; + private ResultSender resultSender; + + public BatchingResultSender(int batchSize, ResultSender resultSender) { + Assert.notNull(resultSender, "resultSender cannot be null"); + Assert.isTrue(batchSize >= 0, "batchSize must be >= 0"); + this.batchSize = batchSize; + this.resultSender = resultSender; + } + + + public void sendResults(Iterable result) { + if (batchSize == 0) { + resultSender.lastResult(result); + return; + } + + List chunk = new ArrayList(batchSize); + + for (Iterator it = result.iterator(); it.hasNext();) { + if (chunk.size() < batchSize) { + chunk.add(it.next()); + } + + if (chunk.size() == batchSize || !it.hasNext()) { + if (it.hasNext()) { + resultSender.sendResult(chunk); + } else { + resultSender.lastResult(chunk); + } + chunk.clear(); + } + } + } + + + public void sendArrayResults(Object result) { + + if (batchSize == 0) { + resultSender.lastResult(result); + return; + } + + Assert.isTrue(ObjectUtils.isArray(result)); + + int length = Array.getLength(result); + + for (int from =0; from < length; from += batchSize) { + int to = Math.min(length,from + batchSize); + Object chunk = copyOfRange(result,from, to); + + if (to == length -1) { + resultSender.lastResult(chunk); + } else { + resultSender.sendResult(chunk); + } + } + } + + + /** + * @param result + * @param from + * @param to + * @return + */ + private Object copyOfRange(Object result, int from, int to) { + + Class arrayClass = result.getClass(); + int size = to - from; + + if (int[].class.isAssignableFrom(arrayClass)) { + int[] array = new int[size]; + for(int i = 0; i < size ; ++i){ + array[i] = Array.getInt(result, from + i); + } + return array; + } + + if (float[].class.isAssignableFrom(arrayClass)) { + float[] array = new float[size]; + for(int i = 0; i < size ; ++i){ + array[i] = Array.getFloat(result, from + i); + } + return array; + } + + if (double[].class.isAssignableFrom(arrayClass)) { + double[] array = new double[size]; + for(int i = 0; i < size ; ++i){ + array[i] = Array.getDouble(result, from + i); + } + return array; + } + + if (boolean[].class.isAssignableFrom(arrayClass)) { + boolean[] array = new boolean[size]; + for(int i = 0; i < size ; ++i){ + array[i] = Array.getBoolean(result, from + i); + } + return array; + } + + if (byte[].class.isAssignableFrom(arrayClass)) { + byte[] array = new byte[size]; + for(int i = 0; i < size ; ++i){ + array[i] = Array.getByte(result, from + i); + } + return array; + } + + if (short[].class.isAssignableFrom(arrayClass)) { + short[] array = new short[size]; + for(int i = 0; i < size ; ++i){ + array[i] = Array.getShort(result, from + i); + } + return array; + } + + if (long[].class.isAssignableFrom(arrayClass)) { + long[] array = new long[size]; + for(int i = 0; i < size ; ++i){ + array[i] = Array.getLong(result, from + i); + } + return array; + } + + if (char[].class.isAssignableFrom(arrayClass)) { + char[] array = new char[size]; + for(int i = 0; i < size ; ++i){ + array[i] = Array.getChar(result, from + i); + } + return array; + } + + return Arrays.copyOfRange((Object[])result, from, to); + + } +} diff --git a/src/main/java/org/springframework/data/gemfire/function/PojoFunctionWrapper.java b/src/main/java/org/springframework/data/gemfire/function/PojoFunctionWrapper.java index 31307c2b..a4e37b50 100644 --- a/src/main/java/org/springframework/data/gemfire/function/PojoFunctionWrapper.java +++ b/src/main/java/org/springframework/data/gemfire/function/PojoFunctionWrapper.java @@ -12,9 +12,11 @@ */ package org.springframework.data.gemfire.function; - import java.lang.reflect.Method; +import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; +import java.util.Iterator; import java.util.List; import org.apache.commons.logging.Log; @@ -22,7 +24,6 @@ import org.apache.commons.logging.LogFactory; import org.springframework.util.ObjectUtils; import org.springframework.util.ReflectionUtils; import org.springframework.util.StringUtils; - import com.gemstone.gemfire.cache.execute.Function; import com.gemstone.gemfire.cache.execute.FunctionContext; import com.gemstone.gemfire.cache.execute.ResultSender; @@ -37,7 +38,7 @@ import com.gemstone.gemfire.cache.execute.ResultSender; * @author David Turanski * */ - + @SuppressWarnings("serial") public class PojoFunctionWrapper implements Function { @@ -45,25 +46,26 @@ public class PojoFunctionWrapper implements Function { private volatile boolean HA; private volatile boolean optimizeForWrite; - private final boolean hasResult; + private volatile boolean hasResult; private final Object target; private final Method method; private final String id; + private volatile int batchSize; private final FunctionArgumentResolver functionArgumentResolver; public PojoFunctionWrapper(Object target, Method method, String id) { - + this.functionArgumentResolver = new FunctionContextInjectingArgumentResolver(method); - + this.id = StringUtils.hasText(id) ? id : method.getName(); this.target = target; this.method = method; this.HA = false; - + this.hasResult = !(method.getReturnType().equals(void.class)); - + this.optimizeForWrite = false; } @@ -94,23 +96,29 @@ public class PojoFunctionWrapper implements Function { public void setOptimizeForWrite(boolean optimizeForWrite) { this.optimizeForWrite = optimizeForWrite; } - + + public void setBatchSize(int batchSize) { + this.batchSize = batchSize; + } + + public void setHasResult(boolean hasResult) { + this.hasResult = hasResult; + } + //@Override public void execute(FunctionContext functionContext) { - + Object[] args = this.functionArgumentResolver.resolveFunctionArguments(functionContext); - Object result = null; result = invokeTargetMethod(args); - + if (hasResult()) { sendResults(functionContext.getResultSender(), result); } } - protected final Object invokeTargetMethod(Object[] args) { if (logger.isDebugEnabled()) { @@ -126,31 +134,16 @@ public class PojoFunctionWrapper implements Function { return (Object) ReflectionUtils.invokeMethod(method, target, (Object[]) args); } - @SuppressWarnings("unchecked") private void sendResults(ResultSender resultSender, Object result) { if (result == null) { resultSender.lastResult(null); return; } - - List results = null; - - if (ObjectUtils.isArray(result)) { - results = Arrays.asList((Object[]) result); - - } else if (List.class.isAssignableFrom(result.getClass())) { - results = (List) result; - } - if (results != null) { - int i = 0; - for (Object item : results) { - if (i++ < results.size() - 1) { - resultSender.sendResult(item); - } else { - resultSender.lastResult(item); - } - } + if (ObjectUtils.isArray(result)) { + new BatchingResultSender(batchSize, resultSender).sendArrayResults(result); + } else if (Iterable.class.isAssignableFrom(result.getClass())) { + new BatchingResultSender(batchSize, resultSender).sendResults((Iterable) result); } else { resultSender.lastResult(result); } diff --git a/src/main/java/org/springframework/data/gemfire/function/config/GemfireFunction.java b/src/main/java/org/springframework/data/gemfire/function/config/GemfireFunction.java index 984beddf..d0a25c03 100644 --- a/src/main/java/org/springframework/data/gemfire/function/config/GemfireFunction.java +++ b/src/main/java/org/springframework/data/gemfire/function/config/GemfireFunction.java @@ -40,4 +40,13 @@ public @interface GemfireFunction { * is the function optimized for write operations */ boolean optimizeForWrite() default false; + /** + * controls the maximum number of results sent at one time + */ + int batchSize() default 0; + /** + * normally follows the method return type, i.e., false if void, true otherwise. This allows overriding + * a void method which uses the resultSender directly. + */ + boolean hasResult() default false; } diff --git a/src/main/java/org/springframework/data/gemfire/function/config/GemfireFunctionUtils.java b/src/main/java/org/springframework/data/gemfire/function/config/GemfireFunctionUtils.java index 9fa3e73e..3f4a8a4c 100644 --- a/src/main/java/org/springframework/data/gemfire/function/config/GemfireFunctionUtils.java +++ b/src/main/java/org/springframework/data/gemfire/function/config/GemfireFunctionUtils.java @@ -51,9 +51,25 @@ public abstract class GemfireFunctionUtils { if (attributes.containsKey("HA")) { function.setHA((Boolean) attributes.get("HA")); } + if (attributes.containsKey("optimizeForWrite")) { function.setOptimizeForWrite((Boolean) attributes.get("optimizeForWrite")); } + + if (attributes.containsKey("batchSize")) { + int batchSize = (Integer) attributes.get("batchSize"); + Assert.isTrue(batchSize >= 0, String.format("batchSize must be a non-negative value %s.%s", + target.getClass().getName(), method.getName())); + function.setBatchSize(batchSize); + } + + if (attributes.containsKey("hasResult")) { + boolean hasResult = (Boolean)attributes.get("hasResult"); + //Only set if true + if (hasResult) { + function.setHasResult(hasResult); + } + } if (FunctionService.isRegistered(function.getId())) { if (overwrite) { @@ -63,6 +79,7 @@ public abstract class GemfireFunctionUtils { FunctionService.unregisterFunction(function.getId()); } } + if (!FunctionService.isRegistered(function.getId())) { FunctionService.registerFunction(function); if (log.isDebugEnabled()) { @@ -70,7 +87,7 @@ public abstract class GemfireFunctionUtils { } } else { if (log.isDebugEnabled()) { - log.debug("function already registered " + function.getId()); + log.debug("function " + function.getId()+ "is already registered"); } } } diff --git a/src/main/java/org/springframework/data/gemfire/function/execution/AbstractFunctionExecution.java b/src/main/java/org/springframework/data/gemfire/function/execution/AbstractFunctionExecution.java index 42362277..53e93aec 100644 --- a/src/main/java/org/springframework/data/gemfire/function/execution/AbstractFunctionExecution.java +++ b/src/main/java/org/springframework/data/gemfire/function/execution/AbstractFunctionExecution.java @@ -100,7 +100,9 @@ abstract class AbstractFunctionExecution { resultCollector = (ResultCollector) execution.execute(function); } - logger.debug("ResultsCollector:" + resultCollector.getClass().getName()); + if (logger.isDebugEnabled()) { + logger.debug("using ResultsCollector:" + resultCollector.getClass().getName()); + } Iterable results = null; diff --git a/src/main/java/org/springframework/data/gemfire/function/execution/GemfireFunctionProxyFactoryBean.java b/src/main/java/org/springframework/data/gemfire/function/execution/GemfireFunctionProxyFactoryBean.java index 3a5fa824..a4e0786e 100644 --- a/src/main/java/org/springframework/data/gemfire/function/execution/GemfireFunctionProxyFactoryBean.java +++ b/src/main/java/org/springframework/data/gemfire/function/execution/GemfireFunctionProxyFactoryBean.java @@ -68,9 +68,9 @@ public class GemfireFunctionProxyFactoryBean implements FactoryBean, Met - protected Iterable invokeFunction(Method method, Object[] args) { + protected Object invokeFunction(Method method, Object[] args) { MethodMetadata mmd = this.methodMetadata.getMethodMetadata(method); - return this.gemfireFunctionOperations.execute(mmd.getFunctionId(), args); + return this.gemfireFunctionOperations.executeAndExtract(mmd.getFunctionId(), args); } @Override @@ -89,9 +89,7 @@ public class GemfireFunctionProxyFactoryBean implements FactoryBean, Met logger.debug("invoking method " + invocation.getMethod().getName()); } - Iterable results = invokeFunction(invocation.getMethod(), invocation.getArguments()); - - return extractResult(results, invocation.getMethod().getReturnType()); + return invokeFunction(invocation.getMethod(), invocation.getArguments()); } diff --git a/src/main/java/org/springframework/data/gemfire/mapping/MappingPdxSerializer.java b/src/main/java/org/springframework/data/gemfire/mapping/MappingPdxSerializer.java index 8b30094d..0f6cb9e7 100644 --- a/src/main/java/org/springframework/data/gemfire/mapping/MappingPdxSerializer.java +++ b/src/main/java/org/springframework/data/gemfire/mapping/MappingPdxSerializer.java @@ -143,7 +143,6 @@ public class MappingPdxSerializer implements PdxSerializer, ApplicationContextAw PdxSerializer customSerializer = getCustomSerializer(persistentProperty.getType()); Object value = null; if (customSerializer != null) { - System.out.println("using custom serializer"); value = customSerializer.fromData(persistentProperty.getType(), reader); } else { value = reader.readField(persistentProperty.getName()); diff --git a/src/main/java/org/springframework/data/gemfire/test/StubCache.java b/src/main/java/org/springframework/data/gemfire/test/StubCache.java index 2f080e9f..29309b69 100644 --- a/src/main/java/org/springframework/data/gemfire/test/StubCache.java +++ b/src/main/java/org/springframework/data/gemfire/test/StubCache.java @@ -808,51 +808,58 @@ public class StubCache implements Cache { String indexName = (String)invocation.getArguments()[0]; String indexedExpression = (String)invocation.getArguments()[1]; String fromClause = (String)invocation.getArguments()[2]; - return mockIndex(indexName, null, indexedExpression, fromClause, null, null); + return mockIndex(indexName, IndexType.FUNCTIONAL, indexedExpression, fromClause, null); } }); when(qs.createIndex(anyString(), anyString(),anyString(),anyString())).thenAnswer(new Answer(){ + @Override + public Index answer(InvocationOnMock invocation) throws Throwable { + String indexName = (String)invocation.getArguments()[0]; + String indexedExpression = (String)invocation.getArguments()[1]; + String fromClause = (String)invocation.getArguments()[2]; + String imports = (String)invocation.getArguments()[3]; + return mockIndex(indexName, IndexType.FUNCTIONAL, indexedExpression, fromClause, imports); + } + }); + + when(qs.createKeyIndex(anyString(), anyString(),anyString())).thenAnswer(new Answer(){ @Override public Index answer(InvocationOnMock invocation) throws Throwable { String indexName = (String)invocation.getArguments()[0]; String indexedExpression = (String)invocation.getArguments()[1]; String fromClause = (String)invocation.getArguments()[2]; - return mockIndex(indexName, null, indexedExpression, fromClause, null, null); + return mockIndex(indexName, IndexType.PRIMARY_KEY, indexedExpression, fromClause, null); } }); - when(qs.createIndex(anyString(), any(IndexType.class), anyString(), anyString())).thenAnswer(new Answer(){ + when(qs.createHashIndex(anyString(), anyString(),anyString())).thenAnswer(new Answer(){ @Override public Index answer(InvocationOnMock invocation) throws Throwable { String indexName = (String)invocation.getArguments()[0]; - IndexType type = (IndexType)invocation.getArguments()[1]; - String indexedExpression = (String)invocation.getArguments()[2]; - String fromClause = (String)invocation.getArguments()[3]; - - return mockIndex(indexName, type, indexedExpression, null, fromClause, null); + String indexedExpression = (String)invocation.getArguments()[1]; + String fromClause = (String)invocation.getArguments()[2]; + + return mockIndex(indexName, IndexType.HASH, indexedExpression, fromClause, null); } }); - - - when(qs.createIndex(anyString(), any(IndexType.class),anyString(), anyString(), anyString())).thenAnswer(new Answer(){ + when(qs.createHashIndex(anyString(), anyString(),anyString(),anyString())).thenAnswer(new Answer(){ @Override public Index answer(InvocationOnMock invocation) throws Throwable { String indexName = (String)invocation.getArguments()[0]; - IndexType type = (IndexType)invocation.getArguments()[1]; - String indexedExpression = (String)invocation.getArguments()[2]; - String fromClause = (String)invocation.getArguments()[3]; - String imports = (String)invocation.getArguments()[4]; - return mockIndex(indexName, type, indexedExpression, null, fromClause, imports); + String indexedExpression = (String)invocation.getArguments()[1]; + String fromClause = (String)invocation.getArguments()[2]; + String imports = (String)invocation.getArguments()[3]; + + return mockIndex(indexName, IndexType.HASH, indexedExpression, fromClause, imports); } }); - return qs; } - Index mockIndex(String indexName, IndexType indexType,String indexedExpression, String regionPath, String fromClause, String imports){ + Index mockIndex(String indexName, IndexType indexType,String indexedExpression, String fromClause, String imports){ Index idx = mock(Index.class); when(idx.getFromClause()).thenReturn(fromClause); when(idx.getIndexedExpression()).thenReturn(indexedExpression); diff --git a/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java b/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java index 01d45801..9935aa80 100644 --- a/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java +++ b/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java @@ -45,6 +45,8 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean< private Boolean persistent; private String diskStoreRef; + + private Boolean parallel; /** * @@ -87,10 +89,12 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean< Assert.isTrue(persistent, "specifying a disk store requires persistent property to be true"); asyncEventQueueFactory.setDiskStoreName(diskStoreRef); } - if (maximumQueueMemory != null) { asyncEventQueueFactory.setMaximumQueueMemory(maximumQueueMemory); } + if( parallel != null ){ + asyncEventQueueFactory.setParallel( parallel ); + } asyncEventQueue = asyncEventQueueFactory.create(getName(), asyncEventListener); } @@ -118,4 +122,8 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean< public void setPersistent(Boolean persistent) { this.persistent = persistent; } + + public void setParallel(Boolean parallel){ + this.parallel = parallel; + } } diff --git a/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java b/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java index d4fc3e15..af83f0f1 100644 --- a/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java +++ b/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java @@ -18,6 +18,7 @@ package org.springframework.data.gemfire.wan; import java.util.Arrays; import java.util.List; +import org.springframework.context.SmartLifecycle; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; @@ -33,7 +34,8 @@ import com.gemstone.gemfire.cache.wan.GatewayTransportFilter; * @author David Turanski * */ -public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean { +public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean +implements SmartLifecycle { private static List validOrderPolicyValues = Arrays.asList("KEY", "PARTITION", "THREAD"); private GatewaySender gatewaySender; @@ -82,12 +84,12 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean getObjectType() { - return SmartLifecycleGatewaySender.class; + return GatewaySender.class; } @Override @@ -233,4 +235,55 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean getGatewayEventFilters() { - return this.delegate.getGatewayEventFilters(); - } - - /* (non-Javadoc) - * @see com.gemstone.gemfire.cache.wan.GatewaySender#getGatewayTransportFilters() - */ - @Override - public List getGatewayTransportFilters() { - return this.delegate.getGatewayTransportFilters(); - } - - /* (non-Javadoc) - * @see com.gemstone.gemfire.cache.wan.GatewaySender#getId() - */ - @Override - public String getId() { - return this.delegate.getId(); - } - - /* (non-Javadoc) - * @see com.gemstone.gemfire.cache.wan.GatewaySender#getMaximumQueueMemory() - */ - @Override - public int getMaximumQueueMemory() { - return this.delegate.getMaximumQueueMemory(); - } - - /* (non-Javadoc) - * @see com.gemstone.gemfire.cache.wan.GatewaySender#getOrderPolicy() - */ - @Override - public OrderPolicy getOrderPolicy() { - return this.delegate.getOrderPolicy(); - } - - /* (non-Javadoc) - * @see com.gemstone.gemfire.cache.wan.GatewaySender#getRemoteDSId() - */ - @Override - public int getRemoteDSId() { - return this.delegate.getRemoteDSId(); - } - - /* (non-Javadoc) - * @see com.gemstone.gemfire.cache.wan.GatewaySender#getSocketBufferSize() - */ - @Override - public int getSocketBufferSize() { - return this.delegate.getSocketBufferSize(); - } - - /* (non-Javadoc) - * @see com.gemstone.gemfire.cache.wan.GatewaySender#getSocketReadTimeout() - */ - @Override - public int getSocketReadTimeout() { - return this.delegate.getSocketReadTimeout(); - } - - /* (non-Javadoc) - * @see com.gemstone.gemfire.cache.wan.GatewaySender#isBatchConflationEnabled() - */ - @Override - public boolean isBatchConflationEnabled() { - return this.delegate.isBatchConflationEnabled(); - } - - /* (non-Javadoc) - * @see com.gemstone.gemfire.cache.wan.GatewaySender#isDiskSynchronous() - */ - @Override - public boolean isDiskSynchronous() { - return this.delegate.isDiskSynchronous(); - } - - /* (non-Javadoc) - * @see com.gemstone.gemfire.cache.wan.GatewaySender#isManualStart() - */ - @Override - public boolean isManualStart() { - return true; - } - - /* (non-Javadoc) - * @see com.gemstone.gemfire.cache.wan.GatewaySender#isParallel() - */ - @Override - public boolean isParallel() { - return this.delegate.isParallel(); - } - - /* (non-Javadoc) - * @see com.gemstone.gemfire.cache.wan.GatewaySender#isPaused() - */ - @Override - public boolean isPaused() { - return this.delegate.isPaused(); - } - - /* (non-Javadoc) - * @see com.gemstone.gemfire.cache.wan.GatewaySender#isPersistenceEnabled() - */ - @Override - public boolean isPersistenceEnabled() { - return this.delegate.isPersistenceEnabled(); - } - - /* (non-Javadoc) - * @see com.gemstone.gemfire.cache.wan.GatewaySender#isRunning() - */ - @Override - public boolean isRunning() { - return this.delegate.isRunning(); - } - - /* (non-Javadoc) - * @see com.gemstone.gemfire.cache.wan.GatewaySender#pause() - */ - @Override - public void pause() { - this.delegate.pause(); - } - - /* (non-Javadoc) - * @see com.gemstone.gemfire.cache.wan.GatewaySender#removeGatewayEventFilter(com.gemstone.gemfire.cache.wan.GatewayEventFilter) - */ - @Override - public void removeGatewayEventFilter(GatewayEventFilter eventFilter) { - this.delegate.removeGatewayEventFilter(eventFilter); - } - - /* (non-Javadoc) - * @see com.gemstone.gemfire.cache.wan.GatewaySender#resume() - */ - @Override - public void resume() { - this.delegate.resume(); - - } - - /* (non-Javadoc) - * @see com.gemstone.gemfire.cache.wan.GatewaySender#start() - */ - @Override - public void start() { - this.delegate.start(); - - } - - /* (non-Javadoc) - * @see com.gemstone.gemfire.cache.wan.GatewaySender#stop() - */ - @Override - public void stop() { - this.delegate.stop(); - } - - /* (non-Javadoc) - * @see org.springframework.context.Phased#getPhase() - */ - @Override - public int getPhase() { - return Integer.MAX_VALUE; - } - - /* (non-Javadoc) - * @see org.springframework.context.SmartLifecycle#isAutoStartup() - */ - @Override - public boolean isAutoStartup() { - return this.autoStartup ; - } - - /* (non-Javadoc) - * @see org.springframework.context.SmartLifecycle#stop(java.lang.Runnable) - */ - @Override - public void stop(Runnable callback) { - stop(); - callback.run(); - } - -} diff --git a/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.3.xsd b/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.3.xsd index e1188eff..dddbc818 100755 --- a/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.3.xsd +++ b/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.3.xsd @@ -2144,12 +2144,11 @@ The name of the index bean definition. If property 'name' is not set, it will be ]]> - + - - + + @@ -2164,7 +2163,13 @@ The name of the index. + use="required"> + + + + - - - - - - - - + + - - - - - + + + + - - - - - + + + + - - - - - + + + + + + + + + - + @@ -2610,26 +2614,26 @@ The id of the cache - default is gemfireCache - - - - - + + + + - - - - - + + + + - - - - + + + + diff --git a/src/test/java/org/springframework/data/gemfire/config/GemfireV7GatewayNamespaceTest.java b/src/test/java/org/springframework/data/gemfire/config/GemfireV7GatewayNamespaceTest.java index afeb7982..af766982 100644 --- a/src/test/java/org/springframework/data/gemfire/config/GemfireV7GatewayNamespaceTest.java +++ b/src/test/java/org/springframework/data/gemfire/config/GemfireV7GatewayNamespaceTest.java @@ -105,10 +105,11 @@ public class GemfireV7GatewayNamespaceTest extends RecreatingContextTest { AsyncEventQueue aseq = ctx.getBean("async-event-queue", AsyncEventQueue.class); assertEquals(10, aseq.getBatchSize()); assertTrue(aseq.isPersistent()); + assertTrue(aseq.isParallel()); assertEquals("diskstore", aseq.getDiskStoreName()); assertEquals(50, aseq.getMaximumQueueMemory()); } - + @Test public void testGatewaySender() throws Exception { GatewaySenderFactoryBean gwsfb = ctx.getBean("&gateway-sender", GatewaySenderFactoryBean.class); diff --git a/src/test/java/org/springframework/data/gemfire/config/IndexNamespaceTest.java b/src/test/java/org/springframework/data/gemfire/config/IndexNamespaceTest.java index 1ed5bdce..e683ccbc 100644 --- a/src/test/java/org/springframework/data/gemfire/config/IndexNamespaceTest.java +++ b/src/test/java/org/springframework/data/gemfire/config/IndexNamespaceTest.java @@ -25,8 +25,6 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; import org.springframework.data.gemfire.test.GemfireTestRunner; import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; - import com.gemstone.gemfire.cache.Cache; import com.gemstone.gemfire.cache.query.Index; import com.gemstone.gemfire.cache.query.IndexType; @@ -71,6 +69,6 @@ public class IndexNamespaceTest { assertEquals("tsi.name", idx.getIndexedExpression()); assertEquals("complex-index", idx.getName()); assertEquals(name, idx.getRegion().getName()); - assertEquals(IndexType.PRIMARY_KEY, idx.getType()); + assertEquals(IndexType.HASH, idx.getType()); } } diff --git a/src/test/java/org/springframework/data/gemfire/function/BatchingResultSenderTest.java b/src/test/java/org/springframework/data/gemfire/function/BatchingResultSenderTest.java new file mode 100644 index 00000000..0d26d479 --- /dev/null +++ b/src/test/java/org/springframework/data/gemfire/function/BatchingResultSenderTest.java @@ -0,0 +1,139 @@ +/* + * Copyright 2002-2013 the original author or authors. + * + * Licensed 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.springframework.data.gemfire.function; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +import org.junit.Test; + +import com.gemstone.gemfire.cache.execute.ResultSender; + +/** + * @author David Turanski + * + */ +public class BatchingResultSenderTest { + + @Test + public void testArrayChunking() { + + testBatchingResultSender(new TestArrayResultSender(),1); + testBatchingResultSender(new TestArrayResultSender(),0); + testBatchingResultSender(new TestArrayResultSender(),9); + testBatchingResultSender(new TestArrayResultSender(),10); + testBatchingResultSender(new TestArrayResultSender(),1000); + } + + + @Test + public void testListChunking() { + testBatchingResultSender(new TestListResultSender(),1); + testBatchingResultSender(new TestListResultSender(),0); + testBatchingResultSender(new TestListResultSender(),9); + testBatchingResultSender(new TestListResultSender(),10); + testBatchingResultSender(new TestListResultSender(),1000); + } + + private void testBatchingResultSender(AbstractTestResultSender resultSender, int batchSize){ + BatchingResultSender brs = new BatchingResultSender(batchSize, resultSender); + + List result = new ArrayList(); + for (int i = 0; i< 100; i++) { + result.add(i); + } + //TODO: Clean this up. Ok for test code + if (resultSender instanceof TestArrayResultSender) { + brs.sendArrayResults(result.toArray(new Integer[100])); + } else { + brs.sendResults(result); + } + + assertEquals(100,resultSender.getResults().size()); + + for(int i=0; i< 100; i++) { + assertEquals(i,resultSender.getResults().get(i)); + } + + } + + public static abstract class AbstractTestResultSender implements ResultSender { + private List results = new ArrayList(); + + /* (non-Javadoc) + * @see com.gemstone.gemfire.cache.execute.ResultSender#lastResult(java.lang.Object) + */ + @Override + public void lastResult(Object arg0) { + if (arg0 == null) { + return; + } + addResults(arg0, results); + } + + /* (non-Javadoc) + * @see com.gemstone.gemfire.cache.execute.ResultSender#sendException(java.lang.Throwable) + */ + @Override + public void sendException(Throwable arg0) { + fail(); + + } + + /* (non-Javadoc) + * @see com.gemstone.gemfire.cache.execute.ResultSender#sendResult(java.lang.Object) + */ + @Override + public void sendResult(Object arg0) { + if (arg0 == null) { + return; + } + addResults(arg0, results); + } + + protected abstract void addResults(Object item, List results); + + public List getResults() { + return this.results; + } + + + } + + public static class TestArrayResultSender extends AbstractTestResultSender { + + protected void addResults(Object arg0, List results) { + assertTrue(arg0.getClass().isArray()); + Object[] array = (Object[]) arg0; + for (Object obj: array) { + results.add(obj); + } + } + } + + public static class TestListResultSender extends AbstractTestResultSender { + protected void addResults(Object arg0, List results) { + if (arg0 == null) { + return; + } + assertTrue(arg0 instanceof Collection); + Collection list = (Collection) arg0; + results.addAll(list); + } + } +} diff --git a/src/test/java/org/springframework/data/gemfire/function/execution/FunctionIntegrationTests.java b/src/test/java/org/springframework/data/gemfire/function/execution/FunctionIntegrationTests.java index 35766d83..9659a8be 100644 --- a/src/test/java/org/springframework/data/gemfire/function/execution/FunctionIntegrationTests.java +++ b/src/test/java/org/springframework/data/gemfire/function/execution/FunctionIntegrationTests.java @@ -102,7 +102,7 @@ public class FunctionIntegrationTests { assertEquals(2,map.get("two").intValue()); assertEquals(3,map.get("three").intValue()); - result = template.execute("collections",Arrays.asList(new Integer[]{1,2,3,4,5})); + result = template.executeAndExtract("collections",Arrays.asList(new Integer[]{1,2,3,4,5})); assertTrue(result.getClass().getName(),result instanceof List); List list = (List)result; @@ -111,7 +111,13 @@ public class FunctionIntegrationTests { assertEquals(i,list.get(i-1)); } } - + + @Test + public void testArrayReturnTypes() { + GemfireOnRegionOperations template = new GemfireOnRegionFunctionTemplate(region); + Object result = template.executeAndExtract("arrays",new int[]{1,2,3,4,5}); + assertTrue(result.getClass().getName(),result instanceof int[]); + } /* * This gets wrapped in a GemFire Function and registered on the forked server. */ @@ -143,5 +149,10 @@ public class FunctionIntegrationTests { } return new HashMap(dataSet); } + + @GemfireFunction(id="arrays",batchSize=2) + public int[] collections(int[] args) { + return args; + } } } diff --git a/src/test/java/org/springframework/data/gemfire/function/execution/GemfireFunctionProxyFactoryBeanTests.java b/src/test/java/org/springframework/data/gemfire/function/execution/GemfireFunctionProxyFactoryBeanTests.java index 887ebf57..577ac5a6 100644 --- a/src/test/java/org/springframework/data/gemfire/function/execution/GemfireFunctionProxyFactoryBeanTests.java +++ b/src/test/java/org/springframework/data/gemfire/function/execution/GemfireFunctionProxyFactoryBeanTests.java @@ -21,6 +21,7 @@ package org.springframework.data.gemfire.function.execution; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.lang.reflect.AccessibleObject; @@ -49,7 +50,6 @@ public class GemfireFunctionProxyFactoryBeanTests { functionOperations = mock(GemfireFunctionOperations.class); } - @SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void testInvokeAndExtractWithAnnotatedFunctionId() throws Throwable { @@ -58,17 +58,17 @@ public class GemfireFunctionProxyFactoryBeanTests { MethodInvocation invocation = new TestInvocation(IFoo.class).withMethodNameAndArgTypes("oneArg",String.class); - List results = Arrays.asList(new Integer[]{1}); + int results = 1; - when(functionOperations.execute("oneArg",invocation.getArguments())).thenReturn(results); - + when(functionOperations.executeAndExtract("oneArg",invocation.getArguments())).thenReturn(results); Object result = proxy.invoke(invocation); - assertTrue(result instanceof Integer); - assertEquals(new Integer(1),result); + verify(functionOperations).executeAndExtract("oneArg",invocation.getArguments()); + assertTrue(result.getClass().getName(), result instanceof Integer); + assertEquals(1,result); } - @SuppressWarnings({ "unchecked", "rawtypes" }) + @SuppressWarnings({ "rawtypes" }) @Test public void testInvoke() throws Throwable { @@ -79,9 +79,9 @@ public class GemfireFunctionProxyFactoryBeanTests { List results = Arrays.asList(new Integer[]{1,2,3}); - when(functionOperations.execute("collections",invocation.getArguments())).thenReturn(results); - + when(functionOperations.executeAndExtract("collections",invocation.getArguments())).thenReturn(results); Object result = proxy.invoke(invocation); + verify(functionOperations).executeAndExtract("collections",invocation.getArguments()); ; assertTrue(result instanceof List); assertEquals(3,((List)result).size()); } diff --git a/src/test/resources/org/springframework/data/gemfire/config/gateway-v7-ns.xml b/src/test/resources/org/springframework/data/gemfire/config/gateway-v7-ns.xml index 7513b1ad..d08e0778 100644 --- a/src/test/resources/org/springframework/data/gemfire/config/gateway-v7-ns.xml +++ b/src/test/resources/org/springframework/data/gemfire/config/gateway-v7-ns.xml @@ -37,7 +37,7 @@ + maximum-queue-memory="50" parallel="true"> diff --git a/src/test/resources/org/springframework/data/gemfire/config/index-ns.xml b/src/test/resources/org/springframework/data/gemfire/config/index-ns.xml index f97b3c5c..ba873f7a 100644 --- a/src/test/resources/org/springframework/data/gemfire/config/index-ns.xml +++ b/src/test/resources/org/springframework/data/gemfire/config/index-ns.xml @@ -17,7 +17,7 @@ - +