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

  21. package org.openspcoop2.pdd.core.node;

  23. import java.sql.Connection;

  25. import org.openspcoop2.core.id.IDSoggetto;
  26. import org.openspcoop2.pdd.config.DBManager;
  27. import org.openspcoop2.pdd.config.OpenSPCoop2Properties;
  28. import org.openspcoop2.pdd.config.Resource;
  29. import org.openspcoop2.pdd.core.AbstractCore;
  30. import org.openspcoop2.pdd.core.GestoreMessaggi;
  31. import org.openspcoop2.pdd.core.PdDContext;
  32. import org.openspcoop2.pdd.core.state.OpenSPCoopStateful;
  33. import org.openspcoop2.pdd.logger.MsgDiagnostico;
  34. import org.openspcoop2.pdd.services.core.RicezioneBuste;
  35. import org.openspcoop2.pdd.services.core.RicezioneBusteMessage;
  36. import org.openspcoop2.pdd.services.core.RicezioneContenutiApplicativi;
  37. import org.openspcoop2.pdd.services.core.RicezioneContenutiApplicativiMessage;
  38. import org.openspcoop2.protocol.engine.constants.Costanti;
  39. import org.openspcoop2.protocol.engine.driver.RepositoryBuste;
  40. import org.openspcoop2.protocol.sdk.state.StateMessage;
  41. import org.openspcoop2.protocol.sdk.state.StatefulMessage;
  42. import org.openspcoop2.utils.Utilities;

  44. /**
  45.  * Classe utilizzata per la ricezione di messaggi contenuti nell'architettura di OpenSPCoop (versione DB).
  46.  *
  47.  * @author Poli Andrea (apoli@link.it)
  48.  * @author $Author$
  49.  * @version $Rev$, $Date$
  50.  */

  52. public class NodeReceiverDB extends AbstractCore implements INodeReceiver{


  55.     private static OpenSPCoop2Properties openspcoopProperties = 
  56.         OpenSPCoop2Properties.getInstance();

  58.     
  59.     
  60.     /**
  61.      * Ricezione di un messaggio  
  62.      *
  63.      * @param codicePorta Codice Porta per cui effettuare la receive
  64.      * @param idModulo Nodo destinatario per cui effettuare la ricezione. 
  65.      * @param timeout Timeout sulla ricezione
  66.      * @param checkInterval Intervallo di check sulla coda
  67.      * @return true se la ricezione JMS e' andata a buon fine, false altrimenti.
  68.      * 
  69.      */
  70.     @Override
  71.     public Object receive(MsgDiagnostico msgDiag, IDSoggetto codicePorta, String idModulo, String idMessaggio, 
  72.             long timeout, long checkInterval) throws NodeException,NodeTimeoutException{

  74.         /* ------------  Lettura parametri del messaggio ricevuto e ValidityCheck -------------- */
  75.         Object objReturn = null;

  77.         // connessione al database
  78.         DBManager dbManager = DBManager.getInstance();
  79.         Resource resource = null;
  80.         Connection connectionDB = null;
  81.                         
  82.         // GestoreMessaggi
  83.         String tipoMessaggio = null;
  84.         if(idModulo.startsWith(RicezioneContenutiApplicativi.ID_MODULO)){
  85.             tipoMessaggio = Costanti.INBOX;
  86.         }
  87.         else if(idModulo.startsWith(RicezioneBuste.ID_MODULO)){
  88.             tipoMessaggio = Costanti.OUTBOX;
  89.         }
  90.         
  91.         OpenSPCoopStateful openspcoopstate = new OpenSPCoopStateful();
  92.         openspcoopstate.setStatoRichiesta(new StatefulMessage(null, null));
  93.         openspcoopstate.setStatoRisposta(new StatefulMessage(null, null));
  94.         
  95.         GestoreMessaggi gestoreMessaggioRichiesta = new GestoreMessaggi(openspcoopstate, false, idMessaggio,tipoMessaggio,msgDiag,null);

  97.         try{
  98.             long attesa = 0;
  99.             String idRisposta = null;
  100.             GestoreMessaggi gestoreMessaggioRisposta = null;
  101.             boolean messaggioPresente = false;
  102.             
  103.             int refreshOnlyCacheCount = 0;
  104.             
  105.             while(attesa<timeout){

  107.                 refreshOnlyCacheCount++;
  108.                 msgDiag.highDebug("Analisi richiesta con ID ["+idMessaggio+"] tipo ["+tipoMessaggio+"] contatoreRif["+refreshOnlyCacheCount+"]");
  109.                 
  110.                 int nodeDBRefresh = NodeReceiverDB.openspcoopProperties.getNodeReceiverCheckDBInterval();
  111.                 int nodeRiferimentoMsgRefresh = NodeReceiverDB.openspcoopProperties.getNodeReceiverCheckDBInterval();
  112.                 try{
  113.                     if(nodeRiferimentoMsgRefresh>2)
  114.                         nodeRiferimentoMsgRefresh = nodeDBRefresh / 2;
  115.                 }catch(Exception e){
  116.                     msgDiag.highDebug("CheckDBInterval (proprieta' 'org.openspcoop.pdd.nodeReceiver.checkDB') non corretto: "+e.getMessage());
  117.                 }
  118.                 
  119.                 msgDiag.highDebug("Proprieta' nodeDBRefresh["+nodeDBRefresh+"] nodeRiferimentoMsgRefresh["+nodeRiferimentoMsgRefresh+"]");
  120.                 
  121.                 // Il riferimento messaggio e il proprietario puo' essere letto dalla cache.
  122.                 // Per migliorare le performance, se non e' presente nella cache si assume anche non presente nel database.
  123.                 // - Per rifMsg se si chiama il metodo mapRiferimentoIntoIDBusta(true) l'id viene cercato solo nella cache
  124.                 // - Per proprietario se si chiama il metodo getProprietario(true) viene cercato solo nella cache
  125.                 // La cache puo' cmq svuotarsi velocemente se la dimensione e' troppo piccola rispetto al numero di msg in parallelo gestiti.
  126.                 // Quindi ogni nodeDBRefresh volte (se checkInterval=500 ogni nodeDBRefresh*500 millisecondi) viene controllato anche il database, oltre alla cache.
  127.                 boolean checkOnlyCache = NodeReceiverDB.openspcoopProperties.isAbilitataCacheGestoreMessaggi();
  128.                 if(refreshOnlyCacheCount==(nodeDBRefresh+1)){
  129.                     msgDiag.highDebug("Re-inizializzo contatore refreshOnlyCacheCount");
  130.                     refreshOnlyCacheCount = 1;
  131.                     checkOnlyCache = false;
  132.                 }
  133.                 
  134.                 // Il riferimentoMessaggio puo' cambiare l'associazione nel tempo.
  135.                 // Es. rispostaSincrona che e' un MessaggioErroreProtocollo.... prima viene salvato l'errore, e poi viene salvato il msg ritornato al servizio applicativo.
  136.                 // Entrambi, in ordine avranno come key il rifMessaggio.
  137.                 // Ogni (nodeDBRefresh/2) volte (se checkInterval=500 ogni (nodeDBRefresh/2)*500 millisecondi) viene fatto il refresh del messaggio
  138.                 if( (refreshOnlyCacheCount%nodeRiferimentoMsgRefresh)==0 ){
  139.                     msgDiag.highDebug("Re-inizializzo idRisposta e gestoreMessaggioRisposta");
  140.                     idRisposta = null;
  141.                     gestoreMessaggioRisposta = null;
  142.                 }
  143.                 
  144.                 // Prendo la connessione solo se mi serve per la ricerca su database
  145.                 boolean needConnection = false;
  146.                 if(NodeReceiverDB.openspcoopProperties.singleConnectionTransactionManager()==false)
  147.                     needConnection = true;
  148.                 else if(resource==null)
  149.                     needConnection = true;
  150.                 if( (checkOnlyCache==false) && needConnection ){
  151.                     msgDiag.highDebug("Prendo Connessione per NodeReceiver");
  152.                     try{
  153.                         resource = dbManager.getResource(codicePorta,idModulo, PdDContext.getValue(org.openspcoop2.core.constants.Costanti.ID_TRANSAZIONE, this.getPddContext()));
  154.                     }catch(Exception e){
  155.                         throw new NodeException("Impossibile ottenere una Risorsa dal DBManager",e);
  156.                     }
  157.                     if(resource==null)
  158.                         throw new NodeException("Risorsa is null");
  159.                     if(resource.getResource() == null)  
  160.                         throw new NodeException("Connessione is null");
  161.                     connectionDB = (Connection) resource.getResource();
  162.                     ((StateMessage)openspcoopstate.getStatoRichiesta()).setConnectionDB(connectionDB);
  163.                     ((StateMessage)openspcoopstate.getStatoRisposta()).setConnectionDB(connectionDB);
  164.                 }
  165.                 
  166.                 // Check esistenza messaggio
  167.                 if(idRisposta==null){
  168.                     msgDiag.highDebug("Analisi richiesta con ID ["+idMessaggio+"] tipo ["+tipoMessaggio+"]: lettura ID Risposta");
  169.                     idRisposta = gestoreMessaggioRichiesta.mapRiferimentoIntoIDBusta(checkOnlyCache); //only cache
  170.                 }
  171.                 if(idRisposta!=null){
  172.                     if(gestoreMessaggioRisposta==null){
  173.                         msgDiag.highDebug("Analisi richiesta con ID ["+idMessaggio+"] tipo ["+tipoMessaggio+"]: Costruisco GestoreRisposta per ID["+idRisposta+"]");
  174.                         gestoreMessaggioRisposta = new GestoreMessaggi(openspcoopstate, false, idRisposta,tipoMessaggio,msgDiag,null);
  175.                         msgDiag.highDebug("Analisi risposta con ID ["+idRisposta+"] tipo ["+tipoMessaggio+"]: existsMessage("+checkOnlyCache+")");
  176.                         if(checkOnlyCache)
  177.                             messaggioPresente = gestoreMessaggioRisposta.existsMessage_onlyCache();
  178.                         else
  179.                             messaggioPresente = gestoreMessaggioRisposta.existsMessage();
  180.                     }else{
  181.                         if(messaggioPresente==false){
  182.                             msgDiag.highDebug("Analisi risposta con ID ["+idRisposta+"] tipo ["+tipoMessaggio+"]: existsMessage("+checkOnlyCache+")");
  183.                             if(checkOnlyCache)
  184.                                 messaggioPresente = gestoreMessaggioRisposta.existsMessage_onlyCache();
  185.                             else
  186.                                 messaggioPresente = gestoreMessaggioRisposta.existsMessage();
  187.                         }
  188.                     }
  189.                 }
  190.                 
  191.                 msgDiag.highDebug("Analisi risposta con ID ["+idRisposta+"] tipo ["+tipoMessaggio+"]: existsMessage="+messaggioPresente);
  192.                 
  193.                 
  194.                 if(messaggioPresente){

  196.                     // read Proprietario
  197.                     /*String proprietario = null;
  198.                     if( idModulo.startsWith(RicezioneContenutiApplicativi.ID_MODULO)) {
  199.                         proprietario = gestoreMessaggioRisposta.getProprietario_SerializableRead(NodeReceiverDB.openspcoopProperties.getGestioneSerializableDB_AttesaAttiva(),NodeReceiverDB.openspcoopProperties.getGestioneSerializableDB_CheckInterval());
  200.                     } else {
  201.                         proprietario = gestoreMessaggioRisposta.getProprietario();
  202.                     }*/
  203.                     msgDiag.highDebug("getProprietario("+checkOnlyCache+")");
  204.                     String proprietario = gestoreMessaggioRisposta.getProprietario(idModulo,checkOnlyCache);
  205.                     msgDiag.highDebug("getProprietario("+checkOnlyCache+") proprietario="+proprietario);
  206.                     
  207.                     // check proprietario
  208.                     if( idModulo.startsWith(RicezioneContenutiApplicativi.ID_MODULO)) {
  209.                         messaggioPresente = idModulo.equals(proprietario);
  210.                     }else if( idModulo.startsWith(RicezioneBuste.ID_MODULO)) {
  211.                         messaggioPresente = idModulo.equals(proprietario);
  212.                     }
  213.                     msgDiag.highDebug("Analisi risposta con ID ["+idRisposta+"] tipo ["+tipoMessaggio+"] proprietario["+proprietario+" existsMessage="+messaggioPresente);
  214.                     
  215.                 }
  216.                 
  217.                 // Gestione messaggio
  218.                 if(messaggioPresente==false){
  219.                     
  220.                     //  rilascio e riprendo la connessione ogni checkInterval fino ad un timeout o alla ricezione di un oggetto
  221.                     if( (NodeReceiverDB.openspcoopProperties.singleConnectionNodeReceiver()==false) && (checkOnlyCache==false) ){
  222.                         msgDiag.highDebug("Rilascio connessione per NodeReceiver");
  223.                         dbManager.releaseResource(codicePorta, idModulo, resource);
  224.                     }
  225.                     
  226.                     msgDiag.highDebug("Sleep...");
  227.                     Utilities.sleep(checkInterval);
  228.                     attesa = attesa + checkInterval;

  230.                 }else{
  231.                     
  232.                     // Prendo informazioni da RepositoryBuste, se non era settata la connesione, la setto
  233.                     if( needConnection && checkOnlyCache){
  234.                         msgDiag.highDebug("Prendo Connessione per NodeReceiver");
  235.                         try{
  236.                             resource = dbManager.getResource(codicePorta,idModulo, PdDContext.getValue(org.openspcoop2.core.constants.Costanti.ID_TRANSAZIONE, this.getPddContext()));
  237.                         }catch(Exception e){
  238.                             throw new NodeException("Impossibile ottenere una Risorsa dal DBManager",e);
  239.                         }
  240.                         if(resource==null)
  241.                             throw new NodeException("Risorsa is null");
  242.                         if(resource.getResource() == null)
  243.                             throw new NodeException("Connessione is null");
  244.                         connectionDB = (Connection) resource.getResource();
  245.                     }
  246.                     
  247.                     StatefulMessage state = new StatefulMessage(connectionDB, null);
  248.                     if(idModulo.startsWith(RicezioneContenutiApplicativi.ID_MODULO)){
  249.                         msgDiag.highDebug("Lettura risposta per RicezioneContenutiApplicativi...");
  250.                         objReturn = new RicezioneContenutiApplicativiMessage();
  251.                         ((RicezioneContenutiApplicativiMessage)objReturn).setIdBustaRisposta(idRisposta);
  252.                         RepositoryBuste repositoryBuste = new RepositoryBuste(state, false,null);
  253.                         ((RicezioneContenutiApplicativiMessage)objReturn).setIdCollaborazione(repositoryBuste.getCollaborazioneFromInBox(idRisposta));
  254.                         ((RicezioneContenutiApplicativiMessage)objReturn).setProfiloCollaborazione(repositoryBuste.getProfiloCollaborazioneFromInBox(idRisposta),
  255.                                 repositoryBuste.getProfiloCollaborazioneValueFromInBox(idRisposta));
  256.                         try{
  257.                             ((RicezioneContenutiApplicativiMessage)objReturn).setPddContext(gestoreMessaggioRisposta.getPdDContext());
  258.                         }catch(Exception e){
  259.                             // ignore
  260.                         }
  261.                         msgDiag.highDebug("Lettura risposta per RicezioneContenutiApplicativi effettuata");
  262.                     }
  263.                     else if(idModulo.startsWith(RicezioneBuste.ID_MODULO)){
  264.                         msgDiag.highDebug("Lettura risposta per RicezioneBuste...");
  265.                         objReturn = new RicezioneBusteMessage();
  266.                         RepositoryBuste repositoryBuste = new RepositoryBuste(state, false,null);
  267.                         if(repositoryBuste.isRegistrataIntoOutBox(idRisposta)){
  268.                             ((RicezioneBusteMessage)objReturn).setBustaRisposta(repositoryBuste.getBustaFromOutBox(idRisposta));
  269.                         }else{
  270.                             ((RicezioneBusteMessage)objReturn).setIdMessaggioSblocco(idRisposta);
  271.                         }
  272.                         try{
  273.                             ((RicezioneBusteMessage)objReturn).setPddContext(gestoreMessaggioRisposta.getPdDContext());
  274.                         }catch(Exception e){
  275.                             // ignore
  276.                         }
  277.                         msgDiag.highDebug("Lettura risposta per RicezioneBuste effettuata");
  278.                     }
  279.                     
  280.                     // rilascio e riprendo la connessione ogni checkInterval fino ad un timeout o alla ricezione di un oggetto
  281.                     if( (NodeReceiverDB.openspcoopProperties.singleConnectionNodeReceiver()==false) && (checkOnlyCache==false) ){
  282.                         msgDiag.highDebug("Rilascio connessione per NodeReceiver");
  283.                         dbManager.releaseResource(codicePorta, idModulo, resource);
  284.                     }
  285.                     
  286.                     msgDiag.highDebug("Fine Lettura");
  287.                     break;
  288.                 }

  290.             }

  292.         } catch (Exception e) {
  293.             throw new NodeException("Riscontrato errore nella ricezione del messaggio di risposta per la gestione della richiesta:"
  294.                     +e.getMessage(),e);
  295.         } finally{
  296.             msgDiag.highDebug("Rilascio connessione per NodeReceiver");
  297.             dbManager.releaseResource(codicePorta, idModulo, resource);
  298.         }
  299.         
  300.         if(objReturn == null){
  301.             throw new NodeTimeoutException("Riscontrato errore durante ricezione del messaggio: Messaggio non ricevuto");
  302.         }
  303.         
  304.         return objReturn;
  305.     }

  307. }
Created with JaCoCo 0.8.8.202204050719