public class PushServlet extends HttpServlet {

    private static final long serialVersionUID = 7504310048722996407L;

    /**
     * transmit the timeline to several hosts
     * @param timeline
     * @param hoststubs a list of host stubs, i.e. ["http://remoteserver.eu"]
     * @return true if the data was transmitted to at least one target peer
     */
    public static boolean push(String[] hoststubs, TwitterTimeline timeline) {
        // transmit the timeline
        String data = timeline.toJSON(false, "search_metadata", "statuses").toString();
        assert data != null;
        boolean transmittedToAtLeastOnePeer = false;
        pushattempts: for (String hoststub: hoststubs) {
            ClientConnection connection = null;
            try {
                if (hoststub.endsWith("/")) hoststub = hoststub.substring(0, hoststub.length() - 1);
                Map<String, byte[]> post = new HashMap<String, byte[]>();
                post.put("data", UTF8.getBytes(data)); // optionally implement a gzipped form here
                JsonSignature.addSignature(post,DAO.private_settings.getPrivateKey());
                connection = new ClientConnection(hoststub + "/api/push.json", post);
                DAO.log("SUCCESS push " + timeline.size() + " messages to backend " + hoststub);
                transmittedToAtLeastOnePeer = true;
                break pushattempts;
            } catch (IOException | JSONException | SignatureException | InvalidKeyException e) {
                DAO.log("FAILED to push " + timeline.size() + " messages to backend " + hoststub + " - " + e.getMessage() + (e.getCause() == null ? "" : " - " + e.getCause().getMessage()));
                DAO.severe(e);
                try {Thread.sleep(1000);} catch (InterruptedException e1) {} // sleep to prevent that same hosts are requested too fast causing a "frequency too high" exception
            } finally {
                if (connection != null) connection.close();
            }
        }
        return transmittedToAtLeastOnePeer;
    }

    /*
     * There are the following sources for data, pushed or retrieved:
     * - twitter (scraped self)
     * - remote (pushed by a remote peer, could be faked by a user)
     * - user (pushed by a user)
     */

    @Override
    protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
        doPost(request, response);
        //response.sendError(400, "your must call this with HTTP POST");
        //return;
    }

    /*
     * call this i.e. with
     * curl -i -F callback=p -F data=@tweets.json http://localhost:9000/api/push.json
     */

    @Override
    protected void doPost(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {

        long timeStart = System.currentTimeMillis();

        Query post = RemoteAccess.evaluate(request);
        String remoteHash = Integer.toHexString(Math.abs(post.getClientHost().hashCode()));
        boolean remoteHashFromPeerId = false;

        // manage DoS
        if (post.isDoS_blackout()) {
            response.sendError(503, "your request frequency is too high");
            return;
        }

        if (!IncomingMessageBuffer.addSchedulerAvailable()) {
            response.sendError(503, "out of server capacity");
            return;
        }

        Map<String, byte[]> m = RemoteAccess.getPostMap(request);
        byte[] data = m.get("data");
        String callback = UTF8.String(m.get("callback"));
        boolean jsonp = callback != null && callback.length() > 0;
        if (data == null || data.length == 0) {response.sendError(400, "your request does not contain a data object. The data object should contain data to be pushed. The format of the data object is JSON; it is exactly the same as the JSON search result"); return;}

        // parse the json data
        int recordCount = 0;//, newCount = 0, knownCount = 0;
        String query = null;
        long timeParsing = 0, timeTimelineStorage = 0, timeQueryStorage = 0;
        String jsonString = new String(data, StandardCharsets.UTF_8);
        JSONObject map = new JSONObject(jsonString);

        timeParsing = System.currentTimeMillis();

        // read metadata
        JSONObject metadata = map.has("search_metadata") ? (JSONObject) map.get("search_metadata") : null;
        // read peer id if they submitted one
        if (metadata != null) {
            String peerid = metadata.has("peerid") ? (String) metadata.get("peerid") : null;
            if (peerid != null && peerid.length() > 3 && peerid.charAt(2) == '_') {
                remoteHash = peerid;
                remoteHashFromPeerId = true;
            }
        }

        // read statuses
        JSONArray statuses = map.has("statuses") ? map.getJSONArray("statuses") : null;
        if (statuses != null) {
            TwitterTimeline tl = new TwitterTimeline(Order.CREATED_AT);
            for (Object tweet_obj: statuses) {
                JSONObject tweet = (JSONObject) tweet_obj;
                recordCount++;
                JSONObject user = (JSONObject) tweet.remove("user");
                if (user == null) continue;
                tweet.put("provider_type", ProviderType.REMOTE.name());
                tweet.put("provider_hash", remoteHash);
                if (!user.has("screen_name") && tweet.has("screen_name")) user.put("screen_name", tweet.getString("screen_name"));
                UserEntry u = new UserEntry(user);
                TwitterTweet t = new TwitterTweet(tweet);
                tl.add(t, u);
                //boolean newtweet = DAO.writeMessage(t, u, true, true, true);
                //if (newtweet) newCount++; else knownCount++;
            }
            IncomingMessageBuffer.addScheduler(tl, true, false);
            //try {DAO.users.bulkCacheFlush();} catch (IOException e) {}
            //try {DAO.messages.bulkCacheFlush();} catch (IOException e) {}

            timeTimelineStorage = System.currentTimeMillis();

            // update query database if query was given in the result list
            if (metadata != null) {
                final String querys = metadata.has("query") ? (String) metadata.get("query") : null;
                if (querys != null) {
                    // update query database concurrently
                    new Thread() {
                        public void run() {
                            QueryEntry qe = null;
                            try {
                                qe = DAO.queries.read(querys);
                            } catch (IOException e1) {
                                e1.printStackTrace();
                            }
                            if (qe != null) {
                                // existing queries are updated
                                qe.update(tl.period(), false);
                                try {
                                    DAO.queries.writeEntryAsync(new IndexEntry<QueryEntry>(querys, qe.getSourceType(), qe));
                                } catch (IOException e) {
                                    DAO.severe(e);
                                }
                            }
                        }
                    }.start();
                }
            }

            timeQueryStorage = System.currentTimeMillis();
        }

        // in case that a peer submitted their peer id, we return also some statistics for that peer
        long messages_from_client = remoteHashFromPeerId ? DAO.countLocalMessages(remoteHash) : -1;

        post.setResponse(response, "application/javascript");

        // generate json
        JSONObject json = new JSONObject(true);
        json.put("status", "ok");
        json.put("records", recordCount);
        json.put("mps", DAO.countLocalMessages(3600000L, true) / 3600L); // to enable client throttling: mps measured by hour
        if (remoteHashFromPeerId) json.put("contribution_message_count", messages_from_client);
        //json.field("new", newCount);
        //json.field("known", knownCount);
        json.put("message", "pushed");

        // write json
        response.setCharacterEncoding("UTF-8");
        PrintWriter sos = response.getWriter();
        if (jsonp) sos.print(callback + "(");
        sos.print(json.toString(2));
        if (jsonp) sos.println(");");
        sos.println();

        long timeResponse = System.currentTimeMillis();

        DAO.log(
                request.getServletPath() + " -> records = " + recordCount +
                //", new = " + newCount +
                //", known = " + knownCount +
                ", from host hash " + remoteHash +
                (query == null ? "" : " for query=" + query) +
                ", timeParsing = " + (timeParsing - timeStart) +
                ", timeTimelineStorage = " + (timeTimelineStorage - timeParsing) +
                ", timeQueryStorage = " + (timeQueryStorage - timeTimelineStorage) +
                ", timeResponse = " + (timeResponse - timeQueryStorage) +
                ", total time = " + (timeResponse - timeStart)
                );

        response.addHeader("Access-Control-Allow-Origin", "*");
        post.finalize();
    }
} public class InstagramProfileScraper extends BaseScraper {

    private static final long serialVersionUID = -3360416757176406604L;
    
    /**
     * instaJsonData is a Pattern object which will be used to match the required pattern
     * with the extracted elements from html page
     */
    private Pattern instaJsonData = Pattern.compile("(\\{\"config\").*(\\});");

    /**
     * Constructor to set baseUrl and scraperName in the super(Base-Interface) for
     * current Search-Scrapper
     */
    public InstagramProfileScraper() {
        super();
        this.baseUrl = "https://www.instagram.com/";
        this.scraperName = "instagram";
    }

    /**
     * Constructor to set the given query and map _extra
     * @param _extra
     * A map with key and value as string
     * @param _query
     * _query in form for string value
     */
    public InstagramProfileScraper(String _query, Map<String, String> _extra) {
        this();
        this.setExtra(_extra);
        this.query = _query;
    }

    /**
     * Constructor to map the given _extra param with key and value as String
     */
    public InstagramProfileScraper(Map<String, String> _extra) {
        this();
        this.setExtra(_extra);
    }

    /**
     * Constructor to set String query
     * @param query as string
     */
    public InstagramProfileScraper(String _query) {
        this();
        this.query = _query;
    }

    /**
     * Method to get api path of the Instagram Profile Scraper
     * @return api endpoint of Instagram Profile Scraper in form of String
     */
	@Override
	public String getAPIPath() {
		return "/api/instagramprofilescraper.json";
	}

	@Override
	public BaseUserRole getMinimalBaseUserRole() {
		return BaseUserRole.ANONYMOUS;
	}

    /**
     * @return null when asked for the default permissions
     * of the given base user role in form of JSONObject
     */
	@Override
	public JSONObject getDefaultPermissions(BaseUserRole baseUserRole) {
		return null;
	}

    /**
     * Method to generate url for search using URIBuilder
     * @return URL in string format
     */
    protected String prepareSearchUrl(String type) {
        URIBuilder url = null;
        try {
            url = new URIBuilder(this.baseUrl + this.query);
        } catch (URISyntaxException e) {
            DAO.log("Invalid Url: baseUrl = " + this.baseUrl + ", mid-URL = " + midUrl + "query = " + this.query + "type = " + type);
            return "";
        }

        return url.toString();
    }

    /**
     * Method to set the "profile" key with value of query if the "profile" key is empty/null
     * and set the "query" key with value of query
     */
    protected void setParam() {
        if("".equals(this.getExtraValue("profile"))) {
            this.setExtraValue("profile", this.query);
        }
        this.setExtraValue("query", this.query);
    }

    /**
     * Method to scrape the given url with key as "profile_posts"
     * @return instaObj as a Post object 
     */
    protected Post scrape(BufferedReader br, String type, String url) {
   	    Post instaObj = new Post(true);
        this.putData(instaObj, "profile_posts", this.scrapeInstagram(br, url));
        return instaObj;
    }

    /**
     * Method to match the given pattern with extracted elements of html page
     * and parse the result for the posts on the given instagram page
     * @return instaProfile as a JSONArray object containing all posts and details of viewer
     */
	public JSONArray scrapeInstagram(BufferedReader br, String url) {
		Document htmlPage = null;
        Post instaObj = null;
        JSONArray instaProfile = new JSONArray();
        try {
            htmlPage = Jsoup.parse(this.bufferedReaderToString(br));
        } catch (IOException e) {
            DAO.trace(e);
		}

		String script = htmlPage.getElementsByTag("script").html();
        Matcher m = instaJsonData.matcher(script);
        m.find();
        int start = m.start(1);
        int end = m.start(2) + 1;
        script = script.substring(start, end);

        //TODO: pre-process the posts captured. At present, complete array of posts are output.
        //Only useful data shall be outputted.
		instaObj = new Post(script, this.query);
        instaProfile.put(instaObj);
       return instaProfile;
	}

} public class QuoraProfileScraper extends BaseScraper {

    private static final long serialVersionUID = -3398701925784347312L;
    private List<String> typeList = null;

    public QuoraProfileScraper() {
        super();
        this.baseUrl = "https://www.quora.com/";
        this.scraperName = "quora";
        this.sourceType = SourceType.QUORA;
    }

    public QuoraProfileScraper(String _query) {
        this();
        this.query = _query;
        this.setExtraValue("query", this.query);
    }

    public QuoraProfileScraper(String _query, Map<String, String> _extra) {
        this();
        this.setExtra(_extra);
        this.query = _query;
        this.setExtraValue("query", this.query);
    }

    public QuoraProfileScraper(Map<String, String> _extra) {
        this();
        this.setExtra(_extra);
    }

    protected void setParam() {
        if(!"".equals(this.getExtraValue("type"))) {
            this.typeList = Arrays.asList(this.getExtraValue("type").trim().split("\\s*,\\s*"));
        } else {
            this.typeList = new ArrayList<String>();
            this.typeList.add("all");
            this.setExtraValue("type", String.join(",", this.typeList));
        }
        this.query = this.getExtraValue("query");
    }

    @Override
    protected void setCacheMap() {
        this.cacheMap = new HashMap<String, Map<String, String>>();

        Map<String, String> getMap = new HashMap<String, String>();
        if(this.typeList.contains("user") && this.typeList.size() == 1) {
            getMap.put("post_type", "user");
            getMap.put("user_name", this.query);
        } else if(this.typeList.contains("question") && this.typeList.size() == 1) {
            getMap.put("post_type", "question");
            getMap.put("post_ques", this.query);
        }
        getMap.put("post_scraper", this.scraperName);

        Map<String, String> alsoGetMap = new HashMap<String, String>();
        if(this.typeList.contains("question") || this.typeList.contains("all")) {
            alsoGetMap.put("post_ques", this.query);
        }
        if(this.typeList.contains("user") || this.typeList.contains("all")) {
            alsoGetMap.put("user_name", this.query);
        }

        this.cacheMap.put("get", getMap);
        this.cacheMap.put("also_get", alsoGetMap);
    }

    @Override
    public String getAPIPath() {
        return "/api/quoraprofilescraper.json";
    }

    @Override
    public BaseUserRole getMinimalBaseUserRole() {
        return BaseUserRole.ANONYMOUS;
    }

    @Override
    public JSONObject getDefaultPermissions(BaseUserRole baseUserRole) {
        return null;
    }

    protected Map<?, ?> getExtra(String _extra) {
        return new HashMap<String, String>();
    }

    protected String prepareSearchUrl(String type) {
        URIBuilder url = null;
        String midUrl = "search/";
        try {
            switch(type) {
                case "user":
                    midUrl = "profile/";
                    url = new URIBuilder(this.baseUrl + midUrl + this.query);
                    break;
                case "question":
                    url = new URIBuilder(this.baseUrl + midUrl);
                    url.addParameter("q", this.query);
                    url.addParameter("type", "question");
                    break;
                case "answer":
                    url = new URIBuilder(this.baseUrl + midUrl);
                    url.addParameter("q", this.query);
                    url.addParameter("type", "answer");
                    break;
                default:
                    url = new URIBuilder("");
                    break;
            }
        } catch (URISyntaxException e) {
            DAO.log("Invalid Url: baseUrl = " + this.baseUrl + ", mid-URL = " + midUrl + "query = " + this.query + "type = " + type);
        }

        return url.toString();
    }

    @Override
    public Post getResults() {
        String url;
        Thread[] dataThreads = new Thread[2];
        PostTimeline postList = new PostTimeline(this.order);

        if(this.typeList.contains("user") || this.typeList.contains("all")) {
            url = prepareSearchUrl("user");
            dataThreads[0] = new ConcurrentScrape(url, "users", postList);
            dataThreads[0].start();
        } else {
            dataThreads[0] = new Thread();
        }
        if(this.typeList.contains("question") || this.typeList.contains("all")) {
            url = prepareSearchUrl("question");
            dataThreads[1] = new ConcurrentScrape(url, "question", postList);
            dataThreads[1].start();
        } else {
            dataThreads[1] = new Thread();
        }
        //TODO: add more types

        int i = 0;
        try {
            for (i = 0; i < dataThreads.length; i++) {
                dataThreads[i].join();
            }
        } catch(InterruptedException e) {
            exceptionOutputGetData(i);
        }

        // Add scraper name
        Post postArray = new Post();
        postArray.put(this.scraperName, postList.toArray());

        return postArray;
    }

    protected class ConcurrentScrape extends Thread {

        private String url = "";
        private String type = "all";
        private PostTimeline postList = null;

        public ConcurrentScrape(String url, String type, PostTimeline postList) {
            this.url = url;
            this.type = type;
            this.postList = postList;
        }

        public void run() {
            try {
                this.postList.addPost(QuoraProfileScraper.this.getDataFromConnection(this.url, this.type));
            } catch (IOException e) {
                DAO.severe("check internet connection for url: " + this.url + " type: " + this.type);
            }
        }
    }

    private void exceptionOutputGetData(int i) {
        String stuck_at = "";
        switch(i) {
            case 0:
                stuck_at = "users";
                break;
            case 1:
                stuck_at = "question";
                break;
            case 2:
                stuck_at = "answer";
                break;
            default:
                stuck_at = "unknown, check the code";
        }
        DAO.severe("Couldn't complete all threads, stuck at scraper: " + this.scraperName + " dataThread: " + stuck_at);
    }

    public PostTimeline scrapeQues(BufferedReader br, String url) {
        Pattern resultBlock = Pattern.compile("<div[^>]*[^>\\s]*[^>]*class=['\"][^>'\"]*(results_list)");
        Pattern quesLink = Pattern.compile("<a[^>]*class=['\"][^>'\"]*question_link[^>'\"]*[\"'][^>]*href=['\"]([^>'\"]*)");
        Pattern quesStart = Pattern.compile("<span[^>]*class=[\'\"][^>\'\"]*question_text[^>\'\"]*[\'\"][^>]*>");
        String ignoreTag = "(<[^>]*>)";
        Matcher matcher;
        String fromTerm = "";
        String uptoTerm = "</a>";
        Post qPost = null;

        PostTimeline quesList = new PostTimeline(this.order);

        String input = "";
        String _qPostId = "";
        String _qPostUrl = "";

        int last = 0;
        try {
            // Get to Result List block
            while(true) {
                input = br.readLine();
                if (input == null) break;

                matcher = resultBlock.matcher(input);
                if (matcher.find()) {
                    input = input.substring(matcher.end());
                    break;
                }
            }
            // Scraping starts
            for(int i = 0;input !=null ; i++) {

                // Get to first result
                while(input != null) {
                    matcher = quesLink.matcher(input);
                    if(matcher.find()) {
                        _qPostId = matcher.group(1).substring(1);
                        _qPostUrl = this.baseUrl + _qPostId;
                        input = input.substring(matcher.end());
                        break;
                    } else {
                        input = br.readLine();
                    }

                }

                qPost = new QuoraPost(_qPostId, i);
                qPost.put("search_url", url);
                qPost.put("post_url", _qPostUrl);
                qPost.put("post_type", "question");
                qPost.put("post_scraper", "quora");
                // Get questions
                while(input != null) {
                    matcher = quesStart.matcher(input);
                    if(matcher.find()) {
                        input = input.substring(matcher.end());
                        break;
                    } else {
                        input = br.readLine();
                    }
                }

                while(input != null) {
                    fromTerm = fromTerm + input;
                    last = input.indexOf(uptoTerm);
                    if(input.indexOf(uptoTerm) > 0) {
                        fromTerm = fromTerm.substring(0, input.indexOf(uptoTerm));
                        break;
                    } else {
                        input = br.readLine();
                    }

                }

                fromTerm = fromTerm.replaceAll(ignoreTag, "");
                input = input.substring(last);
                qPost.put("post_ques", fromTerm);
                fromTerm = "";
                quesList.addPost(qPost);
            }


        } catch(IOException e) {
            qPost = new QuoraPost(this.query, -1);
            qPost.put("error", "Connection error while fetching");
            qPost.put("search_url", url);
            quesList.addPost(qPost);
        } catch(NullPointerException e) { }

        return quesList;
    }

    public PostTimeline scrapeProfile(BufferedReader br, String url) {
        String html;
        Post quoraProfile = new QuoraPost(this.query, 0);
        PostTimeline usersList = new PostTimeline(this.order);
        try {
            html = bufferedReaderToString(br);
        } catch(IOException e) {
            DAO.trace(e);
            html = "";
            //TODO: output error if no output in json
        }

        Document userHTML = Jsoup.parse(html);

        quoraProfile.put("search_url", url);
        quoraProfile.put("post_type", "user");
        quoraProfile.put("post_scraper", this.scraperName);
        String bio = userHTML.getElementsByAttributeValueContaining("class", "ProfileDescription").text();
        quoraProfile.put("bio", bio);

        String profileImage = userHTML.getElementsByAttributeValueContaining("class", "profile_photo_img").attr("src");
        quoraProfile.put("profileImage", profileImage);

        String userName = userHTML.getElementsByAttributeValueContaining("class", "profile_photo_img").attr("alt");
        quoraProfile.put("user_name", userName);

        String rssFeedLink = url + "/rss";
        quoraProfile.put("rss_feed_link", rssFeedLink);

        Elements userBasicInfo = userHTML.getElementsByAttributeValueContaining("class", "UserCredential IdentityCredential");
        for (Element info: userBasicInfo) {
            String infoText = info.text();
            if (infoText.startsWith("Studi")) {
                quoraProfile.put(infoText.split(" ")[0].toLowerCase().trim() + "_at", infoText);
            } else if (infoText.startsWith("Lives")) {
                quoraProfile.put("lives_in", infoText);
            } else {
                quoraProfile.put("works_at", infoText);
            }
        }

        Elements knowsAbout = userHTML.getElementsByAttributeValueContaining("class",  "TopicNameSpan TopicName");
        JSONArray topics = new JSONArray();
        for (Element topic: knowsAbout) {
            topics.put(topic.text());
        }
        quoraProfile.put("knows_about", topics);

        JSONObject feeds = new JSONObject();
        Elements counts = userHTML.getElementsByAttributeValueContaining("class", "list_count");
        for (Element count: counts) {
            String topic = count.parent().text();
            topic = topic.substring(0, topic.indexOf(count.text())).trim();
            feeds.put(topic.toLowerCase() + "_url", baseUrl + count.parent().attr("href").substring(1));
            feeds.put(topic.toLowerCase(), count.text());
        }
        quoraProfile.put("feeds", feeds);

        usersList.addPost(quoraProfile);
        return usersList;
    }

    @Override
    protected Post scrape(BufferedReader br, String type, String url) {
        Post typeArray = new Post(true);

        switch(type) {
            case "users":
                this.putData(typeArray, "users", this.scrapeProfile(br, url));
                break;
            case "question":
                this.putData(typeArray, "question", this.scrapeQues(br, url));
                break;
            default:
                break;
        }
        return typeArray;
    }

    public static class QuoraPost extends Post {

        //quora post-id, for profile it will be username
        private String quoraId;
        private int quoraPostNo;

        public QuoraPost(String _quoraId, int _quoraPostNo) {
            //not UTC, may be error prone
            super();
            this.quoraPostNo = _quoraPostNo;
            this.setPostId(_quoraId);
        }

        public void getQuoraId(String _quoraId) {
            this.quoraId = _quoraId;
        }

        public void getQuoraPostNo(int _quoraPostNo) {
            this.quoraPostNo = _quoraPostNo;
        }

        public String getPostId() {
            return String.valueOf(this.postId);
        }
        //clean data
    }
} public class RSSReaderService extends AbstractAPIHandler implements APIHandler {

	private static final long serialVersionUID = 1463185662941444503L;

    @Override
    public BaseUserRole getMinimalBaseUserRole() { return BaseUserRole.ANONYMOUS; }

    @Override
    public JSONObject getDefaultPermissions(BaseUserRole baseUserRole) {
        return null;
    }

    public String getAPIPath() {
        return "/api/rssreader.json";
    }
    
    @Override
    public JSONObject serviceImpl(Query post, HttpServletResponse response, Authorization rights, final JSONObjectWithDefault permissions) throws APIException {
		String url = post.get("url", "");
		return readRSS(url);
    }
		
    public static SusiThought readRSS(String url) {
    	
		SyndFeedInput input = new SyndFeedInput();
		SyndFeed feed = null;
        
		try {
            ClientConnection connection = new ClientConnection(url, "");
		    XmlReader xmlreader = new XmlReader(connection.getInputStream());
			feed = input.build(xmlreader);
		} catch (Exception e) {
			e.printStackTrace();
			return new SusiThought(); // fail
		}

		@SuppressWarnings("unused")
		int totalEntries = 0;
		int i = 0;

		JSONArray jsonArray = new JSONArray();

		// Reading RSS Feed from the URL
		for (SyndEntry entry : (List<SyndEntry>) feed.getEntries()) {

			JSONObject jsonObject = new JSONObject();
			jsonObject.put("title", entry.getTitle().toString());
			jsonObject.put("link", entry.getLink().toString());
			jsonObject.put("uri", entry.getUri().toString());
			jsonObject.put("guid", Integer.toString(entry.hashCode()));
			if (entry.getPublishedDate() != null) jsonObject.put("pubDate", entry.getPublishedDate().toString());
			if (entry.getUpdatedDate() != null) jsonObject.put("updateDate", entry.getUpdatedDate().toString());
			if (entry.getDescription() != null) jsonObject.put("description", entry.getDescription().getValue().toString());

			jsonArray.put(i, jsonObject);

			i++;
		}

		totalEntries = i;

		SusiThought json = new SusiThought();
		json.setData(jsonArray);
		return json;
	}
} public class SearchServlet extends HttpServlet {

    private static final long serialVersionUID = 563533152152063908L;

    private final static String SEARCH_LOW_COUNT_NAME = "search.count.low";
    private final static String SEARCH_DEFAULT_COUNT_NAME = "search.count.default";
    private final static String SEARCH_MAX_PUBLIC_COUNT_NAME = "search.count.max.public";
    private final static String SEARCH_MAX_LOCALHOST_COUNT_NAME = "search.count.max.localhost";

    private final static int SEARCH_CACHE_THREASHOLD_TIME = 3000;

    private final static AtomicLong last_cache_search_time = new AtomicLong(10L);

    public final static String backend_hash = Integer.toHexString(Integer.MAX_VALUE);
    public final static String frontpeer_hash = Integer.toHexString(Integer.MAX_VALUE - 1);

    // possible values: cache, twitter, all
    public static TwitterTimeline search(final String[] protocolhostportstubs, final String query, final Set<String> filterList, final TwitterTimeline.Order order, final String source, final int count, final int timezoneOffset, final String provider_hash, final long timeout) throws IOException {
        String threadname = Thread.currentThread().getName();

        TwitterTimeline tl = new TwitterTimeline(order);
        if ("".equals(query)) return tl;
        String urlstring = "";
        String filterString = "";

        try {
            byte[] jsonb = null;
            IOException ee = null;
            backendloop: for (String protocolhostportstub: protocolhostportstubs) {
                Thread.currentThread().setName("searchOnOtherPeers on:" + protocolhostportstub);
                ee = null;
                try {
                    urlstring = protocolhostportstub + "/api/search.json?q=" + URLEncoder.encode(query.replace(' ', '+'), "UTF-8") + "&timezoneOffset=" + timezoneOffset + "&maximumRecords=" + count + "&source=" + (source == null ? "all" : source) + "&minified=true&shortlink=false&timeout=" + timeout;

                    if (!"".equals(filterString = String.join(", ", filterList))) {
                        urlstring = urlstring + "&filter=" + filterString;
                    }

                    jsonb = ClientConnection.download(urlstring);
                    break backendloop;
                } catch (IOException e) {
                    ee = e;
                }
            }
            if (jsonb == null || jsonb.length == 0) throw ee == null ? new IOException("empty content from " + protocolhostportstubs) : ee;
            String jsons = UTF8.String(jsonb);
            JSONObject json = new JSONObject(jsons);
            if (json == null || json.length() == 0) return tl;
            JSONArray statuses = json.getJSONArray("statuses");
            if (statuses != null) {
                for (int i = 0; i < statuses.length(); i++) {
                    JSONObject tweet = statuses.getJSONObject(i);
                    JSONObject user = tweet.getJSONObject("user");
                    if (user == null) continue;
                    tweet.remove("user");
                    UserEntry u = new UserEntry(user);
                    TwitterTweet t = new TwitterTweet(tweet);
                    tl.add(t, u);
                }
            }
            JSONObject metadata = json.getJSONObject("search_metadata");
            if (metadata != null) {
                Integer hits = metadata.has("hits") ? (Integer) metadata.get("hits") : null;
                if (hits != null) tl.setHits(hits.intValue());
                String scraperInfo = metadata.has("scraperInfo") ? (String) metadata.get("scraperInfo") : null;
                if (scraperInfo != null) tl.setScraperInfo(scraperInfo);
            }
        } catch (Throwable e) {
            //DAO.severe(e);
            throw new IOException(e.getMessage());
        }
        //System.out.println(parser.text());
        Thread.currentThread().setName(threadname);
        return tl;
    }

    public static TwitterTimeline search(
            final String[] protocolhostportstubs, final String query, final TwitterTimeline.Order order,
            final String source, final int count, final int timezoneOffset,
            final String provider_hash, final long timeout) throws IOException {
        return search(protocolhostportstubs, query, new HashSet<>(), order, source, count, timezoneOffset, provider_hash, timeout);
    }

    @Override
    protected void doPost(final HttpServletRequest request, final HttpServletResponse response) throws ServletException, IOException {
        doGet(request, response);
    }

    @Override
    protected void doGet(final HttpServletRequest request, final HttpServletResponse response) throws ServletException, IOException {
        final long start = System.currentTimeMillis();
        final Query post = RemoteAccess.evaluate(request);
        Map<String, String> inputMap = post.getMap();
        JSONObject metadata = new JSONObject(true);

        final ClientIdentity identity = AbstractAPIHandler.getIdentity(request, response, post);
        JSONArray dataArray = null;
        try {

            // manage DoS
            if (post.isDoS_blackout()) {response.sendError(503, "your (" + post.getClientHost() + ") request frequency is too high"); return;}

            // check call type
            String callType = "json";
            if(request.getServletPath().endsWith(".json")){
                callType = "json";
            } else if(request.getServletPath().endsWith(".rss")) {
                callType = "rss";
            } else if(request.getServletPath().endsWith(".txt")) {
                callType = "txt";
            }

            // evaluate get parameter
            String scraper = post.get("scraper", "");
            String callback = post.get("callback", "");
            boolean jsonp = callback != null && callback.length() > 0;
            boolean minified = post.get("minified", false);
            boolean shortlink_request = post.get("shortlink", true);

            // query parameters
            // naming schema according to SRU 2.0
            // see http://docs.oasis-open.org/search-ws/searchRetrieve/v1.0/os/part3-sru2.0/searchRetrieve-v1.0-os-part3-sru2.0.html
            // for compatibility "query" can be replaced with "q"
            String query = post.get("q", post.get("query", ""));
            query = CharacterCoding.html2unicode(query).replaceAll("\\+", " ");
            // paging options, syntax and meaning as defined by SRU 2.0
            final int startRecord = Math.max(1, post.get("startRecord", 1)); // the first record is number 1, not 0, see http://docs.oasis-open.org/search-ws/searchRetrieve/v1.0/os/part3-sru2.0/searchRetrieve-v1.0-os-part3-sru2.0.html#startAndMax
            final int maximumRecords = Math.min(
                    post.get("count", post.get("maximumRecords", (int) DAO.getConfig(SEARCH_DEFAULT_COUNT_NAME, 10))),
                    (int) (post.isDoS_servicereduction() ?
                            DAO.getConfig(SEARCH_LOW_COUNT_NAME, 10) :
                            post.isLocalhostAccess() ?
                                    DAO.getConfig(SEARCH_MAX_LOCALHOST_COUNT_NAME, 1000) :
                                    DAO.getConfig(SEARCH_MAX_PUBLIC_COUNT_NAME, 100)));

            String filter = post.get("filter", "").replaceAll("\\s","");
            Set<String> filterList = new HashSet<>();
            Arrays.asList(filter.split(",")).forEach(f -> filterList.add(f));

            // create tweet timeline
            final String ordername = post.get("order", Order.CREATED_AT.getMessageFieldName());
            final TwitterTimeline.Order order = TwitterTimeline.parseOrder(ordername);
            TwitterTimeline tl = DAO.timelineCache.getOrCreate(identity, query, startRecord <= 1, order);
            JSONObject hits = new JSONObject(true);
            final JSONObject[] aggregations = new JSONObject[]{null};

            if (!"".equals(scraper)) {
                // start a scraper
                inputMap.put("query", query);
                DAO.log(request.getServletPath() + " scraping with query: "
                        + query + " scraper: " + scraper);
                dataArray = DAO.scrapeLoklak(inputMap, true, true, metadata);
            } else if (tl.size() > 0) {
                // return the timeline from a cached search result
                // in case that the number of available records in the cache is too low, try to get more
                // otherwise there might be nothing to do here
            } else {
                final long timeout = (long) post.get("timeout", DAO.getConfig("search.timeout", 2000));
                String source = post.isDoS_servicereduction() ? "cache" : post.get("source", "all"); // possible values: cache, backend, twitter, all
                int agregation_limit = post.get("limit", 10);
                String[] fields = post.get("fields", new String[0], ",");
                int timezoneOffset = post.get("timezoneOffset", 0);
                if (query.indexOf("id:") >= 0 && ("all".equals(source) || "twitter".equals(source)))
                    source = "cache"; // id's cannot be retrieved from twitter with the scrape-api (yet), only from the cache

                final AtomicInteger cache_hits = new AtomicInteger(0);
                final AtomicInteger count_backend = new AtomicInteger(0);
                final AtomicInteger count_twitter_all = new AtomicInteger(0);
                final AtomicInteger count_twitter_new = new AtomicInteger(0);
                final boolean backend_push = DAO.getConfig("backend.push.enabled", false);
                final QueryEntry.Tokens tokens = new QueryEntry.Tokens(query);

                if ("all".equals(source)) {
                    // start all targets for search concurrently
                    final int timezoneOffsetf = timezoneOffset;
                    final String queryf = query;

                    // start a scraper
                    Thread scraperThread = tokens.raw.length() == 0 ? null : new Thread() {
                        public void run() {
                            final String scraper_query = tokens.translate4scraper();
                            DAO.log(request.getServletPath() + " scraping with query: " + scraper_query);
                            TwitterTimeline twitterTl = DAO.scrapeTwitter(post, filterList, scraper_query, order, timezoneOffsetf, true, timeout, true);
                            count_twitter_new.set(twitterTl.size());
                            tl.putAll(QueryEntry.applyConstraint(twitterTl, tokens, false)); // pre-localized results are not filtered with location constraint any more
                            tl.setScraperInfo(twitterTl.getScraperInfo());
                            post.recordEvent("twitterscraper_time", System.currentTimeMillis() - start);
                        }
                    };
                    if (scraperThread != null) scraperThread.start();

                    // start a local search
                    Thread localThread = queryf == null || queryf.length() == 0 ? null : new Thread() {
                        public void run() {

                            DAO.SearchLocalMessages localSearchResult =
                                    new DAO.SearchLocalMessages(
                                            queryf,
                                            order,
                                            timezoneOffsetf,
                                            last_cache_search_time.get() > SEARCH_CACHE_THREASHOLD_TIME ? Math.min(maximumRecords,
                                            (int) DAO.getConfig(SEARCH_LOW_COUNT_NAME, 10)) : maximumRecords,
                                            agregation_limit,
                                            filterList,
                                            fields
                                    );
                            long time = System.currentTimeMillis() - start;
                            last_cache_search_time.set(time);
                            post.recordEvent("cache_time", time);
                            cache_hits.set(localSearchResult.timeline.getHits());
                            tl.putAll(localSearchResult.timeline);
                            tl.setResultIndex(localSearchResult.timeline.getResultIndex());
                            aggregations[0] = localSearchResult.getAggregations();
                        }
                    };
                    if (localThread != null) localThread.start();

                    // start a backend search, but only if backend_push == true or result from scraper is too bad
                    boolean start_backend_thread = false;
                    if (backend_push) start_backend_thread = true;
                    else {
                        // wait now for termination of scraper thread and local search
                        // to evaluate how many results are available
                        if (scraperThread != null) try {
                            scraperThread.join(Math.max(10000, timeout - System.currentTimeMillis() + start));
                        } catch (InterruptedException e) {
                        }
                        if (localThread != null) try {
                            localThread.join(Math.max(100, timeout - System.currentTimeMillis() + start));
                        } catch (InterruptedException e) {
                        }
                        localThread = null;
                        scraperThread = null;
                        if (tl.size() < maximumRecords) start_backend_thread = true;
                    }
                    Thread backendThread = tokens.original.length() == 0 || !start_backend_thread ? null : new Thread() {
                        public void run() {
                            TwitterTimeline backendTl = DAO.searchBackend(tokens.original, filterList, order, maximumRecords, timezoneOffsetf, "cache", timeout);
                            if (backendTl != null) {
                                tl.putAll(QueryEntry.applyConstraint(backendTl, tokens, true));
                                count_backend.set(tl.size());
                                // TODO: read and aggregate aggregations from backend as well
                            }
                            post.recordEvent("backend_time", System.currentTimeMillis() - start);
                        }
                    };
                    if (backendThread != null) backendThread.start();

                    // wait for termination of all threads
                    if (scraperThread != null) try {
                        scraperThread.join(Math.max(10000, timeout - System.currentTimeMillis() + start));
                    } catch (InterruptedException e) {
                    }

                    // in case that the scraper thread had been started and was successful, we do not wait for the other threads to terminate
                    if (scraperThread == null || tl.getHits() == 0 || query.indexOf(':') >= 0 || query.indexOf('/') >= 0 || fields.length > 0) {
                        if (localThread != null) try {
                            localThread.join(Math.max(100, timeout - System.currentTimeMillis() + start));
                        } catch (InterruptedException e) {
                        }
                    }
                    if (scraperThread == null || tl.getHits() == 0) {
                        if (backendThread != null) try {
                            backendThread.join(Math.max(100, timeout - System.currentTimeMillis() + start));
                        } catch (InterruptedException e) {
                        }
                    }
                } else if ("twitter".equals(source) && tokens.raw.length() > 0) {
                    final String scraper_query = tokens.translate4scraper();
                    DAO.log(request.getServletPath() + " scraping with query: " + scraper_query);
                    TwitterTimeline twitterTl = DAO.scrapeTwitter(post, filterList, scraper_query, order, timezoneOffset, true, timeout, true);

                    count_twitter_new.set(twitterTl.size());
                    tl.putAll(QueryEntry.applyConstraint(twitterTl, tokens, false)); // pre-localized results are not filtered with location constraint any more
                    tl.setScraperInfo(twitterTl.getScraperInfo());
                    post.recordEvent("twitterscraper_time", System.currentTimeMillis() - start);
                    // in this case we use all tweets, not only the latest one because it may happen that there are no new and that is not what the user expects

                } else if ("cache".equals(source)) {
                    DAO.SearchLocalMessages localSearchResult =
                            new DAO.SearchLocalMessages(
                                    query,
                                    order,
                                    timezoneOffset,
                                    last_cache_search_time.get() > SEARCH_CACHE_THREASHOLD_TIME
                                            ? Math.min(maximumRecords,(int) DAO.getConfig(SEARCH_LOW_COUNT_NAME, 10))
                                            : maximumRecords,
                                    agregation_limit,
                                    filterList,
                                    fields
                            );
                    cache_hits.set(localSearchResult.timeline.getHits());
                    tl.putAll(localSearchResult.timeline);
                    tl.setResultIndex(localSearchResult.timeline.getResultIndex());
                    aggregations[0] = localSearchResult.getAggregations();
                    long time = System.currentTimeMillis() - start;
                    last_cache_search_time.set(time);
                    post.recordEvent("cache_time", time);

                } else if ("backend".equals(source) && query.length() > 0) {
                    TwitterTimeline backendTl = DAO.searchBackend(query, filterList, order, maximumRecords, timezoneOffset, "cache", timeout);
                    if (backendTl != null) {
                        tl.putAll(QueryEntry.applyConstraint(backendTl, tokens, true));
                        tl.setScraperInfo(backendTl.getScraperInfo());
                        // TODO: read and aggregate aggregations from backend as well
                        count_backend.set(tl.size());
                    }
                    post.recordEvent("backend_time", System.currentTimeMillis() - start);

                }

                // check the latest user_ids
                DAO.announceNewUserId(tl);

                hits.put("count_twitter_all", count_twitter_all.get());
                hits.put("count_twitter_new", count_twitter_new.get());
                hits.put("count_backend", count_backend.get());
                hits.put("cache_hits", cache_hits.get());
            }

            // create json or xml according to path extension
            int shortlink_iflinkexceedslength = shortlink_request ? (int) DAO.getConfig("shortlink.iflinkexceedslength", 500L) : Integer.MAX_VALUE;
            String shortlink_urlstub = DAO.getConfig("shortlink.urlstub", "http://127.0.0.1:9000");

            dataOutput( response, dataArray, query, startRecord, maximumRecords, minified,
                    callType, jsonp, post, tl, aggregations[0], hits, callback, shortlink_urlstub,
                    shortlink_iflinkexceedslength, filter, metadata);

            post.recordEvent("result_count", tl.size());
            post.recordEvent("postprocessing_time", System.currentTimeMillis() - start);
            post.recordEvent("hits", hits);
            DAO.log(request.getServletPath() + "?" + request.getQueryString()
                    + " -> " + tl.size() + " records returned");
            post.finalize();
        } catch (Throwable e) {
            DAO.severe(e.getMessage(), e);
            //DAO.severe(e);
        }
    }

    private static void dataOutput(HttpServletResponse response, JSONArray dataArray,
            String query, int startRecord, int maximumRecords, boolean minified,
            String callType, boolean jsonp, Query post, TwitterTimeline tl, JSONObject aggregations,
            JSONObject hits, String callback, String shortlink_urlstub,
            int shortlink_iflinkexceedslength, String filter, JSONObject metadata) throws IOException {
        if ("json".equals(callType)) {
            post.setResponse(response, jsonp ? "application/javascript" : "application/json");

            JSONObject m = new JSONObject(true);
            JSONArray statuses = new JSONArray();

            try {
                for (TwitterTweet t : tl.getNextTweets(startRecord - 1, maximumRecords)) {
                    UserEntry u = tl.getUser(t);
                    statuses.put(t.toJSON(u, true, shortlink_iflinkexceedslength, shortlink_urlstub));
                }
            } catch (ConcurrentModificationException e) {
                // late incoming messages from concurrent peer retrieval may cause this
                // we silently do nothing here and return what we listed so far
            }

            if (!minified) setReadMe(m);
            if (dataArray == null) {
                setTwitterMetaData(metadata, startRecord, maximumRecords, tl, query, filter, post, hits);
                m.put("search_metadata", metadata);
                m.put("statuses", statuses);
            } else {
                moreMetaData(metadata, post);
                m.put("results", dataArray);
                m.put("metadata", metadata);
            }
            m.put("aggregations", aggregations);

            // write json
            response.setCharacterEncoding("UTF-8");
            PrintWriter sos = response.getWriter();
            if (jsonp) sos.print(callback + "(");
            sos.print(m.toString(minified ? 0 : 2));
            if (jsonp) sos.println(");");
            sos.println();
        } else if ("rss".equals(callType)) {
            response.setCharacterEncoding("UTF-8");
            post.setResponse(response, "application/rss+xml;charset=utf-8");

            // generate xml
            RSSMessage channel = new RSSMessage();
            channel.setPubDate(new Date());
            channel.setTitle("RSS feed for Twitter search for " + query);
            channel.setDescription("");
            channel.setLink("");
            RSSFeed feed = new RSSFeed(tl.size());
            feed.setChannel(channel);

            try {
                for (TwitterTweet t : tl.getNextTweets(startRecord - 1, maximumRecords)) {
                    UserEntry u = tl.getUser(t);
                    RSSMessage m = new RSSMessage();
                    m.setLink(t.getStatusIdUrl().toExternalForm());
                    m.setAuthor(u.getName() + " @" + u.getScreenName());
                    m.setTitle(u.getName() + " @" + u.getScreenName());
                    m.setDescription(t.moreData.getText(shortlink_iflinkexceedslength, shortlink_urlstub, t.getText(), t.getLinks(), t.getPostId()).text);
                    m.setPubDate(t.getCreatedAt());
                    m.setGuid(t.getPostId());
                    feed.addMessage(m);
                }
            } catch (ConcurrentModificationException e) {
                // late incoming messages from concurrent peer retrieval may cause this
                // we silently do nothing here and return what we listed so far
            }
            String rss = feed.toString();
            //System.out.println("feed has " + feed.size() + " entries");

            // write xml
            response.getOutputStream().write(UTF8.getBytes(rss));
        } else if ("txt".equals(callType)) {
            post.setResponse(response, "text/plain");
            final StringBuilder buffer = new StringBuilder(1000);

            // Write text file
            try {
                for (TwitterTweet t : tl) {
                    UserEntry u = tl.getUser(t);
                    buffer.append(t.getCreatedAt()).append(" ")
                            .append(u.getScreenName())
                            .append(": ")
                            .append(t.moreData.getText(
                                    shortlink_iflinkexceedslength, shortlink_urlstub,
                                    t.getText(), t.getLinks(), t.getPostId()).text)
                            .append('\n');
                }
            } catch (ConcurrentModificationException e) {
                // late incoming messages from concurrent peer retrieval may cause this
                // we silently do nothing here and return what we listed so far
            }
            response.getOutputStream().write(UTF8.getBytes(buffer.toString()));
        }
    }

    private static void setReadMe(JSONObject m) {
        m.put("readme_0", "THIS JSON IS THE RESULT OF YOUR SEARCH QUERY - THERE IS NO WEB PAGE WHICH SHOWS THE RESULT!");
        m.put("readme_1", "loklak.org is the framework for a message search system, not the portal, read: http://api.loklak.org/about.html#notasearchportal");
        m.put("readme_2", "This is supposed to be the back-end of a search portal. For the api, see http://api.loklak.org/api.html");
        m.put("readme_3", "Parameters q=(query), source=(cache|backend|twitter|all), callback=p for jsonp, maximumRecords=(message count), minified=(true|false)");
    }

    private static void moreMetaData(JSONObject metadata, Query post) {
        metadata.put("client", post.getClientHost());
        metadata.put("time", System.currentTimeMillis() - post.getAccessTime());
        metadata.put("servicereduction", post.isDoS_servicereduction() ? "true" : "false");
//        metadata.put("scraper", scraperInfo);
//        String index = tl.getResultIndex() == null ? "" : String.valueOf(tl.getResultIndex());
//        metadata.put("index", index);
    }

    private static void setTwitterMetaData(JSONObject metadata, final int startRecord,
            final int maximumRecords, TwitterTimeline tl, String query, String filter, Query post, JSONObject hits) {
        // the number of the first record (according to SRU set to 1 for very first)
        metadata.put("startRecord", startRecord);
        // number of records within this json result set returned in the api call
        metadata.put("maximumRecords", maximumRecords);
        // number of records available in the search cache (so far, may be increased later > hits)
        metadata.put("count", tl.size());
        // number of records in the search index (so far, may be increased later as well)
        metadata.put("hits", tl.getHits());
        if (tl.getOrder() == Order.CREATED_AT) metadata.put("period", tl.period());
        metadata.put("query", query);
        metadata.put("filter", filter);
        metadata.put("client", post.getClientHost());
        metadata.put("time", System.currentTimeMillis() - post.getAccessTime());
        metadata.put("servicereduction", post.isDoS_servicereduction() ? "true" : "false");
        metadata.putAll(hits);

        String scraperInfo = tl.getScraperInfo() == null ? "" : tl.getScraperInfo();
        metadata.put("scraperInfo", scraperInfo);
        String index = tl.getResultIndex() == null ? "" : String.valueOf(tl.getResultIndex());
        metadata.put("index", index);
    }

    public static void main(String[] args) {
        try {
            TwitterTimeline tl = search(new String[]{"http://root.loklak.org"}, "beer", Order.CREATED_AT, "cache", 20, -120, backend_hash, 10000);
            System.out.println(tl.toJSON(false, "search_metadata", "statuses").toString(2));
        } catch (IOException e) {
            DAO.severe(e);
        }
    }
} public class WeiboUserInfo extends HttpServlet {

	private static final long serialVersionUID = 4653635987712691127L;

	@Override
	protected void doPost(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
		doGet(request, response);
	}

	@Override
	protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
		Query post = RemoteAccess.evaluate(request);

		// manage DoS
		if (post.isDoS_blackout()) {response.sendError(503, "your request frequency is too high"); return;}

		String url = post.get("url", "");
		JSONObject obj = new JSONObject();
		Document doc = Jsoup.connect(url).get();
		Elements infos;
		infos=doc.getElementsByAttributeValue("class", "li_1 clearfix");

		if (infos!=null) {
			Element info;
			String profile;
			for(int i=0;i<infos.size();i++){
				 info=infos.get(i);
				if (info.getElementsByAttributeValueContaining("href", "loc=infblog").size()==0) {
					profile=info.getElementsByAttributeValue("class","pt_detail").first().text().trim();
					obj.put("pro", profile);
					switch(info.getElementsByAttributeValue("class", "pt_title S_txt2").first().text()){
					case "Nickname\uff1a":
						obj.put("username", profile);
						break;
					case "Location\uff1a":
						obj.put("Address", profile);
						break;
					case "Gender\uff1a":
						obj.put("Gender", profile);
						break;
					case "\u6027\u53d6\u5411\uff1a":
						obj.put("Sexuality", profile.replace("t", "").replace("rn", ""));
						break;
					case "\u611f\u60c5\u72b6\u51b5\uff1a":
						obj.put("Relationship", profile.replace("t", "").replace("rn", ""));
						break;
					case "Birthday\uff1a":
						obj.put("Birthday", profile);
						break;
					case "\u8840\u578b\uff1a":
						obj.put("Blood", profile);
						break;
					case "Domain Name\uff1a":
						if(info.getElementsByAttributeValueContaining("href", "loc=infdomain").size()!=0)
						profile=info.select("a").text();
						obj.put("Personaldomain", profile);
						break;
					case "\u7b80\u4ecb\uff1a":
						obj.put("Profile", profile);
						break;
					case "Registration\uff1a":
						obj.put("Registertime", profile.replace("t", "").replace("rn", ""));
						break;
					case "Email\uff1a":
						obj.put("Email", profile);
						break;
					case "QQ\uff1a":
						obj.put("Qq", profile);
						break;
					case "\u5927\u5b66\uff1a":
						obj.put("College", profile.replace("t", "").replace("rn", ""));
						break;
					case "Tags\uff1a":
						obj.put("Tag", profile.replace("t", "").replace("rn", ""));
						break;
					default:
						break;
					}

				} else {
					String blogurl=info.select("a").text();
					obj.put("Blog", blogurl);
				}
			}
		}

		//print JSON
		response.setCharacterEncoding("UTF-8");
		PrintWriter sos = response.getWriter();
		sos.print(obj.toString(2));
		sos.println();
	}
} public class DAO {

    public final static com.fasterxml.jackson.core.JsonFactory jsonFactory = new com.fasterxml.jackson.core.JsonFactory();
    public final static com.fasterxml.jackson.databind.ObjectMapper jsonMapper = new com.fasterxml.jackson.databind.ObjectMapper(DAO.jsonFactory);
    public final static com.fasterxml.jackson.core.type.TypeReference<HashMap<String,Object>> jsonTypeRef = new com.fasterxml.jackson.core.type.TypeReference<HashMap<String,Object>>() {};

    private static Logger logger = LoggerFactory.getLogger(DAO.class);
    
    public final static String MESSAGE_DUMP_FILE_PREFIX = "messages_";
    public final static String ACCOUNT_DUMP_FILE_PREFIX = "accounts_";
    public final static String USER_DUMP_FILE_PREFIX = "users_";
    public final static String ACCESS_DUMP_FILE_PREFIX = "access_";
    public final static String FOLLOWERS_DUMP_FILE_PREFIX = "followers_";
    public final static String FOLLOWING_DUMP_FILE_PREFIX = "following_";
    private static final String IMPORT_PROFILE_FILE_PREFIX = "profile_";

    public static boolean writeDump;

    public final static int CACHE_MAXSIZE =   10000;
    public final static int EXIST_MAXSIZE = 4000000;

    public  static File data_dir, conf_dir, bin_dir, html_dir;
    private static File external_data, assets, dictionaries;
    public static Settings public_settings, private_settings;
    private static Path message_dump_dir, account_dump_dir, import_profile_dump_dir;
    public static Path push_cache_dir;
    public static JsonRepository message_dump;
    private static JsonRepository account_dump;
    private static JsonRepository import_profile_dump;
    public  static JsonDataset user_dump, followers_dump, following_dump;
    public  static AccessTracker access;
    private static File schema_dir, conv_schema_dir;
    public static ElasticsearchClient elasticsearch_client;
    //private static Node elasticsearch_node;
    //private static Client elasticsearch_client;
    public static UserFactory users;
    private static AccountFactory accounts;
    public static MessageFactory messages;
    public static MessageFactory messages_hour;
    public static MessageFactory messages_day;
    public static MessageFactory messages_week;
    public static QueryFactory queries;
    private static ImportProfileFactory importProfiles;
    private static Map<String, String> config = new HashMap<>();
    public  static GeoNames geoNames = null;
    public static Peers peers = new Peers();
    public static OutgoingMessageBuffer outgoingMessages = new OutgoingMessageBuffer();

    // AAA Schema for server usage
    public static JsonTray authentication;
    public static JsonTray authorization;
    public static JsonTray accounting;
    public static UserRoles userRoles;
    public static JsonTray passwordreset;
    public static Map<String, Accounting> accounting_temporary = new HashMap<>();
    public static JsonFile login_keys;
    public static TimelineCache timelineCache;

    public static MQTTPublisher mqttPublisher = null;
    public static boolean streamEnabled = false;
    public static List<String> randomTerms = new ArrayList<>();

    public static enum IndexName {
    	messages_hour("messages.json"), messages_day("messages.json"), messages_week("messages.json"), messages, queries, users, accounts, import_profiles;
        private String schemaFileName;
    	private IndexName() {
    	    schemaFileName = this.name() + ".json";
    	}
    	private IndexName(String filename) {
            schemaFileName = filename;
        }
    	public String getSchemaFilename() {
    	    return this.schemaFileName;
    	}
    }

    /**
     * initialize the DAO
     * @param configMap
     * @param dataPath the path to the data directory
     */
    public static void init(Map<String, String> configMap, Path dataPath) throws Exception{

        log("initializing loklak DAO");

        config = configMap;
        data_dir = dataPath.toFile();
        conf_dir = new File("conf");
        bin_dir = new File("bin");
        html_dir = new File("html");

        writeDump = DAO.getConfig("dump.write_enabled", true);

        // initialize public and private keys
		public_settings = new Settings(new File("data/settings/public.settings.json"));
		File private_file = new File("data/settings/private.settings.json");
		private_settings = new Settings(private_file);
		OS.protectPath(private_file.toPath());

		if(!private_settings.loadPrivateKey() || !public_settings.loadPublicKey()){
        	log("Can't load key pair. Creating new one");

        	// create new key pair
        	KeyPairGenerator keyGen;
			try {
				String algorithm = "RSA";
				keyGen = KeyPairGenerator.getInstance(algorithm);
				keyGen.initialize(2048);
				KeyPair keyPair = keyGen.genKeyPair();
				private_settings.setPrivateKey(keyPair.getPrivate(), algorithm);
				public_settings.setPublicKey(keyPair.getPublic(), algorithm);
			} catch (NoSuchAlgorithmException e) {
				throw e;
			}
			log("Key creation finished. Peer hash: " + public_settings.getPeerHashAlgorithm() + " " + public_settings.getPeerHash());
        }
        else{
        	log("Key pair loaded from file. Peer hash: " + public_settings.getPeerHashAlgorithm() + " " + public_settings.getPeerHash());
        }

        File datadir = dataPath.toFile();
        // check if elasticsearch shall be accessed as external cluster
        String transport = configMap.get("elasticsearch_transport.enabled");
        if (transport != null && "true".equals(transport)) {
            String cluster_name = configMap.get("elasticsearch_transport.cluster.name");
            String transport_addresses_string = configMap.get("elasticsearch_transport.addresses");
            if (transport_addresses_string != null && transport_addresses_string.length() > 0) {
                String[] transport_addresses = transport_addresses_string.split(",");
                elasticsearch_client = new ElasticsearchClient(transport_addresses, cluster_name);
            }
        } else {
            // use all config attributes with a key starting with "elasticsearch." to set elasticsearch settings

        	ESLoggerFactory.setDefaultFactory(new Slf4jESLoggerFactory());
            org.elasticsearch.common.settings.Settings.Builder settings = org.elasticsearch.common.settings.Settings.builder();
            for (Map.Entry<String, String> entry: config.entrySet()) {
                String key = entry.getKey();
                if (key.startsWith("elasticsearch.")) settings.put(key.substring(14), entry.getValue());
            }
            // patch the home path
            settings.put("path.home", datadir.getAbsolutePath());
            settings.put("path.data", datadir.getAbsolutePath());
            settings.build();

            // start elasticsearch
            elasticsearch_client = new ElasticsearchClient(settings);
        }

        // open AAA storage
        Path settings_dir = dataPath.resolve("settings");
        settings_dir.toFile().mkdirs();
        Path authentication_path = settings_dir.resolve("authentication.json");
        authentication = new JsonTray(authentication_path.toFile(), 10000);
        OS.protectPath(authentication_path);
        Path authorization_path = settings_dir.resolve("authorization.json");
        authorization = new JsonTray(authorization_path.toFile(), 10000);
        OS.protectPath(authorization_path);
        Path passwordreset_path = settings_dir.resolve("passwordreset.json");
        passwordreset = new JsonTray(passwordreset_path.toFile(), 10000);
        OS.protectPath(passwordreset_path);
        Path accounting_path = settings_dir.resolve("accounting.json");
        accounting = new JsonTray(accounting_path.toFile(), 10000);
        OS.protectPath(accounting_path);
        Path login_keys_path = settings_dir.resolve("login-keys.json");
        login_keys = new JsonFile(login_keys_path.toFile());
        OS.protectPath(login_keys_path);


        DAO.log("Initializing user roles");

        Path userRoles_path = settings_dir.resolve("userRoles.json");
        userRoles = new UserRoles(new JsonFile(userRoles_path.toFile()));
        OS.protectPath(userRoles_path);

        try{
            userRoles.loadUserRolesFromObject();
            DAO.log("Loaded user roles from file");
        }catch (IllegalArgumentException e){
            DAO.log("Load default user roles");
            userRoles.loadDefaultUserRoles();
        }

        // open index
        Path index_dir = dataPath.resolve("index");
        if (index_dir.toFile().exists()) OS.protectPath(index_dir); // no other permissions to this path

        // define the index factories
        boolean noio = configMap.containsValue("noio") && configMap.get("noio").equals("true");
        messages = new MessageFactory(noio ? null : elasticsearch_client, IndexName.messages.name(), CACHE_MAXSIZE, EXIST_MAXSIZE);
        messages_hour = new MessageFactory(noio ? null : elasticsearch_client, IndexName.messages_hour.name(), CACHE_MAXSIZE, EXIST_MAXSIZE);
        messages_day = new MessageFactory(noio ? null : elasticsearch_client, IndexName.messages_day.name(), CACHE_MAXSIZE, EXIST_MAXSIZE);
        messages_week = new MessageFactory(noio ? null : elasticsearch_client, IndexName.messages_week.name(), CACHE_MAXSIZE, EXIST_MAXSIZE);
        users = new UserFactory(noio ? null : elasticsearch_client, IndexName.users.name(), CACHE_MAXSIZE, EXIST_MAXSIZE);
        accounts = new AccountFactory(noio ? null : elasticsearch_client, IndexName.accounts.name(), CACHE_MAXSIZE, EXIST_MAXSIZE);
        queries = new QueryFactory(noio ? null : elasticsearch_client, IndexName.queries.name(), CACHE_MAXSIZE, EXIST_MAXSIZE);
        importProfiles = new ImportProfileFactory(noio ? null : elasticsearch_client, IndexName.import_profiles.name(), CACHE_MAXSIZE, EXIST_MAXSIZE);

        // create indices and set mapping (that shows how 'elastic' elasticsearch is: it's always good to define data types)
        File mappingsDir = new File(new File(conf_dir, "elasticsearch"), "mappings");
        int shards = Integer.parseInt(configMap.get("elasticsearch.index.number_of_shards"));
        int replicas = Integer.parseInt(configMap.get("elasticsearch.index.number_of_replicas"));
        for (IndexName index: IndexName.values()) {
            log("initializing index '" + index.name() + "'...");
        	try {
        	    elasticsearch_client.createIndexIfNotExists(index.name(), shards, replicas);
        	} catch (Throwable e) {
        		DAO.severe(e);
        	}
            try {
                elasticsearch_client.setMapping(index.name(), new File(mappingsDir, index.getSchemaFilename()));
            } catch (Throwable e) {
            	DAO.severe(e);
            }
        }
        // elasticsearch will probably take some time until it is started up. We do some other stuff meanwhile..

        // create and document the data dump dir
        assets = new File(datadir, "assets").getAbsoluteFile();
        external_data = new File(datadir, "external");
        dictionaries = new File(external_data, "dictionaries");
        dictionaries.mkdirs();

        push_cache_dir = dataPath.resolve("pushcache");
        push_cache_dir.toFile().mkdirs();

        // create message dump dir
        String message_dump_readme =
            "This directory contains dump files for messages which arrived the platform.\n" +
            "There are three subdirectories for dump files:\n" +
            "- own:      for messages received with this peer. There is one file for each month.\n" +
            "- import:   hand-over directory for message dumps to be imported. Drop dumps here and they are imported.\n" +
            "- imported: dump files which had been processed from the import directory are moved here.\n" +
            "You can import dump files from other peers by dropping them into the import directory.\n" +
            "Each dump file must start with the prefix '" + MESSAGE_DUMP_FILE_PREFIX + "' to be recognized.\n";
        message_dump_dir = dataPath.resolve("dump");
        message_dump = new JsonRepository(message_dump_dir.toFile(), MESSAGE_DUMP_FILE_PREFIX, message_dump_readme, JsonRepository.COMPRESSED_MODE, true, Runtime.getRuntime().availableProcessors());

        account_dump_dir = dataPath.resolve("accounts");
        account_dump_dir.toFile().mkdirs();
        OS.protectPath(account_dump_dir); // no other permissions to this path
        account_dump = new JsonRepository(account_dump_dir.toFile(), ACCOUNT_DUMP_FILE_PREFIX, null, JsonRepository.REWRITABLE_MODE, false, Runtime.getRuntime().availableProcessors());

        File user_dump_dir = new File(datadir, "accounts");
        user_dump_dir.mkdirs();
        log("initializing user dump ...");
        user_dump = new JsonDataset(
                user_dump_dir,USER_DUMP_FILE_PREFIX,
                new JsonDataset.Column[]{new JsonDataset.Column("id_str", false), new JsonDataset.Column("screen_name", true)},
                "retrieval_date", DateParser.PATTERN_ISO8601MILLIS,
                JsonRepository.REWRITABLE_MODE, false, Integer.MAX_VALUE);
        log("initializing followers dump ...");
        followers_dump = new JsonDataset(
                user_dump_dir, FOLLOWERS_DUMP_FILE_PREFIX,
                new JsonDataset.Column[]{new JsonDataset.Column("screen_name", true)},
                "retrieval_date", DateParser.PATTERN_ISO8601MILLIS,
                JsonRepository.REWRITABLE_MODE, false, Integer.MAX_VALUE);
        log("initializing following dump ...");
        following_dump = new JsonDataset(
                user_dump_dir, FOLLOWING_DUMP_FILE_PREFIX,
                new JsonDataset.Column[]{new JsonDataset.Column("screen_name", true)},
                "retrieval_date", DateParser.PATTERN_ISO8601MILLIS,
                JsonRepository.REWRITABLE_MODE, false, Integer.MAX_VALUE);

        Path log_dump_dir = dataPath.resolve("log");
        log_dump_dir.toFile().mkdirs();
        OS.protectPath(log_dump_dir); // no other permissions to this path
        access = new AccessTracker(log_dump_dir.toFile(), ACCESS_DUMP_FILE_PREFIX, 60000, 3000);
        access.start(); // start monitor

        timelineCache = new TimelineCache(60000);

        import_profile_dump_dir = dataPath.resolve("import-profiles");
        import_profile_dump = new JsonRepository(import_profile_dump_dir.toFile(), IMPORT_PROFILE_FILE_PREFIX, null, JsonRepository.COMPRESSED_MODE, false, Runtime.getRuntime().availableProcessors());

        // load schema folder
        conv_schema_dir = new File("conf/conversion");
        schema_dir = new File("conf/schema");

        // load dictionaries if they are embedded here
        // read the file allCountries.zip from http://download.geonames.org/export/dump/allCountries.zip
        //File allCountries = new File(dictionaries, "allCountries.zip");
        File cities1000 = new File(dictionaries, "cities1000.zip");
        if (!cities1000.exists()) {
            // download this file
            ClientConnection.download("http://download.geonames.org/export/dump/cities1000.zip", cities1000);
        }

        if(cities1000.exists()){
	        try{
	        	geoNames = new GeoNames(cities1000, new File(conf_dir, "iso3166.json"), 1);
	        }catch(IOException e){
	        	DAO.severe(e.getMessage());
	        	cities1000.delete();
	        	geoNames = null;
	        }
    	}

        // Connect to MQTT message broker
        String mqttAddress = getConfig("stream.mqtt.address", "tcp://127.0.0.1:1883");
        streamEnabled = getConfig("stream.enabled", false);
        if (streamEnabled) {
            mqttPublisher = new MQTTPublisher(mqttAddress);
        }

        // finally wait for healthy status of elasticsearch shards
        ClusterHealthStatus required_status = ClusterHealthStatus.fromString(config.get("elasticsearch_requiredClusterHealthStatus"));
        boolean ok;
        do {
            log("Waiting for elasticsearch " + required_status.name() + " status");
            ok = elasticsearch_client.wait_ready(60000l, required_status);
        } while (!ok);
        /**
        do {
            log("Waiting for elasticsearch green status");
            health = elasticsearch_client.admin().cluster().prepareHealth().setWaitForGreenStatus().execute().actionGet();
        } while (health.isTimedOut());
        **/
        log("elasticsearch has started up!");

        // start the classifier
        new Thread(){
            public void run() {
                log("initializing the classifier...");
                try {
                    Classifier.init(10000, 1000);
                } catch (Throwable e) {
                	DAO.severe(e);
                }
                log("classifier initialized!");
            }
        }.start();

        log("initializing queries...");
        File harvestingPath = new File(datadir, "queries");
        if (!harvestingPath.exists()) harvestingPath.mkdirs();
        String[] list = harvestingPath.list();
        if (list.length < 10) {
            // use the test data instead
            harvestingPath = new File(new File(datadir.getParentFile(), "test"), "queries");
            list = harvestingPath.list();
        }
        for (String queryfile: list) {
            if (queryfile.startsWith(".") || queryfile.endsWith("~")) continue;
            try {
                BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(new File(harvestingPath, queryfile))));
                String line;
                List<IndexEntry<QueryEntry>> bulkEntries = new ArrayList<>();
                while ((line = reader.readLine()) != null) {
                    line = line.trim().toLowerCase();
                    if (line.length() == 0) continue;
                    if (line.charAt(0) <= '9') {
                        // truncate statistic
                        int p = line.indexOf(' ');
                        if (p < 0) continue;
                        line = line.substring(p + 1).trim();
                    }
                    // write line into query database
                    if (!existQuery(line)) {
                        randomTerms.add(line);
                        bulkEntries.add(
                            new IndexEntry<QueryEntry>(
                                line,
                                SourceType.TWITTER,
                                new QueryEntry(line, 0, 60000, SourceType.TWITTER, false))
                        );
                    }
                    if (bulkEntries.size() > 1000) {
                        queries.writeEntries(bulkEntries);
                        bulkEntries.clear();
                    }
                }
                queries.writeEntries(bulkEntries);
                reader.close();
            } catch (IOException e) {
            	DAO.severe(e);
            }
        }
        log("queries initialized.");

        log("finished DAO initialization");
    }

    public static boolean wait_ready(long maxtimemillis) {
        ClusterHealthStatus required_status = ClusterHealthStatus.fromString(config.get("elasticsearch_requiredClusterHealthStatus"));
        return elasticsearch_client.wait_ready(maxtimemillis, required_status);
    }

    public static String pendingClusterTasks() {
        return elasticsearch_client.pendingClusterTasks();
    }

    public static String clusterStats() {
        return elasticsearch_client.clusterStats();
    }

    public static Map<String, String> nodeSettings() {
        return elasticsearch_client.nodeSettings();
    }

    public static File getAssetFile(String screen_name, String id_str, String file) {
        String letter0 = ("" + screen_name.charAt(0)).toLowerCase();
        String letter1 = ("" + screen_name.charAt(1)).toLowerCase();
        Path storage_path = IO.resolvePath(assets.toPath(), letter0, letter1, screen_name);
        return IO.resolvePath(storage_path, id_str + "_" + file).toFile(); // all assets for one user in one file
    }

    public static Collection<File> getTweetOwnDumps(int count) {
        return message_dump.getOwnDumps(count);
    }

    public static void importAccountDumps(int count) throws IOException {
        Collection<File> dumps = account_dump.getImportDumps(count);
        if (dumps == null || dumps.size() == 0) return;
        for (File dump: dumps) {
            JsonReader reader = account_dump.getDumpReader(dump);
            final JsonReader dumpReader = reader;
            Thread[] indexerThreads = new Thread[dumpReader.getConcurrency()];
            for (int i = 0; i < dumpReader.getConcurrency(); i++) {
                indexerThreads[i] = new Thread() {
                    public void run() {
                        JsonFactory accountEntry;
                        try {
                            while ((accountEntry = dumpReader.take()) != JsonStreamReader.POISON_JSON_MAP) {
                                try {
                                    JSONObject json = accountEntry.getJSON();
                                    AccountEntry a = new AccountEntry(json);
                                    DAO.writeAccount(a, false);
                                } catch (IOException e) {
                                	DAO.severe(e);
                                }
                            }
                        } catch (InterruptedException e) {
                        	DAO.severe(e);
                        }
                    }
                };
                indexerThreads[i].start();
            }
            for (int i = 0; i < dumpReader.getConcurrency(); i++) {
                try {indexerThreads[i].join();} catch (InterruptedException e) {}
            }
            account_dump.shiftProcessedDump(dump.getName());
        }
    }

    /**
     * close all objects in this class
     */
    public static void close() {
        DAO.log("closing DAO");

        // close the dump files
        message_dump.close();
        account_dump.close();
        import_profile_dump.close();
        user_dump.close();
        followers_dump.close();
        following_dump.close();

        // close the tracker
        access.close();

        // close the index factories (flushes the caches)
        messages.close();
        messages_hour.close();
        messages_day.close();
        messages_week.close();
        users.close();
        accounts.close();
        queries.close();
        importProfiles.close();

        // close the index
        elasticsearch_client.close();

        DAO.log("closed DAO");
    }

    /**
     * get values from
     * @param key
     * @param default_val
     * @return
     */
    public static String getConfig(String key, String default_val) {
        String value = config.get(key);
        return value == null ? default_val : value;
    }

    public static String[] getConfig(String key, String[] default_val, String delim) {
        String value = config.get(key);
        return value == null || value.length() == 0 ? default_val : value.split(delim);
    }

    public static long getConfig(String key, long default_val) {
        String value = config.get(key);
        try {
            return value == null ? default_val : Long.parseLong(value);
        } catch (NumberFormatException e) {
            return default_val;
        }
    }

    public static double getConfig(String key, double default_val) {
        String value = config.get(key);
        try {
            return value == null ? default_val : Double.parseDouble(value);
        } catch (NumberFormatException e) {
            return default_val;
        }
    }

    public static int getConfig(String key, int default_val) {
        String value = config.get(key);
        try {
            return value == null ? default_val : Integer.parseInt(value);
        } catch (NumberFormatException e) {
            return default_val;
        }
    }
/*
    public static void setConfig(String key, String value) {
        config.put(key, value);
    }

    public static void setConfig(String key, long value) {
        setConfig(key, Long.toString(value));
    }

    public static void setConfig(String key, double value) {
        setConfig(key, Double.toString(value));
    }
*/
    public static JsonNode getSchema(String key) throws IOException {
        File schema = new File(schema_dir, key);
        if (!schema.exists()) {
            throw new FileNotFoundException("No schema file with name " + key + " found");
        }
        return JsonLoader.fromFile(schema);
    }

    public static JSONObject getConversionSchema(String key) throws IOException {
        File schema = new File(conv_schema_dir, key);
        if (!schema.exists()) {
            throw new FileNotFoundException("No schema file with name " + key + " found");
        }
        return new JSONObject(com.google.common.io.Files.toString(schema, Charsets.UTF_8));
    }

    public static boolean getConfig(String key, boolean default_val) {
        String value = config.get(key);
        return value == null ? default_val : value.equals("true") || value.equals("on") || value.equals("1");
    }

    public static Set<String> getConfigKeys() {
        return config.keySet();
    }

    public static class MessageWrapper {
        public TwitterTweet t;
        public UserEntry u;
        public boolean dump;
        public MessageWrapper(TwitterTweet t, UserEntry u, boolean dump) {
            this.t = t;
            this.u = u;
            this.dump = dump;
        }
    }

    /**
     * Store a message together with a user into the search index
     * @param mw a message wrapper
     * @return true if the record was stored because it did not exist, false if it was not stored because the record existed already
     */
    public static boolean writeMessage(MessageWrapper mw) {
        if (mw.t == null) return false;
        try {
            synchronized (DAO.class) {
                // record tweet into search index and check if this is a new entry
                // and check if the message exists
                boolean exists = false;
                if (mw.t.getCreatedAt().after(DateParser.oneHourAgo())) {
                    exists = messages_hour.writeEntry(new IndexEntry<Post>(mw.t.getPostId(), mw.t.getSourceType(), mw.t));
                    if (exists) return false;
                }
                if (mw.t.getCreatedAt().after(DateParser.oneDayAgo())) {
                    exists = messages_day.writeEntry(new IndexEntry<Post>(mw.t.getPostId(), mw.t.getSourceType(), mw.t));
                    if (exists) return false;
                }
                if (mw.t.getCreatedAt().after(DateParser.oneWeekAgo())) {
                    exists = messages_week.writeEntry(new IndexEntry<Post>(mw.t.getPostId(), mw.t.getSourceType(), mw.t));
                    if (exists) return false;
                }
                exists = messages.writeEntry(new IndexEntry<Post>(mw.t.getPostId(), mw.t.getSourceType(), mw.t));
                if (exists) return false;

                // write the user into the index
                users.writeEntryAsync(new IndexEntry<UserEntry>(mw.u.getScreenName(), mw.t.getSourceType(), mw.u));

                // record tweet into text file
                if (mw.dump && writeDump) {
                    message_dump.write(mw.t.toJSON(mw.u, false, Integer.MAX_VALUE, ""), true);
                }
                mw.t.publishToMQTT();
             }

            // teach the classifier
            Classifier.learnPhrase(mw.t.getText());
        } catch (IOException e) {
        	DAO.severe(e);
        }
        return true;
    }

    public static Set<String> writeMessageBulk(Collection<MessageWrapper> mws) {
        List<MessageWrapper> noDump = new ArrayList<>();
        List<MessageWrapper> dump = new ArrayList<>();
        for (MessageWrapper mw: mws) {
            if (mw.t == null) continue;
            if (mw.dump) dump.add(mw); else noDump.add(mw);
        }
        Set<String> createdIDs = new HashSet<>();
        createdIDs.addAll(writeMessageBulkNoDump(noDump));
        // does also do an writeMessageBulkNoDump internally
        createdIDs.addAll(writeMessageBulkDump(dump));
        return createdIDs;
    }

    public static Set<String> writeMessageBulk(Set<PostTimeline> postBulk) {
        for (PostTimeline postList: postBulk) {
            if (postList.size() < 1) continue;
            if(postList.dump) {
                writeMessageBulkDump(postList);
            }
            writeMessageBulkNoDump(postList);
        }
        //TODO: return total dumped, or IDs dumped to create hash of IDs
        return new HashSet<>();
    }

    private static Set<String> writeMessageBulkNoDump(PostTimeline postList) {
        if (postList.size() == 0) return new HashSet<>();
        List<Post> messageBulk = new ArrayList<>();

        for (Post post: postList) {
            if (messages.existsCache(post.getPostId())) {
                continue;
            }
            synchronized (DAO.class) {
                messageBulk.add(post);
            }
        }
        BulkWriteResult result = null;
        try {
            Date limitDate = new Date();
            limitDate.setTime(DateParser.oneHourAgo().getTime());
            List<Post> macc = new ArrayList<Post>();
            final Set<String> existed = new HashSet<>();
            long hourLong = limitDate.getTime();
            for(Post post : messageBulk) {
                if(hourLong <= post.getTimestamp()) macc.add(post);
            }
            
            result = messages_hour.writeEntries(macc);
            for (Post i: macc) if (!(result.getCreated().contains(i.getPostId()))) existed.add(i.getPostId());

            limitDate.setTime(DateParser.oneDayAgo().getTime());
            macc = messageBulk.stream().filter(i -> !(existed.contains(i.getPostId()))).filter(i -> i.getCreated().after(limitDate)).collect(Collectors.toList());
            result = messages_day.writeEntries(macc);
            for (Post i: macc) if (!(result.getCreated().contains(i.getPostId()))) existed.add(i.getPostId());

            limitDate.setTime(DateParser.oneWeekAgo().getTime());
            macc = messageBulk.stream().filter(i -> !(existed.contains(i.getPostId())))
                    .filter(i -> i.getCreated().after(limitDate)).collect(Collectors.toList());
            result = messages_week.writeEntries(macc);
            for (Post i: macc) if (!(result.getCreated().contains(i.getPostId()))) existed.add(i.getPostId());

            macc = messageBulk.stream().filter(i -> !(existed.contains(i.getPostId()))).collect(Collectors.toList());
            result = messages.writeEntries(macc);
            for (Post i: macc) if (!(result.getCreated().contains(i.getPostId()))) existed.add(i.getPostId());
        } catch (IOException e) {
        	DAO.severe(e);
        }
        if (result == null) return new HashSet<String>();
        return result.getCreated();
    }

    /**
     * write messages without writing them to the dump file
     * @param mws a collection of message wrappers
     * @return a set of message IDs which had been created with this bulk write.
     */
    private static Set<String> writeMessageBulkNoDump(Collection<MessageWrapper> mws) {
        if (mws.size() == 0) return new HashSet<>();
        List<IndexEntry<UserEntry>> userBulk = new ArrayList<>();
        List<IndexEntry<Post>> messageBulk = new ArrayList<>();
        for (MessageWrapper mw: mws) {
            if (messages.existsCache(mw.t.getPostId())) continue; // we omit writing this again
            synchronized (DAO.class) {
                // write the user into the index
                userBulk.add(new IndexEntry<UserEntry>(mw.u.getScreenName(), mw.t.getSourceType(), mw.u));

                // record tweet into search index
                messageBulk.add(new IndexEntry<Post>(mw.t.getPostId(), mw.t.getSourceType(), mw.t));
             }

            // teach the classifier
            Classifier.learnPhrase(mw.t.getText());
        }
        BulkWriteResult result = null;
        Set<String> created_ids = new HashSet<>();
        try {
            final Date limitDate = new Date();
            List<IndexEntry<Post>> macc;
            final Set<String> existed = new HashSet<>();

            //DAO.log("***DEBUG messages     INIT: " + messageBulk.size());

            limitDate.setTime(DateParser.oneHourAgo().getTime());
            macc = messageBulk.stream().filter(i -> i.getObject().getCreated().after(limitDate)).collect(Collectors.toList());
            //DAO.log("***DEBUG messages for HOUR: " + macc.size());
            result = messages_hour.writeEntries(macc);
            created_ids.addAll(result.getCreated());
            //DAO.log("***DEBUG messages for HOUR: " + result.getCreated().size() + "  created");
            for (IndexEntry<Post> i: macc) if (!(result.getCreated().contains(i.getId()))) existed.add(i.getId());
            //DAO.log("***DEBUG messages for HOUR: " + existed.size() + "  existed");

            limitDate.setTime(DateParser.oneDayAgo().getTime());
            macc = messageBulk.stream().filter(i -> !(existed.contains(i.getObject().getPostId()))).filter(i -> i.getObject().getCreated().after(limitDate)).collect(Collectors.toList());
            //DAO.log("***DEBUG messages for  DAY : " + macc.size());
            result = messages_day.writeEntries(macc);
            created_ids.addAll(result.getCreated());
            //DAO.log("***DEBUG messages for  DAY: " + result.getCreated().size() + " created");
            for (IndexEntry<Post> i: macc) if (!(result.getCreated().contains(i.getId()))) existed.add(i.getId());
            //DAO.log("***DEBUG messages for  DAY: " + existed.size()  + "  existed");

            limitDate.setTime(DateParser.oneWeekAgo().getTime());
            macc = messageBulk.stream().filter(i -> !(existed.contains(i.getObject().getPostId()))).filter(i -> i.getObject().getCreated().after(limitDate)).collect(Collectors.toList());
            //DAO.log("***DEBUG messages for WEEK: " + macc.size());
            result = messages_week.writeEntries(macc);
            created_ids.addAll(result.getCreated());
            //DAO.log("***DEBUG messages for WEEK: " + result.getCreated().size() + "  created");
            for (IndexEntry<Post> i: macc) if (!(result.getCreated().contains(i.getId()))) existed.add(i.getId());
            //DAO.log("***DEBUG messages for WEEK: " + existed.size()  + "  existed");

            macc = messageBulk.stream().filter(i -> !(existed.contains(i.getObject().getPostId()))).collect(Collectors.toList());
            //DAO.log("***DEBUG messages for  ALL : " + macc.size());
            result = messages.writeEntries(macc);
            created_ids.addAll(result.getCreated());
            //DAO.log("***DEBUG messages for  ALL: " + result.getCreated().size() + "  created");
            for (IndexEntry<Post> i: macc) if (!(result.getCreated().contains(i.getId()))) existed.add(i.getId());
            //DAO.log("***DEBUG messages for  ALL: " + existed.size()  + "  existed");

            users.writeEntries(userBulk);

        } catch (IOException e) {
        	DAO.severe(e);
        }
        return created_ids;
    }

    private static Set<String> writeMessageBulkDump(Collection<MessageWrapper> mws) {
        Set<String> created = writeMessageBulkNoDump(mws);

        for (MessageWrapper mw: mws) try {
            mw.t.publishToMQTT();
            if (!created.contains(mw.t.getPostId())) continue;
            synchronized (DAO.class) {
                
                // record tweet into text file
                if (writeDump) {
                    message_dump.write(mw.t.toJSON(mw.u, false, Integer.MAX_VALUE, ""), true);
                }
            }

            // teach the classifier
            if (randomPicker.nextInt(100) == 0) Classifier.learnPhrase(mw.t.getText());
        } catch (IOException e) {
        	DAO.severe(e);
        }

        return created;
    }

    private static Set<String> writeMessageBulkDump(PostTimeline postList) {
        Set<String> created = writeMessageBulkNoDump(postList);

        for (Post post: postList) try {
            if (!created.contains(post.getPostId())) continue;
            synchronized (DAO.class) {
                // record tweet into text file
                if (writeDump) {
                    message_dump.write(post, true);
                }
            }
        } catch (IOException e) {
        	DAO.severe(e);
        }
        return created;
    }

    /**
     * Store an account together with a user into the search index
     * This method is synchronized to prevent concurrent IO caused by this call.
     * @param a an account
     * @param dump
     * @return true if the record was stored because it did not exist, false if it was not stored because the record existed already
     */
    public static boolean writeAccount(AccountEntry a, boolean dump) {
        try {
            // record account into text file
            if (dump && writeDump) {
                account_dump.write(a.toJSON(null), true);
            }

            // record account into search index
            accounts.writeEntryAsync(new IndexEntry<AccountEntry>(a.getScreenName(), a.getSourceType(), a));
        } catch (IOException e) {
        	DAO.severe(e);
        }
        return true;
    }

    /**
     * Store an import profile into the search index
     * @param i an import profile
     * @return true if the record was stored because it did not exist, false if it was not stored because the record existed already
     */
    public static boolean writeImportProfile(ImportProfileEntry i, boolean dump) {
        try {
            // record import profile into text file
            if (dump && writeDump) {
                import_profile_dump.write(i.toJSON(), true);
            }
            // record import profile into search index
            importProfiles.writeEntryAsync(new IndexEntry<ImportProfileEntry>(i.getId(), i.getSourceType(), i));
        } catch (IOException e) {
        	DAO.severe(e);
        }
        return true;
    }

    private static long countLocalHourMessages(final long millis, boolean created_at) {
        if (millis > DateParser.HOUR_MILLIS) return countLocalDayMessages(millis, created_at);
        if (created_at && millis == DateParser.HOUR_MILLIS) return elasticsearch_client.count(IndexName.messages_hour.name());
        return elasticsearch_client.count(
                created_at ? IndexName.messages_hour.name() : IndexName.messages_day.name(),
                created_at ? AbstractObjectEntry.CREATED_AT_FIELDNAME : AbstractObjectEntry.TIMESTAMP_FIELDNAME,
                millis);
    }

    private static long countLocalDayMessages(final long millis, boolean created_at) {
        if (millis > DateParser.DAY_MILLIS) return countLocalWeekMessages(millis, created_at);
        if (created_at && millis == DateParser.DAY_MILLIS) return elasticsearch_client.count(IndexName.messages_day.name());
        return elasticsearch_client.count(
                created_at ? IndexName.messages_day.name() : IndexName.messages_week.name(),
                created_at ? AbstractObjectEntry.CREATED_AT_FIELDNAME : AbstractObjectEntry.TIMESTAMP_FIELDNAME,
                millis);
    }

    private static long countLocalWeekMessages(final long millis, boolean created_at) {
        if (millis > DateParser.WEEK_MILLIS) return countLocalMessages(millis, created_at);
        if (created_at && millis == DateParser.WEEK_MILLIS) return elasticsearch_client.count(IndexName.messages_week.name());
        return elasticsearch_client.count(
                created_at ? IndexName.messages_week.name() : IndexName.messages.name(),
                created_at ? AbstractObjectEntry.CREATED_AT_FIELDNAME : AbstractObjectEntry.TIMESTAMP_FIELDNAME,
                millis);
    }

    /**
     * count the messages in the local index
     * @param millis number of milliseconds in the past
     * @param created_at field selector: true -> use CREATED_AT, the time when the tweet was created; false -> use TIMESTAMP, the time when the tweet was harvested
     * @return the number of messages in that time span
     */
    public static long countLocalMessages(final long millis, boolean created_at) {
        if (millis == 0) return 0;
        if (millis > 0) {
            if (millis <= DateParser.HOUR_MILLIS) return countLocalHourMessages(millis, created_at);
            if (millis <= DateParser.DAY_MILLIS) return countLocalDayMessages(millis, created_at);
            if (millis <= DateParser.WEEK_MILLIS) return countLocalWeekMessages(millis, created_at);
        }
        return elasticsearch_client.count(
                IndexName.messages.name(),
                created_at ? AbstractObjectEntry.CREATED_AT_FIELDNAME : AbstractObjectEntry.TIMESTAMP_FIELDNAME,
                millis == Long.MAX_VALUE ? -1 : millis);
    }

    public static long countLocalMessages() {
        return elasticsearch_client.count(IndexName.messages.name(), AbstractObjectEntry.TIMESTAMP_FIELDNAME, -1);
    }

    public static long countLocalMessages(String provider_hash) {
        return elasticsearch_client.countLocal(IndexName.messages.name(), provider_hash);
    }

    public static long countLocalUsers() {
        return elasticsearch_client.count(IndexName.users.name(), AbstractObjectEntry.TIMESTAMP_FIELDNAME, -1);
    }

    public static long countLocalQueries() {
        return elasticsearch_client.count(IndexName.queries.name(), AbstractObjectEntry.TIMESTAMP_FIELDNAME, -1);
    }

    public static long countLocalAccounts() {
        return elasticsearch_client.count(IndexName.accounts.name(), AbstractObjectEntry.TIMESTAMP_FIELDNAME, -1);
    }

    public static Post readMessage(String id) throws IOException {
        Post m = null;
        return messages_hour != null && ((m = messages_hour.read(id)) != null) ? m :
               messages_day  != null && ((m = messages_day.read(id))  != null) ? m :
               messages_week != null && ((m = messages_week.read(id)) != null) ? m :
               messages.read(id);
    }

    public static boolean existMessage(String id) {
        return messages_hour != null && messages_hour.exists(id) ||
               messages_day  != null && messages_day.exists(id)  ||
               messages_week != null && messages_week.exists(id) ||
               messages      != null && messages.exists(id);
    }

    public static boolean existUser(String id) {
        return users.exists(id);
    }

    public static boolean existQuery(String id) {
        return queries.exists(id);
    }

    public static boolean deleteQuery(String id, SourceType sourceType) {
        return queries.delete(id, sourceType);
    }
    
    public static String getRandomTerm() {
        return randomTerms.size() == 0 ? null : randomTerms.get(randomPicker.nextInt(randomTerms.size()));
    }

    public static boolean deleteImportProfile(String id, SourceType sourceType) {
        return importProfiles.delete(id, sourceType);
    }

    public static int deleteOld(IndexName indexName, Date createDateLimit) {
        RangeQueryBuilder rangeQuery = QueryBuilders.rangeQuery(AbstractObjectEntry.CREATED_AT_FIELDNAME).to(createDateLimit);
        return elasticsearch_client.deleteByQuery(indexName.name(), rangeQuery);
    }

    public static class SearchLocalMessages {
        public TwitterTimeline timeline;
        public PostTimeline postList;
        public Map<String, List<Map.Entry<String, AtomicLong>>> aggregations;
        public ElasticsearchClient.Query query;

        /**
         * Search the local message cache using a elasticsearch query.
         * @param q - the query, for aggregation this which should include a time frame in the form since:yyyy-MM-dd until:yyyy-MM-dd
         * @param order_field - the field to order the results, i.e. Timeline.Order.CREATED_AT
         * @param timezoneOffset - an offset in minutes that is applied on dates given in the query of the form since:date until:date
         * @param resultCount - the number of messages in the result; can be zero if only aggregations are wanted
         * @param aggregationLimit - the maximum count of facet entities, not search results
         * @param aggregationFields - names of the aggregation fields. If no aggregation is wanted, pass no (zero) field(s)
         * @param filterList - list of filters in String datatype
         */
        public SearchLocalMessages (
                final String q,
                final TwitterTimeline.Order orderField,
                final int timezoneOffset,
                final int resultCount,
                final int aggregationLimit,
                final Set<String> filterList,
                final String... aggregationFields
        ) {
            this.timeline = new TwitterTimeline(orderField);
            QueryEntry.ElasticsearchQuery sq = new QueryEntry.ElasticsearchQuery(q, timezoneOffset, filterList);
            long interval = sq.until.getTime() - sq.since.getTime();
            IndexName resultIndex;
            if (aggregationFields.length > 0 && q.contains("since:")) {
                if (q.contains("since:hour")) {
                    this.query =  elasticsearch_client.query((resultIndex = IndexName.messages_hour).name(), sq.queryBuilder, orderField.getMessageFieldName(), timezoneOffset, resultCount, interval, AbstractObjectEntry.CREATED_AT_FIELDNAME, aggregationLimit, aggregationFields);
                } else if (q.contains("since:day")) {
                    this.query =  elasticsearch_client.query((resultIndex = IndexName.messages_day).name(), sq.queryBuilder, orderField.getMessageFieldName(), timezoneOffset, resultCount, interval, AbstractObjectEntry.CREATED_AT_FIELDNAME, aggregationLimit, aggregationFields);
                } else if (q.contains("since:week")) {
                    this.query =  elasticsearch_client.query((resultIndex = IndexName.messages_week).name(), sq.queryBuilder, orderField.getMessageFieldName(), timezoneOffset, resultCount, interval, AbstractObjectEntry.CREATED_AT_FIELDNAME, aggregationLimit, aggregationFields);
                } else {
                    this.query = elasticsearch_client.query((resultIndex = IndexName.messages).name(), sq.queryBuilder, orderField.getMessageFieldName(), timezoneOffset, resultCount, interval, AbstractObjectEntry.CREATED_AT_FIELDNAME, aggregationLimit, aggregationFields);
                }
            } else {
                // use only a time frame that is sufficient for a result
                this.query = elasticsearch_client.query((resultIndex = IndexName.messages_hour).name(), sq.queryBuilder, orderField.getMessageFieldName(), timezoneOffset, resultCount, interval, AbstractObjectEntry.CREATED_AT_FIELDNAME, aggregationLimit, aggregationFields);
                if (!q.contains("since:hour") && insufficient(this.query, resultCount, aggregationLimit, aggregationFields)) {
                    ElasticsearchClient.Query aq = elasticsearch_client.query((resultIndex = IndexName.messages_day).name(), sq.queryBuilder, orderField.getMessageFieldName(), timezoneOffset, resultCount, interval, AbstractObjectEntry.CREATED_AT_FIELDNAME, aggregationLimit, aggregationFields);
                    this.query.add(aq);
                    if (!q.contains("since:day") && insufficient(this.query, resultCount, aggregationLimit, aggregationFields)) {
                        this.query.add(aq);
                        if (!q.contains("since:week") && insufficient(this.query, resultCount, aggregationLimit, aggregationFields)) {
                            aq = elasticsearch_client.query((resultIndex = IndexName.messages).name(), sq.queryBuilder, orderField.getMessageFieldName(), timezoneOffset, resultCount, interval, AbstractObjectEntry.CREATED_AT_FIELDNAME, aggregationLimit, aggregationFields);
                            this.query.add(aq);
                }}}
            }
                
            timeline.setHits(query.getHitCount());
            timeline.setResultIndex(resultIndex);

            // evaluate search result
            for (Map<String, Object> map: query.getResult()) {
                TwitterTweet tweet = new TwitterTweet(new JSONObject(map));
                try {
                    UserEntry user = users.read(tweet.getScreenName());
                    assert user != null;
                    if (user != null) {
                        timeline.add(tweet, user);
                    }
                } catch (IOException e) {
                	DAO.severe(e);
                }
            }
            this.aggregations = query.getAggregations();
        }

        public SearchLocalMessages (
                final String q,
                final TwitterTimeline.Order orderField,
                final int timezoneOffset,
                final int resultCount,
                final int aggregationLimit,
                final String... aggregationFields
        ) {
            this(
                q,
                orderField,
                timezoneOffset,
                resultCount,
                aggregationLimit,
                new HashSet<>(),
                aggregationFields
            );
        }

        public SearchLocalMessages (
                Map<String, Map<String, String>> inputMap,
                final PostTimeline.Order orderField,
                final int resultCount
        ) {
            this.postList = new PostTimeline(orderField);
            IndexName resultIndex;
            QueryEntry.ElasticsearchQuery sq = new QueryEntry.ElasticsearchQuery(
                    inputMap.get("get"), inputMap.get("not_get"), inputMap.get("also_get"));

            this.query = elasticsearch_client.query(
                    (resultIndex = IndexName.messages_hour).name(),
                    sq.queryBuilder,
                    orderField.getMessageFieldName(),
                    resultCount
            );

            if (this.query.getHitCount() < resultCount) {
                this.query =  elasticsearch_client.query(
                        (resultIndex = IndexName.messages_day).name(),
                        sq.queryBuilder,
                        orderField.getMessageFieldName(),
                        resultCount
                );
            }

            if (this.query.getHitCount() < resultCount) {
                this.query =  elasticsearch_client.query(
                        (resultIndex = IndexName.messages_week).name(),
                        sq.queryBuilder,
                        orderField.getMessageFieldName(),
                        resultCount
                );
            }

            if (this.query.getHitCount() < resultCount) {
                this.query =  elasticsearch_client.query(
                        (resultIndex = IndexName.messages).name(),
                        sq.queryBuilder,
                        orderField.getMessageFieldName(),
                        resultCount
                );
            }

            // Feed search results to postList
            this.postList.setResultIndex(resultIndex);

            addResults();
        }

        private void addResults() {
            Post outputPost;
            for (Map<String, Object> map: this.query.getResult()) {
                outputPost = new Post(map);
                this.postList.addPost(outputPost);
            }
        }

        private static boolean insufficient(
                ElasticsearchClient.Query query,
                int resultCount,
                int aggregationLimit,
                String... aggregationFields
        ) {
            return query.getHitCount() < resultCount || (aggregationFields.length > 0
                    && getAggregationResultLimit(query.getAggregations()) < aggregationLimit);
        }

        public JSONObject getAggregations() {
            JSONObject json = new JSONObject(true);
            if (aggregations == null) return json;
            for (Map.Entry<String, List<Map.Entry<String, AtomicLong>>> aggregation: aggregations.entrySet()) {
                JSONObject facet = new JSONObject(true);
                for (Map.Entry<String, AtomicLong> a: aggregation.getValue()) {
                    if (a.getValue().equals(query)) continue; // we omit obvious terms that cannot be used for faceting, like search for "#abc" -> most hashtag is "#abc"
                    facet.put(a.getKey(), a.getValue().get());
                }
                json.put(aggregation.getKey(), facet);
            }
            return json;
        }

        private static int getAggregationResultLimit(Map<String, List<Map.Entry<String, AtomicLong>>> agg) {
            if (agg == null) return 0;
            int l = 0;
            for (List<Map.Entry<String, AtomicLong>> a: agg.values()) l = Math.max(l, a.size());
            return l;
        }
    }

    public static LinkedHashMap<String, Long> FullDateHistogram(int timezoneOffset) {
        return elasticsearch_client.fullDateHistogram(IndexName.messages.name(), timezoneOffset, AbstractObjectEntry.CREATED_AT_FIELDNAME);
    }

    /**
     * Search the local user cache using a elasticsearch query.
     * @param screen_name - the user id
     */
    public static UserEntry searchLocalUserByScreenName(final String screen_name) {
        try {
            return users.read(screen_name);
        } catch (IOException e) {
        	DAO.severe(e);
            return null;
        }
    }

    public static UserEntry searchLocalUserByUserId(final String user_id) {
        if (user_id == null || user_id.length() == 0) return null;
        Map<String, Object> map = elasticsearch_client.query(IndexName.users.name(), UserEntry.field_user_id, user_id);
        if (map == null) return null;
        return new UserEntry(new JSONObject(map));
    }

    /**
     * Search the local account cache using an elasticsearch query.
     * @param screen_name - the user id
     */
    public static AccountEntry searchLocalAccount(final String screen_name) {
        try {
            return accounts.read(screen_name);
        } catch (IOException e) {
        	DAO.severe(e);
            return null;
        }
    }

    /**
     * Search the local message cache using a elasticsearch query.
     * @param q - the query, can be empty for a matchall-query
     * @param resultCount - the number of messages in the result
     * @param sort_field - the field name to sort the result list, i.e. "query_first"
     * @param sort_order - the sort order (you want to use SortOrder.DESC here)
     */
    public static ResultList<QueryEntry> SearchLocalQueries(final String q, final int resultCount, final String sort_field, final String default_sort_type, final SortOrder sort_order, final Date since, final Date until, final String range_field) {
        ResultList<QueryEntry> queries = new ResultList<>();
        ResultList<Map<String, Object>> result = elasticsearch_client.fuzzyquery(IndexName.queries.name(), "query", q, resultCount, sort_field, default_sort_type, sort_order, since, until, range_field);
        queries.setHits(result.getHits());
        for (Map<String, Object> map: result) {
            QueryEntry qe = new QueryEntry(new JSONObject(map));
            // check a flag value for queries that probably never get new messages
            if (qe.getMessagePeriod() != QueryEntry.DAY_MILLIS) {
                queries.add(qe);
            }
        }
        return queries;
    }

    public static ImportProfileEntry SearchLocalImportProfiles(final String id) {
        try {
            return importProfiles.read(id);
        } catch (IOException e) {
        	DAO.severe(e);
            return null;
        }
    }

    public static Collection<ImportProfileEntry> SearchLocalImportProfilesWithConstraints(final Map<String, String> constraints, boolean latest) throws IOException {
        List<ImportProfileEntry> rawResults = new ArrayList<>();
        List<Map<String, Object>> result = elasticsearch_client.queryWithConstraints(IndexName.import_profiles.name(), "active_status", ImportProfileEntry.EntryStatus.ACTIVE.name().toLowerCase(), constraints, latest);
        for (Map<String, Object> map: result) {
            rawResults.add(new ImportProfileEntry(new JSONObject(map)));
        }

        if (!latest) {
            return rawResults;
        }

        // filter results to display only latest profiles
        Map<String, ImportProfileEntry> latests = new HashMap<>();
        for (ImportProfileEntry entry : rawResults) {
            String uniqueKey;
            if (entry.getImporter() != null) {
                uniqueKey = entry.getSourceUrl() + entry.getImporter();
            } else {
                uniqueKey = entry.getSourceUrl() + entry.getClientHost();
            }
            if (latests.containsKey(uniqueKey)) {
                if (entry.getLastModified().compareTo(latests.get(uniqueKey).getLastModified()) > 0) {
                    latests.put(uniqueKey, entry);
                }
            } else {
                latests.put(uniqueKey, entry);
            }
        }
        return latests.values();
    }

    public static TwitterTimeline scrapeTwitter(
            final Query post,
            final String q,
            final TwitterTimeline.Order order,
            final int timezoneOffset,
            boolean byUserQuery,
            long timeout,
            boolean recordQuery) {

        return scrapeTwitter(post, new HashSet<>(), q, order, timezoneOffset, byUserQuery, timeout, recordQuery);
    }

    /**
     * 
     * @param post
     * @param filterList empty or one/several of image, video
     * @param q
     * @param order
     * @param timezoneOffset
     * @param byUserQuery
     * @param timeout
     * @param recordQuery
     * @return
     */
    public static TwitterTimeline scrapeTwitter(
            final Query post,
            final Set<String> filterList,
            final String q,
            final TwitterTimeline.Order order,
            final int timezoneOffset,
            boolean byUserQuery,
            long timeout,
            boolean recordQuery) {
        // retrieve messages from remote server

        ArrayList<String> remote = DAO.getFrontPeers();
        TwitterTimeline tl;
        if (remote.size() > 0 && (peerLatency.get(remote.get(0)) == null || peerLatency.get(remote.get(0)).longValue() < 3000)) {
            long start = System.currentTimeMillis();
            tl = searchOnOtherPeers(remote, q, filterList, order, 100, timezoneOffset, "all", SearchServlet.frontpeer_hash, timeout); // all must be selected here to catch up missing tweets between intervals
            // at this point the remote list can be empty as a side-effect of the remote search attempt
            if (post != null && remote.size() > 0 && tl != null) post.recordEvent("remote_scraper_on_" + remote.get(0), System.currentTimeMillis() - start);
            if (tl == null || tl.size() == 0) {
                // maybe the remote server died, we try then ourself
                start = System.currentTimeMillis();

                tl = TwitterScraper.search(q, filterList, order, true, true, 400);
                if (post != null) post.recordEvent("local_scraper_after_unsuccessful_remote", System.currentTimeMillis() - start);
            } else {
                tl.writeToIndex();
            }
        } else {
            if (post != null && remote.size() > 0) post.recordEvent("omitted_scraper_latency_" + remote.get(0), peerLatency.get(remote.get(0)));
            long start = System.currentTimeMillis();

            tl = TwitterScraper.search(q, filterList, order, true, true, 400);
            if (post != null) post.recordEvent("local_scraper", System.currentTimeMillis() - start);
        }

        // record the query
        long start2 = System.currentTimeMillis();
        QueryEntry qe = null;
        try {
            qe = queries.read(q);
        } catch (IOException | JSONException e) {
        	DAO.severe(e);
        }

        if (recordQuery && Caretaker.acceptQuery4Retrieval(q) && tl.size() > 0) {
            if (qe == null) {
                // a new query occurred
                qe = new QueryEntry(q, timezoneOffset, tl.period(), SourceType.TWITTER, byUserQuery);
            } else {
                // existing queries are updated
                qe.update(tl.period(), byUserQuery);
            }
            try {
                queries.writeEntryAsync(new IndexEntry<QueryEntry>(q, qe.source_type == null ? SourceType.TWITTER : qe.source_type, qe));
            } catch (IOException e) {
            	DAO.severe(e);
            }
        } else {
            // accept rules may change, we want to delete the query then in the index
            if (qe != null) queries.delete(q, qe.source_type);
        }
        if (post != null) post.recordEvent("query_recorder", System.currentTimeMillis() - start2);
        //log("SCRAPER: TIME LEFT after recording = " + (termination - System.currentTimeMillis()));

        return tl;
    }


    public static JSONArray scrapeLoklak(
            Map<String, String> inputMap,
            boolean byUserQuery,
            boolean recordQuery,
            JSONObject metadata) {
        PostTimeline.Order order= getOrder(inputMap.get("order"));
        PostTimeline dataSet = new PostTimeline(order);
        List<String> scraperList = Arrays.asList(inputMap.get("scraper").trim().split("\\s*,\\s*"));
        List<BaseScraper> scraperObjList = getScraperObjects(scraperList, inputMap);
        ExecutorService scraperRunner = Executors.newFixedThreadPool(scraperObjList.size());

        try{
            for (BaseScraper scraper : scraperObjList) {
                scraperRunner.execute(() -> {
                    dataSet.add(scraper.getData());
                    
                });
            }
        } finally {
            scraperRunner.shutdown();
            try {
                scraperRunner.awaitTermination(3000L, TimeUnit.SECONDS);
            } catch (InterruptedException e) { }
        }
        dataSet.collectMetadata(metadata);
        return dataSet.toArray();
    }

    public static List<BaseScraper> getScraperObjects(List<String> scraperList, Map<String, String> inputMap) {
        //TODO: use SourceType to get this job done
        List<BaseScraper> scraperObjList = new ArrayList<BaseScraper>();
        BaseScraper scraperObj = null;

        if (scraperList.contains("github") || scraperList.contains("all")) {
            scraperObj = new GithubProfileScraper(inputMap);
            scraperObjList.add(scraperObj);
        }
        if (scraperList.contains("quora") || scraperList.contains("all")) {
            scraperObj = new QuoraProfileScraper(inputMap);
            scraperObjList.add(scraperObj);
        }
        if (scraperList.contains("instagram") || scraperList.contains("all")) {
            scraperObj = new InstagramProfileScraper(inputMap);
            scraperObjList.add(scraperObj);
        }
        if (scraperList.contains("youtube") || scraperList.contains("all")) {
            scraperObj = new YoutubeScraper(inputMap);
            scraperObjList.add(scraperObj);
        }
        if (scraperList.contains("wordpress") || scraperList.contains("all")) {
            scraperObj = new WordpressCrawlerService(inputMap);
            scraperObjList.add(scraperObj);
        }
        if (scraperList.contains("twitter") || scraperList.contains("all")) {
            scraperObj = new TweetScraper(inputMap);
            scraperObjList.add(scraperObj);
        }
        //TODO: add more scrapers
        return scraperObjList;
    }

    public static PostTimeline.Order getOrder(String orderString) {
        //TODO: order set according to input
        return PostTimeline.parseOrder("timestamp");
    }

    public static final Random random = new Random(System.currentTimeMillis());
    private static final Map<String, Long> peerLatency = new HashMap<>();
    private static ArrayList<String> getBestPeers(Collection<String> peers) {
        ArrayList<String> best = new ArrayList<>();
        if (peers == null || peers.size() == 0) return best;
        // first check if any of the given peers has unknown latency
        TreeMap<Long, String> o = new TreeMap<>();
        for (String peer: peers) {
            if (peerLatency.containsKey(peer)) {
                o.put(peerLatency.get(peer) * 1000 + best.size(), peer);
            } else {
                best.add(peer);
            }
        }
        best.addAll(o.values());
        return best;
    }
    public static void healLatency(float factor) {
        for (Map.Entry<String, Long> entry: peerLatency.entrySet()) {
            entry.setValue((long) (factor * entry.getValue()));
        }
    }

    private static Set<String> frontPeerCache = ConcurrentHashMap.newKeySet();
    private static Set<String> backendPeerCache = ConcurrentHashMap.newKeySet();

    public static void updateFrontPeerCache(RemoteAccess remoteAccess) {
        if (remoteAccess.getLocalHTTPPort() >= 80) {
            frontPeerCache.add("http://" + remoteAccess.getRemoteHost() + (remoteAccess.getLocalHTTPPort() == 80 ? "" : ":" + remoteAccess.getLocalHTTPPort()));
        } else if (remoteAccess.getLocalHTTPSPort() >= 443) {
            frontPeerCache.add("https://" + remoteAccess.getRemoteHost() + (remoteAccess.getLocalHTTPSPort() == 443 ? "" : ":" + remoteAccess.getLocalHTTPSPort()));
        }
    }

    /**
     * from all known front peers, generate a list of available peers, ordered by the peer latency
     * @return a list of front peers. only the first one shall be used, but the other are fail-over peers
     */
    public static ArrayList<String> getFrontPeers() {
        String[] remote = DAO.getConfig("frontpeers", new String[0], ",");
        ArrayList<String> testpeers = new ArrayList<>();
        if (remote.length > 0) {
            for (String peer: remote) testpeers.add(peer);
            return testpeers;
        }
        /*
        if (frontPeerCache.size() == 0) {
            // add dynamically all peers that contacted myself
            for (Map<String, RemoteAccess> hmap: RemoteAccess.history.values()) {
                for (Map.Entry<String, RemoteAccess> peer: hmap.entrySet()) {
                    updateFrontPeerCache(peer.getValue());
                }
            }
        }
        testpeers.addAll(frontPeerCache);
        */
        return getBestPeers(testpeers);
    }

    public static String[] getBackend() {
        return DAO.getConfig("backend", new String[0], ",");
    }
    
    public static List<String> getBackendPeers() {
        List<String> testpeers = new ArrayList<>();
        if (backendPeerCache.size() == 0) {
            final String[] remote = DAO.getBackend();
            for (String peer: remote) backendPeerCache.add(peer);
        }
        testpeers.addAll(backendPeerCache);
        return getBestPeers(testpeers);
    }

    public static TwitterTimeline searchBackend(final String q,final Set<String> filterList, final TwitterTimeline.Order order, final int count, final int timezoneOffset, final String where, final long timeout) {
        List<String> remote = getBackendPeers();

        if (remote.size() > 0 /*&& (peerLatency.get(remote.get(0)) == null || peerLatency.get(remote.get(0)) < 3000)*/) { // condition deactivated because we need always at least one peer
            TwitterTimeline tt = searchOnOtherPeers(remote, q, filterList, order, count, timezoneOffset, where, SearchServlet.backend_hash, timeout);
            if (tt != null) tt.writeToIndex();
            return tt;
        }
        return null;
    }

    private final static Random randomPicker = new Random(System.currentTimeMillis());

    public static TwitterTimeline searchOnOtherPeers(final List<String> remote, final String q, final Set<String> filterList,final TwitterTimeline.Order order, final int count, final int timezoneOffset, final String source, final String provider_hash, final long timeout) {
        // select remote peer
        while (remote.size() > 0) {
            int pick = randomPicker.nextInt(remote.size());
            String peer = remote.get(pick);
            long start = System.currentTimeMillis();
            try {
                TwitterTimeline tl = SearchServlet.search(new String[]{peer}, q, filterList, order, source, count, timezoneOffset, provider_hash, timeout);
                peerLatency.put(peer, System.currentTimeMillis() - start);
                // to show which peer was used for the retrieval, we move the picked peer to the front of the list
                if (pick != 0) remote.add(0, remote.remove(pick));
                tl.setScraperInfo(tl.getScraperInfo().length() > 0 ? peer + "," + tl.getScraperInfo() : peer);
                return tl;
            } catch (IOException e) {
                DAO.log("searchOnOtherPeers: no IO to scraping target: " + e.getMessage());
                // the remote peer seems to be unresponsive, remove it (temporary) from the remote peer list
                peerLatency.put(peer, DateParser.HOUR_MILLIS);
                frontPeerCache.remove(peer);
                backendPeerCache.remove(peer);
                remote.remove(pick);
            }
        }
        return null;
    }

    public final static Set<Number> newUserIds = ConcurrentHashMap.newKeySet();

    public static void announceNewUserId(TwitterTimeline tl) {
        for (TwitterTweet message: tl) {
            UserEntry user = tl.getUser(message);
            assert user != null;
            if (user == null) continue;
            Number id = user.getUser();
            if (id != null) announceNewUserId(id);
        }
    }

    public static void announceNewUserId(Number id) {
        JsonFactory mapcapsule = DAO.user_dump.get("id_str", id.toString());
        JSONObject map = null;
        try {map = mapcapsule == null ? null : mapcapsule.getJSON();} catch (IOException e) {}
        if (map == null) newUserIds.add(id);
    }

    public static Set<Number> getNewUserIdsChunk() {
        if (newUserIds.size() < 100) return null;
        Set<Number> chunk = new HashSet<>();
        Iterator<Number> i = newUserIds.iterator();
        for (int j = 0; j < 100; j++) {
            chunk.add(i.next());
            i.remove();
        }
        return chunk;
    }

    /**
     * For logging informational events
     */
    public static void log(String line) {
        if (DAO.getConfig("flag.log.dao", "true").equals("true")) {
            logger.info(line);
        }
    }

    /**
     * For events serious enough to inform and log, but not fatal.
     */
    public static void severe(String line) {
        if (DAO.getConfig("flag.severe.dao", "true").equals("true")) {
            logger.warn(line);
        }
    }

    public static void severe(String line, Throwable e) {
        if (DAO.getConfig("flag.severe.dao", "true").equals("true")) {
            logger.warn(line, e);
        }
    }

    public static void severe(Throwable e) {
        if (DAO.getConfig("flag.severe.dao", "true").equals("true")) {
            logger.warn("", e);
        }
    }

    /**
     * For Debugging events (very noisy).
     */
    public static void debug(Throwable e) {
        if (DAO.getConfig("flag.debug.dao", "true").equals("true")) {
            DAO.severe(e);
        }
    }

    /**
     * For Stacktracing exceptions (preferred over debug).
     */
    public static void trace(Throwable e) {
        if(DAO.getConfig("flag.trace.dao", "true").equals("true")) {
            e.printStackTrace();
        }
    }
} public class TwitterHarvester {

    private final static int FETCH_MIN = 20;
    //private final static int HITS_LIMIT_4_QUERIES = 20;
    private final static int MAX_PENDING_CONTEXT = 300; // this could be much larger but we don't want to cache too many of these
    private final static int MAX_HARVESTED_CONTEXT = 10000; // just to prevent a memory leak with possible OOM after a long time we flush that cache after a while
    //private final static Random random = new Random(System.currentTimeMillis());

    private LinkedHashSet<String> pendingQueries = new LinkedHashSet<>();
    private ConcurrentLinkedDeque<String> pendingContext = new ConcurrentLinkedDeque<>();
    private Set<String> harvestedContext = ConcurrentHashMap.newKeySet();

    private final AtomicInteger latest_query_hit_count_on_backend = new AtomicInteger(100);

    private void checkContext(TwitterTimeline tl, boolean front) {
        for (TwitterTweet tweet: tl) {
            for (String user: tweet.getMentions()) checkContext("from:" + user, front);
            for (String hashtag: tweet.getHashtags()) checkContext(hashtag, front);
        }
    }

    private void checkContext(String s, boolean front) {
        if (!front && pendingContext.size() > MAX_PENDING_CONTEXT) return; // queue is full
        if (!harvestedContext.contains(s) && !pendingContext.contains(s)) {
            if (front) pendingContext.addFirst(s); else pendingContext.addLast(s);
        }
        while (pendingContext.size() > MAX_PENDING_CONTEXT) pendingContext.removeLast();
        if (harvestedContext.size() > MAX_HARVESTED_CONTEXT) harvestedContext.clear();
    }

    public int get_harvest_queries() throws IOException {
        String[] backend = DAO.getBackend();

        /*
        if (random.nextInt(100) != 0 && latest_query_hit_count_on_backend.get() < HITS_LIMIT_4_QUERIES && pendingQueries.size() == 0 && pendingContext.size() > 0) {
            // harvest using the collected keys instead using the queries
            String q = pendingContext.removeFirst();
            harvestedContext.add(q);
            TwitterTimeline tl = TwitterScraper.search(q, Order.CREATED_AT, true, true, 400);
            if (tl == null || tl.size() == 0) return -1;

            // find content query strings and store them in the context cache
            checkContext(tl, false);
            DAO.log("retrieval of " + tl.size() + " new messages for q = " + q + ", scheduled push; pendingQueries = " + pendingQueries.size() + ", pendingContext = " + pendingContext.size() + ", harvestedContext = " + harvestedContext.size());
            return tl.size();
        }
         */

        // Load more queries if pendingQueries is empty
        if (pendingQueries.size() == 0) {
            int fetch_random = Math.min(100, Math.max(FETCH_MIN, latest_query_hit_count_on_backend.get() / 10));

            DAO.log("START loading " + fetch_random + " suggestions for harvesting");
            ResultList<QueryEntry> rl = SuggestServlet.suggest(backend, "", "query", fetch_random * 5, "asc", "retrieval_next", DateParser.getTimezoneOffset(), null, "now", "retrieval_next", fetch_random);
            for (QueryEntry qe: rl) {
                pendingQueries.add(qe.getQuery());
            }
            latest_query_hit_count_on_backend.set((int) rl.getHits());
            DAO.log("STOP loading suggestions, got " + rl.size() + " for harvesting from " + latest_query_hit_count_on_backend + " in backend");
        }

        return pendingQueries.size();
    }

    public TwitterTimeline harvest_timeline() {

        if (pendingQueries.size() == 0) return null;

        // take one of the pending queries or pending context and load the tweets
        String q = "";
        try {
            q = pendingQueries.iterator().next();
        } catch (NoSuchElementException e) {
            return null;
        }
        if (q == null || q.length() == 0) return null;
        pendingQueries.remove(q);
        pendingContext.remove(q);
        harvestedContext.add(q);
        TwitterTimeline tl = TwitterScraper.search(q, Order.CREATED_AT, true, false, 400);

        if (tl != null && tl.size() > 0) {
            // find content query strings and store them in the context cache
            tl.setQuery(q);
            checkContext(tl, true);
        }

        return tl;
    }

    public void push_timeline_to_backend(TwitterTimeline tl) {
        String[] backend = DAO.getBackend();
        DAO.log( "starting push to backend; pendingQueries = " + pendingQueries.size() + ", pendingContext = " +
                pendingContext.size() + ", harvestedContext = " + harvestedContext.size());

        boolean success = false;
        for (int i = 0; i < 5; i++) {
            try {
                long start = System.currentTimeMillis();
                success = PushServlet.push(backend, tl);
                if (success) {
                    DAO.log("retrieval of " + tl.size() + " new messages for q = " + tl.getQuery() +
                            ", pushed to backend synchronously in " + (System.currentTimeMillis() - start) + " ms; amount = " + tl.size());
                    return;
                }
            } catch (Throwable e) {
                DAO.log("failed synchronous push to backend, attempt " + i);
                try {Thread.sleep((i + 1) * 3000);} catch (InterruptedException e1) {}
            }
        }
        String q = tl.getQuery();
        tl.setQuery(null);
        DAO.outgoingMessages.transmitTimelineToBackend(tl);
        DAO.log("retrieval of " + tl.size() + " new messages for q = " + q + ", scheduled push");
    }

} public class TwitterScraper {

    public static final ExecutorService executor = Executors.newFixedThreadPool(40);
    public static final Pattern emoji_pattern_span = Pattern.compile("<span [^>]*class=\"Emoji Emoji--forLinks\" [^>]*>[\\n]*[^<]*</span>[\\n]*<span [^>]*class=\"visuallyhidden\" [^>]*aria-hidden=\"true\"[^>]*>[\\n]*([^<]*)[\\n]*</span>");

    public static TwitterTimeline search(
            final String query,
            final Set<String> filterList,
            final TwitterTimeline.Order order,
            final boolean writeToIndex,
            final boolean writeToBackend,
            int jointime) {
        TwitterTimeline[] tl = search(query, filterList, order, writeToIndex, writeToBackend);
        long timeout = System.currentTimeMillis() + jointime;
        long remainingWait = 0;
        for (TwitterTweet tt: tl[1]) {
            remainingWait = Math.max(10, timeout - System.currentTimeMillis());
            if (tt.waitReady(remainingWait)) {
                 // double additions are detected
                tl[0].add(tt, tt.getUser());
            }
        }
        return tl[0];
    }

    public static TwitterTimeline search(
            final String query,
            final TwitterTimeline.Order order,
            final boolean writeToIndex,
            final boolean writeToBackend,
            int jointime) {

        return search(query, new HashSet<>(), order, writeToIndex, writeToBackend, jointime);
    }

    private enum FilterType {
        tweets, image, video, geo, links, mentions, hashtags, media;
    }
    
    private static String prepareSearchUrl(final String query, FilterType type) {
        // check https://twitter.com/search-advanced for a better syntax
        // https://support.twitter.com/articles/71577-how-to-use-advanced-twitter-search#

        // query q
        StringBuilder t = new StringBuilder(query.length());
        for (String s: query.replace('+', ' ').split(" ")) {
            t.append(' ');
            if (s.startsWith("since:") || s.startsWith("until:")) {
                int u = s.indexOf('_');
                t.append(u < 0 ? s : s.substring(0, u));
            } else {
                t.append(s);
            }
        }
        if (type == FilterType.image && t.indexOf("has:images") < 0) t.append(" has:images");
        if (type == FilterType.video && t.indexOf("has:videos") < 0) t.append(" has:videos");
        
        String q;
        try {
            q = t.length() == 0 ? "*" : URLEncoder.encode(t.substring(1), "UTF-8");
        } catch (UnsupportedEncodingException e) {
            q = t.length() == 0 ? "*" : t.substring(1);
        }

        // building url
        // see https://github.com/bisguzar/twitter-scraper/blob/master/twitter_scraper/modules/tweets.py#L12-L29
        return "https://twitter.com/i/search/timeline?f=tweets&vertical=default&q=" + q + "&src=typd&reset_error_state=false&include_available_features=1&include_entities=1&include_new_items_bar=true";
    }

    private static TwitterTimeline[] search(
            final String query,
            final Set<String> filterList,
            final TwitterTimeline.Order order,
            final boolean writeToIndex,
            final boolean writeToBackend) {
        // check
        // https://twitter.com/search-advanced for a better syntax
        // https://support.twitter.com/articles/71577-how-to-use-advanced-twitter-search#
        FilterType tt = FilterType.tweets;
        if (filterList.contains("image")) tt = FilterType.image;
        if (filterList.contains("video")) tt = FilterType.video;
        String https_url = prepareSearchUrl(query, tt);
        TwitterTimeline[] timelines = null;
        try {
            ClientConnection connection = new ClientConnection(https_url, query);
            if (connection.getInputStream() == null) return null;
            try {
                BufferedReader br = new BufferedReader(new InputStreamReader(connection.getInputStream(), StandardCharsets.UTF_8));

                JSONTokener t = new JSONTokener(br);
                JSONObject j = new JSONObject(t);
                String html = j.getString("items_html");
                Document doc = Jsoup.parse(html);
                
                timelines = search(doc, filterList, order, writeToIndex, writeToBackend);
            } catch (IOException e) {
                DAO.severe(e);
            } finally {
                connection.close();
            }
            /*
            HtmlUnitLoader loader = new HtmlUnitLoader(https_url, "scraper");
            ByteArrayInputStream bais = new ByteArrayInputStream(loader.getXml().getBytes(StandardCharsets.UTF_8));
            BufferedReader br = new BufferedReader(new InputStreamReader(bais, StandardCharsets.UTF_8));
            timelines = search(br, filterList, order, writeToIndex, writeToBackend);
            */
        } catch (IOException e) {
            // this could mean that twitter rejected the connection (DoS protection?) or we are offline (we should be silent then)
            // DAO.severe(e);
            if (timelines == null) timelines = new TwitterTimeline[]{new TwitterTimeline(order), new TwitterTimeline(order)};
        };

        // wait until all messages in the timeline are ready
        if (timelines == null) {
            // timeout occurred
            timelines = new TwitterTimeline[]{new TwitterTimeline(order), new TwitterTimeline(order)};
        }
        if (timelines != null) {
            if (timelines[0] != null) timelines[0].setScraperInfo("local");
            if (timelines[1] != null) timelines[1].setScraperInfo("local");
        }
        return timelines;
    }

    /**
     * scrape messages from the reader stream: this already checks if a message is new. There are only new messages returned
     * @param br
     * @param order
     * @return two timelines in one array: Timeline[0] is the one which is finished to be used, Timeline[1] contains messages which are in postprocessing
     * @throws IOException
     */
    private static TwitterTimeline[] search(
            final Document doc,
            final Set<String> filterList,
            final TwitterTimeline.Order order,
            final boolean writeToIndex,
            final boolean writeToBackend) throws IOException {
        TwitterTimeline timelineReady = new TwitterTimeline(order);
        TwitterTimeline timelineWorking = new TwitterTimeline(order);
        Set<String> images = null;
        Set<String> videos = null;
        String place_id = "";
        String place_name = "";
        Set<String> mentions = new LinkedHashSet<>();
        boolean debuglog = DAO.getConfig("flag.debug.twitter_scraper", "false").equals("true");

        // parse
        Elements items = doc.getElementsByClass("stream-item");
        itemloop: for (int itemc = 0; itemc < items.size(); itemc++) {
            Element item = items.get(itemc);
            if (debuglog) System.out.println(item.toString());
    
            Elements streamitem = item.getElementsByClass("stream-item");
            Element tweeti = streamitem.get(0).child(0);
            String tweetID = tweeti.attr("data-tweet-id");
            String conversationID = tweeti.attr("data-conversation-id"); // this is the parent in a conversation
            String mentions_screenname = tweeti.attr("data-mentions"); // the conversationID does not have a link to the user name which produces the actual tweet link. This here is a hint. But several names can be listed here.
            // the first nam in mentions_screenname is the latest in the conversation chain
            Set<String> mentions_screennames = new LinkedHashSet<>(); // to preserve the order we use a linked hash set
            String[] mss= mentions_screenname.split(" ");
            for (int i = 0; i < mss.length; i++) mentions_screennames.add(mss[i]);
            if (conversationID != null && conversationID.equals(tweetID)) {conversationID = ""; mentions_screennames.clear();}
            
            Elements profile = item.getElementsByClass("js-profile-popup-actionable");
            Elements avatare = item.getElementsByClass("js-action-profile-avatar");
            
            String userid = profile.attr("data-user-id");
            String usernickname = profile.attr("data-screen-name");
            String useravatarurl = avatare.attr("src");
            String userfullname = profile.attr("data-name");
            
            UserEntry user = new UserEntry(
                    userid,
                    usernickname,
                    useravatarurl,
                    MessageEntry.html2utf8(userfullname)
            );
    
            Elements timestamp = item.getElementsByClass("_timestamp");
            String rawtext = item.getElementsByClass("tweet-text").text();
            if (mentions_screennames.size() > 0) {
                // attention: this turns the order around! First entry will be conversation starter!
                for (String screenname: mentions_screennames) {
                    if (rawtext.indexOf(screenname) < 0) rawtext = "@" + screenname + " " + rawtext;
                }
            }
            
            String tweettimes = timestamp.attr("data-time-ms");
            if (tweettimes.length() == 0) continue itemloop; // sometimes tweets are not available any more
            long tweettime = Long.parseLong(tweettimes);
            // this assertion holds in most, but unfortunately not in all cases, so it is commented out
            // long snowflaketime = snowflake2millis(Long.parseLong(tweetID));
            // assert tweettime / 1000L == snowflaketime / 1000L : "tweettime = " + tweettime + ", snowflaketime = " + snowflaketime;

            Elements reply = item.getElementsByClass("ProfileTweet-action--reply").get(0).children();
            Elements retweet = item.getElementsByClass("ProfileTweet-action--retweet").get(0).children();
            Elements favourite = item.getElementsByClass("ProfileTweet-action--favorite").get(0).children();
            String tweetreplycount = reply.attr("data-tweet-stat-count");
            String tweetretweetcount = retweet.attr("data-tweet-stat-count");
            String tweetfavouritecount = favourite.attr("data-tweet-stat-count");
            
            String tweetstatusurl = item.getElementsByClass("tweet-timestamp").attr("href");        
            // evaluate
            
            // patch text with non-mentioned mentions (twitter obviously cuts that away)
            mentions.remove(user.getScreenName());
            for (String mention: mentions) {
                String m = "@" + mention;
                if (rawtext.indexOf(m) < 0) rawtext = rawtext + " " + m;
            }
            
            TwitterTweet tweet;
            try {
                tweet = new TwitterTweet(
                        user.getScreenName(),
                        tweettime,
                        tweetID, // like 1284585691259850753
                        conversationID, // like 1284532773563359233
                        mentions_screennames,
                        tweetstatusurl, // like /yacy_search/status/1284585691259850753
                        rawtext,
                        Long.parseLong(tweetreplycount),
                        Long.parseLong(tweetretweetcount),
                        Long.parseLong(tweetfavouritecount),
                        images, videos, place_name, place_id,
                        user, writeToIndex, writeToBackend
                );
    
                place_id = "";
                place_name = "";
                mentions.clear();
                videos = null;
                images = null;
            } catch (NullPointerException e) {
                DAO.severe(e);
                tweet = null;
            }
            
            //if (DAO.messages == null || !DAO.messages.existsCache(tweet.getPostId())) {
                // checking against the exist cache is incomplete. A false negative would just cause that a tweet is
                // indexed again.
                if (tweet.willBeTimeConsuming()) {
                    executor.execute(tweet);
                    //new Thread(tweet).start();
                    // because the executor may run the thread in the current thread it could be possible that the result is here already
                    if (tweet.isReady()) {
    
                        timelineReady.add(tweet, user);
                        //DAO.log("SCRAPERTEST: messageINIT is ready");
                    } else {
                        timelineWorking.add(tweet, user);
                        //DAO.log("SCRAPERTEST: messageINIT unshortening");
                    }
                } else {
                    // no additional thread needed, run the postprocessing in the current thread
                    tweet.run();
                    timelineReady.add(tweet, user);
                }
            //}
        }

        return new TwitterTimeline[]{timelineReady, timelineWorking};
    }

    private static long snowflake2millis(long sf) {
        return (sf >> 22) + 1288834974657L;
    }
    
    private final static Pattern hashtag_pattern = Pattern.compile("<a href=\"/hashtag/.*?\".*?class=\"twitter-hashtag.*?\".*?><s>#</s><b>(.*?)</b></a>");
    private final static Pattern timeline_link_pattern = Pattern.compile("<a href=\"https://(.*?)\".*? data-expanded-url=\"(.*?)\".*?twitter-timeline-link.*?title=\"(.*?)\".*?>.*?</a>");
    private final static Pattern timeline_embed_pattern = Pattern.compile("<a href=\"(https://t.co/\\w+)\" class=\"twitter-timeline-link.*?>pic.twitter.com/(.*?)</a>");
    private final static Pattern emoji_pattern = Pattern.compile("<img .*?class=\"Emoji Emoji--forText\".*?alt=\"(.*?)\".*?>");
    private final static Pattern doublespace_pattern = Pattern.compile("  ");
    private final static Pattern cleanup_pattern = Pattern.compile(
            "</?(s|b|strong)>|" +
                    "<a href=\"/hashtag.*?>|" +
                    "<a.*?class=\"twitter-atreply.*?>|" +
                    "<span.*?span>"
    );

    public static class TwitterTweet extends Post implements Runnable {

        public final Semaphore ready;
        public MessageEntry moreData = new MessageEntry();
        public UserEntry user;
        public boolean writeToIndex;
        public boolean writeToBackend;

        // a time stamp that is given in loklak upon the arrival of the tweet which is the current local time
        public Date timestampDate;
        // the time given in the tweet which is the time when the user created it.
        // This is also use to do the index partition into minute, hour, week
        public Date created_at;
        // on means 'valid from'
        public Date on;
        // 'to' means 'valid_until' and may not be set
        public Date to;

        // where did the message come from
        protected SourceType source_type;
        // who created the message
        protected ProviderType provider_type;

        public String provider_hash, screen_name, retweet_from, postId, conversationID, canonical_id, parent, text;
        public Set<String> conversationUserIDs;
        protected URL status_id_url;
        protected long reply_count, retweet_count, favourite_count;
        public Set<String> images, audios, videos;
        protected String place_name, place_id;

        // the following fields are either set as a common field or generated by extraction from field 'text' or from field 'place_name'
        // coordinate order is [longitude, latitude]
        protected double[] location_point, location_mark;
        // Value in metres
        protected int location_radius;
        protected LocationSource location_source;
        protected PlaceContext place_context;
        protected String place_country;

        // The length of tweets without links, users, hashtags
        // the following can be computed from the tweet data but is stored in the search index
        // to provide statistical data and ranking attributes
        private int without_l_len, without_lu_len, without_luh_len;

        // the arrays of links, users, hashtags
        private List<String> users, hosts, links, mentions, hashtags;

        private boolean enriched;

        public TwitterTweet(
                final String user_screen_name_raw,
                final long created_at_raw,
                final String tweetID, // the ID of the tweet
                final String tweetConversationID, // the ID of a tweet where this is a reply-to tweet
                final Set<String> mentions_screennames,
                final String status_id_url_raw,
                final String text_raw,
                final long replies,
                final long retweets,
                final long favourites,
                final Set<String> images,
                final Set<String> videos,
                final String place_name,
                final String place_id,
                final UserEntry user,
                final boolean writeToIndex,
                final boolean writeToBackend) throws MalformedURLException {
            super();
            this.source_type = SourceType.TWITTER;
            this.provider_type = ProviderType.SCRAPED;
            this.screen_name = user_screen_name_raw;
            this.created_at = new Date(created_at_raw);
            this.status_id_url = new URL("https://twitter.com" + status_id_url_raw);
            int p = status_id_url_raw.lastIndexOf('/');
            this.postId = p >= 0 ? status_id_url_raw.substring(p + 1) : "-1";
            assert this.postId.equals(tweetID);
            this.conversationID = tweetConversationID;
            this.conversationUserIDs = mentions_screennames;
            this.reply_count = replies;
            this.retweet_count = retweets;
            this.favourite_count = favourites;
            this.place_name = place_name;
            this.place_id = place_id;
            this.images = images;
            this.videos = videos;
            this.text = text_raw;
            this.user = user;
            this.writeToIndex = writeToIndex;
            this.writeToBackend = writeToBackend;

            //Date d = new Date(timemsraw);
            //System.out.println(d);

            /* failed to reverse-engineering the place_id :(
            if (place_id.length() == 16) {
                String a = place_id.substring(0, 8);
                String b = place_id.substring(8, 16);
                long an = Long.parseLong(a, 16);
                long bn = Long.parseLong(b, 16);
                System.out.println("place = " + place_name + ", a = " + an + ", b = " + bn);
                // Frankfurt a = 3314145750, b = 3979907708, http://www.openstreetmap.org/#map=15/50.1128/8.6835
                // Singapore a = 1487192992, b = 3578663936
            }
            */

            // this.text MUST be analysed with analyse(); this is not done here because it should be started concurrently; run run();

            this.ready = new Semaphore(0);
        }

        public TwitterTweet(JSONObject json) {
            this.moreData = new MessageEntry();
            Object timestamp_obj = lazyGet(json, AbstractObjectEntry.TIMESTAMP_FIELDNAME);
            this.timestampDate = MessageEntry.parseDate(timestamp_obj);
            this.timestamp = this.timestampDate.getTime();
            Object created_at_obj = lazyGet(json, AbstractObjectEntry.CREATED_AT_FIELDNAME);
            this.created_at = MessageEntry.parseDate(created_at_obj);
            Object on_obj = lazyGet(json, "on");
            this.on = on_obj == null ? null : MessageEntry.parseDate(on);
            Object to_obj = lazyGet(json, "to");
            this.to = to_obj == null ? null : MessageEntry.parseDate(to);
            String source_type_string = (String) lazyGet(json, "source_type");
            try {
                this.source_type = source_type_string == null ? SourceType.GENERIC : SourceType.byName(source_type_string);
            } catch (IllegalArgumentException e) {
                this.source_type = SourceType.GENERIC;
            }
            String provider_type_string = (String) lazyGet(json, "provider_type");
            if (provider_type_string == null) provider_type_string = ProviderType.NOONE.name();
            try {
                this.provider_type = ProviderType.valueOf(provider_type_string);
            } catch (IllegalArgumentException e) {
                this.provider_type = ProviderType.NOONE;
            }
            this.provider_hash = (String) lazyGet(json, "provider_hash");
            this.screen_name = (String) lazyGet(json, "screen_name");
            this.retweet_from = (String) lazyGet(json, "retweet_from");
            this.postId = (String) lazyGet(json, "id_str");
            this.conversationID = (String) lazyGet(json, "conversationid_str");
            this.conversationUserIDs = MessageEntry.parseArrayList(lazyGet(json, "conversation_user"));
            this.text = (String) lazyGet(json, "text");
            try {
                this.status_id_url = new URL((String) lazyGet(json, "link"));
            } catch (MalformedURLException e) {
                this.status_id_url = null;
            }
            this.reply_count = MessageEntry.parseLong((Number) lazyGet(json, "reply_count"));
            this.retweet_count = MessageEntry.parseLong((Number) lazyGet(json, "retweet_count"));
            this.favourite_count = MessageEntry.parseLong((Number) lazyGet(json, "favourites_count")); // inconsitency in naming, but twitter api defines so
            this.images = MessageEntry.parseArrayList(lazyGet(json, "images"));
            this.audios = MessageEntry.parseArrayList(lazyGet(json, "audio"));
            this.videos = MessageEntry.parseArrayList(lazyGet(json, "videos"));
            this.place_id = MessageEntry.parseString((String) lazyGet(json, "place_id"));
            this.place_name = MessageEntry.parseString((String) lazyGet(json, "place_name"));
            this.place_country = MessageEntry.parseString((String) lazyGet(json, "place_country"));

            if (this.place_country != null && this.place_country.length() != 2) this.place_country = null;

            // optional location
            Object location_point_obj = lazyGet(json, "location_point");
            Object location_radius_obj = lazyGet(json, "location_radius");
            Object location_mark_obj = lazyGet(json, "location_mark");
            Object location_source_obj = lazyGet(json, "location_source");
            if (location_point_obj == null || location_mark_obj == null ||
                !(location_point_obj instanceof List<?>) ||
                !(location_mark_obj instanceof List<?>)) {
                this.location_point = null;
                this.location_radius = 0;
                this.location_mark = null;
                this.location_source = null;
            } else {
                this.location_point = new double[]{(Double) ((List<?>) location_point_obj).get(0), (Double) ((List<?>) location_point_obj).get(1)};
                this.location_radius = (int) MessageEntry.parseLong((Number) location_radius_obj);
                this.location_mark = new double[]{(Double) ((List<?>) location_mark_obj).get(0), (Double) ((List<?>) location_mark_obj).get(1)};
                this.location_source = LocationSource.valueOf((String) location_source_obj);
            }
            this.enriched = false;

            // load enriched data
            enrich();

            // may lead to error!!
            this.ready = new Semaphore(0);
            //this.user = null;
            //this.writeToIndex = false;
            //this.writeToBackend = false;
        }

        public TwitterTweet() throws MalformedURLException {
            this.moreData = new MessageEntry();
            this.timestamp = new Date().getTime();
            this.timestampDate = new Date(this.timestamp);
            this.created_at = new Date();
            this.on = null;
            this.to = null;
            this.source_type = SourceType.GENERIC;
            this.provider_type = ProviderType.NOONE;
            this.provider_hash = "";
            this.screen_name = "";
            this.retweet_from = "";
            this.postId = "";
            this.conversationID = "";
            this.conversationUserIDs = new LinkedHashSet<>();
            this.canonical_id = "";
            this.parent = "";
            this.text = "";
            this.status_id_url = null;
            this.reply_count = 0;
            this.retweet_count = 0;
            this.favourite_count = 0;
            this.images = new HashSet<String>();
            this.audios = new HashSet<String>();
            this.videos = new HashSet<String>();
            this.place_id = "";
            this.place_name = "";
            this.place_context = null;
            this.place_country = null;
            this.location_point = null;
            this.location_radius = 0;
            this.location_mark = null;
            this.location_source = null;
            this.without_l_len = 0;
            this.without_lu_len = 0;
            this.without_luh_len = 0;
            this.hosts = new ArrayList<String>();
            this.links = new ArrayList<String>();
            this.mentions = new ArrayList<String>();
            this.hashtags = new ArrayList<String>();
            this.moreData.classifier = null;
            this.enriched = false;

            // may lead to error!!
            this.ready = new Semaphore(0);
            //this.user = null;
            //this.writeToIndex = false;
            //this.writeToBackend = false;
        }

        //TODO: fix the location issue and shift to MessageEntry class
        public void getLocation() {
            if (this.text == null) this.text = "";

            if ((this.location_point == null || this.location_point.length == 0) && DAO.geoNames != null) {
                GeoMark loc = null;
                if (place_name != null && this.place_name.length() > 0 &&
                    (this.location_source == null || this.location_source == LocationSource.ANNOTATION || this.location_source == LocationSource.PLACE)) {
                    loc = DAO.geoNames.analyse(this.place_name, null, 5, Integer.toString(this.text.hashCode()));
                    this.place_context = PlaceContext.FROM;
                    this.location_source = LocationSource.PLACE;
                }
                if (loc == null) {
                    loc = DAO.geoNames.analyse(this.text, this.hashtags.toArray(new String[0]), 5, Integer.toString(this.text.hashCode()));
                    this.place_context = PlaceContext.ABOUT;
                    this.location_source = LocationSource.ANNOTATION;
                }
                if (loc != null) {
                    if (this.place_name == null || this.place_name.length() == 0) this.place_name = loc.getNames().iterator().next();
                    this.location_radius = 0;
                    this.location_point = new double[]{loc.lon(), loc.lat()}; //[longitude, latitude]
                    this.location_mark = new double[]{loc.mlon(), loc.mlat()}; //[longitude, latitude]
                    this.place_country = loc.getISO3166cc();
                }
            }
        }

        /**
         * create enriched data, useful for analytics and ranking:
         * - identify all mentioned users, hashtags and links
         * - count message size without links
         * - count message size without links and without users
         */
        public void enrich() {
            if (this.enriched) return;
            this.moreData.classifier = Classifier.classify(this.text);
        
            StringBuilder text = new StringBuilder(this.text);
            this.links = this.moreData.extractLinks(text.toString());
            text = new StringBuilder(MessageEntry.SPACEX_PATTERN.matcher(text).replaceAll(" ").trim());
            // Text's length without link
            this.without_l_len = text.length();

            this.hosts = this.moreData.extractHosts(links);

            this.videos = this.moreData.getLinksVideo(this.links, this.videos);
            this.images = this.moreData.getLinksImage(this.links, this.images);
            this.audios = this.moreData.getLinksAudio(this.links, this.audios);

            this.users = this.moreData.extractUsers(text.toString());
            text = new StringBuilder(MessageEntry.SPACEX_PATTERN.matcher(text).replaceAll(" ").trim());
            // Text's length without link and users
            this.without_lu_len = text.length();

            this.mentions = new ArrayList<String>();
            for (int i = 0; i < this.users.size(); i++) {
                this.mentions.add(this.users.get(i).substring(1));
            }

            this.hashtags = this.moreData.extractHashtags(text.toString());
            text = new StringBuilder(MessageEntry.SPACEX_PATTERN.matcher(text).replaceAll(" ").trim());
            // Text's length without link, users and hashtags
            this.without_luh_len = text.length();

            getLocation();
            this.enriched = true;
        }

        /**
         * Channels on which the Tweet will be published -
         *      all
         *      twitter
         *      twitter/mention/*username*
         *      twitter/user/*username*         (User who posted the Tweet)
         *      twitter/hashtag/*hashtag*
         *      twitter/country/*country code*
         *      twitter/text/*token*
         * @return Array of channels to publish message to
         */
        @Override
        protected String[] getStreamChannels() {
            ArrayList<String> channels = new ArrayList<>();

            for (String mention : this.mentions) {
                channels.add("twitter/mention/" + mention);
            }

            for (String hashtag : this.hashtags) {
                channels.add("twitter/hashtag/" + hashtag);
            }

            channels.add("twitter/user/" + this.getScreenName());
            if (this.place_country != null) {
                channels.add("twitter/country/" + this.place_country);
            }

            for (String token : Classifier.normalize(this.text)) {
                channels.add("twitter/text/" + token);
            }

            channels.add("all");
            channels.add("twitter");

            return channels.toArray(new String[channels.size()]);
        }

        @Override
        public void run() {
            //long start = System.currentTimeMillis();
            try {
                //DAO.log("TwitterTweet [" + this.postId + "] start");
                this.text = unshorten(this.text);
                this.user.setName(unshorten(this.user.getName()));
                //DAO.log("TwitterTweet [" + this.postId + "] unshorten after " + (System.currentTimeMillis() - start) + "ms");
                this.enrich();

                //DAO.log("TwitterTweet [" + this.postId + "] enrich    after " + (System.currentTimeMillis() - start) + "ms");
                if (this.writeToIndex) IncomingMessageBuffer.addScheduler(this, this.user, true, true);
                //DAO.log("TwitterTweet [" + this.postId + "] write     after " + (System.currentTimeMillis() - start) + "ms");
                if (this.writeToBackend) DAO.outgoingMessages.transmitMessage(this, this.user);
                //DAO.log("TwitterTweet [" + this.postId + "] transmit  after " + (System.currentTimeMillis() - start) + "ms");
            } catch (Throwable e) {
                DAO.severe(e);
            } finally {
                this.ready.release(1000);
            }
        }

        public boolean isReady() {
            if (this.ready == null) throw new RuntimeException("isReady() should not be called if postprocessing is not started");
            return this.ready.availablePermits() > 0;
        }

        public boolean waitReady(long millis) {
            if (this.ready == null) throw new RuntimeException("waitReady() should not be called if postprocessing is not started");
            if (this.ready.availablePermits() > 0) return true;
            try {
                return this.ready.tryAcquire(millis, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                return false;
            }
        }

        public Post toJSON() {
                // very important to include calculated data here because that is written
                // into the index using the abstract index factory
                return toJSON(null, true, Integer.MAX_VALUE, "");
            }

        public Post toJSON(final UserEntry user, final boolean calculatedData, final int iflinkexceedslength, final String urlstub) {

            // tweet data
            this.put(AbstractObjectEntry.TIMESTAMP_FIELDNAME, AbstractObjectEntry.utcFormatter.print(getTimestampDate().getTime()));
            this.put(AbstractObjectEntry.CREATED_AT_FIELDNAME, AbstractObjectEntry.utcFormatter.print(getCreatedAt().getTime()));
            if (this.on != null) this.put("on", AbstractObjectEntry.utcFormatter.print(this.on.getTime()));
            if (this.to != null) this.put("to", AbstractObjectEntry.utcFormatter.print(this.to.getTime()));
            this.put("screen_name", this.screen_name);
            if (this.retweet_from != null && this.retweet_from.length() > 0) this.put("retweet_from", this.retweet_from);
            // the tweet; the cleanup is a helper function which cleans mistakes from the past in scraping
            MessageEntry.TextLinkMap tlm = this.moreData.getText(iflinkexceedslength, urlstub, this.text, this.getLinks(), this.getPostId());
            this.put("text", tlm);
            if (this.status_id_url != null) this.put("link", this.status_id_url.toExternalForm()); // this is the primary key for retrieval in elasticsearch
            this.put("id_str", this.postId);
            this.put("conversation_id", this.conversationID);
            this.put("conversation_user", this.conversationUserIDs);
            if (this.canonical_id != null) this.put("canonical_id", this.canonical_id);
            if (this.parent != null) this.put("parent", this.parent);
            this.put("source_type", this.source_type.toString());
            this.put("provider_type", this.provider_type.name());
            if (this.provider_hash != null && this.provider_hash.length() > 0) this.put("provider_hash", this.provider_hash);
            this.put("reply_count", this.reply_count);
            this.put("retweet_count", this.retweet_count);
            // there is a slight inconsistency here in the plural naming but thats how it is noted in the twitter api
            this.put("favourites_count", this.favourite_count);
            this.put("place_name", this.place_name);
            this.put("place_id", this.place_id);

            // add statistic/calculated data
            if (calculatedData) {

                // text length
                this.put("text_length", this.text.length());

                // location data
                if (this.place_context != null) this.put("place_context", this.place_context.name());
                if (this.place_country != null && this.place_country.length() == 2) {
                    this.put("place_country", DAO.geoNames.getCountryName(this.place_country));
                    this.put("place_country_code", this.place_country);
                    this.put("place_country_center", DAO.geoNames.getCountryCenter(this.place_country));
                }

                // add optional location data. This is written even if calculatedData == false if
                // the source is from REPORT to prevent that it is lost
                if (this.location_point != null && this.location_point.length == 2
                        && this.location_mark != null && this.location_mark.length == 2) {
                    // reference for this format:
                    // https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping-geo-point-type.html#_lat_lon_as_array_5
                    this.put("location_point", this.location_point); // [longitude, latitude]
                    this.put("location_radius", this.location_radius);
                    this.put("location_mark", this.location_mark);
                    this.put("location_source", this.location_source.name());
                }

                // redundant data for enhanced navigation with aggregations
                this.put("hosts", this.hosts);
                this.put("hosts_count", this.hosts.size());
                this.put("links", this.links);
                this.put("links_count", this.links.size());
                this.put("unshorten", tlm.short2long);
                this.put("images", this.images);
                this.put("images_count", this.images.size());
                this.put("audio", this.audios);
                this.put("audio_count", this.audios.size());
                this.put("videos", this.videos);
                this.put("videos_count", this.videos.size());
                this.put("mentions", this.mentions);
                this.put("mentions_count", this.mentions.size());
                this.put("hashtags", this.hashtags);
                this.put("hashtags_count", this.hashtags.size());

                // experimental, for ranking
                this.put("without_l_len", this.without_l_len);
                this.put("without_lu_len", this.without_lu_len);
                this.put("without_luh_len", this.without_luh_len);

                 // text classifier
                if (this.moreData.classifier != null) {
                    for (Map.Entry<Context, Classification<String, Category>> c: this.moreData.classifier.entrySet()) {
                        assert c.getValue() != null;
                        // we don't store non-existing classifications
                        if (c.getValue().getCategory() == Classifier.Category.NONE) continue;
                        this.put("classifier_" + c.getKey().name(), c.getValue().getCategory());
                        this.put("classifier_" + c.getKey().name() + "_probability",
                                c.getValue().getProbability() == Float.POSITIVE_INFINITY
                                        ? Float.MAX_VALUE : c.getValue().getProbability());
                    }
                }
            } else {
                this.remove("text_length");
                this.remove("place_context");
                this.remove("place_country");
                this.remove("place_country_code");
                this.remove("place_country_center");
                this.remove("location_point");
                this.remove("location_radius");
                this.remove("location_mark");
                this.remove("location_source");
                this.remove("hosts");
                this.remove("hosts_count");
                this.remove("links");
                this.remove("links_count");
                this.remove("unshorten");
                this.remove("images");
                this.remove("images_count");
                this.remove("audio");
                this.remove("audio_count");
                this.remove("videos");
                this.remove("videos_count");
                this.remove("mentions");
                this.remove("mentions_count");
                this.remove("hashtags");
                this.remove("hashtags_count");
                this.remove("without_l_len");
                this.remove("without_lu_len");
                this.remove("without_luh_len");
                if (this.moreData.classifier != null) {
                    for (Map.Entry<Context, Classification<String, Category>> c: this.moreData.classifier.entrySet()) {
                        this.remove("classifier_" + c.getKey().name());
                        this.remove("classifier_" + c.getKey().name() + "_probability");
                    }
                }
            }

            // add user
            if (user != null) this.put("user", user.toJSON());
            return this;
        }

        public boolean willBeTimeConsuming() {
            return timeline_link_pattern.matcher(this.text).find();
        }

       public Object lazyGet(JSONObject json, String key) {
            try {
                Object o = json.get(key);
                return o;
            } catch (JSONException e) {
                return null;
            }
        }

        public UserEntry getUser() {
            return this.user;
        }

        public Date getTimestampDate() {
            return this.timestampDate == null ? new Date() : this.timestampDate;
        }

        public Date getCreatedAt() {
            return this.created_at == null ? new Date() : this.created_at;
        }

        public void setCreatedAt(Date created_at) {
            this.created_at = created_at;
        }

        public Date getOn() {
            return this.on;
        }

        public void setOn(Date on) {
            this.on = on;
        }

        public Date getTo() {
            return this.to;
        }

        public void setTo(Date to) {
            this.to = to;
        }

        public SourceType getSourceType() {
            return this.source_type;
        }

        public void setSourceType(SourceType source_type) {
            this.source_type = source_type;
        }

        public ProviderType getProviderType() {
            return provider_type;
        }

        public void setProviderType(ProviderType provider_type) {
            this.provider_type = provider_type;
        }

        public String getProviderHash() {
            return provider_hash;
        }

        public void setProviderHash(String provider_hash) {
            this.provider_hash = provider_hash;
        }

        public String getScreenName() {
            return screen_name;
        }

        public void setScreenName(String user_screen_name) {
            this.screen_name = user_screen_name;
        }

        public String getRetweetFrom() {
            return this.retweet_from;
        }

        public void setRetweetFrom(String retweet_from) {
            this.retweet_from = retweet_from;
        }

        public URL getStatusIdUrl() {
            return this.status_id_url;
        }

        public void setStatusIdUrl(URL status_id_url) {
            this.status_id_url = status_id_url;
        }

        public long getRetweetCount() {
            return retweet_count;
        }

        public void setRetweetCount(long retweet_count) {
            this.retweet_count = retweet_count;
        }

        public long getFavouritesCount() {
            return this.favourite_count;
        }

        public void setFavouritesCount(long favourites_count) {
            this.favourite_count = favourites_count;
        }

        public String getPlaceName() {
            return place_name;
        }

        public void setPlaceName(String place_name, PlaceContext place_context) {
            this.place_name = place_name;
            this.place_context = place_context;
        }

        public PlaceContext getPlaceContext () {
            return place_context;
        }

        public String getPlaceId() {
            return place_id;
        }

        public void setPlaceId(String place_id) {
            this.place_id = place_id;
        }

        /**
         * @return [longitude, latitude]
         */
        public double[] getLocationPoint() {
            return location_point;
        }

        /**
         * set the location
         * @param location_point in the form [longitude, latitude]
         */
        public void setLocationPoint(double[] location_point) {
            this.location_point = location_point;
        }

        public String getPostId() {
            return String.valueOf(this.postId);
        }

        //TODO: to implement this method
        private void setPostId() {
            this.postId = String.valueOf(this.timestamp) + String.valueOf(this.created_at.getTime());
        }

        /**
         * @return [longitude, latitude] which is inside of getLocationRadius() from getLocationPoint()
         */
        public double[] getLocationMark() {
            return location_mark;
        }

        /**
         * Set the location
         * @param location_point in the form [longitude, latitude]
         */
        public void setLocationMark(double[] location_mark) {
            this.location_mark = location_mark;
        }

        /**
         * Get the radius in meter
         * @return radius in meter around getLocationPoint() (NOT getLocationMark())
         */
        public int getLocationRadius() {
            return location_radius;
        }

        public void setLocationRadius(int location_radius) {
            this.location_radius = location_radius;
        }

        public LocationSource getLocationSource() {
            return location_source;
        }

        public void setLocationSource(LocationSource location_source) {
            this.location_source = location_source;
        }

        public String getText() {
            return this.text;
        }

        public void setText(String text) {
            this.text = text;
        }

        public int getTextLength() {
            return this.text.length();
        }

        public long getId() {
            return Long.parseLong(this.postId);
        }

        public List<String> getHosts() {
            return this.hosts;
        }

        public Set<String> getVideos() {
            return this.videos;
        }

        public Set<String> getAudio() {
            return this.audios;
        }

        public Set<String> getImages() {
            return this.images;
        }

        public void setImages(String image) {
            if(this.images == null) {
                this.images = new HashSet<String>();
            }
            this.images.add(image);
        }

        public String[] getMentions() {
            if(this.mentions == null) {
                return new String[0];
            }
            return this.mentions.toArray(new String[0]);
        }

        public String[] getHashtags() {
            return this.hashtags.toArray(new String[0]);
        }

        public String[] getLinks() {
            return this.links.toArray(new String[0]);
        }

        public Classifier.Category getClassifier(Classifier.Context context) {
            return this.moreData.getClassifier(context);
        }

    }

    public static String unshorten(String text) {
        while (true) {
            try {
                Matcher m = emoji_pattern.matcher(text);
                if (m.find()) {
                    String emoji = m.group(1);
                    text = m.replaceFirst(emoji);
                    continue;
                }
            } catch (Throwable e) {
                DAO.severe(e);
                break;
            }
            try {
                Matcher m = emoji_pattern_span.matcher(text);
                if (m.find()) {
                    String emoji = m.group(1);
                    text = m.replaceFirst(emoji);
                    continue;
                }
            } catch (Throwable e) {
                DAO.severe(e);
                break;
            }
            try {
                Matcher m = hashtag_pattern.matcher(text);
                if (m.find()) {
                    text = m.replaceFirst(" #" + m.group(1) + " "); // the extra spaces are needed because twitter removes them if the hashtag is followed with a link
                    continue;
                }
            } catch (Throwable e) {
                DAO.severe(e);
                break;
            }
            try {
                Matcher m = timeline_link_pattern.matcher(text);
                if (m.find()) {
                    String expanded = RedirectUnshortener.unShorten(m.group(2));
                    text = m.replaceFirst(" " + expanded);
                    continue;
                }
            } catch (Throwable e) {
                DAO.severe(e);
                break;
            }
            try {
                Matcher m = timeline_embed_pattern.matcher(text);
                if (m.find()) {
                    text = m.replaceFirst("");
                    continue;
                }
            } catch (Throwable e) {
                DAO.severe(e);
                break;
            }
            break;
        }
        text = cleanup_pattern.matcher(text).replaceAll("");
        text = MessageEntry.html2utf8(text);
        text = doublespace_pattern.matcher(text).replaceAll(" ");
        text = text.trim();
        return text;
    }

    /**
     * Usage: java twitter4j.examples.search.SearchTweets [query]
     *
     * @param args search query
     */
    public static void main(String[] args) {
        //wget --no-check-certificate "https://twitter.com/search?q=eifel&src=typd&f=realtime"
        Set<String> filterList = new HashSet<String>();
        filterList.add("image");
        TwitterTimeline[] result = null;
        result = search(args[0], filterList, Order.CREATED_AT, false, false);
        int all = 0;
        for (int x = 0; x < 2; x++) {
            if (x == 0) System.out.println("Timeline[0] - finished to be used:");
            if (x == 1) System.out.println("Timeline[1] - messages which are in postprocessing");
            all += result[x].size();
            for (TwitterTweet tweet : result[x]) {
                    tweet.waitReady(10000);
                System.out.println(tweet.getCreatedAt().toString() + " from @" + tweet.getScreenName() + " - " + tweet.getText());
            }
        }
        System.out.println("count: " + all);
        System.exit(0);
    }

} public class FileHandler extends ResourceHandler implements Handler {
    
    private final long CACHE_LIMIT = 128L * 1024L;
    
    /**
     * cerate a custom ResourceHandler with more caching
     * @param expiresSeconds the time each file shall stay in the cache
     */
    public FileHandler(int expiresSeconds) {
        //this.setMinMemoryMappedContentLength((int) CACHE_LIMIT);
    }
    
    @Override
    public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException {
        // use the ResourceHandler to handle the request. This method calls doResponseHeaders internally
        super.handle(target, baseRequest, request, response);
    }
    
    public static void setCaching(final HttpServletResponse response, final int expiresSeconds) {
        if (response instanceof org.eclipse.jetty.server.Response) {
            org.eclipse.jetty.server.Response r = (org.eclipse.jetty.server.Response) response;
            HttpFields fields = r.getHttpFields();
            
            // remove the last-modified field since caching otherwise does not work
            /*
               https://www.ietf.org/rfc/rfc2616.txt
               "if the response does have a Last-Modified time, the heuristic
               expiration value SHOULD be no more than some fraction of the interval
               since that time. A typical setting of this fraction might be 10%."
            */
            fields.remove(HttpHeader.LAST_MODIFIED); // if this field is present, the reload-time is a 10% fraction of ttl and other caching headers do not work

            // cache-control: allow shared caching (i.e. proxies) and set expires age for cache
            if(expiresSeconds == 0){
                fields.put(HttpHeader.CACHE_CONTROL, "public, no-store, max-age=" + Integer.toString(expiresSeconds)); // seconds
            }
            else {
                fields.put(HttpHeader.CACHE_CONTROL, "public, max-age=" + Integer.toString(expiresSeconds)); // seconds
            }
        } else {
            response.setHeader(HttpHeader.LAST_MODIFIED.asString(), ""); // not really the best wqy to remove this header but there is no other option
            if(expiresSeconds == 0){
                response.setHeader(HttpHeader.CACHE_CONTROL.asString(), "public, no-store, max-age=" + Integer.toString(expiresSeconds));
            }
            else{
                response.setHeader(HttpHeader.CACHE_CONTROL.asString(), "public, max-age=" + Integer.toString(expiresSeconds));
            }

        }

        // expires: define how long the file shall stay in a cache if cache-control is not used for this information
        response.setDateHeader(HttpHeader.EXPIRES.asString(), System.currentTimeMillis() + expiresSeconds * 1000);
    }

    @Override
    public Resource getResource(String path) {
        Resource resource = super.getResource(path);
        if (resource == null) return null;
        if (!(resource instanceof PathResource) || !resource.exists()) return resource;
        try {
            File f = resource.getFile();
            if (f.isDirectory() && !path.equals("/")) return resource;
            CacheResource cache = resourceCache.get(f);
            if (cache != null) return cache;
            if (f.length() < CACHE_LIMIT || f.getName().endsWith(".html") || path.equals("/")) {
                cache = new CacheResource((PathResource) resource);
                resourceCache.put(f, cache);
                return cache;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return resource;
    }

    private final Map<File, CacheResource> resourceCache = new ConcurrentHashMap<>();
    private final static byte[] SSI_START = "<!--#include file=\"".getBytes();
    private final static byte[] SSI_END   = "\" -->".getBytes();
    
    private static class CacheResource extends Resource {

        private byte[] buffer;
        private long lastModified;
        private File file;
        private List<File> includes;
        
        public CacheResource(PathResource pathResource) throws IOException {
            this.file = pathResource.getFile();
            if (this.file.isDirectory()) this.file = new File(this.file, "index.html");
            this.includes = new ArrayList<>(8);
            initCache(System.currentTimeMillis());
            pathResource.close();
            
        }
        
        private void initCache(long nextLastModified) throws IOException {
            this.buffer = Files.readAllBytes(this.file.toPath());
            if (this.file.getName().endsWith(".html")) this.buffer = insertSSI(this.buffer);
            this.lastModified = nextLastModified;
        }
        
        private byte[] insertSSI(byte[] b) throws IOException {
            this.includes.clear();
            for (int p = findSSI_start(b, 0); p >= 0; p = findSSI_start(b, p)) {
                int q = findSSI_end(b, p);
                if (q < 0) break;
                byte[] f = new byte[q - p - SSI_START.length];
                System.arraycopy(b, p + SSI_START.length, f, 0, f.length);
                File ff = new File(this.file.getParent(), UTF8.String(f)).getCanonicalFile();
                if (!ff.exists()) {
                    byte[] b0 = new byte[b.length - (q - p) - SSI_END.length];
                    System.arraycopy(b, 0, b0, 0, p);
                    System.arraycopy(b, q + SSI_END.length, b0, p, b.length - q - SSI_END.length);
                    b = b0;
                    continue;
                }
                this.includes.add(ff);
                byte[] i = Files.readAllBytes(ff.toPath());
                byte[] b0 = new byte[b.length - (q - p) - SSI_END.length + i.length];
                System.arraycopy(b, 0, b0, 0, p);
                System.arraycopy(i, 0, b0, p, i.length);
                System.arraycopy(b, q + SSI_END.length, b0, p + i.length, b.length - q - SSI_END.length);
                b = b0;
            }
            return b;
        }

        private int findSSI_start(byte[] b, int p) {
            return ByteBuffer.indexOf(b, SSI_START, p);
        }
        
        private int findSSI_end(byte[] b, int p) {
            return ByteBuffer.indexOf(b, SSI_END, p);
        }

        @Override
        public boolean exists() {
            return true;
        }

        @Override
        public boolean isDirectory() {
            return false;
        }

        private long actualLastModified() {
            long l = this.file.lastModified();
            for (File d: this.includes) l = Math.max(l, d.lastModified());
            return l;
        }
        
        @Override
        public long lastModified() {
            long l = actualLastModified();
            if (actualLastModified() > this.lastModified) try {initCache(l);} catch (IOException e) {}
            this.lastModified = l;
            return this.lastModified;
        }

        @Override
        public long length() {
            return this.buffer.length;
        }
        
        @Override public File getFile() throws IOException {
            return null; // if this returns null, the caller will call getInputStream();
        }
        
        @Override
        public InputStream getInputStream() throws IOException {
            long l = actualLastModified();
            if (actualLastModified() > this.lastModified) initCache(l);
            return new ByteArrayInputStream(this.buffer);
        }

        @Override
        public ReadableByteChannel getReadableByteChannel() throws IOException {
            return Channels.newChannel(this.getInputStream());
        }

        @Override
        public String getName() {
            return this.file.getName();
        }

        @Override public void close() {}
        @Override public String[] list() {throw new UnsupportedOperationException();}
        @Override public boolean delete() throws SecurityException {throw new UnsupportedOperationException();}
        @Override public Resource addPath(String arg0) throws IOException, MalformedURLException {throw new UnsupportedOperationException();}
        @Override public URL getURL() {throw new UnsupportedOperationException();}
        @Override public boolean isContainedIn(Resource arg0) throws MalformedURLException {throw new UnsupportedOperationException();}
        @Override public boolean renameTo(Resource arg0) throws SecurityException {throw new UnsupportedOperationException();}
    }
    
    
}
import java.io.*;
import java.net.ServerSocket;
import java.nio.file.FileSystems;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.security.KeyStore;
import java.security.PrivateKey;
import java.security.Security;
import java.security.cert.Certificate;
import java.security.cert.X509Certificate;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.jar.JarEntry;
import java.util.jar.JarFile;
import java.util.regex.Pattern;
import javax.servlet.MultipartConfigElement;
import javax.servlet.Servlet;
import org.apache.logging.log4j.LogManager;
import org.bouncycastle.asn1.pkcs.PrivateKeyInfo;
import org.bouncycastle.cert.X509CertificateHolder;
import org.bouncycastle.cert.jcajce.JcaX509CertificateConverter;
import org.bouncycastle.openssl.PEMParser;
import org.bouncycastle.openssl.jcajce.JcaPEMKeyConverter;
import org.eclipse.jetty.rewrite.handler.RewriteHandler;
import org.eclipse.jetty.rewrite.handler.RewriteRegexRule;
import org.eclipse.jetty.server.Handler;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;
import org.eclipse.jetty.server.SessionIdManager;
import org.eclipse.jetty.server.handler.DefaultHandler;
import org.eclipse.jetty.server.handler.ErrorHandler;
import org.eclipse.jetty.server.handler.HandlerList;
import org.eclipse.jetty.server.handler.IPAccessHandler;
import org.eclipse.jetty.servlet.ServletContextHandler;
import org.eclipse.jetty.servlet.ServletHolder;
import org.eclipse.jetty.server.handler.gzip.GzipHandler;
import org.eclipse.jetty.server.session.DefaultSessionIdManager;
import org.eclipse.jetty.server.session.SessionHandler;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.eclipse.jetty.util.security.Constraint;
import org.eclipse.jetty.webapp.WebAppContext;
import org.eclipse.jetty.server.HttpConfiguration;
import org.eclipse.jetty.server.SecureRequestCustomizer;
import org.eclipse.jetty.server.HttpConnectionFactory;
import org.eclipse.jetty.server.SslConnectionFactory;
import org.eclipse.jetty.security.ConstraintMapping;
import org.eclipse.jetty.security.ConstraintSecurityHandler;
import org.eclipse.jetty.security.authentication.BasicAuthenticator;
import org.loklak.api.admin.AccessServlet;
import org.loklak.api.admin.CampaignServlet;
import org.loklak.api.admin.CrawlerServlet;
import org.loklak.api.admin.LogServlet;
import org.loklak.api.admin.SettingsServlet;
import org.loklak.api.admin.StatusService;
import org.loklak.api.admin.ThreaddumpServlet;
import org.loklak.api.aggregation.ClassifierServlet;
import org.loklak.api.amazon.AmazonProductService;
import org.loklak.api.cms.*;
import org.loklak.api.geo.GeocodeServlet;
import org.loklak.api.iot.FossasiaPushServlet;
import org.loklak.api.iot.FreifunkNodePushServlet;
import org.loklak.api.iot.FreifunkNodeFetchServlet;
import org.loklak.api.iot.NMEAServlet;
import org.loklak.api.iot.NOAAAlertServlet;
import org.loklak.api.iot.GeoJsonPushServlet;
import org.loklak.api.iot.ImportProfileServlet;
import org.loklak.api.iot.NetmonPushServlet;
import org.loklak.api.iot.NodelistPushServlet;
import org.loklak.api.iot.OpenWifiMapPushServlet;
import org.loklak.api.iot.StuffInSpaceServlet;
import org.loklak.api.iot.ValidateServlet;
import org.loklak.api.iot.YahiHazeServlet;
import org.loklak.api.iot.EarthquakeServlet;
import org.loklak.api.p2p.HelloService;
import org.loklak.api.p2p.PeersServlet;
import org.loklak.api.p2p.PushServlet;
import org.loklak.api.search.EventBriteCrawlerService;
import org.loklak.api.search.GenericScraper;
import org.loklak.api.search.GithubProfileScraper;
import org.loklak.api.search.InstagramProfileScraper;
import org.loklak.api.search.LocationWiseTimeService;
import org.loklak.api.search.MeetupsCrawlerService;
import org.loklak.api.search.QuoraProfileScraper;
import org.loklak.api.search.RSSReaderService;
import org.loklak.api.search.SearchServlet;
import org.loklak.api.search.ShortlinkFromTweetServlet;
import org.loklak.api.search.SuggestServlet;
import org.loklak.api.search.TimeAndDateService;
import org.loklak.api.search.TweetScraper;
import org.loklak.api.search.UserServlet;
import org.loklak.api.search.WeiboUserInfo;
import org.loklak.api.search.WikiGeoData;
import org.loklak.api.search.WordpressCrawlerService;
import org.loklak.api.tools.CSVServlet;
import org.loklak.api.tools.XMLServlet;
import org.loklak.api.vis.MapServlet;
import org.loklak.api.vis.MarkdownServlet;
import org.loklak.api.vis.PieChartServlet;
import org.loklak.data.DAO;
import org.loklak.data.IncomingMessageBuffer;
import org.loklak.harvester.TwitterHarvester;
import org.loklak.harvester.TwitterScraper;
import org.loklak.harvester.YoutubeScraper;
import org.loklak.http.RemoteAccess;
import org.loklak.server.APIHandler;
import org.loklak.server.FileHandler;
import org.loklak.server.HttpsMode;
import org.loklak.stream.StreamServlet;
import org.loklak.tools.Browser;
import org.loklak.tools.OS;
                LoginService.class,
                PasswordRecoveryService.class,
                PasswordResetService.class,
                PublicKeyRegistrationService.class,
                SignUpService.class,
                TopMenuService.class,
                UserManagementService.class,
                TwitterAnalysisService.class,
                AmazonProductService.class,
                UserAccountPermissions.class,
                SettingsManagementService.class,

                // geo

                // iot

                // p2p
                HelloService.class,

                // search
                EventBriteCrawlerService.class,
                MeetupsCrawlerService.class,
                RSSReaderService.class,
                WordpressCrawlerService.class,
                GithubProfileScraper.class,
                InstagramProfileScraper.class,
                LocationWiseTimeService.class,
                TimeAndDateService.class,
                WikiGeoData.class,
                QuoraProfileScraper.class,
                YoutubeScraper.class,
                TweetScraper.class
                // tools

                // vis

        };
        for (Class<? extends Servlet> service: services)
            try {
                servletHandler.addServlet(service, ((APIHandler) (service.newInstance())).getAPIPath());
            } catch (InstantiationException | IllegalAccessException e) {
                DAO.severe(service.getName() + " instantiation error", e);
                e.printStackTrace();
            }

        // add servlets
        servletHandler.addServlet(DumpDownloadServlet.class, "/dump/*");
        servletHandler.addServlet(ShortlinkFromTweetServlet.class, "/x");
        servletHandler.addServlet(AccessServlet.class, "/api/access.json");
        servletHandler.addServlet(AccessServlet.class, "/api/access.html");
        servletHandler.addServlet(AccessServlet.class, "/api/access.txt");
        servletHandler.addServlet(ClassifierServlet.class, "/api/classifier.json");
        servletHandler.addServlet(PeersServlet.class, "/api/peers.json");
        servletHandler.addServlet(PeersServlet.class, "/api/peers.csv");
        servletHandler.addServlet(CrawlerServlet.class, "/api/crawler.json");
        servletHandler.addServlet(SearchServlet.class, "/api/search.rss");
        servletHandler.addServlet(SearchServlet.class, "/api/search.json");
        servletHandler.addServlet(SearchServlet.class, "/api/search.txt");
        servletHandler.addServlet(SuggestServlet.class, "/api/suggest.json");
        servletHandler.addServlet(XMLServlet.class, "/api/xml2json.json");
        servletHandler.addServlet(CSVServlet.class, "/api/csv2json.json");
        ServletHolder accountServletHolder = new ServletHolder(AccountService.class);
        accountServletHolder.getRegistration().setMultipartConfig(multipartConfig);
        servletHandler.addServlet(accountServletHolder, "/api/account.json");
        servletHandler.addServlet(UserServlet.class, "/api/user.json");
        servletHandler.addServlet(CampaignServlet.class, "/api/campaign.json");
        servletHandler.addServlet(ImportProfileServlet.class, "/api/import.json");
        servletHandler.addServlet(SettingsServlet.class, "/api/settings.json");
        servletHandler.addServlet(GeocodeServlet.class, "/api/geocode.json");
        servletHandler.addServlet(ProxyServlet.class, "/api/proxy.gif");
        servletHandler.addServlet(ProxyServlet.class, "/api/proxy.png");
        servletHandler.addServlet(ProxyServlet.class, "/api/proxy.jpg");
        servletHandler.addServlet(ValidateServlet.class, "/api/validate.json");
        servletHandler.addServlet(GenericScraper.class, "/api/genericscraper.json");
        servletHandler.addServlet(WeiboUserInfo.class, "/api/weibo.json");
        ServletHolder pushServletHolder = new ServletHolder(PushServlet.class);
        pushServletHolder.getRegistration().setMultipartConfig(multipartConfig);
        servletHandler.addServlet(pushServletHolder, "/api/push.json");
        ServletHolder geojsonPushServletHolder = new ServletHolder(GeoJsonPushServlet.class);
        geojsonPushServletHolder.getRegistration().setMultipartConfig(multipartConfig);
        servletHandler.addServlet(geojsonPushServletHolder, "/api/push/geojson.json");
        servletHandler.addServlet(FossasiaPushServlet.class, "/api/push/fossasia.json");
        servletHandler.addServlet(OpenWifiMapPushServlet.class, "/api/push/openwifimap.json");
        servletHandler.addServlet(NodelistPushServlet.class, "/api/push/nodelist.json");
        servletHandler.addServlet(FreifunkNodePushServlet.class, "/api/push/freifunknode.json");
        servletHandler.addServlet(FreifunkNodeFetchServlet.class, "/api/freifunkfetch.json");
        servletHandler.addServlet(NetmonPushServlet.class, "/api/push/netmon.xml");
        servletHandler.addServlet(NMEAServlet.class, "/api/nmea.txt");
        servletHandler.addServlet(NOAAAlertServlet.class, "/api/noaa.json");
        servletHandler.addServlet(StuffInSpaceServlet.class, "/api/stuffinspace.json");
        servletHandler.addServlet(YahiHazeServlet.class, "/api/yahi.json");
        servletHandler.addServlet(EarthquakeServlet.class, "/api/earthquake.json");
        ServletHolder assetServletHolder = new ServletHolder(AssetServlet.class);
        assetServletHolder.getRegistration().setMultipartConfig(multipartConfig);
        servletHandler.addServlet(assetServletHolder, "/api/asset");
        servletHandler.addServlet(Sitemap.class, "/api/sitemap.xml");
        servletHandler.addServlet(ThreaddumpServlet.class, "/api/threaddump.txt");
        servletHandler.addServlet(LogServlet.class, "/api/log.txt");
        servletHandler.addServlet(MarkdownServlet.class, "/vis/markdown.gif");
        servletHandler.addServlet(MarkdownServlet.class, "/vis/markdown.gif.base64");
        servletHandler.addServlet(MarkdownServlet.class, "/vis/markdown.png");
        servletHandler.addServlet(MarkdownServlet.class, "/vis/markdown.png.base64");
        servletHandler.addServlet(MarkdownServlet.class, "/vis/markdown.jpg");
        servletHandler.addServlet(MarkdownServlet.class, "/vis/markdown.jpg.base64");
        servletHandler.addServlet(MapServlet.class, "/vis/map.gif");
        servletHandler.addServlet(MapServlet.class, "/vis/map.gif.base64");
        servletHandler.addServlet(MapServlet.class, "/vis/map.png");
        servletHandler.addServlet(MapServlet.class, "/vis/map.png.base64");
        servletHandler.addServlet(MapServlet.class, "/vis/map.jpg");
        servletHandler.addServlet(MapServlet.class, "/vis/map.jpg.base64");
        servletHandler.addServlet(PieChartServlet.class, "/vis/piechart.png");
        if (DAO.streamEnabled) {
            servletHandler.addServlet(StreamServlet.class, "/api/stream.json");
        }
        servletHandler.setMaxFormContentSize(10 * 1024 * 1024); // 10 MB

        ErrorHandler errorHandler = new ErrorHandler();
        errorHandler.setShowStacks(true);
        servletHandler.setErrorHandler(errorHandler);

        FileHandler fileHandler = new FileHandler(Integer.parseInt(DAO.getConfig("www.expires","600")));