root/BH13SPARQLBuilder/src/hozo/sparql/CrossSparqlAccessor.java @ 16

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;
                }
        }

}