PAGOPA-1745 adding dynamic mechanism to retrieve station and broker

.env.sample

@@ -11,4 +11,6 @@ PAA_PASSWORD=PAA_PASSWORD
 NCRON_SCHEDULE_BATCH=*/45 * * * * *
 MAX_RETRY_QUEUING= 5
 QUEUE_RETENTION_SEC= 86400
-QUEUE_DELAY_SEC= 3600
+QUEUE_DELAY_SEC= 3600
+CACHE_CLIENT_HOST=https://localhost:8080
+CACHE_API_KEY=yyy

local.settings.json.example

@@ -15,6 +15,8 @@
       "PAA_PASSWORD":"PAA_PASSWORD",
       "MAX_RETRY_QUEUING": 5,
       "QUEUE_RETENTION_SEC": 86400,
-      "QUEUE_DELAY_SEC": 3600
+      "QUEUE_DELAY_SEC": 3600,
+      "CACHE_CLIENT_HOST": "https://localhost:8080",
+      "CACHE_API_KEY": "yyy"
     }
   }

pom.xml

@@ -19,6 +19,7 @@
         <azure.storage.queue.version>12.11.3</azure.storage.queue.version>
         <functionAppName>com.microsoft.azure-20220215182005862</functionAppName>
         <resteasy.version>3.15.3.Final</resteasy.version>
+        <google-api-client.version>2.0.1</google-api-client.version>
     </properties>
 
     <dependencies>
@@ -150,6 +151,18 @@
             <version>2.13.1</version>
         </dependency>
 
+        <dependency>
+            <groupId>com.google.api-client</groupId>
+            <artifactId>google-api-client-gson</artifactId>
+            <version>${google-api-client.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.google.http-client</groupId>
+            <artifactId>google-http-client-jackson2</artifactId>
+            <version>1.42.3</version>
+        </dependency>
+
         <!-- Jackson END-->
 
         <!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
@@ -170,8 +183,8 @@
                 <artifactId>maven-compiler-plugin</artifactId>
                 <version>3.8.1</version>
                 <configuration>
-                    <source>${java.version}</source>
-                    <target>${java.version}</target>
+                    <source>16</source>
+                    <target>16</target>
                     <encoding>${project.build.sourceEncoding}</encoding>
                 </configuration>
             </plugin>

src/main/java/it/gov/pagopa/reporting/RetrieveFlows.java

@@ -6,15 +6,24 @@
 import com.microsoft.azure.functions.annotation.FunctionName;
 import com.microsoft.azure.functions.annotation.QueueTrigger;
 import com.sun.xml.ws.client.ClientTransportException;
+import it.gov.pagopa.reporting.client.ApiConfigClient;
+import it.gov.pagopa.reporting.exception.Cache4XXException;
+import it.gov.pagopa.reporting.exception.Cache5XXException;
 import it.gov.pagopa.reporting.models.OrganizationsMessage;
+import it.gov.pagopa.reporting.models.cache.CacheResponse;
+import it.gov.pagopa.reporting.models.cache.CreditorInstitutionStation;
+import it.gov.pagopa.reporting.models.cache.Station;
 import it.gov.pagopa.reporting.service.FlowsService;
 import it.gov.pagopa.reporting.service.NodoChiediElencoFlussi;
 import it.gov.pagopa.reporting.service.OrganizationsService;
 import it.gov.pagopa.reporting.servicewsdl.FaultBean;
 import it.gov.pagopa.reporting.servicewsdl.TipoElencoFlussiRendicontazione;
 
+import java.io.IOException;
 import java.time.LocalDateTime;
 import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -32,6 +41,8 @@ public class RetrieveFlows {
     private final String initialVisibilityDelayInSeconds = System.getenv("QUEUE_DELAY_SEC");
     private final String maxRetryQueuing = System.getenv("MAX_RETRY_QUEUING");
 
+    private static CacheResponse cacheContent;
+
     /**
      * This function will be invoked when a new message is detected in the queue
      */
@@ -45,15 +56,25 @@ public void run(
 
         NodoChiediElencoFlussi nodeClient = this.getNodeClientInstance(logger);
         FlowsService flowsService = this.getFlowsServiceInstance(logger);
+        ApiConfigClient cacheClient = this.getCacheClientInstance();
+        if(cacheContent == null || (cacheContent.getRetrieveDate() != null && cacheContent.getRetrieveDate().isBefore(LocalDateTime.now()))) {
+            synchronized (RetrieveFlows.class) {
+                setCache(cacheClient, logger);
+            }
+        }
 
         try {
             OrganizationsMessage organizationsMessage = new ObjectMapper().readValue(message, OrganizationsMessage.class);
 
             Arrays.stream(organizationsMessage.getIdPA())
                     .forEach((organization -> {
                         try {
+                            Station stationBroker = getPAStationIntermediario(organization)
+                                    .orElseThrow(() -> new RuntimeException(String.format("No data present in api config database for PA %s", organization)));
+                            String idStation = stationBroker.getStationCode();
+                            String idBroker = stationBroker.getBrokerCode();
                             // call NODO dei pagamenti
-                            nodeClient.nodoChiediElencoFlussiRendicontazione(organization);
+                            nodeClient.nodoChiediElencoFlussiRendicontazione(organization, idBroker, idStation);
 
                             // retrieve result
                             FaultBean faultBean = nodeClient.getNodoChiediElencoFlussiRendicontazioneFault();
@@ -83,6 +104,10 @@ public void run(
         }
     }
 
+    public ApiConfigClient getCacheClientInstance() {
+        return ApiConfigClient.getInstance();
+    }
+
     public NodoChiediElencoFlussi getNodeClientInstance(Logger logger) {
         return new NodoChiediElencoFlussi(logger);
     }
@@ -94,4 +119,37 @@ public FlowsService getFlowsServiceInstance(Logger logger) {
     public OrganizationsService getOrganizationsServiceInstance(Logger logger) {
         return new OrganizationsService(this.storageConnectionString, this.organizationsTable, this.organizationsQueue, Integer.parseInt(timeToLiveInSeconds), Integer.parseInt(initialVisibilityDelayInSeconds), logger);
     }
+
+    public Optional<Station> getPAStationIntermediario(String idPa) {
+        List<String> stationPa = getStazioni(idPa);
+        return cacheContent.getStations().stream()
+                .filter(station -> stationPa.contains(station.getStationCode()))
+                .filter(Station::getEnabled)
+                .findFirst();
+    }
+
+    public List<String> getStazioni(String idPa) {
+        return cacheContent.getCreditorInstitutionStations().stream()
+                .filter(creditorInstitutionStation -> creditorInstitutionStation.getCreditorInstitutionCode().equals(idPa))
+                .map(CreditorInstitutionStation::getStationCode).toList();
+    }
+
+    public synchronized void setCache(ApiConfigClient cacheClient, Logger logger) {
+        try {
+            if(cacheContent == null) {
+                cacheContent = cacheClient.getCache();
+                cacheContent.setRetrieveDate(LocalDateTime.now());
+            }
+        } catch (Cache4XXException | Cache5XXException e) {
+            cacheContent = null;
+            logger.log(Level.SEVERE, e.getMessage());
+        } catch (IOException e) {
+            cacheContent = null;
+            logger.log(Level.SEVERE, e.getMessage());
+            throw new RuntimeException(e.getMessage());
+        } catch (Exception e) {
+            cacheContent = null;
+            logger.log(Level.SEVERE, e.getMessage());
+        }
+    }
 }

src/main/java/it/gov/pagopa/reporting/client/ApiConfigClient.java

@@ -0,0 +1,134 @@
+package it.gov.pagopa.reporting.client;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.api.client.http.*;
+import com.google.api.client.http.javanet.NetHttpTransport;
+import com.google.api.client.json.JsonFactory;
+import com.google.api.client.json.JsonObjectParser;
+import com.google.api.client.json.gson.GsonFactory;
+import com.google.api.client.util.ExponentialBackOff;
+import com.google.gson.reflect.TypeToken;
+import it.gov.pagopa.reporting.exception.Cache4XXException;
+import it.gov.pagopa.reporting.exception.Cache5XXException;
+import it.gov.pagopa.reporting.models.cache.CacheResponse;
+import it.gov.pagopa.reporting.models.cache.CreditorInstitutionStation;
+import it.gov.pagopa.reporting.models.cache.Station;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class ApiConfigClient {
+
+    private static ApiConfigClient instance = null;
+
+    private final HttpTransport httpTransport = new NetHttpTransport();
+    private final JsonFactory jsonFactory = new GsonFactory();
+    private final String apiConfigCacheHost = System.getenv("CACHE_CLIENT_HOST"); // es: https://api.xxx.platform.pagopa.it
+    private final String getCacheDetails =
+            System.getenv("CACHE_PATH") != null ? System.getenv("CACHE_PATH") : "/cache?keys=creditorInstitutionStations,stations";
+    private final String apiKey = System.getenv("CACHE_API_KEY");
+
+
+    // Retry ExponentialBackOff config
+    private final boolean enableRetry =
+            System.getenv("ENABLE_CLIENT_RETRY") != null ? Boolean.parseBoolean(System.getenv("ENABLE_CLIENT_RETRY")) : Boolean.FALSE;
+    private final int initialIntervalMillis =
+            System.getenv("INITIAL_INTERVAL_MILLIS") != null ? Integer.parseInt(System.getenv("INITIAL_INTERVAL_MILLIS")) : 500;
+    private final int maxElapsedTimeMillis =
+            System.getenv("MAX_ELAPSED_TIME_MILLIS") != null ? Integer.parseInt(System.getenv("MAX_ELAPSED_TIME_MILLIS")) : 1000;
+    private final int maxIntervalMillis  =
+            System.getenv("MAX_INTERVAL_MILLIS") != null ? Integer.parseInt(System.getenv("MAX_INTERVAL_MILLIS")) : 1000;
+    private final double multiplier  =
+            System.getenv("MULTIPLIER") != null ? Double.parseDouble(System.getenv("MULTIPLIER")) : 1.5;
+    private final double randomizationFactor  =
+            System.getenv("RANDOMIZATION_FACTOR") != null ? Double.parseDouble(System.getenv("RANDOMIZATION_FACTOR")) : 0.5;
+
+    public static ApiConfigClient getInstance() {
+        if (instance == null) {
+            instance = new ApiConfigClient();
+        }
+        return instance;
+    }
+
+    public CacheResponse getCache() throws IOException, IllegalArgumentException, Cache5XXException, Cache4XXException {
+        GenericUrl url = new GenericUrl(apiConfigCacheHost + getCacheDetails);
+        HttpRequest request = this.buildGetRequestToApiConfigCache(url);
+
+        if (enableRetry) {
+            this.setRequestRetry(request);
+        }
+
+        return this.executeCallToApiConfigCache(request);
+    }
+
+    public HttpRequest buildGetRequestToApiConfigCache(GenericUrl url) throws IOException {
+
+        HttpRequestFactory requestFactory = httpTransport.createRequestFactory(
+                (HttpRequest request) ->
+                        request.setParser(new JsonObjectParser(jsonFactory))
+        );
+
+        HttpRequest request = requestFactory.buildGetRequest(url);
+        HttpHeaders headers = request.getHeaders();
+        headers.set("Ocp-Apim-Subscription-Key", apiKey);
+        return request;
+    }
+
+    public void setRequestRetry(HttpRequest request) {
+        /**
+         * Retry section config
+         */
+        ExponentialBackOff backoff = new ExponentialBackOff.Builder()
+                .setInitialIntervalMillis(initialIntervalMillis)
+                .setMaxElapsedTimeMillis(maxElapsedTimeMillis)
+                .setMaxIntervalMillis(maxIntervalMillis)
+                .setMultiplier(multiplier)
+                .setRandomizationFactor(randomizationFactor)
+                .build();
+
+        // Exponential Backoff is turned off by default in HttpRequest -> it's necessary include an instance of HttpUnsuccessfulResponseHandler to the HttpRequest to activate it
+        // The default back-off on anabnormal HTTP response is BackOffRequired.ON_SERVER_ERROR (5xx)
+        request.setUnsuccessfulResponseHandler(
+                new HttpBackOffUnsuccessfulResponseHandler(backoff));
+    }
+
+    public CacheResponse executeCallToApiConfigCache(HttpRequest request) throws IOException, IllegalArgumentException, Cache5XXException, Cache4XXException {
+
+        Type type = new TypeToken<List<CreditorInstitutionStation>>() {}.getType();
+
+        ObjectMapper mapper = new ObjectMapper();
+        CacheResponse cacheResponse = CacheResponse.builder().build();
+        List<CreditorInstitutionStation> creditorInstitutionStationList = new ArrayList<>();
+        List<Station> stationList = new ArrayList<>();
+        try {
+            InputStream resIs = request.execute().getContent();
+            Map<String,Object> responseMap = mapper.readValue(resIs, HashMap.class);
+            Map<String,Object> creditorInstitutionStations = (HashMap) responseMap.get("creditorInstitutionStations");
+            for (Map.Entry<String, Object> creditorInstitutionStation : creditorInstitutionStations.entrySet()) {
+                creditorInstitutionStationList.add(mapper.readValue(mapper.writeValueAsString(creditorInstitutionStation.getValue()), CreditorInstitutionStation.class));
+            }
+            Map<String,Object> stations = (HashMap) responseMap.get("stations");
+            for (Map.Entry<String, Object> station : stations.entrySet()) {
+                stationList.add(mapper.readValue(mapper.writeValueAsString(station.getValue()), Station.class));
+            }
+            cacheResponse.setStations(stationList);
+            cacheResponse.setCreditorInstitutionStations(creditorInstitutionStationList);
+        } catch (HttpResponseException e) {
+            if (e.getStatusCode() / 100 == 4) {
+                String message = String.format("Error %s calling the service URL %s", e.getStatusCode(), request.getUrl());
+                throw new Cache4XXException(message);
+
+            } else if (e.getStatusCode() / 100 == 5) {
+                String message = String.format("Error %s calling the service URL %s", e.getStatusCode(), request.getUrl());
+                throw new Cache5XXException(message);
+
+            }
+        }
+        return cacheResponse;
+    }
+}

src/main/java/it/gov/pagopa/reporting/exception/AppException.java

@@ -0,0 +1,18 @@
+package it.gov.pagopa.reporting.exception;
+
+public class AppException extends Exception {
+
+	/**
+	 * generated serialVersionUID
+	 */
+	private static final long serialVersionUID = -7564079264281462536L;
+
+	public AppException() {
+		super();
+	}
+
+	public AppException(String message) {
+		super(message);
+	}
+
+}

src/main/java/it/gov/pagopa/reporting/exception/Cache4XXException.java

@@ -0,0 +1,18 @@
+package it.gov.pagopa.reporting.exception;
+
+public class Cache4XXException extends Exception {
+
+	/**
+	 * generated serialVersionUID
+	 */
+	private static final long serialVersionUID = -7564079264281462536L;
+
+	public Cache4XXException() {
+		super();
+	}
+
+	public Cache4XXException(String message) {
+		super(message);
+	}
+
+}

src/main/java/it/gov/pagopa/reporting/exception/Cache5XXException.java

@@ -0,0 +1,18 @@
+package it.gov.pagopa.reporting.exception;
+
+public class Cache5XXException extends Exception {
+
+	/**
+	 * generated serialVersionUID
+	 */
+	private static final long serialVersionUID = -7564079264281462536L;
+
+	public Cache5XXException() {
+		super();
+	}
+
+	public Cache5XXException(String message) {
+		super(message);
+	}
+
+}

src/main/java/it/gov/pagopa/reporting/models/cache/CacheResponse.java

@@ -0,0 +1,27 @@
+package it.gov.pagopa.reporting.models.cache;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.*;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+@NoArgsConstructor
+@AllArgsConstructor
+@Getter
+@Setter
+@ToString
+@JsonIgnoreProperties(ignoreUnknown = true)
+@Builder
+public class CacheResponse {
+
+    @JsonProperty(value = "stations")
+    private List<Station> stations;
+
+    @JsonProperty(value = "creditorInstitutionStations")
+    private List<CreditorInstitutionStation> creditorInstitutionStations;
+
+    private LocalDateTime retrieveDate;
+
+}