package jp.ac.osaka_u.sanken.sparql;

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

import com.hp.hpl.jena.rdf.model.RDFNode;

public class CrossSparqlAccessor implements ThreadedSparqlAccessor {

	private List<EndpointSettings> settings;

	private Map<EndpointSettings, SparqlAccessor> accessorHash;

	private Map<Integer, CrossOffset> crossOffsetMap;

	public CrossSparqlAccessor(List<EndpointSettings> settings, SparqlQueryListener listener){
		accessorHash = new LinkedHashMap<EndpointSettings, SparqlAccessor>();
		for (EndpointSettings setting : settings){
			accessorHash.put(setting, SparqlAccessorFactory.createSparqlAccessor(setting, listener));
		}

		this.settings = settings;

		init();
	}

	public CrossSparqlAccessor(List<EndpointSettings> settings){
		accessorHash = new LinkedHashMap<EndpointSettings, SparqlAccessor>();
		for (EndpointSettings setting : settings){
			accessorHash.put(setting, SparqlAccessorFactory.createSparqlAccessor(setting));
		}
		this.settings = settings;

		init();
	}

	private void init(){
		crossOffsetMap = new LinkedHashMap<Integer, CrossSparqlAccessor.CrossOffset>();
	}

	@Override
	public List<Map<String, RDFNode>> executeQuery(String queryString)
			throws Exception {

		// TODO 未対応で良い？
		throw new UnsupportedOperationException("Can't Execute Query");
	}

	@Override
	public SparqlResultSet findSubject(String word, boolean fullMatch,
			Integer limit, Integer offset, int type, String[] propList) throws Exception {
		SparqlResultSet ret = new SparqlResultSet(new ArrayList<Map<String, RDFNode>>());
		Integer limit_ = limit;
		Integer offset_ = offset;

		CrossOffset cOffset = null;
		if (offset != null){
			crossOffsetMap.get(new Integer(offset));
			if (cOffset == null){
				cOffset = new CrossOffset(0, offset);
			}
		}

		for (int i=0; i<settings.size(); i++){
			if (cOffset != null){
				i = cOffset.endpointIndex;
			}
			EndpointSettings setting = settings.get(i);
			SparqlAccessor sa = accessorHash.get(setting);
			SparqlResultSet set = sa.findSubject(word, fullMatch, limit_, (cOffset == null ? null : cOffset.offset), type, propList);
			ret.addResult(setting.getEndpoint(), set.getDefaultResult());
			if (cOffset != null){
				if (ret.getDefaultResult().size() < limit){
					limit_ = limit - set.getDefaultResult().size();
					offset_ = 0;
				} else {
					cOffset = new CrossOffset(i, limit_ + offset_);
					crossOffsetMap.put(new Integer(limit + offset), cOffset);
					break;
				}
				if (i == settings.size() - 1 && set.isHasNext()){
					ret.setHasNext(true);
				}
			}
		}

		return ret;
	}

	@Override
	public List<Map<String, RDFNode>> findTripleFromSubject(String subject)
			throws Exception {
		List<Map<String, RDFNode>> ret = new ArrayList<Map<String,RDFNode>>();

		for (EndpointSettings setting : settings){
			boolean hit = false;
			for (String ns : setting.getNamespaceList()){
				if (subject.startsWith(ns)){
					hit = true;
				}
			}
			if (hit){
				SparqlAccessor sa = accessorHash.get(setting);
				List<Map<String, RDFNode>> set = sa.findTripleFromSubject(subject);
				ret.addAll(set);
			}

		}

		return ret;
	}

	private boolean isThereTargetEndpoint(String subject){
		for (EndpointSettings setting : settings){
			for (String ns : setting.getNamespaceList()){
				if (subject.startsWith(ns)){
					return true;
				}
			}
		}
		return false;
	}

	@Override
	public boolean executeQuery(String queryString,
			SparqlResultListener resultListener) {
		// TODO 未対応で良い？
		throw new UnsupportedOperationException("Can't Execute Query");

	}

	@Override
	public boolean findSubject(String word, boolean fullMatch, Integer limit,
			Integer offset, int type, String[] propList, SparqlResultListener resultListener) {
		Thread thread = new QueryThread(word, new Object[]{new Boolean(fullMatch), limit, offset, new Integer(type), propList}, resultListener){
			public void run(){
				try {
					Boolean fullMatch = (Boolean)((Object[])getOption())[0];
					Integer limit = (Integer)((Object[])getOption())[1];
					Integer offset = (Integer)((Object[])getOption())[2];
					Integer type = (Integer)((Object[])getOption())[3];
					String[] propList = (String[])((Object[])getOption())[4];
					getSparqlResultListener().resultReceived(findSubject(getQueryString(), fullMatch, limit, offset, type, propList));
				} catch(Exception e){
					throw new RuntimeException(e);
				}
			}
		};
		thread.setUncaughtExceptionHandler(resultListener);
		thread.start();

		return true;
	}

	@Override
	public boolean findTripleFromSubject(String subject,
			SparqlResultListener listener) {
		if (!isThereTargetEndpoint(subject)){
			// 対象endpointなし
			return false;
		}

		Thread thread = new QueryThread(subject, listener){
			public void run(){
				try {
					getSparqlResultListener().resultReceived(new SparqlResultSet(findTripleFromSubject(getQueryString())));
				} catch(Exception e){
					throw new RuntimeException(e);
				}
			}
		};
		thread.setUncaughtExceptionHandler(listener);
		thread.start();
		return true;

	}


	@Override
	public List<Map<String, RDFNode>> findPropertyList() throws Exception {
		List<Map<String, RDFNode>> ret = new ArrayList<Map<String,RDFNode>>();
		Map<String, RDFNode> result = new LinkedHashMap<String, RDFNode>();
		ret.add(result);
		for (EndpointSettings setting : settings){
			SparqlAccessor sa = accessorHash.get(setting);
			List<Map<String, RDFNode>> set = sa.findPropertyList();
			for (Map<String, RDFNode> r : set){
				for (String key : r.keySet()){
					result.put(key, r.get(key));
				}
			}
		}

		return ret;
	}

	@Override
	public boolean findPropertyList(SparqlResultListener listener) {
		Thread thread = new QueryThread(null, listener){
			public void run(){
				try {
					getSparqlResultListener().resultReceived(new SparqlResultSet(findPropertyList()));
				} catch(Exception e){
					throw new RuntimeException(e);
				}
			}
		};
		thread.setUncaughtExceptionHandler(listener);
		thread.start();
		return true;
	}

	private class CrossOffset {

		public int endpointIndex;
		public int offset;

		public CrossOffset(int endpointIndex, int offset){
			this.endpointIndex = endpointIndex;
			this.offset = offset;
		}
	}

}