DATAGEODE-151 - BatchingResultSender does not handle empty collections.
The current o.s.d.gemfire.function.BatchingResultSendera does not handle size/length 0 arrays or collections when batch size is larger than 0. Also fixed issue where lastResult was not sent with arrays, do to logic issue, when the lastResult is to be sent.
This commit is contained in:
Udo Kohlmeyer
@@ -12,159 +12,164 @@
|
||||
*/
|
||||
package org.springframework.data.gemfire.function;
|
||||
|
||||
import org.apache.geode.cache.execute.ResultSender;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.ObjectUtils;
|
||||
|
||||
import java.lang.reflect.Array;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
|
||||
import org.apache.geode.cache.execute.ResultSender;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.ObjectUtils;
|
||||
|
||||
/**
|
||||
* Sends collection results using a {@link ResultSender} in chunks determined by batchSize
|
||||
*
|
||||
* @author David Turanski
|
||||
* @author Udo Kohlmeyer
|
||||
* @since 1.3.0
|
||||
*/
|
||||
class BatchingResultSender {
|
||||
private final int batchSize;
|
||||
private ResultSender<Object> resultSender;
|
||||
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 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;
|
||||
}
|
||||
public void sendResults(Iterable<?> result) {
|
||||
//Handle both batchSize == 0 or empty Iterable
|
||||
if (batchSize == 0 || !result.iterator().hasNext()) {
|
||||
resultSender.lastResult(result);
|
||||
return;
|
||||
}
|
||||
|
||||
List<Object> chunk = new ArrayList<Object>(batchSize);
|
||||
List<Object> chunk = new ArrayList<Object>(batchSize);
|
||||
|
||||
for (Iterator<?> it = result.iterator(); it.hasNext();) {
|
||||
if (chunk.size() < batchSize) {
|
||||
chunk.add(it.next());
|
||||
}
|
||||
for (Iterator<?> it = result.iterator(); it.hasNext(); ) {
|
||||
if (chunk.size() < batchSize) {
|
||||
chunk.add(it.next());
|
||||
}
|
||||
|
||||
if (chunk.size() == batchSize || !it.hasNext()) {
|
||||
if (chunk.size() == batchSize || !it.hasNext()) {
|
||||
if (it.hasNext()) {
|
||||
resultSender.sendResult(chunk);
|
||||
resultSender.sendResult(chunk);
|
||||
} else {
|
||||
resultSender.lastResult(chunk);
|
||||
resultSender.lastResult(chunk);
|
||||
}
|
||||
chunk.clear();
|
||||
}
|
||||
}
|
||||
}
|
||||
chunk.clear();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public void sendArrayResults(Object result) {
|
||||
public void sendArrayResults(Object result) {
|
||||
Assert.isTrue(ObjectUtils.isArray(result));
|
||||
|
||||
if (batchSize == 0) {
|
||||
resultSender.lastResult(result);
|
||||
return;
|
||||
}
|
||||
if (batchSize == 0) {
|
||||
resultSender.lastResult(result);
|
||||
return;
|
||||
}
|
||||
|
||||
Assert.isTrue(ObjectUtils.isArray(result));
|
||||
int length = Array.getLength(result);
|
||||
|
||||
int length = Array.getLength(result);
|
||||
if (length == 0) {
|
||||
resultSender.lastResult(result);
|
||||
return;
|
||||
}
|
||||
|
||||
for (int from =0; from < length; from += batchSize) {
|
||||
int to = Math.min(length,from + batchSize);
|
||||
Object chunk = copyOfRange(result,from, to);
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (to == length) {
|
||||
resultSender.lastResult(chunk);
|
||||
} else {
|
||||
resultSender.sendResult(chunk);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* @param result
|
||||
* @param from
|
||||
* @param to
|
||||
* @return
|
||||
*/
|
||||
private Object copyOfRange(Object result, int from, int to) {
|
||||
/**
|
||||
* @param result
|
||||
* @param from
|
||||
* @param to
|
||||
* @return
|
||||
*/
|
||||
private Object copyOfRange(Object result, int from, int to) {
|
||||
Class<?> arrayClass = result.getClass();
|
||||
int size = to - from;
|
||||
|
||||
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){
|
||||
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;
|
||||
}
|
||||
return array;
|
||||
}
|
||||
|
||||
if (float[].class.isAssignableFrom(arrayClass)) {
|
||||
float[] array = new float[size];
|
||||
for(int i = 0; i < size ; ++i){
|
||||
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;
|
||||
}
|
||||
return array;
|
||||
}
|
||||
|
||||
if (double[].class.isAssignableFrom(arrayClass)) {
|
||||
double[] array = new double[size];
|
||||
for(int i = 0; i < size ; ++i){
|
||||
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;
|
||||
}
|
||||
return array;
|
||||
}
|
||||
|
||||
if (boolean[].class.isAssignableFrom(arrayClass)) {
|
||||
boolean[] array = new boolean[size];
|
||||
for(int i = 0; i < size ; ++i){
|
||||
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;
|
||||
}
|
||||
return array;
|
||||
}
|
||||
|
||||
if (byte[].class.isAssignableFrom(arrayClass)) {
|
||||
byte[] array = new byte[size];
|
||||
for(int i = 0; i < size ; ++i){
|
||||
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;
|
||||
}
|
||||
return array;
|
||||
}
|
||||
|
||||
if (short[].class.isAssignableFrom(arrayClass)) {
|
||||
short[] array = new short[size];
|
||||
for(int i = 0; i < size ; ++i){
|
||||
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;
|
||||
}
|
||||
return array;
|
||||
}
|
||||
|
||||
if (long[].class.isAssignableFrom(arrayClass)) {
|
||||
long[] array = new long[size];
|
||||
for(int i = 0; i < size ; ++i){
|
||||
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;
|
||||
}
|
||||
return array;
|
||||
}
|
||||
|
||||
if (char[].class.isAssignableFrom(arrayClass)) {
|
||||
char[] array = new char[size];
|
||||
for(int i = 0; i < size ; ++i){
|
||||
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 array;
|
||||
}
|
||||
|
||||
return Arrays.copyOfRange((Object[])result, from, to);
|
||||
return Arrays.copyOfRange((Object[]) result, from, to);
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user