GemfireAsyncEventQueueProvisioner.java

package com.demo.aggregate;

import com.gemstone.gemfire.cache.Cache;
import com.gemstone.gemfire.cache.DataPolicy;
import com.gemstone.gemfire.cache.Scope;

public class GemfireAggregatesProvisioner {

    private static String AGGREGATE_EVENT_QUEUE = "async-aggregate-queue";
    private static String AGGREGATE_EVENT_REGION = "aggregateEvent";
    private static String AGGREGATE_REGION = "AggregateRegion";

    public static void provisionIfAbsent(Cache cache) {

        if (cache.getAsyncEventQueue(AGGREGATE_EVENT_QUEUE) == null) {
            cache.createAsyncEventQueueFactory()
                .setPersistent(true)
                //            .setDiskStoreName(UUID.randomUUID().toString())
                .setParallel(false)
                .setBatchSize(10)
                .setBatchTimeInterval(100)
                .create(DAY_AGGREGATE_EVENT_QUEUE, new AggregateEventListener());
        }

        if (cache.getRegion(AGGREGATE_EVENT_REGION) == null) {
            cache.createRegionFactory()
                // We could add more listeners here calculating other kind of aggregates
                .addAsyncEventQueueId(AGGREGATE_EVENT_QUEUE)
                .setScope(Scope.DISTRIBUTED_ACK)
                .setDataPolicy(DataPolicy.REPLICATE)
                .create(AGGREGATE_EVENT_REGION);
        }

        if (cache.getRegion(DAY_AGGREGATE_REGION) == null) {
            cache.createRegionFactory()
                .setScope(Scope.DISTRIBUTED_ACK)
                .setDataPolicy(DataPolicy.REPLICATE)
                .create(AGGREGATE_REGION);
        }
    }

}