Project

General

Profile

1
/**
2
 *  '$RCSfile$'
3
 *    Purpose: A Class that tracks sessions for MetaCatServlet users.
4
 *  Copyright: 2000 Regents of the University of California and the
5
 *             National Center for Ecological Analysis and Synthesis
6
 *    Authors: Matt Jones
7
 *
8
 *   '$Author: tao $'
9
 *     '$Date: 2007-03-07 17:01:53 -0800 (Wed, 07 Mar 2007) $'
10
 * '$Revision: 3199 $'
11
 *
12
 * This program is free software; you can redistribute it and/or modify
13
 * it under the terms of the GNU General Public License as published by
14
 * the Free Software Foundation; either version 2 of the License, or
15
 * (at your option) any later version.
16
 *
17
 * This program is distributed in the hope that it will be useful,
18
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
19
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
20
 * GNU General Public License for more details.
21
 *
22
 * You should have received a copy of the GNU General Public License
23
 * along with this program; if not, write to the Free Software
24
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
25
 */
26

    
27
package edu.ucsb.nceas.metacat;
28

    
29
import java.sql.PreparedStatement;
30
import java.sql.ResultSet;
31
import java.sql.SQLException;
32
import java.util.Vector;
33
import java.util.HashMap;
34
import java.lang.Comparable;
35
import edu.ucsb.nceas.metacat.MetaCatUtil;
36
import org.apache.log4j.Logger;
37

    
38
public class IndexingQueue {
39

    
40
	private Logger logMetacat = Logger.getLogger(IndexingQueue.class);
41
	//	 Map used to keep tracks of docids to be indexed
42
	private HashMap indexingMap = new HashMap();     
43
	private Vector currentThreads = new Vector();
44
	public Vector currentDocidsBeingIndexed = new Vector();
45
	private boolean metacatRunning = true;
46
	
47
	private static IndexingQueue instance = null;
48

    
49
	final static int NUMBEROFINDEXINGTHREADS = 
50
		Integer.parseInt(MetaCatUtil.getOption("numberOfIndexingThreads"));
51
	private int sleepTime = 2000; 
52

    
53
    private IndexingQueue() {
54
	    for (int i = 0; i < NUMBEROFINDEXINGTHREADS; i++) {
55
	      Thread thread = new IndexingTask();
56
	      thread.start();
57
	      currentThreads.add(thread);
58
	    }
59
    }
60
	
61
	public static synchronized IndexingQueue getInstance(){
62
		if (instance == null) {
63
			instance = new IndexingQueue();
64
		}
65
		return instance;
66
	}//getInstance
67

    
68
    public void add(String docid, String rev) {
69
    	add(new IndexingQueueObject(docid, rev, 0));
70
    }
71
    
72
    protected void add(IndexingQueueObject queueObject) {
73
	    synchronized (indexingMap) {
74
	    	if(!indexingMap.containsKey(queueObject.getDocid())){
75
	    		indexingMap.put(queueObject.getDocid(), queueObject);
76
	    		indexingMap.notify();
77
	    	} else {
78
	    		IndexingQueueObject oldQueueObject = 
79
	    			(IndexingQueueObject) indexingMap.get(queueObject.getDocid());
80
	    		if(oldQueueObject.compareTo(queueObject) < 0){
81
	    	  		indexingMap.put(queueObject.getDocid(), queueObject);
82
		    		indexingMap.notify();  			
83
	    		}
84
	    	}
85
	    }
86
	  }
87
    
88
	public boolean getMetacatRunning(){
89
		return this.metacatRunning;
90
	}
91
	
92
	public void setMetacatRunning(boolean metacatRunning){
93
		this.metacatRunning = metacatRunning;
94
		
95
		if(!metacatRunning){
96
			for(int count=0; count<currentThreads.size(); count++){
97
				((IndexingTask)currentThreads.get(count)).metacatRunning = false;
98
				((Thread)currentThreads.get(count)).interrupt();
99
			}
100
		}
101
	}
102
	
103
    protected IndexingQueueObject getNext() {
104
    	IndexingQueueObject returnVal = null;
105
        synchronized (indexingMap) {
106
          while (indexingMap.isEmpty() && metacatRunning) {
107
            try {
108
            	indexingMap.wait();
109
            } catch (InterruptedException ex) {
110
              System.err.println("Interrupted");
111
            }
112
          }
113
          
114
          if(metacatRunning){
115
        	  String docid = (String) indexingMap.keySet().iterator().next();
116
        	  returnVal = (IndexingQueueObject)indexingMap.get(docid);
117
        	  indexingMap.remove(docid);
118
          }
119
        }
120
        return returnVal;
121
      }
122

    
123
}
124

    
125
class IndexingTask extends Thread {
126
  	  private Logger logMetacat = Logger.getLogger(IndexingTask.class);
127
      protected final long MAXIMUMINDEXDELAY = Integer.
128
      	parseInt(MetaCatUtil.getOption("maximumIndexDelay"));;
129
      protected boolean metacatRunning = true;
130
      	
131
	  public void run() {
132
	    while (metacatRunning) {
133
	      // blocks until job
134
	      IndexingQueueObject returnVal = 
135
	    	  IndexingQueue.getInstance().getNext();
136

    
137
	      if(returnVal != null){
138
	    	  if(!IndexingQueue.getInstance().
139
	    			currentDocidsBeingIndexed.contains(returnVal.getDocid())){
140
    		  try {
141
    			  IndexingQueue.getInstance().
142
    			  		currentDocidsBeingIndexed.add(returnVal.getDocid());
143
    		      String docid = returnVal.getDocid() + "." + returnVal.getRev();
144
    			  if(checkDocumentTable(docid, "xml_documents")){
145
    				  logMetacat.warn("Calling buildIndex for " + docid);
146
    				  DocumentImpl doc = new DocumentImpl(docid, false);
147
    				  doc.buildIndex();
148
    				  logMetacat.warn("finish building index for doicd "+docid);
149
    			  } else {
150
    				  logMetacat.warn("Couldn't find the docid:" + docid 
151
	                			+ " in xml_documents table");
152
	                  sleep(MAXIMUMINDEXDELAY);
153
	                  throw(new Exception("Couldn't find the docid:" + docid 
154
	                			+ " in xml_documents table"));
155
    			  }
156
    		  	} catch (Exception e) {
157
    			  logMetacat.warn("Exception: " + e);
158
    			  e.printStackTrace();
159
	        
160
    			  if(returnVal.getCount() < 25){
161
    				  returnVal.setCount(returnVal.getCount()+1);
162
    				  // add the docid back to the list
163
    				  IndexingQueue.getInstance().add(returnVal);
164
    			  } else {
165
    				  logMetacat.fatal("Docid " + returnVal.getDocid() 
166
	        			+ " has been inserted to IndexingQueue "
167
	        			+ "more than 25 times. Not adding the docid to"
168
	        			+ " the queue again.");
169
    			  }
170
    		  	} finally {
171
    			  	IndexingQueue.getInstance().currentDocidsBeingIndexed
172
    			  		.remove(returnVal.getDocid());	    	  
173
    		  	}
174
	    	  } else {
175
	    		  returnVal.setCount(returnVal.getCount()+1);
176
	    		  IndexingQueue.getInstance().add(returnVal);    		  
177
	    	  }
178
	      }
179
	    }
180
	  }
181
	  
182
      private boolean checkDocumentTable(String docid, String tablename) throws Exception{
183
	        DBConnection dbConn = null;
184
	        int serialNumber = -1;
185
	        boolean inxmldoc = false;
186
            
187
	        String revision = docid.substring(docid.lastIndexOf(".")+1,docid.length());
188
	        docid = docid.substring(0,docid.lastIndexOf("."));
189

    
190
	        logMetacat.info("Checking if document exsists in xml_documents: docid is " 
191
	        		+ docid + " and revision is " + revision);
192

    
193
	        try {
194
	            // Opening separate db connection for writing XML Index
195
	            dbConn = DBConnectionPool
196
	                    .getDBConnection("DBSAXHandler.checkDocumentTable");
197
	            serialNumber = dbConn.getCheckOutSerialNumber();
198

    
199
	            String xmlDocumentsCheck = "SELECT distinct docid FROM " + tablename
200
	                        + " WHERE docid ='" + docid
201
	                        + "' AND rev ='" + revision + "'";
202

    
203
                PreparedStatement xmlDocCheck = dbConn.prepareStatement(xmlDocumentsCheck);
204
                // Increase usage count
205
	            
206
                dbConn.increaseUsageCount(1);
207
	            xmlDocCheck.execute();
208
	            ResultSet doccheckRS = xmlDocCheck.getResultSet();
209
	            boolean tableHasRows = doccheckRS.next();
210
	            if (tableHasRows) {
211
	            	inxmldoc = true;
212
	            }
213
	            doccheckRS.close();
214
	            xmlDocCheck.close();
215
	        } catch (SQLException e) {
216
	        	   e.printStackTrace();
217
	        } finally {
218
	            DBConnectionPool.returnDBConnection(dbConn, serialNumber);
219
	        }//finally
220
	        
221
	        return inxmldoc;
222
      }
223
	}
224

    
225
class IndexingQueueObject implements Comparable{
226
	// the docid of the document to be indexed. 
227
	private String docid;
228
	// the docid of the document to be indexed. 
229
	private String rev;
230
	// the count of number of times the document has been in the queue
231
	private int count;
232
	
233
	IndexingQueueObject(String docid, String rev, int count){
234
		this.docid = docid;
235
		this.rev = rev;
236
		this.count = count;
237
	}
238
	
239
	public int getCount(){
240
		return count;
241
	}
242

    
243
	public String getDocid(){
244
		return docid;
245
	}
246

    
247
	public String getRev(){
248
		return rev;
249
	}
250

    
251
	public void setCount(int count){
252
		this.count = count;
253
	}
254
	
255
	public void setDocid(String docid){
256
		this.docid = docid;
257
	}
258

    
259
	public void setRev(String rev){
260
		this.rev = rev;
261
	}
262
	
263
	public int compareTo(Object o){
264
		if(o instanceof IndexingQueueObject){
265
			int revision = Integer.parseInt(rev);
266
			int oRevision = Integer.parseInt(((IndexingQueueObject)o).getRev());
267
			
268
			if(revision == oRevision) {
269
				return 0;
270
			} else if (revision > oRevision) {
271
				return 1;
272
			} else {
273
				return -1;
274
			}
275
		} else {
276
			throw new java.lang.ClassCastException();
277
		}
278
	}
279
}
(40-40/66)