AlfrescoTransactionSupport


1   /*
2    * Copyright (C) 2005 Alfresco, Inc.
3    *
4    * Licensed under the Mozilla Public License version 1.1 
5    * with a permitted attribution clause. You may obtain a
6    * copy of the License at
7    *
8    *   http://www.alfresco.org/legal/license.txt
9    *
10   * Unless required by applicable law or agreed to in writing,
11   * software distributed under the License is distributed on an
12   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
13   * either express or implied. See the License for the specific
14   * language governing permissions and limitations under the
15   * License.
16   */
17  package org.alfresco.repo.transaction;
18  
19  import java.util.ArrayList  ;
20  import java.util.HashMap  ;
21  import java.util.HashSet  ;
22  import java.util.List  ;
23  import java.util.Map  ;
24  import java.util.Set  ;
25  
26  import org.alfresco.error.AlfrescoRuntimeException;
27  import org.alfresco.repo.node.db.NodeDaoService;
28  import org.alfresco.repo.node.integrity.IntegrityChecker;
29  import org.alfresco.repo.search.impl.lucene.LuceneIndexerAndSearcherFactory;
30  import org.alfresco.service.cmr.rule.RuleService;
31  import org.alfresco.util.GUID;
32  import org.apache.commons.logging.Log;
33  import org.apache.commons.logging.LogFactory;
34  import org.springframework.orm.hibernate3.SessionFactoryUtils;
35  import org.springframework.transaction.support.TransactionSynchronization;
36  import org.springframework.transaction.support.TransactionSynchronizationAdapter;
37  import org.springframework.transaction.support.TransactionSynchronizationManager;
38  
39  /**
40   * Helper class to manage transaction synchronization.  This provides helpers to
41   * ensure that the necessary <code>TransactionSynchronization</code> instances
42   * are registered on behalf of the application code.
43   * 
44   * @author Derek Hulley
45   */
46  public abstract class AlfrescoTransactionSupport
47  {
48      /*
49       * The registrations of services is very explicit on the interface.  This
50       * is to convey the idea that the execution of these services when the
51       * transaction completes is very explicit.  As we only have a finite
52       * list of types of services that need registration, this is still
53       * OK.
54       */
55      
56      /**
57       * The order of synchronization set to be 100 less than the Hibernate synchronization order
58       */
59      public static final int SESSION_SYNCHRONIZATION_ORDER =
60          SessionFactoryUtils.SESSION_SYNCHRONIZATION_ORDER - 100;
61  
62      /** resource key to store the transaction synchronizer instance */
63      private static final String   RESOURCE_KEY_TXN_SYNCH = "txnSynch";
64      
65      private static Log logger = LogFactory.getLog(AlfrescoTransactionSupport.class);
66      
67      /**
68       * Get a unique identifier associated with each transaction of each thread.  Null is returned if
69       * no transaction is currently active.
70       * 
71       * @return Returns the transaction ID, or null if no transaction is present
72       */
73      public static String   getTransactionId()
74      {
75          /*
76           * Go direct to the synchronizations as we don't want to register a resource if one doesn't exist.
77           * This method is heavily used, so the simple Map lookup on the ThreadLocal is the fastest.
78           */
79          
80          TransactionSynchronizationImpl txnSynch =
81                  (TransactionSynchronizationImpl) TransactionSynchronizationManager.getResource(RESOURCE_KEY_TXN_SYNCH);
82          if (txnSynch == null)
83          {
84              if (TransactionSynchronizationManager.isSynchronizationActive())
85              {
86                  // need to lazily register synchronizations
87                  return registerSynchronizations().getTransactionId();
88              }
89              else
90              {
91                  return null;   // not in a transaction
92              }
93          }
94          else
95          {
96              return txnSynch.getTransactionId();
97          }
98      }
99      
100     /**
101      * Are there any pending changes which must be synchronized with the store?
102      * 
103      * @return true => changes are pending
104      */
105     public static boolean isDirty()
106     {
107         TransactionSynchronizationImpl synch = getSynchronization();
108         
109         Set  <NodeDaoService> services = synch.getNodeDaoServices();
110         for (NodeDaoService service : services)
111         {
112            if (service.isDirty())
113            {
114                return true;
115            }
116         }
117         
118         return false;
119     }
120     
121     /**
122      * Gets a resource associated with the current transaction, which must be active.
123      * <p>
124      * All necessary synchronization instances will be registered automatically, if required.
125      * 
126      *  
127      * @param key the thread resource map key
128      * @return Returns a thread resource of null if not present
129      */
130     public static Object   getResource(Object   key)
131     {
132         // get the synchronization
133         TransactionSynchronizationImpl txnSynch = getSynchronization();
134         // get the resource
135         Object   resource = txnSynch.resources.get(key);
136         // done
137         if (logger.isDebugEnabled())
138         {
139             logger.debug("Fetched resource: \n" +
140                     "   key: " + key + "\n" +
141                     "   resource: " + resource);
142         }
143         return resource;
144     }
145     
146     /**
147      * Binds a resource to the current transaction, which must be active.
148      * <p>
149      * All necessary synchronization instances will be registered automatically, if required.
150      * 
151      * @param key
152      * @param resource
153      */
154     public static void bindResource(Object   key, Object   resource)
155     {
156         // get the synchronization
157         TransactionSynchronizationImpl txnSynch = getSynchronization();
158         // bind the resource
159         txnSynch.resources.put(key, resource);
160         // done
161         if (logger.isDebugEnabled())
162         {
163             logger.debug("Bound resource: \n" +
164                     "   key: " + key + "\n" +
165                     "   resource: " + resource);
166         }
167     }
168     
169     /**
170      * Unbinds a resource from the current transaction, which must be active.
171      * <p>
172      * All necessary synchronization instances will be registered automatically, if required.
173      * 
174      * @param key
175      */
176     public static void unbindResource(Object   key)
177     {
178         // get the synchronization
179         TransactionSynchronizationImpl txnSynch = getSynchronization();
180         // remove the resource
181         txnSynch.resources.remove(key);
182         // done
183         if (logger.isDebugEnabled())
184         {
185             logger.debug("Unbound resource: \n" +
186                     "   key: " + key);
187         }
188     }
189     
190     /**
191      * Method that registers a <tt>NodeDaoService</tt> against the transaction.
192      * Setting this will ensure that the pre- and post-commit operations perform
193      * the necessary cleanups against the <tt>NodeDaoService</tt>.
194      * <p>
195      * This method can be called repeatedly as long as the service being bound
196      * implements <tt>equals</tt> and <tt>hashCode</tt>.
197      * 
198      * @param nodeDaoService
199      */
200     public static void bindNodeDaoService(NodeDaoService nodeDaoService)
201     {
202         // get transaction-local synchronization
203         TransactionSynchronizationImpl synch = getSynchronization();
204         
205         // bind the service in
206         boolean bound = synch.getNodeDaoServices().add(nodeDaoService);
207         
208         // done
209         if (logger.isDebugEnabled())
210         {
211             logBoundService(nodeDaoService, bound); 
212         }
213     }
214 
215     /**
216      * Method that registers an <tt>IntegrityChecker</tt> against the transaction.
217      * Setting this will ensure that the pre- and post-commit operations perform
218      * the necessary cleanups against the <tt>IntegrityChecker</tt>.
219      * <p>
220      * This method can be called repeatedly as long as the service being bound
221      * implements <tt>equals</tt> and <tt>hashCode</tt>.
222      * 
223      * @param integrityChecker
224      */
225     public static void bindIntegrityChecker(IntegrityChecker integrityChecker)
226     {
227         // get transaction-local synchronization
228         TransactionSynchronizationImpl synch = getSynchronization();
229         
230         // bind the service in
231         boolean bound = synch.getIntegrityCheckers().add(integrityChecker);
232         
233         // done
234         if (logger.isDebugEnabled())
235         {
236             logBoundService(integrityChecker, bound); 
237         }
238     }
239 
240     /**
241      * Method that registers a <tt>LuceneIndexerAndSearcherFactory</tt> against
242      * the transaction.
243      * <p>
244      * Setting this will ensure that the pre- and post-commit operations perform
245      * the necessary cleanups against the <tt>LuceneIndexerAndSearcherFactory</tt>.
246      * <p>
247      * Although bound within a <tt>Set</tt>, it would still be better for the caller
248      * to only bind once per transaction, if possible.
249      * 
250      * @param indexerAndSearcher the Lucene indexer to perform transaction completion
251      *      tasks on
252      */
253     public static void bindLucene(LuceneIndexerAndSearcherFactory indexerAndSearcher)
254     {
255         // get transaction-local synchronization
256         TransactionSynchronizationImpl synch = getSynchronization();
257         
258         // bind the service in
259         boolean bound = synch.getLucenes().add(indexerAndSearcher);
260         
261         // done
262         if (logger.isDebugEnabled())
263         {
264             logBoundService(indexerAndSearcher, bound); 
265         }
266     }
267     
268     /**
269      * Method that registers a <tt>LuceneIndexerAndSearcherFactory</tt> against
270      * the transaction.
271      * <p>
272      * Setting this will ensure that the pre- and post-commit operations perform
273      * the necessary cleanups against the <tt>LuceneIndexerAndSearcherFactory</tt>.
274      * <p>
275      * Although bound within a <tt>Set</tt>, it would still be better for the caller
276      * to only bind once per transaction, if possible.
277      * 
278      * @param indexerAndSearcher the Lucene indexer to perform transaction completion
279      *      tasks on
280      */
281     public static void bindListener(TransactionListener listener)
282     {
283         // get transaction-local synchronization
284         TransactionSynchronizationImpl synch = getSynchronization();
285         
286         // bind the service in
287         boolean bound = synch.getListeners().add(listener);
288         
289         // done
290         if (logger.isDebugEnabled())
291         {
292             logBoundService(listener, bound); 
293         }
294     }
295     
296     /**
297      * Use as part of a debug statement
298      * 
299      * @param service the service to report 
300      * @param bound true if the service was just bound; false if it was previously bound
301      */
302     private static void logBoundService(Object   service, boolean bound)
303     {
304         if (bound)
305         {
306             logger.debug("Bound service: \n" +
307                     "   transaction: " + getTransactionId() + "\n" +
308                     "   service: " + service);
309         }
310         else
311         {
312             logger.debug("Service already bound: \n" +
313                     "   transaction: " + getTransactionId() + "\n" +
314                     "   service: " + service);
315         }
316     }
317     
318     /**
319      * Flush in-transaction resources.  A transaction must be active.
320      * <p>
321      * The flush may include:
322      * <ul>
323      *   <li>{@link NodeDaoService#flush()}</li>
324      *   <li>{@link RuleService#executePendingRules()}</li>
325      *   <li>{@link IntegrityChecker#checkIntegrity()}</li>
326      * </ul>
327      *
328      */
329     public static void flush()
330     {
331         // get transaction-local synchronization
332         TransactionSynchronizationImpl synch = getSynchronization();
333         // flush
334         synch.flush();
335     }
336 
337     /**
338      * Gets the current transaction synchronization instance, which contains the locally bound
339      * resources that are available to {@link #getResource(Object) retrieve} or
340      * {@link #bindResource(Object, Object) add to}.
341      * <p>
342      * This method also ensures that the transaction binding has been performed.
343      * 
344      * @return Returns the common synchronization instance used
345      */
346     private static TransactionSynchronizationImpl getSynchronization()
347     {
348         // ensure synchronizations
349         return registerSynchronizations();
350     }
351     
352     /**
353      * Binds the Alfresco-specific to the transaction resources
354      * 
355      * @return Returns the current or new synchronization implementation
356      */
357     private static TransactionSynchronizationImpl registerSynchronizations()
358     {
359         /*
360          * No thread synchronization or locking required as the resources are all threadlocal
361          */
362         if (!TransactionSynchronizationManager.isSynchronizationActive())
363         {
364             throw new AlfrescoRuntimeException("Transaction must be active and synchronization is required");
365         }
366         TransactionSynchronizationImpl txnSynch =
367             (TransactionSynchronizationImpl) TransactionSynchronizationManager.getResource(RESOURCE_KEY_TXN_SYNCH);
368         if (txnSynch != null)
369         {
370             // synchronization already registered
371             return txnSynch;
372         }
373         // we need a unique ID for the transaction
374         StringBuilder   sb = new StringBuilder  (56);
375         sb.append(System.currentTimeMillis()).append(":").append(GUID.generate());
376         String   txnId = sb.toString();
377         // register the synchronization
378         txnSynch = new TransactionSynchronizationImpl(txnId);
379         TransactionSynchronizationManager.registerSynchronization(txnSynch);
380         // register the resource that will ensure we don't duplication the synchronization
381         TransactionSynchronizationManager.bindResource(RESOURCE_KEY_TXN_SYNCH, txnSynch);
382         // done
383         if (logger.isDebugEnabled())
384         {
385             logger.debug("Bound txn synch: " + txnSynch);
386         }
387         return txnSynch;
388     }
389     
390     /**
391      * Cleans out transaction resources if present
392      */
393     private static void clearSynchronization()
394     {
395         if (TransactionSynchronizationManager.hasResource(RESOURCE_KEY_TXN_SYNCH))
396         {
397             Object   txnSynch = TransactionSynchronizationManager.unbindResource(RESOURCE_KEY_TXN_SYNCH);
398             // done
399             if (logger.isDebugEnabled())
400             {
401                 logger.debug("Unbound txn synch:" + txnSynch);
402             }
403         }
404     }
405     
406     /**
407      * Helper method to rebind the synchronization to the transaction
408      * 
409      * @param txnSynch
410      */
411     private static void rebindSynchronization(TransactionSynchronizationImpl txnSynch)
412     {
413         TransactionSynchronizationManager.bindResource(RESOURCE_KEY_TXN_SYNCH, txnSynch);
414         if (logger.isDebugEnabled())
415         {
416             logger.debug("Bound txn synch: " + txnSynch);
417         }
418     }
419     
420     /**
421      * Handler of txn synchronization callbacks specific to internal
422      * application requirements
423      */
424     private static class TransactionSynchronizationImpl extends TransactionSynchronizationAdapter
425     {
426         private final String   txnId;
427         private final Set  <NodeDaoService> nodeDaoServices;
428         private final Set  <IntegrityChecker> integrityCheckers;
429         private final Set  <LuceneIndexerAndSearcherFactory> lucenes;
430         private final Set  <TransactionListener> listeners;
431         private final Map  <Object  , Object  > resources;
432         
433         /**
434          * Sets up the resource map
435          * 
436          * @param txnId
437          */
438         public TransactionSynchronizationImpl(String   txnId)
439         {
440             this.txnId = txnId;
441             nodeDaoServices = new HashSet  <NodeDaoService>(3);
442             integrityCheckers = new HashSet  <IntegrityChecker>(3);
443             lucenes = new HashSet  <LuceneIndexerAndSearcherFactory>(3);
444             listeners = new HashSet  <TransactionListener>(5);
445             resources = new HashMap  <Object  , Object  >(17);
446         }
447         
448         public String   getTransactionId()
449         {
450             return txnId;
451         }
452 
453         /**
454          * @return Returns a set of <tt>NodeDaoService</tt> instances that will be called
455          *      during end-of-transaction processing
456          */
457         public Set  <NodeDaoService> getNodeDaoServices()
458         {
459             return nodeDaoServices;
460         }
461         
462         /**
463          * @return Returns a set of <tt>IntegrityChecker</tt> instances that will be called
464          *      during end-of-transaction processing
465          */
466         public Set  <IntegrityChecker> getIntegrityCheckers()
467         {
468             return integrityCheckers;
469         }
470 
471         /**
472          * @return Returns a set of <tt>LuceneIndexerAndSearcherFactory</tt> that will be called
473          *      during end-of-transaction processing
474          */
475         public Set  <LuceneIndexerAndSearcherFactory> getLucenes()
476         {
477             return lucenes;
478         }
479         
480         /**
481          * @return Returns a set of <tt>TransactionListener<tt> instances that will be called
482          *      during end-of-transaction processing
483          */
484         public Set  <TransactionListener> getListeners()
485         {
486             return listeners;
487         }
488         
489         /**
490          * @return Returns the listeners in a list disconnected from the original set
491          */
492         private List  <TransactionListener> getListenersIterable()
493         {
494             return new ArrayList  <TransactionListener>(listeners);
495         }
496 
497         public String   toString()
498         {
499             StringBuilder   sb = new StringBuilder  (50);
500             sb.append("TransactionSychronizationImpl")
501               .append("[ txnId=").append(txnId)
502               .append(", node service=").append(nodeDaoServices.size())
503               .append(", integrity=").append(integrityCheckers.size())
504               .append(", indexers=").append(lucenes.size())
505               .append(", resources=").append(resources)
506               .append("]");
507             return sb.toString();
508         }
509 
510         /**
511          * Performs the in-transaction flushing.  Typically done during a transaction or
512          * before commit.
513          */
514         public void flush()
515         {
516             // check integrity
517             for (IntegrityChecker integrityChecker : integrityCheckers)
518             {
519                 integrityChecker.checkIntegrity();
520             }
521             // flush listeners
522             for (TransactionListener listener : getListenersIterable())
523             {
524                 listener.flush();
525             }
526         }
527         
528         /**
529          * @see AlfrescoTransactionSupport#SESSION_SYNCHRONIZATION_ORDER
530          */
531         @Override  
532         public int getOrder()
533         {
534             return AlfrescoTransactionSupport.SESSION_SYNCHRONIZATION_ORDER;
535         }
536 
537         @Override  
538         public void suspend()
539         {
540             if (logger.isDebugEnabled())
541             {
542                 logger.debug("Suspending transaction: " + this);
543             }
544             AlfrescoTransactionSupport.clearSynchronization();
545         }
546 
547         @Override  
548         public void resume()
549         {
550             if (logger.isDebugEnabled())
551             {
552                 logger.debug("Resuming transaction: " + this);
553             }
554             AlfrescoTransactionSupport.rebindSynchronization(this);
555         }
556 
557         /**
558          * Pre-commit cleanup.
559          * <p>
560          * Ensures that the session resources are {@link #flush() flushed}.
561          * The Lucene indexes are then prepared.
562          */
563         @Override  
564         public void beforeCommit(boolean readOnly)
565         {
566             if (logger.isDebugEnabled())
567             {
568                 logger.debug("Before commit " + (readOnly ? "read-only" : "" ) + ": " + this);
569             }
570             // get the txn ID
571             TransactionSynchronizationImpl synch = (TransactionSynchronizationImpl)
572                     TransactionSynchronizationManager.getResource(RESOURCE_KEY_TXN_SYNCH);
573             if (synch == null)
574             {
575                 throw new AlfrescoRuntimeException("No synchronization bound to thread");
576             }
577 
578             // These are still considered part of the transaction so are executed here
579             for (TransactionListener listener : getListenersIterable())
580             {
581                 listener.beforeCommit(readOnly);
582             }
583 
584             // flush
585             flush();
586             // prepare the indexes
587             for (LuceneIndexerAndSearcherFactory lucene : lucenes)
588             {
589                 lucene.prepare();
590             }
591         }
592         
593         @Override  
594         public void beforeCompletion()
595         {
596             if (logger.isDebugEnabled())
597             {
598                 logger.debug("Before completion: " + this);
599             }
600             // notify listeners
601             for (TransactionListener listener : getListenersIterable())
602             {
603                 listener.beforeCompletion();
604             }
605         }
606                
607 
608         @Override  
609         public void afterCompletion(int status)
610         {
611             String   statusStr = "unknown";
612             switch (status)
613             {
614                 case TransactionSynchronization.STATUS_COMMITTED:
615                     statusStr = "committed";
616                     break;
617                 case TransactionSynchronization.STATUS_ROLLED_BACK:
618                     statusStr = "rolled-back";
619                     break;
620                 default:
621             }
622             if (logger.isDebugEnabled())
623             {
624                 logger.debug("After completion (" + statusStr + "): " + this);
625             }
626             
627             // commit/rollback Lucene
628             for (LuceneIndexerAndSearcherFactory lucene : lucenes)
629             {
630                 try
631                 {
632                     if (status  == TransactionSynchronization.STATUS_COMMITTED)
633                     {
634                         lucene.commit();
635                     }
636                     else
637                     {
638                         lucene.rollback();
639                     }
640                 }
641                 catch (RuntimeException   e)
642                 {
643                     logger.error("After completion (" + statusStr + ") Lucene exception", e);
644                 }
645             }
646             
647             List  <TransactionListener> iterableListeners = getListenersIterable();
648             // notify listeners
649             if (status  == TransactionSynchronization.STATUS_COMMITTED)
650             {
651                 for (TransactionListener listener : iterableListeners)
652                 {
653                     try
654                     {
655                         listener.afterCommit();
656                     }
657                     catch (RuntimeException   e)
658                     {
659                         logger.error("After completion (" + statusStr + ") listener exception: \n" +
660                                 "   listener: " + listener,
661                                 e);
662                     }
663                 }
664             }
665             else
666             {
667                 for (TransactionListener listener : iterableListeners)
668                 {
669                     try
670                     {
671                         listener.afterRollback();
672                     }
673                     catch (RuntimeException   e)
674                     {
675                         logger.error("After completion (" + statusStr + ") listener exception: \n" +
676                                 "   listener: " + listener,
677                                 e);
678                     }
679                 }
680             }
681             
682             // clear the thread's registrations and synchronizations
683             AlfrescoTransactionSupport.clearSynchronization();
684         }
685     }
686 }
687
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags