SGF-149 implement chunking in PojoFunctionWrapper

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/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<Integer> result = new ArrayList<Integer>();
+		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<Object> {
+		private List<Object> results = new ArrayList<Object>();
+		
+		/* (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<Object> results);
+		
+		public List<Object> getResults() {
+			return this.results;
+		}
+		
+		
+	}
+	
+	public static class TestArrayResultSender extends AbstractTestResultSender {
+	
+		protected void addResults(Object arg0, List<Object> 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<Object> results) {
+			if (arg0 == null) {
+				return;
+			}
+			assertTrue(arg0 instanceof Collection);
+			Collection<?> list  = (Collection<?>) arg0;
+			results.addAll(list);
+		}		
+	}
+}

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<String, Integer>(dataSet);
 		}
+	    
+	    @GemfireFunction(id="arrays",batchSize=2)  
+	    	public int[] collections(int[] args) {
+				return args;
+	    }
 	}
 }

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