dbsearchindexer.java

Jive Forums 1.0 src

                    long nextIndex = lastIndexed + updateInterval;
                    if (now > nextIndex) {
                        synchronized(indexLock) {
                            updateIndex(lastIndexed, now);
                            lastIndexed = now;
                            //Save the time as a Jive property.
                            PropertyManager.setProperty("DbSearchIndexer.lastIndexed",
                                "" + lastIndexed);
                        }
                    }
                }
            }
            //sleep for 1 minute and then check again.
            try {
                this.sleep(60000);
            }
            catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * Returns the wordID of the passed-in word. It will check the local word
     * cache for the ID first. If that fails, it will use the connection to
     * look up the ID. The connection is assumed to be open when passed in
     * and will remain open after the method is done executing.
     */
    private int getWordID(String word, Connection con) {
        int wordID = -1;
        //look for word in cache
        if (wordCache.containsKey(word)) {
            wordID = ((Integer)wordCache.get(word)).intValue();
        }
        else {
            PreparedStatement pstmt = null;
            try {
                pstmt = con.prepareStatement(FIND_WORD_ID);
                pstmt.setString(1, word);
                ResultSet rs = pstmt.executeQuery();
                //word is already in db, save the id to cache and return it
                if (rs.next()) {
                    wordID = rs.getInt(1);
                }
                //word is not in db, insert it then put it in cache
                else {
                    try {  pstmt.close(); }
                    catch (Exception e) { e.printStackTrace(); }
                    int nextWordID = DbSequenceManager.nextID("WordID");
                    pstmt = con.prepareStatement(INSERT_WORD);
                    pstmt.setInt(1, nextWordID);
                    pstmt.setString(2, word);
                    pstmt.executeUpdate();
                    wordCache.put(word, new Integer(nextWordID));
                    wordID = nextWordID;
                }
            }
            catch( SQLException sqle ) {
                sqle.printStackTrace();
            }
            finally {
                try {  pstmt.close(); }
                catch (Exception e) { e.printStackTrace(); }
            }
        }
        return wordID;
    }

    /**
     * Indexes an indivual message. The connection is assumed to be open when
     * passed in and will remain open after the method is done executing.
     */
    protected final void indexMessage(ForumMessage message, Connection con) {
        String [] subjectWords = StringUtils.toLowerCaseWordArray(message.getSubject());
        subjectWords = StringUtils.removeCommonWords(subjectWords);
        String [] bodyWords = StringUtils.toLowerCaseWordArray(message.getBody());
        bodyWords = StringUtils.removeCommonWords(bodyWords);

        Map words = new HashMap(subjectWords.length);
        Map wordCount = new HashMap(subjectWords.length);

        //Loop through all words in subject.
        for (int i=0; i<subjectWords.length; i++) {
            String currentWord = subjectWords[i];
            boolean exists = (words.get(currentWord) != null);
            //Check if we already have an entry for this word in the cache
            if (exists) {
                //increment the count since we found the word again
                int count = ((Integer)wordCount.get(currentWord)).intValue();
                count+=2; //count words in subject doubly
                wordCount.remove(currentWord);
                wordCount.put(currentWord, new Integer(count));
            }
            else {
                //first time we've found the word, store it into cache
                int count = 0;
                int wordID = getWordID(currentWord, con);
                words.put(currentWord, new Integer(wordID));
                wordCount.put(currentWord, new Integer(count));
            }
        }
        //Loop through all words in body.
        for (int i=0; i<bodyWords.length; i++) {
            String currentWord = bodyWords[i];
            boolean exists = (words.get(currentWord) != null);
            if (exists) {
                int count = ((Integer)wordCount.get(currentWord)).intValue();
                count++;
                wordCount.remove(currentWord);
                wordCount.put(currentWord, new Integer(count));
            }
            else {
                int count = 0;
                int wordID = getWordID(currentWord, con);
                words.put(currentWord, new Integer(wordID));
                wordCount.put(currentWord, new Integer(count));
            }
        }

        //Now add all the index entries into the database.
        PreparedStatement pstmt = null;
        int messageID = message.getID();
        int forumID = message.getForumThread().getForum().getID();
        try {
            pstmt = con.prepareStatement(INSERT_INDEX_RECORD);
            //insert all subject words
            Iterator iter = words.keySet().iterator();
            while(iter.hasNext()) {
                String word = (String)iter.next();
                int tempWordID = ((Integer)words.get(word)).intValue();
                int tempWordCount = ((Integer)wordCount.get(word)).intValue();
                pstmt.setInt(1, messageID);
                pstmt.setInt(2, tempWordID);
                pstmt.setInt(3, tempWordCount);
                pstmt.setInt(4, forumID);
                pstmt.execute();
            }
        }
        catch( SQLException sqle ) {
            System.err.println("Error in DbSearchIndexer:indexMessage-" + sqle);
            sqle.printStackTrace();
        }
        finally {
            try {  pstmt.close(); }
            catch (Exception e) { e.printStackTrace(); }
        }
    }

    /**
     * Rebuilds the search index from scratch. It deletes the entire index
     * and word tables and then indexes every message up to the end time.
     */
    protected final void rebuildIndex(long end) {
        Connection con = null;
        PreparedStatement pstmt = null;
        try {
            //For a clean rebuild we delete the whole index table and also
            //the word table.
            con = DbConnectionManager.getConnection();
            pstmt = con.prepareStatement(DELETE_INDEX1);
            pstmt.execute();
            pstmt.close();
            pstmt = con.prepareStatement(DELETE_INDEX2);
            pstmt.execute();
            pstmt.close();

            //Now, find all messages that need to inserted
            pstmt = con.prepareStatement(MESSAGES_BEFORE_DATE_COUNT);
            pstmt.setString(1, ""+end);
            ResultSet rs = pstmt.executeQuery();
            rs.next();
            int messageCount = rs.getInt(1);
            pstmt.close();

            int [] messages = new int[messageCount];
            pstmt = con.prepareStatement(MESSAGES_BEFORE_DATE);
            pstmt.setString(1, ""+end);
            rs = pstmt.executeQuery();
            for (int i=0; i<messages.length; i++) {
                rs.next();
                messages[i] = rs.getInt("messageID");
            }
            pstmt.close();

            //Finally, index all messages;
            for (int i=0; i<messages.length; i++) {
                ForumMessage message = factory.getMessage(messages[i]);
                indexMessage(message, con);
            }
        }
        catch( ForumMessageNotFoundException fmnfe) {
            fmnfe.printStackTrace();
        }
        catch( SQLException sqle ) {
            System.err.println("Error in DbSearchIndexer:updateIndex()-" + sqle);
            sqle.printStackTrace();
        }
        finally {
            try {  con.close();   }
            catch (Exception e) { e.printStackTrace(); }
        }
        //Clear out word cache until next operation
        wordCache.clear();

        wordCache = new HashMap();
    }

    /**
     * Updates the index. It first deletes any messages in the index between
     * the start and end times, and then adds all messages to the index that
     * are between the start and end times.
     */
    protected final void updateIndex(long start, long end) {
        Connection con = null;
        PreparedStatement pstmt = null;

        try {
            con = DbConnectionManager.getConnection();
            //For a clean update, we need to make sure that we first delete
            //and index entries that were made since we last updated. This
            //might happen if a process was calling indexMessage() between runs
            //of this method. For this reason, the two types of indexing should
            //not be intermixed. However, we still perform this deletion to be
            //safe.
            pstmt = con.prepareStatement(MESSAGES_SINCE_DATE_COUNT);
            pstmt.setString(1, ""+start);
            pstmt.setString(2, ""+end);
            ResultSet rs = pstmt.executeQuery();
            rs.next();
            int messageCount = rs.getInt(1);
            int [] messages = new int[messageCount];
            pstmt.close();
            pstmt = con.prepareStatement(MESSAGES_SINCE_DATE);
            pstmt.setString(1, ""+start);
            pstmt.setString(2, ""+end);
            rs = pstmt.executeQuery();
            for (int i=0; i<messages.length; i++) {
                rs.next();
                messages[i] = rs.getInt("messageID");
            }
            pstmt.close();
            pstmt = con.prepareStatement(DELETE_PARTIAL_INDEX);
            for (int i=0; i<messages.length; i++) {
                pstmt.setInt(1, messages[i]);
                pstmt.execute();
            }
            pstmt.close();

            //Finally, index all messages;
            for (int i=0; i<messages.length; i++) {
                ForumMessage message = factory.getMessage(messages[i]);
                indexMessage(message, con);
            }
        }
        catch( ForumMessageNotFoundException fmnfe) {
            fmnfe.printStackTrace();
        }
        catch( SQLException sqle ) {
            System.err.println("Error in DbSearchIndexer:updateIndex()-" + sqle);
            sqle.printStackTrace();
        }
        finally {
            try {  con.close();   }
            catch (Exception e) { e.printStackTrace(); }
        }

        //Clear out word cache until next operation

        wordCache.clear();

        wordCache = new HashMap();

    }
}