Java > Open Source Codes > org > jboss > cache > loader > JDBCCacheLoader


1 /*
2  * JBoss, the OpenSource J2EE webOS
3  *
4  * Distributable under LGPL license.
5  * See terms of license at gnu.org.
6  */
7 package org.jboss.cache.loader;
8
9 import org.apache.commons.logging.Log;
10 import org.apache.commons.logging.LogFactory;
11 import org.jboss.cache.Fqn;
12 import org.jboss.cache.Modification;
13 import org.jboss.cache.config.CacheLoaderConfig.IndividualCacheLoaderConfig;
14
15 import javax.naming.InitialContext ;
16 import javax.naming.NamingException ;
17 import javax.sql.DataSource ;
18 import java.io.ByteArrayInputStream ;
19 import java.io.ByteArrayOutputStream ;
20 import java.io.IOException ;
21 import java.io.InputStream ;
22 import java.io.ObjectInputStream ;
23 import java.io.ObjectOutputStream ;
24 import java.sql.Connection ;
25 import java.sql.DatabaseMetaData ;
26 import java.sql.DriverManager ;
27 import java.sql.PreparedStatement ;
28 import java.sql.ResultSet ;
29 import java.sql.SQLException ;
30 import java.sql.Statement ;
31 import java.sql.Types ;
32 import java.util.ArrayList ;
33 import java.util.Collection ;
34 import java.util.Collections ;
35 import java.util.HashMap ;
36 import java.util.HashSet ;
37 import java.util.List ;
38 import java.util.Map ;
39 import java.util.Properties ;
40 import java.util.Set ;
41
42 /**
43  * JDBC CacheLoader implementation.
44  * <p/>
45  * This implementation uses one table. The table consists of three columns:
46  * <ul>
47  * <li>text column for fqn (which is also a primary key)</li>
48  * <li>blob column for attributes (can contain null)</li>
49  * <li>text column for parent fqn (can contain null)</li>
50  * </ul>
51  * <p/>
52  * The configuration options are:
53  * <p/>
54  * <b>Table configuration</b>
55  * <ul>
56  * <li><b>cache.jdbc.table.name</b> - the table name (default is <i>jbosscache</i>)</li>
57  * <li><b>cache.jdbc.table.create</b> - should be true or false, indicates whether to create the table at start phase</li>
58  * <li><b>cache.jdbc.table.drop</b> - should be true or false, indicates whether to drop the table at stop phase</li>
59  * <li><b>cache.jdbc.table.primarykey</b> - the name for the table primary key (default is <i>jbosscache_pk</i>)</li>
60  * <li><b>cache.jdbc.fqn.column</b> - the name for the fqn column (default is <i>fqn</i>)</li>
61  * <li><b>cache.jdbc.fqn.type</b> - the type for the fqn column (default is <i>varchar(255)</i>)</li>
62  * <li><b>cache.jdbc.node.column</b> - the name for the node's contents column (default is <i>node</i>)</li>
63  * <li><b>cache.jdbc.node.type</b> - the type for the node's contents column (default is <i>blob</i>)</li>
64  * <li><b>cache.jdbc.parent.column</b> - the name for the parent fqn column (default is <i>parent</i>)</li>
65  * </ul>
66  * <p/>
67  * <b>DataSource configuration</b>
68  * <ul>
69  * <li><b>cache.jdbc.datasource</b> - the JNDI name of the datasource</li>
70  * </ul>
71  * <p/>
72  * <b>JDBC driver configuration (used when DataSource is not configured)</b>
73  * <ul>
74  * <li><b>cache.jdbc.driver</b> - fully qualified JDBC driver name</li>
75  * <li><b>cache.jdbc.url</b> - URL to connect to the database</li>
76  * <li><b>cache.jdbc.user</b> - the username to use to connect to the database</li>
77  * <li><b>cache.jdbc.password</b> - the password to use to connect to the database</li>
78  * </ul>
79  *
80  * @author <a HREF="mailto:alex@jboss.org">Alexey Loubyansky</a>
81  * @author <a HREF="mailto:hmesha@novell.com">Hany Mesha </a>
82  * @version <tt>$Revision: 1.26 $</tt>
83  */
84 public class JDBCCacheLoader extends AbstractCacheLoader
85 {
86    private static final Log log = LogFactory.getLog(JDBCCacheLoader.class);
87
88    private static final ThreadLocal connection = new ThreadLocal ();
89
90    private JDBCCacheLoaderConfig config;
91    private ConnectionFactory cf;
92    private String driverName;
93
94    public void setConfig(IndividualCacheLoaderConfig base)
95    {
96       if (config instanceof JDBCCacheLoaderConfig)
97       {
98          config = (JDBCCacheLoaderConfig) base;
99       }
100       else
101       {
102          config = new JDBCCacheLoaderConfig(base);
103       }
104
105       if (config.getDatasourceName() == null)
106       {
107          this.cf = new NonManagedConnectionFactory(config.getJdbcURL(), config.getJdbcUser(), config.getJdbcPassword());
108       }
109       // else we wait until the start method to do a JNDI lookup
110 // of the datasource, since that's when its registered in its lifecycle
111 }
112
113    public IndividualCacheLoaderConfig getConfig()
114    {
115       return config;
116    }
117
118    /**
119     * Fetches child node names (not pathes).
120     *
121     * @param fqn parent fqn
122     * @return a set of child node names or null if there are not children found for the fqn
123     * @throws Exception
124     */
125    public Set <String > getChildrenNames(Fqn fqn) throws Exception
126    {
127       Set children = null;
128       Connection con = null;
129       PreparedStatement ps = null;
130       ResultSet rs = null;
131       try
132       {
133          if (log.isDebugEnabled())
134          {
135             log.debug("executing sql: " + config.getSelectChildNamesSql() + " (" + fqn + ")");
136          }
137
138          con = cf.getConnection();
139          ps = con.prepareStatement(config.getSelectChildNamesSql());
140          ps.setString(1, fqn.toString());
141          rs = ps.executeQuery();
142          if (rs.next())
143          {
144             children = new HashSet ();
145             do
146             {
147                String child = rs.getString(1);
148                int slashInd = child.lastIndexOf('/');
149                String name = child.substring(slashInd + 1);
150                //Fqn childFqn = Fqn.fromString(child);
151 //String name = (String) childFqn.get(childFqn.size() - 1);
152 children.add(name);
153             }
154             while (rs.next());
155          }
156       }
157       catch (SQLException e)
158       {
159          log.error("Failed to get children names for fqn " + fqn, e);
160          throw new IllegalStateException ("Failed to get children names for fqn " + fqn + ": " + e.getMessage());
161       }
162       finally
163       {
164          safeClose(rs);
165          safeClose(ps);
166          cf.close(con);
167       }
168
169       return children == null ? null : Collections.unmodifiableSet(children);
170    }
171
172    // See http://jira.jboss.com/jira/browse/JBCACHE-118 for why this is commented out.
173
174    /**
175     * Loads an attribute from the database.
176     *
177     * @param name node's fqn
178     * @param key attribute's key
179     * @return attribute's value. Null is returned if
180     * <ul>
181     * <li>there is no value for the attribute key in the node</li>
182     * <li>there is a row in the table for the fqn but the node column contains null</li>
183     * <li>there is no row in table for the fqn (should this result in an exception?)</li>
184     * </ul>
185     * @throws Exception
186     */
187    // public Object get(Fqn name, Object key) throws Exception
188 // {
189 // Map node = loadNode(name);
190 // return node == null || node == NULL_NODE_IN_ROW ? null : node.get(key);
191 // }
192
193    /**
194     * Returns a map representing a node.
195     *
196     * @param name node's fqn
197     * @return node
198     * @throws Exception
199     */
200    public Map get(Fqn name) throws Exception
201    {
202       final Map node = loadNode(name);
203       return node == NULL_NODE_IN_ROW ? new HashMap (0) : node;
204    }
205
206    /**
207     * Checks that there is a row for the fqn in the database.
208     *
209     * @param name node's fqn
210     * @return true if there is a row in the database for the given fqn even if the node column is null.
211     * @throws Exception
212     */
213    public boolean exists(Fqn name) throws Exception
214    {
215       final Map node = loadNode(name);
216       return node != null;// && node != NULL_NODE_IN_ROW;
217 }
218
219    /**
220     * Adds/overrides a value in a node for a key.
221     * If the node does not exist yet, the node will be created.
222     * If parent nodes do not exist for the node, empty parent nodes will be created.
223     *
224     * @param name node's fqn
225     * @param key attribute's key
226     * @param value attribute's value
227     * @return old value associated with the attribute's key or null if there was no value previously
228     * associated with the attribute's key
229     * @throws Exception
230     */
231    public Object put(Fqn name, Object key, Object value) throws Exception
232    {
233       Map oldNode = loadNode(name);
234       Object oldValue;
235       Map node;
236
237       if (oldNode == null || oldNode == NULL_NODE_IN_ROW)
238       {
239          node = new HashMap ();
240       }
241       else
242       {
243          node = oldNode;
244       }
245       oldValue = node.put(key, value);
246
247       if (oldNode != null)
248       {
249          updateNode(name, node);
250       }
251       else
252       {
253          if (name.size() > 1)
254          {
255             for (int i = 1; i < name.size(); ++i)
256             {
257                final Fqn parent = name.getFqnChild(i);
258                if (!exists(parent))
259                {
260                   insertNode(parent, null);
261                }
262             }
263          }
264          insertNode(name, node);
265       }
266
267       return oldValue;
268    }
269
270    /**
271     * Adds attributes from the passed in map to the existing node.
272     * If there is no node for the fqn, a new node will be created.
273     *
274     * @param name node's fqn
275     * @param attributes attributes
276     * @throws Exception
277     */
278    public void put(Fqn name, Map attributes) throws Exception
279    {
280       put(name, attributes, false);
281    }
282
283    /**
284     * Removes attribute's value for a key. If after removal the node contains no attributes, the node is nullified.
285     *
286     * @param name node's name
287     * @param key attribute's key
288     * @return removed value or null if there was no value for the passed in key
289     * @throws Exception
290     */
291    public Object remove(Fqn name, Object key) throws Exception
292    {
293       Object removedValue = null;
294       Map node = loadNode(name);
295       if (node != null && node != NULL_NODE_IN_ROW)
296       {
297          removedValue = node.remove(key);
298          if (node.isEmpty())
299          {
300             updateNode(name, null);
301          }
302          else
303          {
304             updateNode(name, node);
305          }
306       }
307       return removedValue;
308    }
309
310    /**
311     * Removes a node and all its children.
312     * Uses the same connection for all the db work.
313     *
314     * @param name node's fqn
315     * @throws Exception
316     */
317    public void remove(Fqn name) throws Exception
318    {
319       Connection con = null;
320       PreparedStatement ps = null;
321       try
322       {
323          if (name.size() == 0)
324          {
325             if (log.isDebugEnabled())
326             {
327                log.debug("executing sql: " + config.getDeleteAllSql());
328             }
329
330             con = cf.getConnection();
331             ps = con.prepareStatement(config.getDeleteAllSql());
332             int deletedRows = ps.executeUpdate();
333
334             if (log.isDebugEnabled())
335             {
336                log.debug("total rows deleted: " + deletedRows);
337             }
338          }
339          else
340          {
341             StringBuffer sql = new StringBuffer (300);
342             sql.append("delete from ").append(config.getTable()).append(" where fqn in (");
343             //sql2.append("delete from " + table + " where fqn=? or parent in (");
344 List fqns = new ArrayList ();
345
346             addChildrenToDeleteSql(name.toString(), sql, fqns);
347
348             sql.append(')');
349
350             if (fqns.size() == 1)
351             {
352                if (log.isDebugEnabled())
353                {
354                   log.debug("executing sql: " + config.getDeleteNodeSql() + "(" + name + ")");
355                }
356
357                con = cf.getConnection();
358                ps = con.prepareStatement(config.getDeleteNodeSql());
359                ps.setString(1, name.toString());
360             }
361             else
362             {
363                if (log.isDebugEnabled())
364                {
365                   log.debug("executing sql: " + sql + " " + fqns);
366                }
367
368                con = cf.getConnection();
369                ps = con.prepareStatement(sql.toString());
370                for (int i = 0; i < fqns.size(); ++i)
371                {
372                   ps.setString(i + 1, (String ) fqns.get(i));
373                }
374             }
375
376             int deletedRows = ps.executeUpdate();
377
378             if (log.isDebugEnabled())
379             {
380                log.debug("total rows deleted: " + deletedRows);
381             }
382          }
383       }
384       catch (SQLException e)
385       {
386          log.error("Failed to remove node " + name, e);
387          throw new IllegalStateException ("Failed to remove node " + name + ": " + e.getMessage());
388       }
389       finally
390       {
391          safeClose(ps);
392          cf.close(con);
393       }
394    }
395
396    /**
397     * Nullifies the node.
398     *
399     * @param name node's fqn
400     * @throws Exception
401     */
402    public void removeData(Fqn name) throws Exception
403    {
404       updateNode(name, null);
405    }
406
407    /**
408     * First phase in transaction commit process. The changes are committed if only one phase if requested.
409     * All the modifications are committed using the same connection.
410     *
411     * @param tx something representing transaction
412     * @param modifications a list of modifications
413     * @param one_phase indicates whether it's one or two phase commit transaction
414     * @throws Exception
415     */
416    public void prepare(Object tx, List <Modification> modifications, boolean one_phase) throws Exception
417    {
418       // start a tx
419 //JBCACHE-346 fix, we don't need to prepare a DataSource object (Managed connection)
420 if (cf instanceof NonManagedConnectionFactory)
421       {
422          Connection con = cf.prepare(tx);
423          if (log.isTraceEnabled())
424          {
425             log.trace("openned tx connection: tx=" + tx + ", con=" + con);
426          }
427       }
428
429       try
430       {
431          put(modifications);
432
433          // commit if it's one phase only
434 if (one_phase)
435          {
436             commit(tx);
437          }
438       }
439       catch (Exception e)
440       {
441          // todo should I rollback it here or rollback is supposed to be invoke by someone from outside?
442 rollback(tx);
443          // is this ok?
444 throw e;
445       }
446    }
447
448    /**
449     * Commits a transaction.
450     *
451     * @param tx the tx to commit
452     * @throws Exception
453     */
454    public void commit(Object tx) throws Exception
455    {
456       cf.commit(tx);
457    }
458
459    /**
460     * Rolls back a transaction.
461     *
462     * @param tx the tx to rollback
463     */
464    public void rollback(Object tx)
465    {
466       cf.rollback(tx);
467    }
468
469    // Service implementation
470
471    public void create() throws Exception
472    {
473    }
474
475    public void start() throws Exception
476    {
477       if (config.getDriverClass() != null)
478       {
479          loadDriver(config.getDriverClass());
480       }
481       else
482       {
483          // A datasource will be registered in JNDI in the start portion of
484 // its lifecycle, so now that we are in start() we can look it up
485 InitialContext ctx = null;
486          try
487          {
488             ctx = new InitialContext ();
489             DataSource dataSource = (DataSource ) ctx.lookup(config.getDatasourceName());
490             this.cf = new ManagedConnectionFactory(dataSource);
491          }
492          catch (NamingException e)
493          {
494             log.error("Failed to lookup datasource " + config.getDatasourceName() + ": " + e.getMessage(), e);
495             throw new IllegalStateException ("Failed to lookup datasource " + config.getDatasourceName() + ": " + e.getMessage());
496          }
497          finally
498          {
499             if (ctx != null)
500             {
501                try
502                {
503                   ctx.close();
504                }
505                catch (NamingException e)
506                {
507                   log.warn("Failed to close naming context.", e);
508                }
509             }
510          }
511       }
512
513       Connection con = null;
514       Statement st = null;
515
516       try
517       {
518          con = cf.getConnection();
519          driverName = getDriverName(con);
520          if (config.getCreateTable())
521          {
522             if (!tableExists(config.getTable(), con))
523             {
524                if (log.isDebugEnabled())
525                {
526                   log.debug("executing ddl: " + config.getCreateTableDDL());
527                }
528                st = con.createStatement();
529                st.executeUpdate(config.getCreateTableDDL());
530             }
531          }
532       }
533       finally
534       {
535          safeClose(st);
536          cf.close(con);
537       }
538    }
539
540    public void stop()
541    {
542       if (config.getDropTable())
543       {
544          Connection con = null;
545          Statement st = null;
546          try
547          {
548             if (log.isDebugEnabled())
549             {
550                log.debug("executing ddl: " + config.getDropTableDDL());
551             }
552
553             con = cf.getConnection();
554             st = con.createStatement();
555             st.executeUpdate(config.getDropTableDDL());
556             safeClose(st);
557          }
558          catch (SQLException e)
559          {
560             log.error("Failed to drop table: " + e.getMessage(), e);
561          }
562          finally
563          {
564             safeClose(st);
565             cf.close(con);
566          }
567       }
568    }
569
570    public void destroy()
571    {
572    }
573
574    // Private
575
576    private void addChildrenToDeleteSql(String name, StringBuffer sql, List fqns)
577            throws SQLException
578    {
579       // for now have to use connection per method, i.e. can't pass the same connection to recursive
580 // invocations because buggy PointBase driver invalidates result sets.
581 Connection con = null;
582       PreparedStatement selChildrenPs = null;
583       ResultSet rs = null;
584       try
585       {
586          if (log.isDebugEnabled())
587          {
588             log.debug("executing sql: " + config.getSelectChildFqnsSql() + "(" + name + ")");
589          }
590
591          con = cf.getConnection();
592          selChildrenPs = con.prepareStatement(config.getSelectChildFqnsSql());
593          selChildrenPs.setString(1, name);
594          rs = selChildrenPs.executeQuery();
595
596          if (rs.next())
597          {
598             do
599             {
600                String childStr = rs.getString(1);
601                addChildrenToDeleteSql(childStr, sql, fqns);
602             }
603             while (rs.next());
604          }
605
606          if (fqns.size() == 0)
607          {
608             sql.append("?");
609          }
610          else
611          {
612             sql.append(", ?");
613          }
614          fqns.add(name);
615       }
616       finally
617       {
618          safeClose(rs);
619          safeClose(selChildrenPs);
620          cf.close(con);
621       }
622    }
623
624    public void put(Fqn name, Map attributes, boolean override) throws Exception
625    {
626       // JBCACHE-769 -- make a defensive copy
627 Map attrs = (attributes == null ? null : new HashMap (attributes));
628
629       Map oldNode = loadNode(name);
630       if (oldNode != null)
631       {
632          if (!override && oldNode != NULL_NODE_IN_ROW && attrs != null)
633          {
634             attrs.putAll(oldNode);
635          }
636          updateNode(name, attrs);
637       }
638       else
639       {
640          if (name.size() > 1)
641          {
642             for (int i = 1; i < name.size(); ++i)
643             {
644                final Fqn parent = name.getFqnChild(i);
645                if (!exists(parent))
646                {
647                   insertNode(parent, null);
648                }
649             }
650          }
651          insertNode(name, attrs);
652       }
653    }
654
655    /**
656     * Inserts a node into the database
657     *
658     * @param name the fqn
659     * @param node the node
660     */
661    private void insertNode(Fqn name, Map node)
662    {
663       Connection con = null;
664       PreparedStatement ps = null;
665       try
666       {
667          if (log.isDebugEnabled())
668          {
669             log.debug("executing sql: " + config.getInsertNodeSql() + " (" + name + ")");
670          }
671
672          con = cf.getConnection();
673          ps = con.prepareStatement(config.getInsertNodeSql());
674
675          ps.setString(1, name.toString());
676
677          if (node != null)
678          {
679 // Object marshalledNode = new MarshalledValue(node);
680 ByteArrayOutputStream baos = new ByteArrayOutputStream ();
681             ObjectOutputStream oos = new ObjectOutputStream (baos);
682 // oos.writeObject(marshalledNode);
683 //
684 oos.writeObject(node);
685
686             ByteArrayInputStream bais = new ByteArrayInputStream (baos.toByteArray());
687             ps.setBinaryStream(2, bais, baos.size());
688          }
689          else
690          {
691             // a hack to handles the incomp. of SQL server jdbc driver prior to SQL SERVER 2005
692 if (driverName != null && (driverName.contains("SQLSERVER")
693                     || driverName.contains("POSTGRESQL")))
694             {
695                ps.setNull(2, Types.LONGVARBINARY);
696             }
697             else
698             {
699                ps.setNull(2, Types.BLOB);
700             }
701             //ps.setNull(2, Types.LONGVARBINARY);
702 }
703
704          if (name.size() == 0)
705          {
706             ps.setNull(3, Types.VARCHAR);
707          }
708          else
709          {
710             ps.setString(3, name.getFqnChild(name.size() - 1).toString());
711          }
712
713          int rows = ps.executeUpdate();
714          if (rows != 1)
715          {
716             throw new IllegalStateException ("Expected one insert row but got " + rows);
717          }
718       }
719       catch (RuntimeException e)
720       {
721          throw e;
722       }
723       catch (Exception e)
724       {
725          log.error("Failed to insert node: " + e.getMessage(), e);
726          throw new IllegalStateException ("Failed to insert node: " + e.getMessage());
727       }
728       finally
729       {
730          safeClose(ps);
731          cf.close(con);
732       }
733    }
734
735    /**
736     * Updates a node in the database.
737     *
738     * @param name the fqn
739     * @param node new node value
740     */
741    private void updateNode(Fqn name, Map node)
742    {
743       Connection con = null;
744       PreparedStatement ps = null;
745       try
746       {
747          if (log.isDebugEnabled())
748          {
749             log.debug("executing sql: " + config.getUpdateNodeSql());
750          }
751
752          con = cf.getConnection();
753          ps = con.prepareStatement(config.getUpdateNodeSql());
754
755          if (node == null)
756          {
757             //ps.setNull(1, Types.BLOB);
758 // ps.setNull(1, Types.LONGVARBINARY);
759 // don't set it to null - simply use an empty hash map.
760 node = new HashMap (0);
761          }
762          // else
763 // {
764 // Object marshalledNode = new MarshalledValue(node);
765 ByteArrayOutputStream baos = new ByteArrayOutputStream ();
766          ObjectOutputStream oos = new ObjectOutputStream (baos);
767 // oos.writeObject(marshalledNode);
768 oos.writeObject(node);
769
770          ByteArrayInputStream bais = new ByteArrayInputStream (baos.toByteArray());
771          ps.setBinaryStream(1, bais, baos.size());
772          // }
773
774          ps.setString(2, name.toString());
775
776          int rows = ps.executeUpdate();
777 // if (rows != 1)
778 // {
779 // throw new IllegalStateException("Expected one updated row but got " + rows);
780 // }
781 }
782       catch (Exception e)
783       {
784          log.error("Failed to update node for fqn " + name + ": " + e.getMessage(), e);
785          throw new IllegalStateException ("Failed to update node for fqn " + name + ": " + e.getMessage());
786       }
787       finally
788       {
789          safeClose(ps);
790          cf.close(con);
791       }
792    }
793
794    /**
795     * Loads a node from the database.
796     *
797     * @param name the fqn
798     * @return non-null Map representing the node,
799     * null if there is no row with the fqn in the table,
800     * NULL_NODE_IN_ROW if there is a row in the table with the fqn but the node column contains null.
801     */
802    private Map loadNode(Fqn name)
803    {
804       boolean rowExists = false;
805       Map oldNode = null;
806       Connection con = null;
807       PreparedStatement ps = null;
808       ResultSet rs = null;
809       try
810       {
811          if (log.isDebugEnabled())
812          {
813             log.debug("executing sql: " + config.getSelectNodeSql() + " (" + name + ")");
814          }
815
816          con = cf.getConnection();
817          ps = con.prepareStatement(config.getSelectNodeSql());
818          ps.setString(1, name.toString());
819
820          rs = ps.executeQuery();
821
822          if (rs.next())
823          {
824             rowExists = true;
825             InputStream is = rs.getBinaryStream(1);
826             if (is != null && !rs.wasNull())
827             {
828                ObjectInputStream ois = null;
829                try
830                {
831                   // deserialize result
832 ois = new ObjectInputStream (is);
833                   Object marshalledNode = ois.readObject();
834
835                   // de-marshall value if possible
836 // if (marshalledNode instanceof MarshalledValue)
837 // {
838 // oldNode = (Map) ((MarshalledValue) marshalledNode).get();
839 // }
840 // else if (marshalledNode instanceof MarshalledObject)
841 // {
842 // oldNode = (Map) ((MarshalledObject) marshalledNode).get();
843 // }
844 oldNode = (Map ) marshalledNode;
845                }
846                catch (IOException e)
847                {
848                   throw new SQLException ("Unable to load to deserialize result: " + e);
849                }
850                catch (ClassNotFoundException e)
851                {
852                   throw new<