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 Mapper {
		public List map(Object arg);
	}

	public interface Handler extends Mapper {
		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,Mapper mapper) {
		MapReduce mr = instances.get(key);
		if( mr == null ) {
			mr = new MapReduce();
			instances.put(key,mr);
		}
		mr.mappers.add(mapper);
		return mr;
	}

	private Set<Mapper> mappers = Collections.newSetFromMap(new WeakHashMap<Mapper, Boolean>());

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