
package forager.client;

import java.io.IOException;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

import forager.events.ForagerEventMap;
import forager.events.JoinEvent;
import forager.events.TaskRequest;
import forager.events.TaskSpec;

import galileo.event.EventContext;
import galileo.event.EventHandler;
import galileo.event.EventReactor;
import galileo.net.ClientMessageRouter;
import galileo.net.NetworkDestination;

public class Forager {

    private NetworkDestination server;
    private ClientMessageRouter messageRouter;

    private ForagerEventMap eventMap = new ForagerEventMap();
    private EventReactor eventReactor = new EventReactor(this, eventMap);

    protected ThreadPoolExecutor threadPool;

    public Forager(NetworkDestination server) {
        this(server, 4);
    }

    public Forager(NetworkDestination server, int threads) {
        this.server = server;
        this.threadPool
            = (ThreadPoolExecutor) Executors.newFixedThreadPool(threads);
    }

    public void start()
    throws Exception {
        messageRouter = new ClientMessageRouter();
        messageRouter.addListener(eventReactor);

        /* Join the network */
        JoinEvent join = new JoinEvent();
        messageRouter.sendMessage(server, eventReactor.wrapEvent(join));

        /* Start the monitor thread */
        Thread monitor = new Thread(new StatusMonitor(this));
        monitor.start();

        while (true) {
            eventReactor.processNextEvent();
        }
    }

    protected synchronized void submitTaskRequest(int numTasks)
    throws IOException {
        TaskRequest tr = new TaskRequest(numTasks);
        messageRouter.sendMessage(server, eventReactor.wrapEvent(tr));
    }

    @EventHandler
    public void processTaskSpec(TaskSpec taskSpec, EventContext context) {
        System.out.println("Starting job: " + taskSpec);
        Task task = new Task(taskSpec.command);
        threadPool.submit(task);
    }

    public static void main(String[] args)
    throws Exception {
        if (args.length < 2) {
            System.out.println("Usage: forager.client.Forager <host> <port>");
            System.exit(1);
        }

        String host = args[0];
        int port = Integer.parseInt(args[1]);
        NetworkDestination overlord = new NetworkDestination(host, port);

        Forager forager = new Forager(overlord);
        forager.start();
    }
}

@@ -83,17 +83,23 @@
 Executor
-s
+Service
 ;%0Aimport
@@ -120,34 +120,24 @@
 current.
-ThreadPool
 Executor
 ;%0A%0Aimpor
@@ -128,16 +128,17 @@
 Executor
+s
 ;%0A%0Aimpor
@@ -715,34 +715,57 @@
   pr
-otected ThreadPool
+ivate int activeTasks = 0;%0A    protected 
 Executor
  thr
@@ -760,16 +760,23 @@
 Executor
+Service
  threadP
@@ -979,43 +979,10 @@
 Pool
-%0A            = (ThreadPoolExecutor)
+ =
  Exe
@@ -1746,24 +1746,257 @@
 r));%0A    %7D%0A%0A
+    protected synchronized void finalizeTask() %7B%0A        System.out.println(%22Received task completion notification%22);%0A        activeTasks--;%0A    %7D%0A%0A    protected synchronized int getActiveCount() %7B%0A        return activeTasks;%0A    %7D%0A%0A
     @EventHa
@@ -2133,16 +2133,16 @@
 kSpec);%0A
-
         
@@ -2177,24 +2177,47 @@
 c.command);%0A
+        activeTasks++;%0A
         thre

Implement active task count