package goodjava.util;

import java.util.Collections;
import java.util.List;
import java.util.ArrayList;
import java.util.Map;
import java.util.HashMap;
import java.util.WeakHashMap;
import java.util.Set;


public class MapReduce {

	public interface Handler {
		public List map(Object arg);
		public List reduce(List<List> lists);
	}

	private static Map<String,MapReduce> instances = new HashMap<String,MapReduce>();

	// an active handler will be arbitrarily chosen for reduce
	public static synchronized MapReduce register(String key,Handler handler) {
		MapReduce mr = instances.get(key);
		if( mr == null ) {
			mr = new MapReduce();
			instances.put(key,mr);
		}
		mr.handlers.add(handler);
		return mr;
	}

	private Set<Handler> handlers = Collections.newSetFromMap(new WeakHashMap<Handler, Boolean>());

	public List run(Object arg) {
		List<List> lists = new ArrayList<List>();
		Handler lastHandler = null;
		Handler[] snapshot = handlers.toArray(new Handler[0]);
		for( Handler h : snapshot ) {
			lists.add( h.map(arg) );
			lastHandler = h;
		}
		return lastHandler==null ? Collections.emptyList() : lastHandler.reduce(lists);
	}
}