icegem-cache-utils/src/main/java/com/googlecode/icegem/cacheutils/regioncomparator/GetKeysFunction.java
@Override
    public void execute(FunctionContext functionContext) {
        Map<String, Object> args = (HashMap) functionContext.getArguments();
        String regionName = (String) args.get("regionName");
        Region region = CacheFactory.getAnyInstance().getRegion(regionName);
        if (region == null) {
            functionContext.getResultSender().lastResult(new HashSet());
            return;
        }

        int bucket = 0;
        if (PartitionRegionHelper.isPartitionedRegion(region)) {
            bucket = (Integer) args.get("bucket");                                              //todo: NPE if actual region is different from parameter (replicate vs partition)
            PartitionedRegion pr = (PartitionedRegion) region;
            Set keys = new HashSet();
            if (pr.getDataStore().getAllLocalPrimaryBucketIds().contains(bucket)) {
                try {
                    keys.addAll(pr.getDataStore().getKeysLocally(bucket));
                } catch (Exception e) {
                    throw new RuntimeException("error getting local keys for bucket " + bucket, e);
                }
            }
            functionContext.getResultSender().lastResult((HashSet) keys);
            return;
        }

        //todo: it's ugly. better: to ask a particular batch of keys (between oldBatch upper bound and nextBatch lower bound)
        Set keys = region.keySet();
        Iterator iterator = keys.iterator();
        Set keysBatch = new HashSet(BATCH_SIZE);
        while(iterator.hasNext()) {
            keysBatch.add(iterator.next());
            if ((keysBatch.size() + 1) % BATCH_SIZE  == 0) {
                functionContext.getResultSender().sendResult((HashSet) keysBatch);
                keysBatch = new HashSet(BATCH_SIZE);
            }
        }
        functionContext.getResultSender().lastResult((HashSet) keysBatch);
    }
@Override
    public void execute(FunctionContext functionContext) {
        Map<String, Object> args = (HashMap) functionContext.getArguments();
        String regionName = (String) args.get("regionName");
        Region region = CacheFactory.getAnyInstance().getRegion(regionName);
        if (region == null) {
            functionContext.getResultSender().lastResult(new HashSet());
            return;
        }

        int bucket = 0;
        if (PartitionRegionHelper.isPartitionedRegion(region)) {
            bucket = (Integer) args.get("bucket");                                              //todo: NPE if actual region is different from parameter (replicate vs partition)
            PartitionedRegion pr = (PartitionedRegion) region;
            Set keys = new HashSet();
            if (pr.getDataStore().getAllLocalPrimaryBucketIds().contains(bucket)) {
                try {
                    keys.addAll(pr.getDataStore().getKeysLocally(bucket));
                } catch (Exception e) {
                    throw new RuntimeException("error getting local keys for bucket " + bucket, e);
                }
            }
            functionContext.getResultSender().lastResult((HashSet) keys);
            return;
        }

        //todo: it's ugly. better: to ask a particular batch of keys (between oldBatch upper bound and nextBatch lower bound)
        Set keys = region.keySet();
        Iterator iterator = keys.iterator();
        Set keysBatch = new HashSet(BATCH_SIZE);
        while(iterator.hasNext()) {
            keysBatch.add(iterator.next());
            if ((keysBatch.size() + 1) % BATCH_SIZE  == 0) {
                functionContext.getResultSender().sendResult((HashSet) keysBatch);
                keysBatch = new HashSet(BATCH_SIZE);
            }
        }
        functionContext.getResultSender().lastResult((HashSet) keysBatch);
    }
todo: how to customize this parameter. also it exists in regioninfofunction. ugly.