Java > Open Source Codes > org > lateralnz > c3d > DatabaseEngine


1 /* ====================================================================
2  * The LateralNZ Software License, Version 1.0
3  *
4  * Copyright (c) 2003 LateralNZ. All rights reserved.
5  *
6  * Redistribution and use in source and binary forms, with or without
7  * modification, are permitted provided that the following conditions
8  * are met:
9  *
10  * 1. Redistributions of source code must retain the above copyright
11  * notice, this list of conditions and the following disclaimer.
12  *
13  * 2. Redistributions in binary form must reproduce the above copyright
14  * notice, this list of conditions and the following disclaimer in
15  * the documentation and/or other materials provided with the
16  * distribution.
17  *
18  * 3. The end-user documentation included with the redistribution,
19  * if any, must include the following acknowledgment:
20  * "This product includes software developed by
21  * LateralNZ (http://www.lateralnz.org/) and other third parties."
22  * Alternately, this acknowledgment may appear in the software itself,
23  * if and wherever such third-party acknowledgments normally appear.
24  *
25  * 4. The names "LateralNZ" must not be used to endorse or promote
26  * products derived from this software without prior written
27  * permission. For written permission, please
28  * contact oss@lateralnz.org.
29  *
30  * 5. Products derived from this software may not be called "Panther",
31  * or "Lateral" or "LateralNZ", nor may "PANTHER" or "LATERAL" or
32  * "LATERALNZ" appear in their name, without prior written
33  * permission of LateralNZ.
34  *
35  * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED
36  * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
37  * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
38  * DISCLAIMED. IN NO EVENT SHALL THE APACHE SOFTWARE FOUNDATION OR
39  * ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
40  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
41  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
42  * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
43  * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
44  * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
45  * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
46  * SUCH DAMAGE.
47  * ====================================================================
48  *
49  * This software consists of voluntary contributions made by many
50  * individuals on behalf of LateralNZ. For more
51  * information on Lateral, please see http://www.lateralnz.com/ or
52  * http://www.lateralnz.org
53  *
54  */
55 package org.lateralnz.c3d;
56
57 import java.io.IOException ;
58 import java.sql.CallableStatement ;
59 import java.sql.Connection ;
60 import java.sql.DatabaseMetaData ;
61 import java.sql.DriverManager ;
62 import java.sql.PreparedStatement ;
63 import java.sql.ResultSet ;
64 import java.sql.ResultSetMetaData ;
65 import java.sql.SQLException ;
66 import java.sql.Statement ;
67 import java.util.HashMap ;
68 import java.util.Iterator ;
69 import java.util.ArrayList ;
70 import java.util.List ;
71 import java.util.Properties ;
72 import java.util.StringTokenizer ;
73 import java.util.Timer ;
74 import java.util.TimerTask ;
75 import javax.naming.NamingException ;
76
77 import org.apache.log4j.Logger;
78
79 import org.lateralnz.common.util.Constants;
80 import org.lateralnz.common.util.DAOUtils;
81 import org.lateralnz.common.util.JNDIUtils;
82 import org.lateralnz.common.util.StringUtils;
83
84 import org.lateralnz.messaging.Message;
85 import org.lateralnz.messaging.MessageListener;
86 import org.lateralnz.messaging.MessageHandler;
87
88 import org.lateralnz.c3d.util.CacheOperation;
89 import org.lateralnz.c3d.util.Column;
90 import org.lateralnz.c3d.util.DBUtils;
91 import org.lateralnz.c3d.util.ResultWrapper;
92
93 /**
94  * core 'database' functionality -- basically this is the interceptor for all SQL
95  * statements so we can cache resultsets, refresh the cache and so on.
96  */
97 public class DatabaseEngine implements Constants, MessageListener {
98   private static final Logger log = Logger.getLogger(DatabaseEngine.class.getName());
99
100   private static final Class HASHMAP_CLASS = HashMap .class;
101   private static final Class ARRAYLIST_CLASS = ArrayList .class;
102   
103   private static final String CACHE_BEGIN_TAG = "cache(";
104   private static final String CACHE_END_TAG = ")";
105   private static final String CACHES = "caches";
106   private static final String DATABASE_CACHE = "database_cache_";
107   private static final String SELECT_FROM = "select * from ";
108   private static final String WHERE = " where ";
109   
110   private static final String RESULTS_TO_CACHE_LINK = "RESULTS_TO_CACHE_LINK";
111
112   private static HashMap dbengines = new HashMap ();
113   
114   private static MessageHandler mh = null;
115   
116   private String dbname; // the name of this database cache/bridge
117 private String messageGroup;
118   private Properties props; // properties of this databases
119
120   private Timer timer = new Timer ();
121   
122   private HashMap users = new HashMap ();
123   private HashMap datacache = new HashMap ();
124   private HashMap caches = new HashMap ();
125   private HashMap tableCheck = new HashMap (); // map of primary tables
126 private HashMap updateCheck = new HashMap (); // map of all monitored tables
127 private HashMap operationsAwaitingCommit = new HashMap ();
128   
129   
130   class CacheProperties {
131     boolean resetOnInsert = false;
132     String primaryTable;
133     ArrayList primaryKeys = new ArrayList ();
134     ArrayList deleteCheck = new ArrayList ();
135     long timeout = Long.MIN_VALUE;
136     
137     public String toString() {
138       return "resetinsert=" + resetOnInsert + ",primarykeys=" + primaryKeys.toString();
139     }
140   }
141   
142   class CacheCleaner extends TimerTask {
143     private String cacheName;
144     private DatabaseEngine dbengine;
145     private long last = 0;
146     private long period;
147     
148     CacheCleaner(DatabaseEngine dbengine, String cacheName, long period) {
149       super();
150       this.dbengine = dbengine;
151       this.cacheName = cacheName;
152       this.period = period;
153     }
154     
155     public boolean cancel() {
156       return false;
157     }
158     
159     public void run() {
160       last = System.currentTimeMillis();
161       try {
162         List l = dbengine.getCachedResultsetNames(cacheName);
163         Iterator iter = l.iterator();
164         while (iter.hasNext()) {
165           String sql = (String )iter.next();
166           DCResultSet dcrs = (DCResultSet)dbengine.getCachedResultSet(cacheName, sql);
167           if (last - dcrs.lastAccessed > period) {
168             if (log.isDebugEnabled()) {
169               log.debug("clearing timed out data (" + cacheName + ") " + sql);
170             }
171             dbengine.clearResultsByKey(cacheName, sql);
172           }
173         }
174       }
175       catch (Exception e) {
176         e.printStackTrace();
177         log.error(e);
178       }
179     }
180     
181     public long scheduledExecutionTime() {
182       return last;
183     }
184   }
185
186   
187   private DatabaseEngine(String dbname, Properties props) throws SQLException {
188     this.dbname = dbname;
189     this.messageGroup = DATABASE_CACHE + dbname;
190     this.props = props;
191     String tmp = getProperty(CACHES) + COMMA;
192     StringTokenizer st = new StringTokenizer (tmp, COMMA);
193     while (st.hasMoreTokens()) {
194       // cache name
195 String cache = st.nextToken();
196       CacheProperties cprops = new CacheProperties();
197       
198       // do we reset this cache when an insert statement is executed?
199 cprops.resetOnInsert = Boolean.valueOf(getProperty(cache + ".reset_on_insert")).booleanValue();
200       if (log.isInfoEnabled()) {
201         log.info(cache + " reset on insert is " + cprops.resetOnInsert);
202       }
203       cprops.primaryTable = getProperty(cache + ".primary_table");
204
205       // get the meta data for the primary table
206 // and add the primary keys into our watch list
207 Connection conn = DriverManager.getConnection(getProperty("dburl"), getProperty("dbuser"), getProperty("dbpassword"));
208       DatabaseMetaData dbmd = conn.getMetaData();
209       ResultSet rs = dbmd.getPrimaryKeys(EMPTY, EMPTY, cprops.primaryTable);
210       while (rs.next()) {
211         cprops.primaryKeys.add(rs.getString(4));
212       }
213       
214       ArrayList al = (ArrayList )DBUtils.getObjectFromMap(cprops.primaryTable, updateCheck, ARRAYLIST_CLASS);
215       al.add(cache);
216       al = (ArrayList )DBUtils.getObjectFromMap(cprops.primaryTable, tableCheck, ARRAYLIST_CLASS);
217       al.add(cache);
218       
219       StringUtils.toList(getProperty(cache + ".delete_check"), COMMA, cprops.deleteCheck);
220       Iterator iter = cprops.deleteCheck.iterator();
221       while (iter.hasNext()) {
222         String tab = (String )iter.next();
223         al = (ArrayList )DBUtils.getObjectFromMap(tab, tableCheck, ARRAYLIST_CLASS);
224         al.add(cache);
225       }
226       
227       if (log.isInfoEnabled()) {
228         log.info("delete check for " + cache + " is " + cprops.deleteCheck);
229       }
230       
231       String timeout = getProperty(cache + ".timeout");
232       if (!StringUtils.isEmpty(timeout)) {
233         // schedule regular cache cleanouts
234 cprops.timeout = 60000L * 60L * Integer.parseInt(timeout);
235
236         timer.schedule(new CacheCleaner(this, cache, cprops.timeout), cprops.timeout, cprops.timeout);
237       }
238       
239       caches.put(cache, cprops);
240       if (log.isInfoEnabled()) {
241         log.info("db(" + dbname + "), cache(" + cache + ") initialised with properties: " + cprops.toString());
242       }
243     }
244     
245     // user access to the database
246 String usersstr = props.getProperty("users");
247     StringUtils.toMap(users, usersstr, AMPERSAND);
248     
249     // messaging for cache changes
250 try {
251       String msgservice = getProperty("messaging_service_name");
252       if (!StringUtils.isEmpty(msgservice)) {
253         mh = (MessageHandler)JNDIUtils.get(msgservice);
254         mh.addListener(messageGroup, this);
255       }
256     }
257     catch (NamingException ne) {
258       throw new SQLException ("unable to get message service");
259     }
260   }
261   
262   public static final DatabaseEngine getInstance(String dbname, Properties props) throws SQLException {
263     if (!dbengines.containsKey(dbname)) {
264       synchronized (dbengines) {
265         if (!dbengines.containsKey(dbname)) {
266           if (props == null) {
267             throw new SQLException ("no such database " + dbname);
268           }
269           DatabaseEngine dbengine = new DatabaseEngine(dbname, props);
270           dbengines.put(dbname, dbengine);
271         }
272       }
273     }
274     
275     
276     return (DatabaseEngine)dbengines.get(dbname);
277   }
278   
279   private final void clearAll() {
280     if (log.isDebugEnabled()) {
281       log.debug("clearing all cached data");
282     }
283     synchronized (datacache) {
284       datacache.clear();
285     }
286     
287     Runtime.getRuntime().gc();
288   }
289   
290   private final void clearCache(String cacheName) {
291     if (log.isDebugEnabled()) {
292       log.debug("clearing data for " + cacheName);
293     }
294     
295     HashMap hm = (HashMap )datacache.get(cacheName);
296
297     if (hm != null) {
298       synchronized (hm) {
299         hm.clear();
300       }
301     }
302   }
303
304
305   private final void clearResultsByKey(String cache, String key) throws SQLException {
306     HashMap hm = (HashMap )DBUtils.getObjectFromMap(cache, datacache, HASHMAP_CLASS);
307     HashMap rtc = (HashMap )DBUtils.getObjectFromMap(RESULTS_TO_CACHE_LINK, hm, HASHMAP_CLASS);
308     
309     ArrayList ll = (ArrayList )rtc.get(key);
310     if (ll != null) {
311       synchronized (ll) {
312         Iterator iter = ll.iterator();
313         while (iter.hasNext()) {
314           String sql = (String )iter.next();
315           iter.remove();
316           
317           if (hm.remove(sql) != null) {
318             if (log.isDebugEnabled()) {
319               log.debug("remove data for cache " + cache + "/key " + sql + " succeeded");
320             }
321           }
322         }
323       }
324     }
325     rtc.remove(key);
326   }
327
328
329   protected void commit(DCConnection conn) throws SQLException {
330     ArrayList ll = (ArrayList )operationsAwaitingCommit.get(conn);
331     if (log.isDebugEnabled()) {
332       log.debug("committing " + conn + " with ops " + (ll != null ? Integer.toString(ll.size()) : "0"));
333     }
334     operationsAwaitingCommit.remove(conn);
335     
336     if (ll != null && ll.size() > 0) {
337       triggerCacheChanges(ll);
338
339       if (mh != null) {
340         try {
341           mh.send(new Message(1, messageGroup, ll));
342         }
343         catch (IOException ioe) {
344           log.error("commit error: " + ioe.getMessage(), ioe);
345         }
346       }
347     }
348     
349     if (log.isDebugEnabled()) {
350       log.debug(Integer.toString(operationsAwaitingCommit.size()) + " operations (after commit)");
351     }
352   }
353   
354   
355   protected ResultWrapper execute(DCStatement st) throws SQLException {
356     String sql = st.getSQL();
357     List l = DBUtils.splitSQL(sql);
358     if (DBUtils.countSelects(l) > 1) {
359       throw new SQLException ("multiple resultsets are not supported");
360     }
361     
362     ResultWrapper rw = new ResultWrapper();
363     Iterator iter = l.iterator();
364     while (iter.hasNext()) {
365       sql = (String )iter.next();
366       
367       if (DBUtils.isStatementType(sql, DBUtils.SELECT_STATEMENT)) {
368         processSelect(st, sql, rw);
369       }
370       else if (DBUtils.isStatementType(sql, DBUtils.INSERT_STATEMENT)) {
371         processInsert(st, sql, rw);
372       }
373       else if (DBUtils.isStatementType(sql, DBUtils.UPDATE_STATEMENT)) {
374         processUpdateOrDelete(true, st, sql, rw);
375       }
376       else if (DBUtils.isStatementType(sql, DBUtils.DELETE_STATEMENT)) {
377         processUpdateOrDelete(false, st, sql, rw);
378       }
379       else if (DBUtils.isStatementType(sql, DBUtils.CLEAR_CACHE_STATEMENT)) {
380         processClearCache(st, sql, rw);
381       }
382       else if (DBUtils.isStatementType(sql, DBUtils.QUERY_CACHE_STATEMENT)) {
383         processQueryCache(st, sql, rw);
384       }
385       else if (DBUtils.isStatementType(sql, DBUtils.QUERY_STATS_STATEMENT)) {
386         processQueryStats(st, sql, rw);
387       }
388       else {
389         if (log.isDebugEnabled()) {
390           log.debug("passing thru unrecognised statement: " + sql);
391         }
392
393         boolean results = false;
394         switch (st.statementType) {
395           case DCStatement.STATEMENT:
396             results = st.getRealStatement().execute(sql);
397             break;
398           case DCStatement.PREPARED:
399             results = ((PreparedStatement )st.getRealStatement()).execute();
400             break;
401           case DCStatement.CALLABLE:
402             results = ((CallableStatement )st.getRealStatement()).execute();
403         }
404         if (results) {
405           rw.rs = st.getRealStatement().getResultSet();
406         }
407         else {
408           rw.updateCount = st.getRealStatement().getUpdateCount();
409         }
410       }
411     }
412     
413     if (st.conn.getAutoCommit()) {
414       st.conn.commit();
415     }
416     
417     return rw;
418   }
419   
420   
421   public List getCacheNames() {
422     ArrayList ll = new ArrayList ();
423     ll.addAll(caches.keySet());
424     return ll;
425   }
426   
427   public List getCachedResultsetNames(String cacheName) throws SQLException {
428     ArrayList ll = new ArrayList ();
429     HashMap hm = (HashMap )DBUtils.getObjectFromMap(cacheName, datacache, HASHMAP_CLASS);
430     if (hm != null) {
431       ll.addAll(hm.keySet());
432     }
433     
434     return ll;
435   }
436   
437
438   
439  /**
440   * return the column data based on an array of primary key columns.
441   * For example, if the primary keys are in columns 1, 5 and 6 of the resultset,
442   * this will return the data in those columns as a 3 element array.
443   */
444   protected final String [] getKeyColumnData(int[] keyColumns, ResultSet rs) throws SQLException {
445     String [] keyData = new String [keyColumns.length];
446     for (int i = 0; i < keyColumns.length; i++) {
447       keyData[i] = rs.getString(keyColumns[i]);
448     }
449     return keyData;
450   }
451   
452   private final ArrayList getOperations(DCConnection conn) {
453     ArrayList ll = (ArrayList )operationsAwaitingCommit.get(conn);
454     if (ll == null) {
455       synchronized (operationsAwaitingCommit) {
456         ll = (ArrayList )operationsAwaitingCommit.get(conn);
457         if (ll == null) {
458           ll = new ArrayList ();
459           operationsAwaitingCommit.put(conn, ll);
460         }
461       }
462     }
463     return ll;
464   }
465   
466
467   protected int[] getPrimaryKeyColumns(String cacheName, ResultSetMetaData meta) throws SQLException {
468     CacheProperties cprops = (CacheProperties)caches.get(cacheName);
469     if (cprops == null) {
470       throw new SQLException (cacheName + " is not a valid cache");
471     }
472     int[] rtn = new int[cprops.primaryKeys.size()];
473     int j = 0;
474     for (int i = 1; i < meta.getColumnCount(); i++) {
475       if (cprops.primaryKeys.contains(meta.getColumnName(i))) {
476         rtn[j] = i;
477         j++;
478       }
479       if (j >= rtn.length) {
480         break;
481       }
482     }
483     if (j != rtn.length) {
484       log.warn("warning: unable to match returned data with key columns for cache " + cacheName);
485       return null;
486     }
487     else {
488       return rtn;
489     }
490   }
491   
492   public ResultSet getCachedResultSet(String cacheName, String sql) throws SQLException {
493     HashMap hm = (HashMap )DBUtils.getObjectFromMap(cacheName, datacache, HASHMAP_CLASS);
494
495     if (hm != null && hm.containsKey(sql)) {
496       DCResultSet dcrs = (DCResultSet)hm.get(sql);
497       dcrs.lastAccessed = System.currentTimeMillis();
498       return dcrs;
499     }
500     
501     return null;
502   }
503     
504   protected String getProperty(String name) {
505     return props.getProperty(name);
506   }
507   
508   public void handle(Message msg) {
509     ArrayList ll = (ArrayList )msg.getValue();
510     try {
511       triggerCacheChanges(ll);
512     }
513     catch (SQLException se) {
514       log.error("sqlerror: " + se.getMessage(), se);
515     }
516   }
517   
518   
519   private final void processClearCache(DCStatement st, String sql, ResultWrapper rw) {
520     String cacheName = DBUtils.getTargetName(sql, DBUtils.CLEAR_CACHE_STATEMENT);
521
522     ArrayList operations = getOperations(st.conn);
523     if (StringUtils.isEmpty(cacheName)) {
524       operations.add(new CacheOperation(CacheOperation.ALL_CHANGE, null, null));
525     }
526     else {
527       operations.add(new CacheOperation(CacheOperation.CACHE_CHANGE, cacheName, null));
528     }
529     
530     rw.updateCount = 1;
531     
532   }
533   
534  /**
535   * query the contents of the cache
536   */
537   private final void processQueryCache(DCStatement st, String sql, ResultWrapper rw) throws SQLException {
538     String cacheName = DBUtils.getTargetName(sql, DBUtils.QUERY_CACHE_STATEMENT);
539     
540     // if no cachename specified just dump the list of caches
541 if (StringUtils.isEmpty(cacheName)) {
542       
543       String [] cols = new String []{ "cache_name", "reset_on_insert", "timeout" };
544       ArrayList al = new ArrayList ();
545       Iterator iter = caches.keySet().iterator();
546       int i = 0;
547       
548       while (iter.hasNext()) {
549         String cache = (String )iter.next();
550         CacheProperties cprops = (CacheProperties)caches.get(cache);
551         
552         al.add(new Column[]{ DBUtils.createColumn(cache),
553                              DBUtils.createColumn((cprops.resetOnInsert ? TRUE : FALSE)),
554                              DBUtils.createColumn((cprops.timeout > 0 ? Long.toString(cprops.timeout) : EMPTY)) });
555       }
556
557       rw.rs = new DCResultSet(cols, al, this);
558     }
559     else {
560       HashMap hm = (HashMap )DBUtils.getObjectFromMap(cacheName, datacache, HASHMAP_CLASS);
561
562       // either return the contents of a particular cached resultset
563 if (DBUtils.isStatementType(sql, DBUtils.QUERY_CACHE_DUMP_STATEMENT)) {
564         String rname = DBUtils.getTargetName(sql, DBUtils.QUERY_CACHE_DUMP_STATEMENT);
565         
566         rw.rs = getCachedResultSet(cacheName, rname);
567       }
568       // or return the list of resultsets stored in a particular cache
569 else {
570         String [] cols = new String []{ "cache_name", "result_name" };
571
572         ArrayList al = new ArrayList ();
573         Iterator iter = hm.keySet().iterator();
574         while (iter.hasNext()) {
575           String result = (String )iter.next();
576           if (!result.equals(RESULTS_TO_CACHE_LINK)) {
577             al.add(new Column[]{ DBUtils.createColumn(cacheName), DBUtils.createColumn(result) });
578           }
579         }
580         rw.rs = new DCResultSet(cols, al, this);
581       }
582
583     }
584   }
585   
586   private final void processQueryStats(DCStatement st, String sql, ResultWrapper rw) throws SQLException {
587     String [] cols = new String []{ "component", "subcomponent", "size", "extra" };
588     ArrayList al = new ArrayList ();
589
590     al.add(DBUtils.createRow(new String []{ "caches", null, Integer.toString(caches.size()), null }));
591     al.add(DBUtils.createRow(new String []{ "table_check", null, Integer.toString(tableCheck.size()), null }));
592     al.add(DBUtils.createRow(new String []{ "update_check", null, Integer.toString(updateCheck.size()), null }));
593     al.add(DBUtils.createRow(new String []{ "ops_awaiting_commit", null, Integer.toString(operationsAwaitingCommit.size()), null }));
594
595     al.add(DBUtils.createRow(new String []{ "data_cache", null, Integer.toString(datacache.size()), null }));
596     
597     Iterator iter = datacache.keySet().iterator();
598     while (iter.hasNext()) {
599       String key = (String )iter.next();
600       HashMap hm = (HashMap )datacache.get(key);
601
602       al.add(DBUtils.createRow(new String []{ "data_cache", key, Integer.toString((hm.size() > 1 ? hm.size()-1 : 0)), null }));
603     }
604
605     HashMap hm = (HashMap )operationsAwaitingCommit.clone();
606     iter = hm.keySet().iterator();
607     while (iter.hasNext()) {
608       Object key = iter.next();
609       ArrayList oal = (ArrayList )hm.get(key);
610       
611       al.add(DBUtils.createRow(new String []{ "ops_awaiting_commit", (String )key, Integer.toString(oal.size()), oal.toString() }));
612     }
613
614     rw.rs = new DCResultSet(cols, al, this);
615   }
616   
617  /**
618   * process a select statement
619   */
620   private final void processSelect(DCStatement st, String sql, ResultWrapper rw) throws SQLException {
621     Statement rst = null;
622     ResultSet rrs = null;
623     boolean close = true;
624     try {
625       String cacheName = null;
626       String params = EMPTY;
627        
628       // if we are caching
629 if (!st.nocaching && !st.conn.blockCache) {
630         cacheName = StringUtils.getTagValue(sql, CACHE_BEGIN_TAG, CACHE_END_TAG);
631         
632         if (!StringUtils.isEmpty(cacheName)) {
633           if (st.statementType != DCStatement.STATEMENT) {
634             params = DBUtils.SEP + DBUtils.flatten(((DCPreparedStatement)st).params, PIPE);
635           }
636           
637           // lookup the cached result set and return if it's found
638 DCResultSet tmp = (DCResultSet)getCachedResultSet(cacheName, sql + params);
639           if (tmp != null) {
640             if (tmp.getRowCount() > 0) {
641               if (log.isDebugEnabled()) {
642                 log.debug("return cached data for " + sql + params);
643               }
644               rw.rs = tmp;
645               rw.updateCount = -1;
646               return;
647             }
648             else {
649               if (log.isInfoEnabled()) {
650                 log.info("clearing empty result set (" + cacheName + ") " + sql + params);
651               }
652               clearResultsByKey(cacheName, sql + params);
653             }
654           }
655         }
656       }
657
658       // create a real statement
659 if (st.statementType == DCStatement.STATEMENT) {
660         rst = st.getRealStatement();
661         rrs = rst.executeQuery(sql);
662       }
663       else if (st.statementType == DCStatement.PREPARED) {
664         rst = st.getRealStatement();
665         PreparedStatement ps = (PreparedStatement )rst;
666         rrs = ps.executeQuery();
667       }
668       else {
669         // shouldn't get here unless something has gone bizarrely wrong
670 throw new SQLException ("system error. unknown statement type");
671       }
672
673       // if we're not caching the results then just return the real resultset
674 //if (StringUtils.isEmpty(cacheName) || rrs.getFetchSize() > getMaxRowsForCache(cacheName)) {
675 if (StringUtils.isEmpty(cacheName)) {
676         if (log.isDebugEnabled()) {
677           log.debug("pass through SQL " + sql);
678         }
679         rw.rs = rrs;
680         rw.updateCount = -1;
681         close = false;
682       }
683       else {
684         // otherwise we need to process the results and cache the data
685 if (log.isDebugEnabled()) {
686           log.debug("caching SQL " + sql + params);
687         }
688
689         DCResultSet dcrs = new DCResultSet(cacheName, sql + params, rrs, this, rrs.getConcurrency(), rrs.getType());
690         rw.rs = dcrs;
691         rw.updateCount = -1;
692       }
693     }
694     finally {
695       if (close) {
696         DAOUtils.close(rrs);
697         DAOUtils.close(rst);
698       }
699     }
700   }
701   
702   
703  /**
704   * process an insert statement
705   */
706   private final void processInsert(DCStatement st, String sql, ResultWrapper rw) throws SQLException {
707     Statement rst = null;
708     try {
709       rst = st.getRealStatement();
710
711       // execute the update depending upon what params have been set in the statement
712 if (st.autoGeneratedKeys != Integer.MIN_VALUE) {
713         rw.updateCount = rst.executeUpdate(sql, st.autoGeneratedKeys);
714       }
715       else if (st.insertKeyColumnIndexes != null) {
716         rw.updateCount = rst.executeUpdate(sql, st.insertKeyColumnIndexes);
717       }
718       else if (st.insertKeyColumns != null) {
719         rw.updateCount = rst.executeUpdate(sql, st.insertKeyColumns);
720       }
721       else if (st.statementType == DCStatement.PREPARED) {
722         DCPreparedStatement dcp = (DCPreparedStatement)st;
723         dcp.realPS.execute();
724         rw.updateCount = dcp.realPS.getUpdateCount();
725       }
726       else {
727         rw.updateCount = rst.executeUpdate(sql);
728       }
729
730       // if an insert has occurred
731 if (rw.updateCount > 0) {
732         // get the table name for the insert
733 String table = DBUtils.getTargetName(sql, DBUtils.INSERT_STATEMENT);
734         
735         // if we are watching this table
736 if (!StringUtils.isEmpty(table) && updateCheck.containsKey(table)) {
737           if (log.isDebugEnabled()) {
738             log.debug("found a watched table " + table);
739           }
740           
741           // get the list of caches linked to this table
742 ArrayList ll = (ArrayList )updateCheck.get(table);
743           if (ll != null) {
744             // we'll store any cache operations here until a commit or rollback
745 ArrayList operations = getOperations(st.conn);
746             
747             // loop through the caches
748 Iterator citer = ll.iterator();
749             while (citer.hasNext()) {
750               String cache = (String )citer.next();
751               CacheProperties cprops = (CacheProperties)caches.get(cache);
752
753               if (cprops != null && cprops.resetOnInsert) {
754                 // add a remove cache op
755 operations.add(new CacheOperation(CacheOperation.CACHE_CHANGE, cache, null));
756               }
757             }
758           }
759         }
760       }
761       
762     }
763     finally {
764       DAOUtils.close(rst);
765     }
766   }
767   
768   private final void processUpdateOrDelete(boolean update, DCStatement st, String sql, ResultWrapper rw) throws SQLException {
769     String table;
770     if (update) {
771       table = DBUtils.getTargetName(sql, DBUtils.UPDATE_STATEMENT);
772     }
773     else {
774       table = DBUtils.getTargetName(sql, DBUtils.DELETE_STATEMENT);
775     }
776     
777     Statement rst = st.getRealStatement();
778     ResultSet rrs = null;
779     DCResultSet dcrs = null;
780     ResultSetMetaData rsmd = null;
781     boolean docache = true;
782     try {
783       if (updateCheck.containsKey(table)) {
784         String whereClause = DBUtils.getWhereClause(sql);
785
786         if (StringUtils.isEmpty(whereClause)) {
787           // no where clause, so wipe the entire cache
788 ArrayList operations = getOperations(st.conn);
789           ArrayList ll = (ArrayList )updateCheck.get(table);
790           Iterator iter = ll.iterator();
791           while (iter.hasNext()) {
792             operations.add(new CacheOperation(CacheOperation.CACHE_CHANGE, (String )iter.next(), null));
793           }
794           docache = false;
795         }
796         else {
797           String qurSQL = SELECT_FROM + table + WHERE + whereClause;
798           int paramsCount = StringUtils.countOccurrences(whereClause, '?');
799           // find out what data we need to clear from the cache
800 try {
801             if (st.statementType == DCStatement.PREPARED) {
802               // we try to reuse a prepared statement if possible
803 DCPreparedStatement dcp = (DCPreparedStatement)st;
804               
805               if (dcp.queryUpdateRowsSQL != null && !dcp.queryUpdateRowsSQL.equals(qurSQL)) {
806                 // but if the SQL isn't the same we need to create a new one
807 DAOUtils.close(dcp.queryUpdateRowsPS);
808                 dcp.queryUpdateRowsPS = null;
809               }
810               
811               if (dcp.queryUpdateRowsPS == null) {
812                 dcp.queryUpdateRowsPS = dcp.conn.getRealConnection().prepareStatement(qurSQL);
813                 dcp.queryUpdateRowsSQL = qurSQL;
814               }
815
816               DBUtils.setParams(dcp.queryUpdateRowsPS, dcp.params, dcp.params.length - paramsCount, paramsCount);
817               rrs = dcp.queryUpdateRowsPS.executeQuery();
818             }
819             else {
820               rrs = rst.executeQuery(qurSQL);
821             }
822
823             dcrs = new DCResultSet(null, null, rrs, this, rrs.getConcurrency(), rrs.getType());
824
825             rsmd = dcrs.getMetaData();
826
827             DAOUtils.close(rrs);
828           }
829           catch (SQLException se) {
830             se.printStackTrace();
831             log.warn("invalid SQL " + qurSQL);
832           }
833         }
834       }
835
836       if (st.statementType == DCStatement.PREPARED) {
837         DCPreparedStatement dcp = (DCPreparedStatement)st;
838         rw.updateCount = ((PreparedStatement )dcp.getRealStatement()).executeUpdate();
839       }
840       else {
841         rw.updateCount = st.getRealStatement().executeUpdate(sql);
842       }
843
844       if (rw.updateCount > 0 && docache && tableCheck.containsKey(table)) {
845         st.conn.blockCache = true;
846         ArrayList ll = (ArrayList )tableCheck.get(table);
847         Iterator citer = ll.iterator();
848         while (citer.hasNext()) {
849           String cache = (String )citer.next();
850           CacheProperties cp = (CacheProperties)caches.get(cache);
851           if (dcrs == null || (!update && cp.deleteCheck.contains(table))) {
852             // this could be failed SQL query, i.e. one where we have no idea what will
853 // have been updated, so we must clear the entire cache.
854 // or it is in our 'delete check' list. in which case we also clear the cache
855 ArrayList operations = getOperations(st.conn);
856             operations.add(new CacheOperation(CacheOperation.CACHE_CHANGE, cache, null));
857           }
858           else if (cp.primaryTable.equals(table)) {
859             // otherwise for an update of the primary table we'll only refresh
860 // the cache for specific key changes
861
862             dcrs.beforeFirst();
863             int[] keyColumns = this.getPrimaryKeyColumns(cache, rsmd);
864             if (keyColumns == null) {
865               // if we can't work out the key columns, we'll have to reset the
866 // cache as well
867 if (log.isDebugEnabled()) {
868                 log.debug("unable to find key columns, resetting cache " + cache);
869               }
870               ArrayList operations = getOperations(st.conn);
871               operations.add(new CacheOperation(CacheOperation.CACHE_CHANGE, cache, null));
872             }
873             else {
874                // otherwise we'll just remove the results that are potentially
875 // affected by the update
876 dcrs.beforeFirst();
877               while (dcrs.next()) {
878                 String key = DBUtils.getResultSetDataKey(cache, getKeyColumnData(keyColumns, dcrs));
879                 if (log.isDebugEnabled()) {
880                   log.debug("clearing data " + key);
881                 }
882
883                 ArrayList operations = getOperations(st.conn);
884                 operations.add(new CacheOperation(CacheOperation.KEY_CHANGE, cache, key));
885               }
886             }
887           }
888         }
889       }
890     }
891     finally {
892       DAOUtils.close(rrs);
893     }
894   }
895   
896   
897   public void triggerCacheChanges(ArrayList ll) throws SQLException {
898     if (ll != null) {
899       Iterator iter = ll.iterator();
900       while (iter.hasNext()) {
901         CacheOperation op = (CacheOperation)iter.next();
902         if (op.type == CacheOperation.CACHE_CHANGE) {
903           if (log.isDebugEnabled()) {
904             log.debug("removing all data for cache " + op.cache);
905           }
906           clearCache(op.cache);
907         }
908         else if (op.type == CacheOperation.KEY_CHANGE) {
909           if (log.isDebugEnabled()) {
910             log.debug("removing all data for key " + op.data);
911           }
912           clearResultsByKey(op.cache, op.data);
913         }
914         else if (op.type == CacheOperation.ALL_CHANGE) {
915           if (log.isDebugEnabled()) {
916             log.debug("removing all data");
917           }
918           clearAll();
919         }
920       }
921     }
922   }
923   
924  /**
925   * rollback a connection (get rid of operations list)
926   */
927   protected void rollback(DCConnection conn) {
928     operationsAwaitingCommit.remove(conn);
929     if (log.isDebugEnabled()) {
930       log.debug(Integer.toString(operationsAwaitingCommit.size()) + " operations (after rollback)");
931     }
932   }
933   
934   
935  /**
936   * link a key to a resultset
937   * @param cacheName the cache we're linking in
938   * @param rsKeys the primary key data to use as key
939   * @param sql the sql query used to generate the resultset
940   */
941   protected void setKeyToResultSetLink(String cacheName, String [] rsKeys, String sql) throws SQLException {
942     String key = DBUtils.getResultSetDataKey(cacheName, rsKeys);
943     
944     HashMap hm = (HashMap )DBUtils.getObjectFromMap(cacheName, datacache, HASHMAP_CLASS);
945     HashMap rtc = (HashMap )DBUtils.getObjectFromMap(RESULTS_TO_CACHE_LINK, hm, HASHMAP_CLASS);
946     ArrayList ll = (ArrayList )DBUtils.getObjectFromMap(key, rtc, ARRAYLIST_CLASS);
947     
948     ll.add(sql);
949   }
950   
951   
952  /**
953   * add a resultset to the cache
954   */
955   protected void setCachedResultSet(String cacheName, String sql, ResultSet rs) throws SQLException {
956     if (log.isDebugEnabled()) {
957       log.debug("adding cache result for " + cacheName + "::" + sql);
958     }
959     
960     HashMap hm = (HashMap )DBUtils.getObjectFromMap(cacheName, datacache, HASHMAP_CLASS);
961     synchronized (hm) {
962       hm.put(sql, rs);
963     }
964   }
965
966   
967  /**
968   * validate user access to this 'database'
969   */
970   protected boolean validate(String user, String password) {
971     String pass = (String )users.get(user);
972     return (!StringUtils.isEmpty(pass) && pass.equals(password));
973   }
974
975 }

A to Z: JavaDoc & Examples

---

Daily Java News & Articles

---

Open Source Projects

---

Open Source Codes

---

Free Computer Books

---

Remove Frame

---

Popular Tags