Merge branch 'master' into datasource

incorporate updates to master branch

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<Index> 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;
 	}
 

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<K, V> extends RegionLookupFactoryBean<K, V> 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<K, V> extends RegionLookupFactoryBean<K, V> 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<K, V> extends RegionLookupFactoryBean<K, V> imple
 		stop();
 		callback.run();
 	}
-
 }

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);

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<Object> resultSender;
+	
+	public BatchingResultSender(int batchSize, ResultSender<Object> 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<Object> chunk = new ArrayList<Object>(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);
+	 
+	}
+}

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<Object> resultSender, Object result) {
 		if (result == null) {
 			resultSender.lastResult(null);
 			return;
 		}
-	 
-		List<Object> results = null;
-		
-		if (ObjectUtils.isArray(result)) {
-			results = Arrays.asList((Object[]) result);
-		
-		} else if (List.class.isAssignableFrom(result.getClass())) {
-			results = (List<Object>) 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);
 		}

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;
 }

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");
 			}
 		}
 	}

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<T> results = null;
 		

src/main/java/org/springframework/data/gemfire/function/execution/GemfireFunctionProxyFactoryBean.java

@@ -68,9 +68,9 @@ public class GemfireFunctionProxyFactoryBean implements FactoryBean<Object>, 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<Object>, 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());
 	}
 
 

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());

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<Index>(){
+			@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<Index>(){
 			@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<Index>(){
+		when(qs.createHashIndex(anyString(), anyString(),anyString())).thenAnswer(new Answer<Index>(){
 			@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<Index>(){
+		when(qs.createHashIndex(anyString(), anyString(),anyString(),anyString())).thenAnswer(new Answer<Index>(){
 			@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);

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;
+	}
 }

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<GatewaySender>  {
+public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<GatewaySender>
+implements SmartLifecycle {
 	private static List<String> validOrderPolicyValues = Arrays.asList("KEY", "PARTITION", "THREAD");
 
 	private GatewaySender gatewaySender;
@@ -82,12 +84,12 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
 
 	@Override
 	public GatewaySender getObject() throws Exception {
-		return new SmartLifecycleGatewaySender(gatewaySender, !manualStart);
+		return gatewaySender;
 	}
 
 	@Override
 	public Class<?> getObjectType() {
-		return SmartLifecycleGatewaySender.class;
+		return GatewaySender.class;
 	}
 
 	@Override
@@ -233,4 +235,55 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
 	public void setSocketReadTimeout(Integer socketReadTimeout) {
 		this.socketReadTimeout = socketReadTimeout;
 	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.context.Lifecycle#start()
+	 */
+	@Override
+	public synchronized void start() {
+		if (!gatewaySender.isRunning()){
+			gatewaySender.start();
+		}
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.context.Lifecycle#stop()
+	 */
+	@Override
+	public void stop() {
+		gatewaySender.stop();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.context.Lifecycle#isRunning()
+	 */
+	@Override
+	public boolean isRunning() {
+		return gatewaySender.isRunning();
+	}
+
+	/* (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 !manualStart;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.context.SmartLifecycle#stop(java.lang.Runnable)
+	 */
+	@Override
+	public void stop(Runnable callback) {
+		stop();
+		callback.run();
+	}
 }

src/main/java/org/springframework/data/gemfire/wan/SmartLifecycleGatewaySender.java

@@ -1,277 +0,0 @@
-/*
- * 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.wan;
-
-import java.util.List;
-
-import org.springframework.context.SmartLifecycle;
-import org.springframework.util.Assert;
-
-import com.gemstone.gemfire.cache.util.Gateway.OrderPolicy;
-import com.gemstone.gemfire.cache.wan.GatewayEventFilter;
-import com.gemstone.gemfire.cache.wan.GatewaySender;
-import com.gemstone.gemfire.cache.wan.GatewayTransportFilter;
-
-/**
- * A {@link GatewaySender} controlled by {@link SmartLifecycle}
- * @author David Turanski
- *
- */
-public class SmartLifecycleGatewaySender implements GatewaySender, SmartLifecycle {
-	
-	private final GatewaySender delegate;
-	private final boolean autoStartup;
-
-	public SmartLifecycleGatewaySender(GatewaySender delegate, boolean autoStartup) {
-		Assert.notNull(delegate);
-		this.delegate = delegate;
-		this.autoStartup = autoStartup; 
-	}
-
-	/* (non-Javadoc)
-	 * @see com.gemstone.gemfire.cache.wan.GatewaySender#addGatewayEventFilter(com.gemstone.gemfire.cache.wan.GatewayEventFilter)
-	 */
-	@Override
-	public void addGatewayEventFilter(GatewayEventFilter eventFilter) {
-		this.delegate.addGatewayEventFilter(eventFilter);
-		
-	}
-
-	/* (non-Javadoc)
-	 * @see com.gemstone.gemfire.cache.wan.GatewaySender#getAlertThreshold()
-	 */
-	@Override
-	public int getAlertThreshold() {
-		return this.delegate.getAlertThreshold();
-	}
-
-	/* (non-Javadoc)
-	 * @see com.gemstone.gemfire.cache.wan.GatewaySender#getBatchSize()
-	 */
-	@Override
-	public int getBatchSize() {
-		return this.delegate.getBatchSize();
-	}
-
-	/* (non-Javadoc)
-	 * @see com.gemstone.gemfire.cache.wan.GatewaySender#getBatchTimeInterval()
-	 */
-	@Override
-	public int getBatchTimeInterval() {
-		return this.delegate.getBatchTimeInterval();
-	}
-
-	/* (non-Javadoc)
-	 * @see com.gemstone.gemfire.cache.wan.GatewaySender#getDiskStoreName()
-	 */
-	@Override
-	public String getDiskStoreName() {
-		return this.delegate.getDiskStoreName();
-	}
-
-	/* (non-Javadoc)
-	 * @see com.gemstone.gemfire.cache.wan.GatewaySender#getDispatcherThreads()
-	 */
-	@Override
-	public int getDispatcherThreads() {
-		return this.delegate.getDispatcherThreads();
-	}
-
-	/* (non-Javadoc)
-	 * @see com.gemstone.gemfire.cache.wan.GatewaySender#getGatewayEventFilters()
-	 */
-	@Override
-	public List<GatewayEventFilter> getGatewayEventFilters() {
-		return this.delegate.getGatewayEventFilters();
-	}
-
-	/* (non-Javadoc)
-	 * @see com.gemstone.gemfire.cache.wan.GatewaySender#getGatewayTransportFilters()
-	 */
-	@Override
-	public List<GatewayTransportFilter> 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();
-	}
-
-}