package aj.hadoop.monitor.mapreduce;

import java.io.*;
import java.net.*;
import java.util.*;
import java.text.*;
import java.lang.*;
import java.lang.management.*;
import java.util.StringTokenizer;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.annotation.After;
import org.aspectj.lang.annotation.Before;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.Signature;

import aj.hadoop.monitor.util.PickerClient;

@Aspect
public abstract class MapperMonitor {

	private String HOST = "localhost";
	private int PORT = 9999;

	@Pointcut ("call(void org.apache.hadoop.mapreduce.Mapper.Context+.write" +
			"(" +
			"java.lang.Object, " +
			"java.lang.Object" +
			"))" +
			"&& args(key, value)")
	public void pointcut_mapper_out(Object key, Object value){}

	@Before ("pointcut_mapper_out( key, value )")
	public void logging( JoinPoint thisJoinPoint,
			Object key,
			Object value) {
		PickerClient client = new PickerClient();
		client.setHost(this.HOST);
		client.setPORT(this.PORT);
		client.send((String)key);
	}
}

@@ -646,16 +646,120 @@
 Client;%0A
+import java.util.logging.Logger;%0Aimport java.util.logging.Level;%0Aimport java.util.logging.FileHandler;%0A%0A
 %0A@Aspect
@@ -859,16 +859,502 @@
  = 9999;
+%0A%09%0A%09/** log file */%0A%09public static final String LOGFILE = %22Hanoi-MethodTraceMonitor.log%22;%0A%09%0A%09/** logger */%0A%09private Logger logger = null;%0A%09%0A%09/** file handler */%0A%09private FileHandler fh = null;%0A%09%0A%09/**%0A%09 * initialize%0A%09 */%0A%09public MapperMonitor() %7B%0A%09%09logger = Logger.getLogger(this.getClass().getName());%0A%09%09try %7B%0A%09%09%09fh = new FileHandler(this.LOGFILE, true);%0A%09%09%09logger.addHandler(fh);%0A%09%09%7D catch (IOException e) %7B%0A%09%09%09e.printStackTrace();%0A%09%09%7D%0A%09%09%0A%09%09logger.log(Level.INFO, %22Start trace...%22);%0A%09%7D
 %0A%0A%09@Poin
@@ -1806,24 +1806,24 @@
 this.PORT);%0A
-
 %09%09client.sen
@@ -1838,13 +1838,66 @@
 g)key);%0A
+%09%09System.err.println(%22** %5BPOINTCUT%5D%22 + (String)key);%0A
 %09%7D%0A%7D%0A

add logger.