GestorePolicyAttiveInMemory.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.pdd.core.controllo_traffico.policy.driver;

  22. import java.io.File;
  23. import java.io.InputStream;
  24. import java.io.OutputStream;
  25. import java.util.HashMap;
  26. import java.util.Iterator;
  27. import java.util.Map;
  28. import java.util.Map.Entry;
  29. import java.util.Set;
  30. import java.util.zip.ZipEntry;
  31. import java.util.zip.ZipFile;
  32. import java.util.zip.ZipOutputStream;

  34. import javax.activation.FileDataSource;

  36. import org.openspcoop2.core.controllo_traffico.AttivazionePolicy;
  37. import org.openspcoop2.core.controllo_traffico.ConfigurazionePolicy;
  38. import org.openspcoop2.core.controllo_traffico.beans.ActivePolicy;
  39. import org.openspcoop2.core.controllo_traffico.beans.ConfigurazioneControlloTraffico;
  40. import org.openspcoop2.core.controllo_traffico.beans.DatiCollezionati;
  41. import org.openspcoop2.core.controllo_traffico.beans.DatiTransazione;
  42. import org.openspcoop2.core.controllo_traffico.beans.IDUnivocoGroupByPolicy;
  43. import org.openspcoop2.core.controllo_traffico.beans.UniqueIdentifierUtilities;
  44. import org.openspcoop2.core.controllo_traffico.constants.TipoRisorsa;
  45. import org.openspcoop2.core.controllo_traffico.driver.IGestorePolicyAttive;
  46. import org.openspcoop2.core.controllo_traffico.driver.IPolicyGroupByActiveThreads;
  47. import org.openspcoop2.core.controllo_traffico.driver.IPolicyGroupByActiveThreadsInMemory;
  48. import org.openspcoop2.core.controllo_traffico.driver.PolicyException;
  49. import org.openspcoop2.core.controllo_traffico.driver.PolicyGroupByActiveThreadsType;
  50. import org.openspcoop2.core.controllo_traffico.driver.PolicyNotFoundException;
  51. import org.openspcoop2.core.controllo_traffico.driver.PolicyShutdownException;
  52. import org.openspcoop2.core.controllo_traffico.utils.serializer.JaxbDeserializer;
  53. import org.openspcoop2.core.controllo_traffico.utils.serializer.JaxbSerializer;
  54. import org.openspcoop2.pdd.config.DynamicClusterManager;
  55. import org.openspcoop2.pdd.config.OpenSPCoop2Properties;
  56. import org.openspcoop2.pdd.core.controllo_traffico.policy.driver.hazelcast.HazelcastManager;
  57. import org.openspcoop2.pdd.core.controllo_traffico.policy.driver.hazelcast.PolicyGroupByActiveThreadsDistributedLocalCache;
  58. import org.openspcoop2.pdd.core.controllo_traffico.policy.driver.hazelcast.PolicyGroupByActiveThreadsDistributedNearCache;
  59. import org.openspcoop2.pdd.core.controllo_traffico.policy.driver.hazelcast.PolicyGroupByActiveThreadsDistributedNearCacheWithoutEntryProcessorPutAsync;
  60. import org.openspcoop2.pdd.core.controllo_traffico.policy.driver.hazelcast.PolicyGroupByActiveThreadsDistributedNearCacheWithoutEntryProcessorPutSync;
  61. import org.openspcoop2.pdd.core.controllo_traffico.policy.driver.hazelcast.PolicyGroupByActiveThreadsDistributedNoCache;
  62. import org.openspcoop2.pdd.core.controllo_traffico.policy.driver.hazelcast.PolicyGroupByActiveThreadsDistributedReplicatedMap;
  63. import org.openspcoop2.pdd.core.controllo_traffico.policy.driver.redisson.PolicyGroupByActiveThreadsDistributedRedis;
  64. import org.openspcoop2.pdd.core.controllo_traffico.policy.driver.redisson.RedissonManager;
  65. import org.openspcoop2.pdd.services.OpenSPCoop2Startup;
  66. import org.openspcoop2.protocol.basic.Costanti;
  67. import org.openspcoop2.protocol.sdk.state.IState;
  68. import org.openspcoop2.utils.SemaphoreLock;
  69. import org.openspcoop2.utils.Utilities;
  70. import org.openspcoop2.utils.io.ZipUtilities;
  71. import org.openspcoop2.utils.resources.FileSystemUtilities;
  72. import org.redisson.api.RedissonClient;
  73. import org.slf4j.Logger;

  75. /**     
  76.  * GestorePolicyAttiveInMemory
  77.  *
  78.  * @author Poli Andrea (poli@link.it)
  79.  * @author $Author$
  80.  * @version $Rev$, $Date$
  81.  */
  82. public class GestorePolicyAttiveInMemory implements IGestorePolicyAttive {


  85.     /** 
  86.      * Threads allocati sulle Policy. La chiave รจ l'active-policy-id
  87.      **/
  88.     private Map<String, IPolicyGroupByActiveThreadsInMemory> mapActiveThreadsPolicy = 
  89.             new HashMap<String, IPolicyGroupByActiveThreadsInMemory>();
  90.     private final org.openspcoop2.utils.Semaphore lock = new org.openspcoop2.utils.Semaphore("GestorePolicyAttiveInMemory");
  91.     
  92.     private static final String IMPL_DESCR = "Implementazione InMemory IGestorePolicyAttive";
  93.     public static String getImplDescr(){
  94.         return IMPL_DESCR;
  95.     }
  96.     
  97.     private Logger log;
  98.     private PolicyGroupByActiveThreadsType type;
  99.     private boolean useCountersWithLock = false;
  100.     @Override
  101.     public void initialize(Logger log, boolean isStartupGovWay, PolicyGroupByActiveThreadsType type, Object ... params) throws PolicyException{
  102.         this.log = log;
  103.         this.type = type;
  104.         if(this.type==null) {
  105.             this.type = PolicyGroupByActiveThreadsType.LOCAL;
  106.         }
  107.         
  108.         switch (this.type) {
  109.         case LOCAL:
  110.             break;
  111.         case LOCAL_DIVIDED_BY_NODES:
  112.             if(!DynamicClusterManager.isInitialized()) {
  113.                 try {
  114.                     DynamicClusterManager.initStaticInstance();
  115.                     DynamicClusterManager.getInstance().setRateLimitingGestioneCluster(true);
  116.                     DynamicClusterManager.getInstance().register(log);
  117.                     OpenSPCoop2Startup.startTimerClusterDinamicoThread();
  118.                 }catch(Exception e) {
  119.                     throw new PolicyException(e.getMessage(),e);
  120.                 }
  121.             }
  122.             else {
  123.                 try {
  124.                     DynamicClusterManager.getInstance().setRateLimitingGestioneCluster(true);
  125.                 }catch(Exception e) {
  126.                     throw new PolicyException(e.getMessage(),e);
  127.                 }
  128.             }
  129.             break;
  130.             
  131.         case DATABASE:
  132.             break;
  133.         
  134.         case HAZELCAST_MAP:
  135.         case HAZELCAST_NEAR_CACHE:
  136.         case HAZELCAST_NEAR_CACHE_UNSAFE_ASYNC_MAP:
  137.         case HAZELCAST_NEAR_CACHE_UNSAFE_SYNC_MAP:
  138.         case HAZELCAST_PNCOUNTER:
  139.         case HAZELCAST_ATOMIC_LONG:
  140.         case HAZELCAST_ATOMIC_LONG_ASYNC:
  141.         case HAZELCAST_REPLICATED_MAP:
  142.             HazelcastManager.getInstance(this.type);
  143.             break;
  144.         case HAZELCAST_LOCAL_CACHE:
  145.             HazelcastManager.getInstance(this.type);
  146.             if(!OpenSPCoop2Startup.isStartedTimerClusteredRateLimitingLocalCache()) {
  147.                 try {
  148.                     OpenSPCoop2Startup.startTimerClusteredRateLimitingLocalCache(this);
  149.                 }catch(Exception e) {
  150.                     throw new PolicyException(e.getMessage(),e);
  151.                 }
  152.             }
  153.             break;

  155.         case REDISSON_MAP:
  156.         case REDISSON_ATOMIC_LONG:
  157.         case REDISSON_LONGADDER:
  158.             
  159.             boolean throwInitializingException = true;
  160.             if(isStartupGovWay) {
  161.                 throwInitializingException = OpenSPCoop2Properties.getInstance().isControlloTrafficoGestorePolicyInMemoryRedisThrowExceptionIfRedisNotReady();
  162.             }
  163.             
  164.             try {
  165.                 RedissonManager.getRedissonClient(throwInitializingException);
  166.             }catch(Exception e) {
  167.                 throw new PolicyException(e.getMessage(),e);
  168.             }
  169.             break;
  170.         }
  171.         
  172.         if(this.type.isHazelcastCounters() || this.type.isRedisCounters()) {
  173.             this.useCountersWithLock=OpenSPCoop2Properties.getInstance().isControlloTrafficoGestorePolicyInMemoryRemoteCountersUseLocalLock();
  174.         }
  175.     }
  176.     
  177.     private boolean isStop = false;
  178.     
  179.     
  180.     @Override
  181.     public PolicyGroupByActiveThreadsType getType() {
  182.         return this.type;
  183.     }
  184.     
  185.     @Override
  186.     public IPolicyGroupByActiveThreads getActiveThreadsPolicy(ActivePolicy activePolicy,DatiTransazione datiTransazione, Object state) throws PolicyShutdownException,PolicyException {     
  187.         
  188.         String uniqueIdMap = UniqueIdentifierUtilities.getUniqueId(activePolicy.getInstanceConfiguration());
  189.                 
  190.         /**synchronized (this.mapActiveThreadsPolicy) {*/
  191.         SemaphoreLock slock = this.lock.acquireThrowRuntime("getActiveThreadsPolicy(ActivePolicy)");
  192.         try {
  193.             
  194.             if(this.isStop){
  195.                 throw new PolicyShutdownException("Policy Manager shutdown");
  196.             }
  197.             
  198.             IPolicyGroupByActiveThreadsInMemory active = null;
  199.             /**System.out.println("@@@ getActiveThreadsPolicy["+uniqueIdMap+"] contains["+this.mapActiveThreadsPolicy.containsKey(uniqueIdMap)+"]...");*/
  200.             if(this.mapActiveThreadsPolicy.containsKey(uniqueIdMap)){
  201.                 active = this.mapActiveThreadsPolicy.get(uniqueIdMap);
  202.                 /**System.out.println("@@@ getActiveThreadsPolicy["+uniqueIdMap+"] GET");*/
  203.             }
  204.             else{
  205.                 active = newPolicyGroupByActiveThreadsInMemory(activePolicy, uniqueIdMap, datiTransazione, state);
  206.                 this.mapActiveThreadsPolicy.put(uniqueIdMap, active);
  207.                 /**System.out.println("@@@ getActiveThreadsPolicy["+uniqueIdMap+"] CREATE");*/
  208.             }
  209.             return active;
  210.         }finally {
  211.             this.lock.release(slock, "getActiveThreadsPolicy(ActivePolicy)");
  212.         }
  213.     }
  214.     @Override
  215.     public IPolicyGroupByActiveThreads getActiveThreadsPolicy(String uniqueIdMap) throws PolicyShutdownException,PolicyException,PolicyNotFoundException { // usata per la remove
  216.         /**synchronized (this.mapActiveThreadsPolicy) {*/
  217.         SemaphoreLock slock = this.lock.acquireThrowRuntime("getActiveThreadsPolicy(uniqueIdMap)");
  218.         try {
  219.             
  220.             if(this.isStop){
  221.                 throw new PolicyShutdownException("Policy Manager shutdown");
  222.             }
  223.             
  224.             IPolicyGroupByActiveThreads active = null;
  225.             /**System.out.println("@@@ getActiveThreadsPolicy["+uniqueIdMap+"] contains["+this.mapActiveThreadsPolicy.containsKey(uniqueIdMap)+"]...");*/
  226.             if(this.mapActiveThreadsPolicy.containsKey(uniqueIdMap)){
  227.                 active = this.mapActiveThreadsPolicy.get(uniqueIdMap);
  228.                 /**System.out.println("@@@ getActiveThreadsPolicy["+uniqueIdMap+"] GET");*/
  229.             }
  230.             else{
  231.                 throw new PolicyNotFoundException("ActivePolicy ["+uniqueIdMap+"] notFound");
  232.             }
  233.             return active;
  234.         }finally{
  235.             this.lock.release(slock, "getActiveThreadsPolicy(uniqueIdMap)");
  236.         }
  237.     }
  238.     
  239.     @Override
  240.     public long sizeActivePolicyThreads(boolean sum) throws PolicyShutdownException,PolicyException{
  241.         /**synchronized (this.mapActiveThreadsPolicy) {*/
  242.         SemaphoreLock slock = this.lock.acquireThrowRuntime("sizeActivePolicyThreads");
  243.         try {
  244.             
  245.             if(this.isStop){
  246.                 throw new PolicyShutdownException("Policy Manager shutdown");
  247.             }
  248.             
  249.             if(sum){
  250.                 long sumLong = 0;
  251.                 if(this.mapActiveThreadsPolicy!=null && !this.mapActiveThreadsPolicy.isEmpty()) {
  252.                     for (String idPolicy : this.mapActiveThreadsPolicy.keySet()) {
  253.                         sumLong = sumLong +this.mapActiveThreadsPolicy.get(idPolicy).getActiveThreads();
  254.                     }
  255.                 }
  256.                 return sumLong;
  257.             }else{
  258.                 return this.mapActiveThreadsPolicy.size();
  259.             }
  260.         }finally {
  261.             this.lock.release(slock, "sizeActivePolicyThreads");
  262.         }
  263.     }
  264.     
  265.     @Override
  266.     public String printKeysPolicy(String separator) throws PolicyShutdownException, PolicyException{
  267.         /**synchronized (this.mapActiveThreadsPolicy) {*/
  268.         SemaphoreLock slock = this.lock.acquireThrowRuntime("printKeysPolicy");
  269.         try {
  270.             
  271.             if(this.isStop){
  272.                 throw new PolicyShutdownException("Policy Manager shutdown");
  273.             }
  274.             
  275.             StringBuilder bf = new StringBuilder();
  276.             if(this.mapActiveThreadsPolicy!=null && !this.mapActiveThreadsPolicy.isEmpty()) {
  277.                 int i = 0;
  278.                 for (String idPolicy : this.mapActiveThreadsPolicy.keySet()) {
  279.                     String key = idPolicy;
  280.                     if(i>0){
  281.                         bf.append(separator);
  282.                     }
  283.                     bf.append("Cache-"+this.type+"["+i+"]=["+key+"]");
  284.                     i++;
  285.                 }
  286.             }
  287.             return bf.toString();
  288.         }finally {
  289.             this.lock.release(slock, "printKeysPolicy");
  290.         }
  291.     }
  292.     
  293.     @Override
  294.     public String printInfoPolicy(String id, String separatorGroups) throws PolicyShutdownException,PolicyException,PolicyNotFoundException{
  295.         IPolicyGroupByActiveThreadsInMemory activeThreads = (IPolicyGroupByActiveThreadsInMemory) this.getActiveThreadsPolicy(id);  
  296.         try{
  297.             return activeThreads.printInfos(this.log, separatorGroups);
  298.         }catch(Exception e){
  299.             throw new PolicyException(e.getMessage(),e);
  300.         }
  301.     }
  302.     
  303.     @Override
  304.     public void removeActiveThreadsPolicy(String idActivePolicy) throws PolicyShutdownException, PolicyException{
  305.         /**synchronized (this.mapActiveThreadsPolicy) {*/
  306.         SemaphoreLock slock = this.lock.acquireThrowRuntime("removeActiveThreadsPolicy");
  307.         try {
  308.             
  309.             if(this.isStop){
  310.                 throw new PolicyShutdownException("Policy Manager shutdown");
  311.             }
  312.             
  313.             if(this.mapActiveThreadsPolicy.containsKey(idActivePolicy)){
  314.                 this.mapActiveThreadsPolicy.remove(idActivePolicy);
  315.             }
  316.         }finally {
  317.             this.lock.release(slock, "removeActiveThreadsPolicy");
  318.         }
  319.     }
  320.     
  321.     @Override
  322.     public void removeActiveThreadsPolicyUnsafe(String idActivePolicy) throws PolicyShutdownException,PolicyException{
  323.         if(this.isStop){
  324.             throw new PolicyShutdownException("Policy Manager shutdown");
  325.         }
  326.         
  327.         IPolicyGroupByActiveThreadsInMemory policy = this.mapActiveThreadsPolicy.remove(idActivePolicy);
  328.         if(policy!=null) {
  329.             try {
  330.                 policy.remove();
  331.             }catch(Throwable e) {
  332.                 this.log.error("removeActiveThreadsPolicyUnsafe failed: "+e.getMessage(),e);
  333.             }
  334.         }
  335.     }
  336.     
  337.     @Override
  338.     public void removeAllActiveThreadsPolicy() throws PolicyShutdownException, PolicyException{
  339.         /**synchronized (this.mapActiveThreadsPolicy) {*/
  340.         SemaphoreLock slock = this.lock.acquireThrowRuntime("removeAllActiveThreadsPolicy");
  341.         try {
  342.             
  343.             if(this.isStop){
  344.                 throw new PolicyShutdownException("Policy Manager shutdown");
  345.             }
  346.             
  347.             this.mapActiveThreadsPolicy.clear();
  348.         }finally {
  349.             this.lock.release(slock, "removeAllActiveThreadsPolicy");
  350.         }
  351.     }
  352.     
  353.     @Override
  354.     public void resetCountersActiveThreadsPolicy(String idActivePolicy) throws PolicyShutdownException, PolicyException{
  355.         /**synchronized (this.mapActiveThreadsPolicy) {*/
  356.         SemaphoreLock slock = this.lock.acquireThrowRuntime("resetCountersActiveThreadsPolicy");
  357.         try {
  358.             
  359.             if(this.isStop){
  360.                 throw new PolicyShutdownException("Policy Manager shutdown");
  361.             }
  362.             
  363.             if(this.mapActiveThreadsPolicy.containsKey(idActivePolicy)){
  364.                 this.mapActiveThreadsPolicy.get(idActivePolicy).resetCounters();
  365.             }
  366.         }finally {
  367.             this.lock.release(slock, "resetCountersActiveThreadsPolicy");
  368.         }
  369.     }
  370.     
  371.     @Override
  372.     public void resetCountersAllActiveThreadsPolicy() throws PolicyShutdownException, PolicyException{
  373.         /**synchronized (this.mapActiveThreadsPolicy) {*/
  374.         SemaphoreLock slock = this.lock.acquireThrowRuntime("resetCountersAllActiveThreadsPolicy");
  375.         try {
  376.             
  377.             if(this.isStop){
  378.                 throw new PolicyShutdownException("Policy Manager shutdown");
  379.             }
  380.             
  381.             if(this.mapActiveThreadsPolicy.size()>0){
  382.                 for (String key : this.mapActiveThreadsPolicy.keySet()) {
  383.                     this.mapActiveThreadsPolicy.get(key).resetCounters();
  384.                 }
  385.             }
  386.         }finally {
  387.             this.lock.release(slock, "resetCountersAllActiveThreadsPolicy");
  388.         }
  389.     }
  390.     
  391.     public Set<Entry<String, IPolicyGroupByActiveThreadsInMemory>> entrySet() throws PolicyShutdownException, PolicyException{
  392.         Set<Entry<String, IPolicyGroupByActiveThreadsInMemory>> activeThreadsPolicies;
  393.         
  394.         SemaphoreLock slock = this.lock.acquireThrowRuntime("updateLocalCacheMap");
  395.         try {
  396.             
  397.             if(this.isStop){
  398.                 throw new PolicyShutdownException("Policy Manager shutdown");
  399.             }
  400.             activeThreadsPolicies = this.mapActiveThreadsPolicy.entrySet();
  401.         } finally {
  402.             this.lock.release(slock, "updateLocalCacheMap");
  403.         }
  404.         
  405.         return activeThreadsPolicies;
  406.     }
  407.     
  408.     
  409.     // ---- Per salvare
  410.     
  411.     private static final String ZIP_POLICY_PREFIX = "policy-";
  412.     private static final String ZIP_POLICY_ID_ACTIVE_SUFFIX = "-id-active.txt";
  413.     private static final String ZIP_POLICY_CONFIGURAZIONE_POLICY_SUFFIX = "ConfigurazionePolicy.xml";
  414.     private static final String ZIP_POLICY_ATTIVAZIONE_POLICY_SUFFIX = "AttivazionePolicy.xml";
  415.     private static final String ZIP_POLICY_ID_DATI_COLLEZIONATI_POLICY_SUFFIX = "-id-datiCollezionati.txt";
  416.     private static final String ZIP_POLICY_DATI_COLLEZIONATI_POLICY_SUFFIX = "-datiCollezionati.txt";
  417.     
  418.     @Override
  419.     public void serialize(OutputStream out) throws PolicyException{
  420.                 
  421.         /**synchronized (this.mapActiveThreadsPolicy) {*/
  422.         SemaphoreLock slock = null;
  423.         try {
  424.             slock = this.lock.acquireThrowRuntime("serialize");
  425.             
  426.             if(this.isStop){
  427.                 throw new PolicyException("Already serialized");
  428.             }
  429.             this.isStop = true;
  430.             
  431.             if(this.mapActiveThreadsPolicy==null || this.mapActiveThreadsPolicy.size()<=0){
  432.                 return;
  433.             }
  434.             
  435.             ZipOutputStream zipOut = null;
  436.             try{
  437.                 zipOut = new ZipOutputStream(out);

  439.                 String rootPackageDir = "";
  440.                 // Il codice dopo fissa il problema di inserire una directory nel package.
  441.                 // Commentare la riga sotto per ripristinare il vecchio comportamento.
  442.                 rootPackageDir = Costanti.OPENSPCOOP2_ARCHIVE_ROOT_DIR+File.separatorChar;
  443.                 
  444.                 // indice 
  445.                 int index = 1;
  446.                 
  447.                 // Chiavi possiedono la policy id
  448.                 for (String idActivePolicy : this.mapActiveThreadsPolicy.keySet()) {
  449.                     
  450.                     // Id File
  451.                     String idFileActivePolicy = ZIP_POLICY_PREFIX+index;
  452.                     
  453.                     // File contenente l'identificativo della policy attivata
  454.                     String nomeFile = idFileActivePolicy+ZIP_POLICY_ID_ACTIVE_SUFFIX;
  455.                     zipOut.putNextEntry(new ZipEntry(rootPackageDir+nomeFile));
  456.                     zipOut.write(idActivePolicy.getBytes());
  457.                     
  458.                     // GroupByThread
  459.                     IPolicyGroupByActiveThreadsInMemory active = this.mapActiveThreadsPolicy.get(idActivePolicy);
  460.                     if(active!=null){
  461.                         
  462.                         ActivePolicy activePolicy = active.getActivePolicy();
  463.                         JaxbSerializer serializer = new JaxbSerializer();
  464.                         
  465.                         byte[] cPolicy = null;
  466.                         if(activePolicy.getConfigurazionePolicy()!=null){
  467.                             try {
  468.                                 cPolicy = serializer.toByteArray(activePolicy.getConfigurazionePolicy());
  469.                             }catch(Throwable t) {
  470.                                 this.log.error("["+this.type+"] Serializzazione configurazione policy '"+activePolicy.getConfigurazionePolicy().getIdPolicy()+"' fallita: "+t.getMessage(),t);
  471.                             }
  472.                             if(cPolicy==null) {
  473.                                 continue;
  474.                             }
  475.                         }
  476.                         
  477.                         byte[] aPolicy = null;
  478.                         if(activePolicy.getInstanceConfiguration()!=null){
  479.                             try {
  480.                                 aPolicy = serializer.toByteArray(activePolicy.getInstanceConfiguration());
  481.                             }catch(Throwable t) {
  482.                                 this.log.error("["+this.type+"] Serializzazione attivazione policy '"+activePolicy.getInstanceConfiguration().getAlias()+"' ("+activePolicy.getInstanceConfiguration().getIdActivePolicy()+") fallita: "+t.getMessage(),t);
  483.                             }
  484.                             if(aPolicy==null) {
  485.                                 continue;
  486.                             }
  487.                         }
  488.                         
  489.                         
  490.                         // ConfigurazionePolicy
  491.                         if(activePolicy.getConfigurazionePolicy()!=null){
  492.                             nomeFile = idFileActivePolicy+File.separatorChar+ZIP_POLICY_CONFIGURAZIONE_POLICY_SUFFIX;
  493.                             zipOut.putNextEntry(new ZipEntry(rootPackageDir+nomeFile));
  494.                             zipOut.write(cPolicy);
  495.                         }
  496.                         
  497.                         // AttivazionePolicy
  498.                         if(activePolicy.getInstanceConfiguration()!=null){
  499.                             nomeFile = idFileActivePolicy+File.separatorChar+ZIP_POLICY_ATTIVAZIONE_POLICY_SUFFIX;
  500.                             zipOut.putNextEntry(new ZipEntry(rootPackageDir+nomeFile));
  501.                             zipOut.write(aPolicy);
  502.                         }
  503.                         
  504.                         Map<IDUnivocoGroupByPolicy, DatiCollezionati> map = active.getMapActiveThreads();
  505.                         if(map!=null && map.size()>0){
  506.                             
  507.                             // indice 
  508.                             int indexDatoCollezionato = 1;
  509.                             
  510.                             // Chiavi dei raggruppamenti
  511.                             for (IDUnivocoGroupByPolicy idUnivocoGroupByPolicy : map.keySet()) {
  512.                                 
  513.                                 // Id Raggruppamento
  514.                                 String idFileRaggruppamento = idFileActivePolicy+File.separatorChar+"groupBy"+File.separatorChar+"groupBy-"+indexDatoCollezionato;
  515.                                 
  516.                                 String id = null;
  517.                                 try {
  518.                                     id = IDUnivocoGroupByPolicy.serialize(idUnivocoGroupByPolicy);
  519.                                 }catch(Throwable t) {
  520.                                     this.log.error("["+this.type+"] Serializzazione idUnivocoGroupByPolicy ("+idUnivocoGroupByPolicy+") della policy '"+activePolicy.getInstanceConfiguration().getAlias()+"' ("+activePolicy.getInstanceConfiguration().getIdActivePolicy()+") fallita: "+t.getMessage(),t);
  521.                                 }
  522.                                 if(id==null) {
  523.                                     continue;
  524.                                 }
  525.                                 
  526.                                 DatiCollezionati datiCollezionati = map.get(idUnivocoGroupByPolicy);
  527.                                 String dati = null;
  528.                                 try {
  529.                                     dati = DatiCollezionati.serialize(datiCollezionati);
  530.                                 }catch(Throwable t) {
  531.                                     this.log.error("["+this.type+"] Serializzazione dati per idUnivocoGroupByPolicy ("+idUnivocoGroupByPolicy+") della policy '"+activePolicy.getInstanceConfiguration().getAlias()+"' ("+activePolicy.getInstanceConfiguration().getIdActivePolicy()+") fallita: "+t.getMessage(),t);
  532.                                 }
  533.                                 if(dati==null) {
  534.                                     continue;
  535.                                 }
  536.                                 
  537.                                 // File contenente l'identificativo del raggruppamento
  538.                                 nomeFile = idFileRaggruppamento+ZIP_POLICY_ID_DATI_COLLEZIONATI_POLICY_SUFFIX;
  539.                                 zipOut.putNextEntry(new ZipEntry(rootPackageDir+nomeFile));
  540.                                 zipOut.write(id.getBytes());
  541.                                 
  542.                                 // DatiCollezionati
  543.                                 // NOTA: l'ulteriore directory serve a garantire il corretto ordine di ricostruzione
  544.                                 nomeFile = idFileRaggruppamento+File.separatorChar+"dati"+ZIP_POLICY_DATI_COLLEZIONATI_POLICY_SUFFIX;
  545.                                 zipOut.putNextEntry(new ZipEntry(rootPackageDir+nomeFile));
  546.                                 zipOut.write(dati.getBytes());
  547.                                 
  548.                                 // increment
  549.                                 indexDatoCollezionato++;
  550.                             }
  551.                             
  552.                         }
  553.                     }
  554.                     
  555.                     // increment
  556.                     index++;
  557.                 }
  558.                 
  559.                 zipOut.flush();

  561.             }catch(Exception e){
  562.                 throw new PolicyException(e.getMessage(),e);
  563.             }finally{
  564.                 try{
  565.                     if(zipOut!=null)
  566.                         zipOut.close();
  567.                 }catch(Exception eClose){}
  568.             }

  570.         }finally {
  571.             this.lock.release(slock, "serialize");
  572.         }
  573.     }
  574.     
  575.     @Override
  576.     public void initialize(InputStream in,ConfigurazioneControlloTraffico configurazioneControlloTraffico) throws PolicyException{
  577.         
  578.         /**synchronized (this.mapActiveThreadsPolicy) {*/
  579.         SemaphoreLock slock = null;
  580.         try {
  581.             slock = this.lock.acquireThrowRuntime("initialize");
  582.             
  583.             if(in==null){
  584.                 return;
  585.             }
  586.             
  587.             File f = null;
  588.             ZipFile zipFile = null;
  589.             String entryName = null;
  590.             try{
  591.                 
  592.                 // Leggo InputStream
  593.                 byte [] bytesIn = Utilities.getAsByteArray(in);
  594.                 in.close();
  595.                 in = null;
  596.                 if(bytesIn==null || bytesIn.length<=0){
  597.                     return;
  598.                 }
  599.                 f = FileSystemUtilities.createTempFile("controlloTraffico", ".tmp");
  600.                 FileSystemUtilities.writeFile(f, bytesIn);
  601.                 
  602.                 // Leggo Struttura ZIP
  603.                 try {
  604.                     zipFile = new ZipFile(f);
  605.                 }catch(Throwable t) {
  606.                     this.log.error("Inizializzazione immagine ControlloTraffico precedente allo shutdown non ripristinabile, immagine corrotta: "+t.getMessage(),t);
  607.                     return;
  608.                 }
  609.                 
  610.                 JaxbDeserializer deserializer = new JaxbDeserializer();
  611.                 
  612.                 String rootPackageDir = Costanti.OPENSPCOOP2_ARCHIVE_ROOT_DIR+File.separatorChar;
  613.                 
  614.                 String rootDir = null;
  615.                 
  616.                 String idActivePolicy = null;
  617.                 ConfigurazionePolicy configurazionePolicy = null;
  618.                 AttivazionePolicy attivazionePolicy = null;
  619.                 Map<IDUnivocoGroupByPolicy, DatiCollezionati> map = null;
  620.                 
  621.                 IDUnivocoGroupByPolicy idDatiCollezionati = null;
  622.                 
  623.                 Iterator<ZipEntry> it = ZipUtilities.entries(zipFile, true);
  624.                 while (it.hasNext()) {
  625.                     ZipEntry zipEntry = (ZipEntry) it.next();
  626.                     entryName = ZipUtilities.operativeSystemConversion(zipEntry.getName());
  627.                     
  628.                     //System.out.println("FILE NAME:  "+entryName);
  629.                     //System.out.println("SIZE:  "+entry.getSize());

  631.                     // Il codice dopo fissa il problema di inserire una directory nel package.
  632.                     // Commentare la riga sotto per ripristinare il vecchio comportamento.
  633.                     if(rootDir==null){
  634.                         // Calcolo ROOT DIR
  635.                         rootDir=ZipUtilities.getRootDir(entryName);
  636.                     }
  637.                     
  638.                     if(zipEntry.isDirectory()) {
  639.                         continue; // directory
  640.                     }
  641.                     else {
  642.                         FileDataSource fds = new FileDataSource(entryName);
  643.                         String nome = fds.getName();
  644.                         String tipo = nome.substring(nome.lastIndexOf(".")+1,nome.length()); 
  645.                         tipo = tipo.toUpperCase();
  646.                         //System.out.println("VERIFICARE NAME["+nome+"] TIPO["+tipo+"]");
  647.                         
  648.                         InputStream inputStream = zipFile.getInputStream(zipEntry);
  649.                         byte[]content = Utilities.getAsByteArray(inputStream);
  650.                         
  651.                         try{
  652.                             
  653.                             if(entryName.startsWith((rootPackageDir+ZIP_POLICY_PREFIX)) ){
  654.                                 
  655.                                 if(entryName.endsWith(ZIP_POLICY_ID_ACTIVE_SUFFIX)){
  656.                                     
  657.                                     if(idActivePolicy!=null){
  658.                                         // salvo precedente immagine
  659.                                         this.mapActiveThreadsPolicy.put(idActivePolicy, 
  660.                                                 this.buildPolicyGroupByActiveThreads(configurazionePolicy, attivazionePolicy, map, configurazioneControlloTraffico));
  661.                                         //System.out.println("@@@ RICOSTRUITO ID ACTIVE POLICY ["+idActivePolicy+"]");
  662.                                         idActivePolicy = null;
  663.                                         configurazionePolicy = null;
  664.                                         attivazionePolicy = null;
  665.                                         map = null;
  666.                                         idDatiCollezionati = null;
  667.                                     }
  668.                                     
  669.                                     idActivePolicy = new String(content);
  670.                                     map=new HashMap<IDUnivocoGroupByPolicy, DatiCollezionati>();
  671.                                     
  672.                                     //System.out.println("ENTRY ["+idActivePolicy+"] NUOVO ID ["+entryName+"]");
  673.                                     
  674.                                 }
  675.                                 else if(entryName.endsWith(ZIP_POLICY_CONFIGURAZIONE_POLICY_SUFFIX)){
  676.                                 
  677.                                     configurazionePolicy = deserializer.readConfigurazionePolicy(content);
  678.                                     
  679.                                     //System.out.println("ENTRY ["+idActivePolicy+"] CONFIGURAZIONE POLICY ["+entryName+"]");
  680.                                     
  681.                                 }
  682.                                 else if(entryName.endsWith(ZIP_POLICY_ATTIVAZIONE_POLICY_SUFFIX)){
  683.                                     
  684.                                     attivazionePolicy = deserializer.readAttivazionePolicy(content);
  685.                                     
  686.                                     //System.out.println("ENTRY ["+idActivePolicy+"] ATTIVAZIONE POLICY ["+entryName+"]");
  687.                                     
  688.                                 }
  689.                                 else if(entryName.endsWith(ZIP_POLICY_ID_DATI_COLLEZIONATI_POLICY_SUFFIX)){
  690.                                     
  691.                                     idDatiCollezionati  = IDUnivocoGroupByPolicy.deserialize( new String(content) );
  692.                                     
  693.                                     //System.out.println("ENTRY ["+idActivePolicy+"] ID DATI COLLEZIONATI POLICY ["+entryName+"]");
  694.                                     
  695.                                 }
  696.                                 else if(entryName.endsWith(ZIP_POLICY_DATI_COLLEZIONATI_POLICY_SUFFIX)){
  697.                                     
  698.                                     if(idDatiCollezionati==null){
  699.                                         throw new Exception("Identificativo di group by not found");
  700.                                     }
  701.                                     map.put(idDatiCollezionati, DatiCollezionati.deserialize( new String(content) ));
  702.                                                                         
  703.                                     //System.out.println("ENTRY ["+idActivePolicy+"] DATI COLLEZIONATI POLICY ["+entryName+"]");
  704.                                     
  705.                                 }
  706.                                 
  707.                             }
  708.                             else{
  709.                                 throw new Exception("Entry unknown");
  710.                             }
  711.                             
  712.                         }finally{
  713.                             try{
  714.                                 if(inputStream!=null){
  715.                                     inputStream.close();
  716.                                 }
  717.                             }catch(Exception eClose){}
  718.                         }
  719.                     }
  720.                     
  721.                 }
  722.                 
  723.                 if(idActivePolicy!=null){
  724.                     // salvo precedente immagine ?
  725.                     this.mapActiveThreadsPolicy.put(idActivePolicy, 
  726.                             this.buildPolicyGroupByActiveThreads(configurazionePolicy, attivazionePolicy, map, configurazioneControlloTraffico));
  727.                     //System.out.println("@@@ RICOSTRUITO FINALE ID ACTIVE POLICY ["+idActivePolicy+"]");
  728.                     idActivePolicy = null;
  729.                     configurazionePolicy = null;
  730.                     attivazionePolicy = null;
  731.                     map = null;
  732.                     idDatiCollezionati = null;
  733.                 }
  734.                 
  735.             }catch(Exception e){
  736.                 throw new PolicyException("["+entryName+"] "+e.getMessage(),e);
  737.             }
  738.             finally{
  739.                 try{
  740.                     if(zipFile!=null)
  741.                         zipFile.close();
  742.                 }catch(Exception eClose){}
  743.                 try{
  744.                     if(f!=null) {
  745.                         if(!f.delete()) {
  746.                             // ignore
  747.                         }
  748.                     }
  749.                 }catch(Exception eClose){}
  750.                 try{
  751.                     if(in!=null)
  752.                         in.close();
  753.                 }catch(Exception eClose){}
  754.             }
  755.             
  756.             

  758.         }finally {
  759.             this.lock.release(slock, "initialize");
  760.         }
  761.     }
  762.     
  763.     @Override
  764.     public void cleanOldActiveThreadsPolicy() throws PolicyException{
  765.         SemaphoreLock slock =this.lock.acquireThrowRuntime("cleanOldActiveThreadsPolicy");
  766.         try {
  767.             if(this.mapActiveThreadsPolicy!=null && !this.mapActiveThreadsPolicy.isEmpty()) {
  768.                 for (String uniqueIdMap : this.mapActiveThreadsPolicy.keySet()) {
  769.                     for (PolicyGroupByActiveThreadsType tipo : GestorePolicyAttive.getTipiGestoriAttivi()) {
  770.                         if(!tipo.equals(this.type)) {
  771.                             /**System.out.println("======== RIPULISCO DALLA PARTENZA in '"+tipo+"' ["+uniqueIdMap+"] !!!");*/
  772.                             GestorePolicyAttive.getInstance(tipo).removeActiveThreadsPolicyUnsafe(uniqueIdMap);
  773.                         }
  774.                     }
  775.                 }
  776.             }
  777.         }catch(Exception e){
  778.             throw new PolicyException(e.getMessage(),e);
  779.         }   
  780.         finally {
  781.             this.lock.release(slock, "cleanOldActiveThreadsPolicy");
  782.         }
  783.     }
  784.     
  785.     private IPolicyGroupByActiveThreadsInMemory buildPolicyGroupByActiveThreads(ConfigurazionePolicy configurazionePolicy,
  786.             AttivazionePolicy attivazionePolicy, Map<IDUnivocoGroupByPolicy, DatiCollezionati> map,
  787.             ConfigurazioneControlloTraffico configurazioneControlloTraffico) throws Exception{
  788.         
  789.         if(configurazionePolicy==null){
  790.             throw new PolicyException("ConfigurazionePolicy non presente");
  791.         }
  792.         if(attivazionePolicy==null){
  793.             throw new PolicyException("AttivazionePolicy non presente");
  794.         }
  795.         if(configurazioneControlloTraffico==null){
  796.             throw new PolicyException("ConfigurazioneControlloTraffico non presente");
  797.         }
  798.         
  799.         ActivePolicy activePolicy = new ActivePolicy();
  800.         activePolicy.setConfigurazioneControlloTraffico(configurazioneControlloTraffico);
  801.         activePolicy.setConfigurazionePolicy(configurazionePolicy);
  802.         activePolicy.setInstanceConfiguration(attivazionePolicy);
  803.         activePolicy.setTipoRisorsaPolicy(TipoRisorsa.toEnumConstant(configurazionePolicy.getRisorsa(), true));     
  804.         
  805.         String uniqueIdMap = UniqueIdentifierUtilities.getUniqueId(activePolicy.getInstanceConfiguration());
  806.         IPolicyGroupByActiveThreadsInMemory p = newPolicyGroupByActiveThreadsInMemory(activePolicy, uniqueIdMap,
  807.                 null, null);
  808.         
  809.         if(map!=null && map.size()>0){
  810.             for (IDUnivocoGroupByPolicy id : map.keySet()) {
  811.                 map.get(id).initActiveRequestCounter();
  812.             }
  813.             p.initMap(map);
  814.         }
  815.         
  816.         return p;
  817.         
  818.     }
  819.     
  820.     private IPolicyGroupByActiveThreadsInMemory newPolicyGroupByActiveThreadsInMemory(ActivePolicy activePolicy, String uniqueIdMap,
  821.             DatiTransazione datiTransazione, Object state) throws PolicyException {
  822.         switch (this.type) {
  823.         case LOCAL:
  824.             return new PolicyGroupByActiveThreads(activePolicy, this.type);
  825.         case LOCAL_DIVIDED_BY_NODES:
  826.             return new PolicyGroupByActiveThreads(activePolicy, this.type);
  827.         case DATABASE:
  828.             return new PolicyGroupByActiveThreadsDB(activePolicy, this.type, uniqueIdMap, 
  829.                     (state!=null && state instanceof IState) ? ((IState)state) : null, 
  830.                     datiTransazione!=null ? datiTransazione.getDominio() : null, 
  831.                     datiTransazione!=null ? datiTransazione.getIdTransazione() : null);
  832.         case HAZELCAST_LOCAL_CACHE:
  833.             return new PolicyGroupByActiveThreadsDistributedLocalCache(activePolicy, uniqueIdMap, HazelcastManager.getInstance(this.type));
  834.         case HAZELCAST_NEAR_CACHE:
  835.             return new PolicyGroupByActiveThreadsDistributedNearCache(activePolicy, uniqueIdMap, HazelcastManager.getInstance(this.type));
  836.         case HAZELCAST_NEAR_CACHE_UNSAFE_SYNC_MAP:
  837.             return new PolicyGroupByActiveThreadsDistributedNearCacheWithoutEntryProcessorPutSync(activePolicy, uniqueIdMap, HazelcastManager.getInstance(this.type));
  838.         case HAZELCAST_NEAR_CACHE_UNSAFE_ASYNC_MAP:
  839.             return new PolicyGroupByActiveThreadsDistributedNearCacheWithoutEntryProcessorPutAsync(activePolicy, uniqueIdMap, HazelcastManager.getInstance(this.type));
  840.         case HAZELCAST_MAP:
  841.             return new PolicyGroupByActiveThreadsDistributedNoCache(activePolicy, uniqueIdMap, HazelcastManager.getInstance(this.type));
  842.         case HAZELCAST_REPLICATED_MAP:
  843.             return new PolicyGroupByActiveThreadsDistributedReplicatedMap(activePolicy, uniqueIdMap, HazelcastManager.getInstance(this.type));
  844.         case HAZELCAST_PNCOUNTER:
  845.         case HAZELCAST_ATOMIC_LONG:
  846.         case HAZELCAST_ATOMIC_LONG_ASYNC:
  847.         case REDISSON_ATOMIC_LONG:
  848.         case REDISSON_LONGADDER:
  849.              BuilderDatiCollezionatiDistributed builder = BuilderDatiCollezionatiDistributed.getBuilder(this.type);
  850.              if(this.useCountersWithLock) {
  851.                  return new PolicyGroupByActiveThreadsDistributedCountersWithLock(activePolicy, uniqueIdMap,  builder);
  852.              }
  853.              else {
  854.                  return new PolicyGroupByActiveThreadsDistributedCounters(activePolicy, uniqueIdMap,  builder);
  855.              }
  856.         case REDISSON_MAP:
  857.             
  858.             boolean throwInitializingException = true;
  859.             RedissonClient redissonClient = null;
  860.             try {
  861.                 redissonClient = RedissonManager.getRedissonClient(throwInitializingException);
  862.             }catch(Exception e) {
  863.                 throw new PolicyException(e.getMessage(),e);
  864.             }
  865.             
  866.             return new PolicyGroupByActiveThreadsDistributedRedis(activePolicy, uniqueIdMap, redissonClient);
  867.         }
  868.         throw new PolicyException("Unsupported type '"+this.type+"'");
  869.     }
  870.     
  871. }
Created with JaCoCo 0.8.8.202204050719