SmistatoreThread.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.  */


  22. package org.openspcoop2.web.ctrlstat.gestori;

  24. import java.sql.Connection;
  25. import java.util.List;
  26. import java.util.Map;
  27. import java.util.Properties;

  29. import javax.jms.JMSException;
  30. import javax.jms.ObjectMessage;
  31. import javax.jms.Queue;
  32. import javax.jms.QueueConnection;
  33. import javax.jms.QueueConnectionFactory;
  34. import javax.jms.QueueReceiver;
  35. import javax.jms.QueueSession;
  36. import javax.naming.InitialContext;

  38. import org.openspcoop2.core.registry.constants.PddTipologia;
  39. import org.openspcoop2.pdd.config.OpenSPCoop2ConfigurationException;
  40. import org.openspcoop2.utils.Utilities;
  41. import org.openspcoop2.utils.transport.jms.ExceptionListenerJMS;
  42. import org.openspcoop2.web.ctrlstat.config.ConsoleProperties;
  43. import org.openspcoop2.web.ctrlstat.config.DatasourceProperties;
  44. import org.openspcoop2.web.ctrlstat.core.ControlStationLogger;
  45. import org.openspcoop2.web.ctrlstat.core.DBManager;
  46. import org.openspcoop2.web.ctrlstat.core.OperazioneDaSmistare;
  47. import org.openspcoop2.web.ctrlstat.costanti.CostantiControlStation;
  48. import org.openspcoop2.web.ctrlstat.costanti.OperationsParameter;
  49. import org.openspcoop2.web.ctrlstat.costanti.TipoOggettoDaSmistare;
  50. import org.openspcoop2.web.ctrlstat.dao.PdDControlStation;
  51. import org.openspcoop2.web.ctrlstat.servlet.pdd.PddCore;
  52. import org.openspcoop2.web.lib.queue.ClassQueue;
  53. import org.openspcoop2.web.lib.queue.QueueOperation;
  54. import org.openspcoop2.web.lib.queue.QueueParameter;
  55. import org.openspcoop2.web.lib.queue.config.QueueProperties;
  56. import org.openspcoop2.web.lib.queue.costanti.Operazione;
  57. import org.openspcoop2.web.lib.queue.costanti.TipoOperazione;
  58. import org.slf4j.Logger;

  60. /**
  61.  * SmistatoreThread
  62.  * 
  63.  * @author Andrea Poli (apoli@link.it)
  64.  * @author Stefano Corallo (corallo@link.it)
  65.  * @author Sandra Giangrandi (sandra@link.it)
  66.  * @author $Author$
  67.  * @version $Rev$, $Date$
  68.  * 
  69.  */
  70. public class SmistatoreThread extends Thread {

  72.     /** Logger utilizzato per debug. */
  73.     private static Logger log = null;
  74.     
  75.     /** run */
  76.     private boolean stop = false;
  77.     private boolean isRunning = false;
  78.     public boolean isRunning() {
  79.         return this.isRunning;
  80.     }

  82.     private DBManager dbm;
  83.     private Connection con;

  85.     private ExceptionListenerJMS exceptionListenerJMS = new ExceptionListenerJMS();

  87.     private ConsoleProperties consoleProperties;
  88.     
  89.     private QueueProperties queueProperties;
  90.     
  91.     private DatasourceProperties datasourceProperties;
  92.     
  93.     /** Costruttore 
  94.      * @throws OpenSPCoop2ConfigurationException */
  95.     public SmistatoreThread() throws OpenSPCoop2ConfigurationException {

  97.         // configuro il logger
  98.         SmistatoreThread.log = ControlStationLogger.getSmistatoreLogger();

  100.         this.dbm = DBManager.getInstance();
  101.         
  102.         this.consoleProperties = ConsoleProperties.getInstance();
  103.         
  104.         this.queueProperties = QueueProperties.getInstance();
  105.         
  106.         this.datasourceProperties = DatasourceProperties.getInstance();
  107.     }

  109.     /**
  110.      * Metodo che fa partire il Thread.
  111.      * 
  112.      * @since 0.4
  113.      */
  114.     @Override
  115.     public void run() {

  117.         this.isRunning = true;
  118.         
  119.         // Controllo se dbmanager inizializzato
  120.         if (!DBManager.isInitialized()) {
  121.             SmistatoreThread.log.info("Inizializzazione di " + this.getClass().getSimpleName() + " non riuscito perche' DBManager non INIZIALIZZATO");
  122.             SmistatoreThread.log.info(this.getClass().getName() + " Non AVVIATO!");
  123.             return;
  124.         }

  126.         String jmsConnectionFactory = null;
  127.         Properties jmsConnectionFactoryContext = null;
  128.         
  129.         String smistatoreQueue = null;
  130.         String registroServiziQueue = null;
  131.         String gestoreEventiQueue = null;
  132.         String pddQueuePrefix = null;
  133.         
  134.         boolean enginePDD = false;
  135.         boolean engineRegistro = false;
  136.         boolean engineGestoreEventi = false;
  137.         
  138.         boolean singlePdD = true;
  139.         String tipoDatabase = null;
  140.         
  141.         try{
  142.             // Leggo le informazioni da queue.properties
  143.             jmsConnectionFactory = this.queueProperties.getConnectionFactory();
  144.             jmsConnectionFactoryContext = this.queueProperties.getConnectionFactoryContext();
  145.             
  146.             // Leggo le informazioni da console.properties
  147.             
  148.             // nomi code
  149.             smistatoreQueue = this.consoleProperties.getGestioneCentralizzataNomeCodaSmistatore();
  150.             registroServiziQueue = this.consoleProperties.getGestioneCentralizzataNomeCodaRegistroServizi();
  151.             gestoreEventiQueue = this.consoleProperties.getGestioneCentralizzataNomeCodaGestoreEventi();
  152.             pddQueuePrefix = this.consoleProperties.getGestioneCentralizzataPrefissoNomeCodaConfigurazionePdd();
  153.             
  154.             // Abilitazione Engine
  155.             enginePDD = this.consoleProperties.isGestioneCentralizzataSincronizzazionePdd();
  156.             engineRegistro = this.consoleProperties.isGestioneCentralizzataSincronizzazioneRegistro();
  157.             engineGestoreEventi = this.consoleProperties.isGestioneCentralizzataSincronizzazioneGestoreEventi();
  158.             
  159.             // Altre informazioni
  160.             singlePdD = this.consoleProperties.isSinglePdD();
  161.             
  162.             // Database Info
  163.             tipoDatabase = this.datasourceProperties.getTipoDatabase();
  164.             
  165.         }catch(Exception e){
  166.             SmistatoreThread.log.info("Smistatore non avviato, sono stati rilevati errori durante la lettura delle configurazione: "+e.getMessage(),e);
  167.             return;
  168.         }
  169.         
  170.         if (singlePdD) {
  171.             SmistatoreThread.log.info("Smistatore non avviato: govwayConsole avviata in singlePdD mode.");
  172.             return;
  173.         }

  175.         // Configurazione JMS
  176.         SmistatoreThread.log.debug("Smistatore: Avvio Servizio di Gestione Operazioni, Registro[" + engineRegistro + "] Pdd[" + enginePDD + "] GestoreEventi[" + engineGestoreEventi + "]");
  177.         QueueReceiver receiver = null;
  178.         Queue queue = null;
  179.         QueueConnectionFactory qcf = null;
  180.         QueueConnection qc = null;
  181.         QueueSession qs = null;
  182.         boolean trovato = false;
  183.         int i = 0;
  184.         SmistatoreThread.log.debug("Smistatore: Inizializzazione Receiver ...");
  185.         while (!trovato && (i < 600000)) {
  186.             try {
  187.                 InitialContext ctx = new InitialContext(jmsConnectionFactoryContext);
  188.                 queue = (Queue) ctx.lookup(smistatoreQueue);
  189.                 qcf = (QueueConnectionFactory) ctx.lookup(jmsConnectionFactory);
  190.                 qc = qcf.createQueueConnection();
  191.                 qc.setExceptionListener(this.exceptionListenerJMS);
  192.                 qs = qc.createQueueSession(true, -1);
  193.                 receiver = qs.createReceiver(queue);
  194.                 qc.start();
  195.                 ctx.close();
  196.                 SmistatoreThread.log.debug("Smistatore: Inizializzazione Receiver effettuata.");
  197.                 trovato = true;
  198.             } catch (Exception e) {
  199.                 i = i + 10000;
  200.                 Utilities.sleep(10000);
  201.             }
  202.         }

  204.         if (!trovato) {
  205.             SmistatoreThread.log.error("Smistatore: Inizializzazione Receiver non effettuata");
  206.             return;
  207.         }

  209.         // Avvio Gestione Operazioni
  210.         boolean riconnessioneConErrore = false;
  211.         while (this.stop == false) {

  213.             try {

  215.                 // riconnessione precedente non riuscita.....
  216.                 if (riconnessioneConErrore) {
  217.                     throw new JMSException("RiconnessioneJMS non riuscita...");
  218.                 }
  219.                 // Controllo ExceptionListenerJMS
  220.                 if (this.exceptionListenerJMS.isConnessioneCorrotta()) {
  221.                     SmistatoreThread.log.error("ExceptionJMSListener ha rilevato una connessione jms corrotta", this.exceptionListenerJMS.getException());
  222.                     throw new JMSException("ExceptionJMSListener ha rilevato una connessione jms corrotta: " + this.exceptionListenerJMS.getException().getMessage());
  223.                 }

  225.                 SmistatoreThread.log.info("Smistatore: Ricezione operazione...");
  226.                 ObjectMessage richiesta = null;
  227.                 while (this.stop == false) {
  228.                     richiesta = (ObjectMessage) receiver.receive(CostantiControlStation.INTERVALLO_RECEIVE);
  229.                     if (richiesta != null) {
  230.                         break;
  231.                     }
  232.                 }
  233.                 if (this.stop == true) {
  234.                     break;
  235.                 }

  237.                 // Ricezione Operazione
  238.                 OperazioneDaSmistare operazione = null;
  239.                 try {
  240.                     operazione = (OperazioneDaSmistare) richiesta.getObject();
  241.                 } catch (Exception e) {
  242.                     SmistatoreThread.log.error("Smistatore: Ricevuta richiesta con tipo errato:" + e.toString());
  243.                     qs.commit();
  244.                     continue;
  245.                 }

  247.                 String idOperazione = richiesta.getStringProperty("ID");

  249.                 SmistatoreThread.log.info(CostantiControlStation.OPERATIONS_DELIMITER+"Smistatore: Ricevuta richiesta di operazione con ID: " + idOperazione);
  250.                 SmistatoreThread.log.debug("Smistatore: Dati operazione ricevuta idTab[" + operazione.getIDTable() + "] operazione[" + operazione.getOperazione() + "] pdd[" + operazione.getPdd() + "] oggetto[" + operazione.getOggetto() + "]");

  252.                 if ((operazione.getOperazione() == null)) {
  253.                     SmistatoreThread.log.error("Smistatore: Ricevuta richiesta con parametri scorretti.");
  254.                     qs.commit();
  255.                     continue;
  256.                 }

  258.                 if ((Operazione.change.equals(operazione.getOperazione()) == false) && (Operazione.add.equals(operazione.getOperazione()) == false) && (Operazione.del.equals(operazione.getOperazione()) == false)) {
  259.                     SmistatoreThread.log.error("Smistatore: Operazione [" + operazione.getOperazione() + "] non supportata dal gestore");
  260.                     qs.commit();
  261.                     continue;
  262.                 }

  264.                 if ( (operazione.getOggetto() == null) || (operazione.getIDTable() < 0)) {
  265.                     SmistatoreThread.log.error("Smistatore: Ricevuta richiesta con parametri scorretti.");
  266.                     qs.commit();
  267.                     continue;
  268.                 }

  270.                 // Guardo che tipo di operazione ho in coda...
  271.                 // Se e' un'operazione per il registro, la metto nella coda
  272.                 // OperazioniGestoreRegistroServizi
  273.                 // Se e' un'operazione per la pdd, la metto nella coda del pdd
  274.                 // interessato

  276.                 this.con = this.dbm.getConnection();

  278.                 Operazione operazioneTipologia = operazione.getOperazione();
  279.                 String su = operazione.getSuperuser();
  280.                 TipoOggettoDaSmistare tipoOggettoDaSmistare = operazione.getOggetto();
  281.                 String pdd = operazione.getPdd();

  283.                 // Preparo un oggetto di tipo PetraOperation e poi chiamo
  284.                 // la insertQueue
  285.                 QueueOperation queueOperation = new QueueOperation();
  286.                 queueOperation.setTipoOperazione(TipoOperazione.webService);
  287.                 queueOperation.setOperazione(operazioneTipologia);
  288.                 queueOperation.setSuperuser(su);

  290.                 // disabilito il commit
  291.                 this.con.setAutoCommit(false);

  293.                 // OggettoClassQueue
  294.                 ClassQueue cq = null;
  295.                 try {
  296.                     cq = new ClassQueue(this.con, tipoDatabase, qs);
  297.                 } catch (Exception e) {
  298.                     SmistatoreThread.log.error("Smistatore: Inizializzazione ClassQueue non effettuata: " + e.getMessage());
  299.                     qs.rollback();
  300.                     continue;
  301.                 }

  303.                 if (tipoOggettoDaSmistare != null) {
  304.                     long idTable = operazione.getIDTable();

  306.                     String filter = "[" + operazione.getIDTable() + "]";
  307.                     filter += "[" + tipoOggettoDaSmistare.name() + "]";
  308.                     filter += "[" + operazione.getOperazione() + "]";

  310.                     queueOperation.addParametro(new QueueParameter("Oggetto", tipoOggettoDaSmistare.name()));
  311.                     queueOperation.addParametro(new QueueParameter("IDTable", "" + idTable));

  313.                     Map<OperationsParameter, List<String>> params = operazione.getParameters();
  314.                     if(params!=null && !params.isEmpty()) {
  315.                         for (OperationsParameter key : params.keySet()) {
  316.                             // Per ogni parametro presente nell'operazione da smistare
  317.                             // creo un nuovo PetraParameter con nome key.getNome() e
  318.                             // valore il valore associato nella tabella
  319.     
  320.                             List<String> values = params.get(key);
  321.                             for (String value : values) {
  322.                                 queueOperation.addParametro(new QueueParameter(key.getNome(), value));
  323.                                 filter += "[" + value + "]";
  324.                             }
  325.     
  326.                         }
  327.                     }

  329.                     // Smisto l'operazione

  331.                     /* ***** REGISTRO ***** */
  332.                     
  333.                     if (tipoOggettoDaSmistare.equals(TipoOggettoDaSmistare.soggetto) || 
  334.                             tipoOggettoDaSmistare.equals(TipoOggettoDaSmistare.servizio) || 
  335.                             tipoOggettoDaSmistare.equals(TipoOggettoDaSmistare.ruolo) || 
  336.                             tipoOggettoDaSmistare.equals(TipoOggettoDaSmistare.accordo) || 
  337.                             tipoOggettoDaSmistare.equals(TipoOggettoDaSmistare.accordoCooperazione) || 
  338. //                          tipoOggettoDaSmistare.equals(TipoOggettoDaSmistare.fruitore) || 
  339.                             tipoOggettoDaSmistare.equals(TipoOggettoDaSmistare.pdd)) {
  340.                         // Operazione per il registro
  341.                         if (engineRegistro) {
  342.                             QueueOperation queueOperationRegistro = (QueueOperation) queueOperation.clone();
  343.                             queueOperationRegistro.setTipoOperazione(TipoOperazione.webService);
  344.                             if (cq.insertQueue(registroServiziQueue, queueOperationRegistro, filter) == 0) {
  345.                                 SmistatoreThread.log.error("Smistatore: Si e' verificato un problema durante l'inserimento in coda.");
  346.                                 qs.rollback();
  347.                                 this.con.rollback();
  348.                                 this.dbm.releaseConnection(this.con);
  349.                                 continue;
  350.                             }
  351.                         } else {
  352.                             SmistatoreThread.log.info("Smistatore: sincronizzazione Registro Servizi non abilitata.");
  353.                         }
  354.                     }

  356.                     /* ***** PDD ***** */
  357.                     
  358.                     if ((pdd != null) && !pdd.equals("") && !pdd.equals("-") ) {
  359.                         // Operazione per il pdd
  360.                         if (enginePDD) {
  361.                             // Qualcuno avra' provveduto a creare una coda per
  362.                             // il pdd, che si chiama come il pdd stesso
  363.                             PddCore pddCore = new PddCore();
  364.                             PdDControlStation myPdd = pddCore.getPdDControlStation(pdd);
  365.                             String tipoPdd = myPdd.getTipo();

  367.                             if (PddTipologia.OPERATIVO.toString().equals(tipoPdd)) {
  368.                                 QueueOperation queueOperationPdD = (QueueOperation) queueOperation.clone();
  369.                                 queueOperationPdD.setTipoOperazione(TipoOperazione.webService);
  370.                                 if (cq.insertQueue(pddQueuePrefix + pdd, queueOperationPdD, filter) == 0) {
  371.                                     SmistatoreThread.log.error("Smistatore: Si e' verificato un problema durante l'inserimento in coda.");
  372.                                     qs.rollback();
  373.                                     this.con.rollback();
  374.                                     this.dbm.releaseConnection(this.con);
  375.                                     continue;
  376.                                 }
  377.                             } else {
  378.                                 SmistatoreThread.log.warn("Smistatore: Inserimento in coda non effettuato causa NAL [" + pdd + "] Tipo [" + tipoPdd + "] ");
  379.                             }
  380.                         } else {
  381.                             SmistatoreThread.log.info("Smistatore: sincronizzazione Nal non abilitata.");
  382.                         }
  383.                     }

  385.                     /* ***** Gestore Eventi ***** */
  386.                     
  387.                     if ( ((gestoreEventiQueue != null) && !gestoreEventiQueue.equals("")) && 
  388.                             (
  389.                             tipoOggettoDaSmistare.equals(TipoOggettoDaSmistare.soggetto) || 
  390.                             tipoOggettoDaSmistare.equals(TipoOggettoDaSmistare.servizio) || 
  391.                             tipoOggettoDaSmistare.equals(TipoOggettoDaSmistare.mappingFruizionePD)// || 
  392.                             //tipoOggettoDaSmistare.equals(TipoOggettoDaSmistare.fruitore) 
  393.                             )
  394.                         ){
  395.                         // Operazione per il registro
  396.                         if (engineGestoreEventi) {
  397.                             QueueOperation queueOperationGestoreEventi= (QueueOperation) queueOperation.clone();
  398.                             queueOperationGestoreEventi.setTipoOperazione(TipoOperazione.webService);
  399.                             if (cq.insertQueue(gestoreEventiQueue, queueOperationGestoreEventi, filter) == 0) {
  400.                                 SmistatoreThread.log.error("Smistatore: Si e' verificato un problema durante l'inserimento in coda.");
  401.                                 qs.rollback();
  402.                                 this.con.rollback();
  403.                                 this.dbm.releaseConnection(this.con);
  404.                                 continue;
  405.                             }
  406.                         } else {
  407.                             SmistatoreThread.log.info("Smistatore: sincronizzazione Gestore Eventi non abilitata.");
  408.                         }
  409.                     }
  410.                 }

  412.                 this.con.commit();
  413.                 this.con.setAutoCommit(true);
  414.                 this.dbm.releaseConnection(this.con);

  416.                 qs.commit();

  418.                 SmistatoreThread.log.info("Smistatore: Operazione [" + idOperazione + "] completata.");

  420.             } catch (JMSException e) {
  421.                 try {
  422.                     qs.rollback();
  423.                     this.con.rollback();
  424.                     this.dbm.releaseConnection(this.con);
  425.                 } catch (Exception er) {
  426.                 }
  427.                 SmistatoreThread.log.error("Smistatore: Riscontrato erroreJMS durante la gestione di una richiesta: " + e.toString());
  428.                 try {
  429.                     Utilities.sleep(5000);
  430.                     SmistatoreThread.log.debug("Smistatore: Re-Inizializzazione Receiver ...");
  431.                     try {
  432.                         receiver.close();
  433.                     } catch (Exception eclose) {
  434.                     }
  435.                     try {
  436.                         qs.close();
  437.                     } catch (Exception eclose) {
  438.                     }
  439.                     try {
  440.                         qc.close();
  441.                     } catch (Exception eclose) {
  442.                     }
  443.                     qc = qcf.createQueueConnection();
  444.                     // Ripristino stato Exception Listener
  445.                     if (this.exceptionListenerJMS.isConnessioneCorrotta()) {
  446.                         this.exceptionListenerJMS.setConnessioneCorrotta(false);
  447.                         this.exceptionListenerJMS.setException(null);
  448.                     }
  449.                     qc.setExceptionListener(this.exceptionListenerJMS);
  450.                     qs = qc.createQueueSession(true, -1);
  451.                     receiver = qs.createReceiver(queue);
  452.                     qc.start();
  453.                     SmistatoreThread.log.debug("Smistatore: Re-Inizializzazione Receiver effettuata.");
  454.                     riconnessioneConErrore = false;

  456.                 } catch (Exception er) {
  457.                     SmistatoreThread.log.error("Smistatore: Re-Inizializzazione Receiver non effettuata:" + er.toString());
  458.                     riconnessioneConErrore = true;
  459.                 }
  460.             } catch (Exception e) {
  461.                 try {
  462.                     qs.rollback();
  463.                     this.con.rollback();
  464.                     this.dbm.releaseConnection(this.con);
  465.                 } catch (Exception er) {
  466.                 }
  467.                 SmistatoreThread.log.error("Smistatore: Riscontrato errore durante la gestione di una richiesta: " + e.toString(), e);
  468.             } finally {

  470.                 try {
  471.                     this.dbm.releaseConnection(this.con);
  472.                 } catch (Exception e) {

  474.                 }
  475.             }
  476.         }

  478.         // Chiusura connessione
  479.         try {
  480.             if (receiver != null) {
  481.                 receiver.close();
  482.             }
  483.             if (qs != null) {
  484.                 qs.rollback();
  485.                 qs.close();
  486.             }
  487.             if (qc != null) {
  488.                 qc.stop();
  489.                 qc.close();
  490.             }
  491.         } catch (Exception e) {
  492.             try {
  493.                 SmistatoreThread.log.error("Smistatore: Riscontrato errore durante la chiusura del Thread: " + e.toString());
  494.             } catch (Exception eLogger) {
  495.             }
  496.         }
  497.         
  498.         this.isRunning = false;
  499.         log.debug("Thread terminato");
  500.     }

  502.     public void stopGestore() {
  503.         this.stop = true;
  504.         
  505.         SmistatoreThread.log.debug("Fermo il thread ...");
  506.         int timeout = 60;
  507.         for (int i = 0; i < timeout; i++) {
  508.             if(this.isRunning()){
  509.                 Utilities.sleep(1000);
  510.             }
  511.             else{
  512.                 break;
  513.             }
  514.         }
  515.         if(this.isRunning){
  516.             SmistatoreThread.log.debug("Sono trascorsi 60 secondi ed il thread non รจ ancora terminato??");
  517.         }
  518.     }
  519. }
Created with JaCoCo 0.8.8.202204050719