package appengine_commons.event_logger;

import com.google.appengine.api.taskqueue.Queue;
import com.google.appengine.api.taskqueue.QueueFactory;
import com.google.appengine.api.taskqueue.TaskHandle;
import com.google.appengine.repackaged.com.google.gson.Gson;
import com.google.appengine.repackaged.com.google.gson.reflect.TypeToken;

import java.io.UnsupportedEncodingException;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Created by kaiinui on 2015/05/02.
 */
public class WriteEventPullTaskDelegate {
    static final int MAX_FETCH_SIZE = 1000;

    public static void flushEvents(EventFlusher... flushers) throws UnsupportedEncodingException {
        final Queue queue = QueueFactory.getQueue(Events.PULL_QUEUE_NAME);
        final int enumerateCount = (queue.fetchStatistics().getNumTasks() / MAX_FETCH_SIZE) + 1;
        for (int i = 0; i < enumerateCount; i++) {
            final List<TaskHandle> tasks = queue.leaseTasks(3600, TimeUnit.SECONDS, MAX_FETCH_SIZE);
            final List<Events.Event> events = makeEventsFromTasks(tasks);

            for (EventFlusher flusher : flushers) {
                flusher.flushEvents(events);
            }

            queue.deleteTaskAsync(tasks);
        }
    }

    private static List<Events.Event> makeEventsFromTasks(List<TaskHandle> taskHandles) throws UnsupportedEncodingException {
        final ArrayList<Events.Event> events = new ArrayList<Events.Event>(taskHandles.size());
        for (TaskHandle taskHandle : taskHandles) {
            final Type mapT = new TypeToken<Map<String, Object>>() {}.getType();
            final String tag = taskHandle.getTag();
            final Map<String, Object> params = new Gson().fromJson(new String(taskHandle.getPayload()), mapT);
            events.add(new Events.Event(tag, params));
        }
        return events;
    }

    public static interface EventFlusher {
        public void flushEvents(List<Events.Event> events);
    }
}

@@ -520,95 +520,489 @@
 it;%0A
-import java.util.concurrent.atomic.AtomicLong;%0A%0A/**%0A * Created by kaiinui on 2015/05/02
+%0A/**%0A * A class provides a static method that handles event flushing cron job.%0A *%0A * With following code, you can flushes the events buffered in pull-queue.%0A *%0A * %3Cpre%3E%3Ccode%3E%0A *     // cron-job-targeted servlet%0A *     protected void doGet(HttpServletRequest request, HttpServletResponse response) %7B%0A *         WriteEventPullTaskDelegate.flushEvents(new DatastoreEventFlusher());%0A *     %7D%0A * %3C/code%3E%3C/pre%3E%0A *%0A * To publish events, refer %7B@link appengine_commons.event_logger.Events%7D
 .%0A *

Add some javadoc