TransactionLibrary.java

  1. /*
  2.  * GovWay - A customizable API Gateway 
  3.  * https://govway.org
  4.  * 
  5.  * Copyright (c) 2005-2025 Link.it srl (https://link.it).
  6.  * 
  7.  * This program is free software: you can redistribute it and/or modify
  8.  * it under the terms of the GNU General Public License version 3, as published by
  9.  * the Free Software Foundation.
  10.  *
  11.  * This program is distributed in the hope that it will be useful,
  12.  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13.  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14.  * GNU General Public License for more details.
  15.  *
  16.  * You should have received a copy of the GNU General Public License
  17.  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  18.  *
  19.  */
  20. package org.openspcoop2.monitor.engine.transaction;

  22. import java.util.ArrayList;
  23. import java.util.Date;
  24. import java.util.HashMap;
  25. import java.util.List;
  26. import java.util.concurrent.ExecutorService;
  27. import java.util.concurrent.Executors;

  29. import org.openspcoop2.core.commons.dao.DAOFactory;
  30. import org.openspcoop2.core.id.IDServizio;
  31. import org.openspcoop2.core.transazioni.Transazione;
  32. import org.openspcoop2.core.transazioni.TransazioneInfo;
  33. import org.openspcoop2.core.transazioni.dao.ITransazioneInfoService;
  34. import org.openspcoop2.core.transazioni.dao.ITransazioneInfoServiceSearch;
  35. import org.openspcoop2.core.transazioni.dao.ITransazioneServiceSearch;
  36. import org.openspcoop2.generic_project.beans.NonNegativeNumber;
  37. import org.openspcoop2.generic_project.exception.NotFoundException;
  38. import org.openspcoop2.generic_project.expression.IExpression;
  39. import org.openspcoop2.generic_project.expression.IPaginatedExpression;
  40. import org.openspcoop2.generic_project.expression.SortOrder;
  41. import org.openspcoop2.monitor.engine.config.BasicServiceLibrary;
  42. import org.openspcoop2.monitor.engine.config.BasicServiceLibraryReader;
  43. import org.openspcoop2.monitor.engine.config.TransactionServiceLibrary;
  44. import org.openspcoop2.monitor.engine.config.TransactionServiceLibraryReader;
  45. import org.openspcoop2.utils.Utilities;
  46. import org.slf4j.Logger;

  48. /**
  49.  * TransactionLibrary
  50.  *
  51.  * @author Poli Andrea (apoli@link.it)
  52.  * @author $Author$
  53.  * @version $Rev$, $Date$
  54.  */
  55. public class TransactionLibrary {
  56.     
  57.     private static final String ID_TRANSACTION_PROCESSOR = TransactionLibrary.class.getName();
  58.     
  59.     private static HashMap<String,TransactionServiceLibrary> pluginTransazioni = new HashMap<String, TransactionServiceLibrary>();

  61.     public static TransactionServiceLibrary getTransactionServiceLibrary(IDServizio idServizio,BasicServiceLibraryReader basicServiceLibraryReader,
  62.             TransactionServiceLibraryReader transactionServiceLibraryReader,Logger log) throws Exception{
  63.         TransactionServiceLibrary transactionServiceLibrary = null;
  64.         if(pluginTransazioni.containsKey(idServizio.toString())){
  65.             transactionServiceLibrary = pluginTransazioni.get(idServizio.toString());
  66.         }
  67.         else{
  68.             transactionServiceLibrary = initAndGetTransactionServiceLibrary(idServizio, basicServiceLibraryReader, transactionServiceLibraryReader, log);
  69.         }
  70.         return transactionServiceLibrary;
  71.     }
  72.     private static synchronized TransactionServiceLibrary initAndGetTransactionServiceLibrary(IDServizio idServizio,BasicServiceLibraryReader basicServiceLibraryReader,
  73.             TransactionServiceLibraryReader transactionServiceLibraryReader,Logger log) throws Exception{
  74.         TransactionServiceLibrary transactionServiceLibrary = null;
  75.         if(pluginTransazioni.containsKey(idServizio.toString())){
  76.             transactionServiceLibrary = pluginTransazioni.get(idServizio.toString());
  77.         }
  78.         else{
  79.             BasicServiceLibrary basicServiceLibrary = basicServiceLibraryReader.read(idServizio, log);
  80.             if(basicServiceLibrary!=null){
  81.                 transactionServiceLibrary = transactionServiceLibraryReader.readConfigurazioneTransazione(basicServiceLibrary, log);
  82.             }
  83.             if(transactionServiceLibrary==null){
  84.                 transactionServiceLibrary = new TransactionServiceLibrary();
  85.             }
  86.             pluginTransazioni.put(idServizio.toString(), transactionServiceLibrary);
  87.         }
  88.         return transactionServiceLibrary;
  89.     }
  90.     
  91.     public static void process(Logger logCore, DAOFactory daoFactory, boolean debug,
  92.             int poolSize, int msgForThread) {
  93.         
  94.         int totaleTransazioni = 0;
  95.         ExecutorService threadsPool = null;
  96.         try {
  97.             
  98.             org.openspcoop2.core.transazioni.dao.IServiceManager serviceManagerTransazioni = (org.openspcoop2.core.transazioni.dao.IServiceManager) 
  99.                     daoFactory.getServiceManager(org.openspcoop2.core.transazioni.utils.ProjectInfo.getInstance());
  100.             org.openspcoop2.core.plugins.dao.IServiceManager serviceManagerPluginsBase = (org.openspcoop2.core.plugins.dao.IServiceManager) 
  101.                     daoFactory.getServiceManager(org.openspcoop2.core.plugins.utils.ProjectInfo.getInstance());
  102.             org.openspcoop2.monitor.engine.config.transazioni.dao.IServiceManager serviceManagerPluginsTransazioni = (org.openspcoop2.monitor.engine.config.transazioni.dao.IServiceManager) 
  103.                     daoFactory.getServiceManager(org.openspcoop2.monitor.engine.config.transazioni.utils.ProjectInfo.getInstance());
  104.             org.openspcoop2.core.commons.search.dao.IServiceManager serviceManagerUtils = (org.openspcoop2.core.commons.search.dao.IServiceManager) 
  105.                     daoFactory.getServiceManager(org.openspcoop2.core.commons.search.utils.ProjectInfo.getInstance());
  106.             
  107.             threadsPool = Executors.newFixedThreadPool(poolSize);
  108.             
  109.             
  110.             // ServiceLibraryManager
  111.             BasicServiceLibraryReader basicServiceLibraryReader = new BasicServiceLibraryReader(serviceManagerPluginsBase, serviceManagerUtils, debug);
  112.             TransactionServiceLibraryReader transactionServiceLibraryReader = new TransactionServiceLibraryReader(serviceManagerPluginsTransazioni, debug);
  113.             
  114.             
  115.             // Devo recuperare l'ultima data di esecuzione del processor.
  116.             ITransazioneInfoService transazioneInfoDAO = serviceManagerTransazioni.getTransazioneInfoService();
  117.             ITransazioneInfoServiceSearch  transazioneInfoSearchDAO = serviceManagerTransazioni.getTransazioneInfoServiceSearch();
  118.             IExpression exprTransactionInfo = transazioneInfoSearchDAO.newExpression();
  119.             exprTransactionInfo.
  120.                 equals(TransazioneInfo.model().TIPO, ID_TRANSACTION_PROCESSOR);
  121.             TransazioneInfo tInfo = null;
  122.             Date now = new Date();
  123.             Date lastRunning = null;
  124.             try{
  125.                 tInfo = transazioneInfoSearchDAO.find(exprTransactionInfo);
  126.                 lastRunning = tInfo.getData();
  127.             }catch(NotFoundException notFound){
  128.                 // ignore
  129.             }
  130.             
  131.             
  132.             // Esamino le transazioni da processare
  133.             ITransazioneServiceSearch transazioneSearchDAO = serviceManagerTransazioni.getTransazioneServiceSearch();
  134.             IPaginatedExpression exprTransazioni = transazioneSearchDAO.newPaginatedExpression();
  135.             exprTransazioni.and();
  136.             if(lastRunning!=null){
  137.                 exprTransazioni.greaterThan(Transazione.model().DATA_INGRESSO_RICHIESTA, lastRunning);
  138.             }
  139.             exprTransazioni.lessEquals(Transazione.model().DATA_INGRESSO_RICHIESTA, now);
  140.                 
  141.             NonNegativeNumber countTransazioniObject = transazioneSearchDAO.count(transazioneSearchDAO.toExpression(exprTransazioni));
  142.             long countTransazioni = countTransazioniObject.longValue();
  143.             logCore.info("Trovate ["+countTransazioni+"] transazioni da processare ...");
  144.             
  145.             
  146.             exprTransazioni.sortOrder(SortOrder.ASC).addOrder(Transazione.model().DATA_INGRESSO_RICHIESTA);
  147.             int limit = msgForThread*poolSize;
  148.             exprTransazioni.limit(limit).offset(0);
  149.             List<String> list = transazioneSearchDAO.findAllIds(exprTransazioni);
  150.             
  151.             Date lastTransactionProcessed = null;
  152.             
  153.             int threadNumber = 1;
  154.             
  155.             while(list!=null && list.size()>0){
  156.                 
  157.                 threadNumber = 1;
  158.                 
  159.                 lastTransactionProcessed = null;
  160.                 
  161.                 List<String> listIdTransazioni = new ArrayList<>();
  162.                 
  163.                 List<TransactionProcessorThread> processorThreads = new ArrayList<TransactionProcessorThread>();
  164.                 
  165.                 for (String idTransazione : list) {

  167.                     listIdTransazioni.add(idTransazione);
  168.                     
  169.                     if(listIdTransazioni.size()==msgForThread){
  170.                         TransactionProcessorThread pt = new TransactionProcessorThread(listIdTransazioni, logCore, 
  171.                                 basicServiceLibraryReader, transactionServiceLibraryReader, daoFactory,
  172.                                 threadNumber, debug);
  173.                         threadsPool.execute(pt);
  174.                         processorThreads.add(pt);
  175.                         logCore.info("Avviato Thread #"+threadNumber+" ...");
  176.                         listIdTransazioni = new ArrayList<>();
  177.                         threadNumber++;
  178.                     }
  179.                     
  180.                 }

  182.                 if(listIdTransazioni.size()>0){
  183.                     // sono rimasti messaggi da far gestire ad un thread.
  184.                     TransactionProcessorThread pt = new TransactionProcessorThread(listIdTransazioni, logCore,
  185.                             basicServiceLibraryReader, transactionServiceLibraryReader, daoFactory,
  186.                             threadNumber, debug);
  187.                     threadsPool.execute(pt);
  188.                     processorThreads.add(pt);
  189.                     logCore.info("Avviato (coda msg non piena) Thread #"+threadNumber+" ...");
  190.                     listIdTransazioni = new ArrayList<>();
  191.                     threadNumber++;
  192.                 }
  193.                 
  194.                 int timeout = 10;
  195.                 boolean terminated = false;
  196.                 while(terminated == false){
  197.                     logCore.info((threadNumber-1)+" threads avviati correttamente, attendo terminazione (timeout "+timeout+"s) ...");
  198.                     for (int i = 0; i < timeout*4; i++) {
  199.                         boolean tmpTerminated = true;
  200.                         for (TransactionProcessorThread processorThread : processorThreads) {
  201.                             if(processorThread.isFinished()==false){
  202.                                 tmpTerminated = false;
  203.                                 break;
  204.                             }
  205.                         }
  206.                         if(tmpTerminated==false){
  207.                             Utilities.sleep(250);
  208.                         }
  209.                         else{
  210.                             terminated = true;
  211.                         }
  212.                     }
  213.                 }
  214.                 
  215.                 
  216.                 if(debug)
  217.                     logCore.debug("Check Last Update ...");
  218.                     
  219.                 String error = null;
  220.                 for (TransactionProcessorThread processorThread : processorThreads) {
  221.                     if(processorThread.getLastDateTransaction()!=null){
  222.                         if(lastTransactionProcessed==null){
  223.                             lastTransactionProcessed = processorThread.getLastDateTransaction();
  224.                         }
  225.                         else if(processorThread.getLastDateTransaction().after(lastTransactionProcessed)){
  226.                             lastTransactionProcessed = processorThread.getLastDateTransaction();
  227.                         }
  228.                     }
  229.                     if(error==null)
  230.                         error = processorThread.getError();
  231.                 }
  232.                 if(error!=null){
  233.                     logCore.error("Threads terminati con errori");
  234.                     return;
  235.                 }
  236.                 if(lastTransactionProcessed!=null){
  237.                     if(tInfo==null){
  238.                         tInfo = new TransazioneInfo();
  239.                         tInfo.setData(lastTransactionProcessed);
  240.                         tInfo.setTipo(ID_TRANSACTION_PROCESSOR);
  241.                         transazioneInfoDAO.create(tInfo);
  242.                     }else{
  243.                         tInfo.setData(lastTransactionProcessed);
  244.                         transazioneInfoDAO.update(tInfo);
  245.                     }
  246.                 }
  247.                 
  248.                 totaleTransazioni = totaleTransazioni + list.size();
  249.                     
  250.                 // al termine del ciclo si aggiorna il timestamp dell'ultima transazione processata
  251.                 logCore.info(totaleTransazioni+"/["+countTransazioni+"] transazioni processate (lastUpdate:"+lastTransactionProcessed+") ...");
  252.                 exprTransazioni.offset(totaleTransazioni);
  253.                 list = transazioneSearchDAO.findAllIds(exprTransazioni);
  254.             }
  255.                         
  256.             logCore.info(countTransazioni+"] transazioni processate correttamente");
  257.             return;
  258.             
  259.         } catch (Exception e) {
  260.             logCore.error("TransactionProcessor ha riscontrato un errore: "+e.getMessage(),e);
  261.             return;
  262.         }finally{
  263.             try{
  264.                 logCore.info("Shutdown pool ...");
  265.                 threadsPool.shutdown(); 
  266.                 logCore.info("Shutdown pool ok");
  267.             }catch(Throwable e){
  268.                 // ignore
  269.             }
  270.         }


  273.     }

  275. }
Created with JaCoCo 0.8.8.202204050719