Cde 297 modifier traitement dans bestppnapi2

src/main/java/fr/abes/bestppn/configuration/KafkaConfig.java

@@ -4,7 +4,9 @@
 import fr.abes.LigneKbartImprime;
 import io.confluent.kafka.serializers.KafkaAvroSerializer;
 import io.confluent.kafka.serializers.KafkaAvroSerializerConfig;
+import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
@@ -14,15 +16,23 @@
 import org.springframework.kafka.annotation.EnableKafka;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.core.*;
+import org.springframework.kafka.listener.CommonErrorHandler;
+import org.springframework.kafka.listener.MessageListenerContainer;
+import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
 import org.springframework.kafka.transaction.KafkaTransactionManager;
 
 import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Semaphore;
 
 @Configuration
 @EnableKafka
 public class KafkaConfig {
+    @Value("${spring.kafka.concurrency.nbThread}")
+    private int nbThread;
+
     @Value("${spring.kafka.consumer.bootstrap-servers}")
     private String bootstrapAddress;
 
@@ -53,6 +63,7 @@ public ConsumerFactory<String, String> consumerKbartFactory() {
         return new DefaultKafkaConsumerFactory<>(props);
     }
 
+
     @Bean
     public ConcurrentKafkaListenerContainerFactory<String, String>
     kafkaKbartListenerContainerFactory() {
@@ -99,8 +110,7 @@ public ProducerFactory<String, LigneKbartImprime> producerFactoryLigneKbartImpri
 
     @Bean
     public ProducerFactory<String, String> producerFactory() {
-        DefaultKafkaProducerFactory<String, String> factory = new DefaultKafkaProducerFactory<>(producerConfigs());
-        return factory;
+        return new DefaultKafkaProducerFactory<>(producerConfigs());
     }
 
     @Bean
@@ -122,4 +132,10 @@ public KafkaTransactionManager<String, LigneKbartImprime> kafkaTransactionManage
     public KafkaTemplate<String, String> kafkatemplateEndoftraitement(final ProducerFactory producerFactory) {
         return new KafkaTemplate<>(producerFactory);
     }
+
+    @Bean
+    public Semaphore semaphore() {
+        return new Semaphore(1);
+    }
+
 }

src/main/java/fr/abes/bestppn/controller/BestPpnController.java

@@ -2,7 +2,7 @@
 
 import fr.abes.bestppn.dto.kafka.LigneKbartDto;
 import fr.abes.bestppn.exception.BestPpnException;
-import fr.abes.bestppn.exception.IllegalPpnException;
+import fr.abes.bestppn.exception.IllegalDoiException;
 import fr.abes.bestppn.service.BestPpnService;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.media.Content;
@@ -41,7 +41,7 @@ public String bestPpn(@RequestParam(name = "provider") String provider, @Request
                           @RequestParam(name = "publication_type") String publicationType, @RequestParam(name = "online_identifier", required = false) String onlineIdentifier,
                           @RequestParam(name = "print_identifier", required = false) String printIdentifier, @RequestParam(name = "titleUrl", required = false) String titleUrl,
                           @RequestParam(name = "date_monograph_published_online", required = false) String dateMonographPublishedOnline, @RequestParam(name = "date_monograph_published_print", required = false) String dateMonographPublishedPrint,
-                          @RequestParam(name = "first_author", required = false) String firstAuthor, @RequestParam(name = "force", required = false) Boolean force) throws IOException, IllegalPpnException {
+                          @RequestParam(name = "first_author", required = false) String firstAuthor, @RequestParam(name = "force", required = false) Boolean force) throws IOException {
         try {
             LigneKbartDto ligneKbartDto = new LigneKbartDto();
             ligneKbartDto.setPublicationType(publicationType);
@@ -56,7 +56,7 @@ public String bestPpn(@RequestParam(name = "provider") String provider, @Request
             return service.getBestPpn(ligneKbartDto, provider, injectKafka).getPpn();
         } catch (URISyntaxException e) {
             throw new IllegalArgumentException("Une url dans le champ title_url du kbart n'est pas correcte");
-        } catch (BestPpnException | RestClientException | IllegalArgumentException e) {
+        } catch (BestPpnException | RestClientException | IllegalArgumentException | IllegalDoiException e) {
             return e.getMessage();
         }
     }

src/main/java/fr/abes/bestppn/entity/ExecutionReport.java

@@ -2,36 +2,65 @@
 
 import lombok.Data;
 
-@Data
+import java.util.concurrent.atomic.AtomicInteger;
+
+
 public class ExecutionReport {
-    private int nbtotalLines = 0;
+    private int nbtotalLines;
 
-    private int nbBestPpnFind = 0;
+    private AtomicInteger nbBestPpnFind;
 
-    private int nbLinesWithInputDataErrors = 0;
+    private AtomicInteger nbLinesWithInputDataErrors;
 
-    private int nbLinesWithErrorsInBestPPNSearch = 0;
+    private AtomicInteger nbLinesWithErrorsInBestPPNSearch;
 
-    public int getNbLinesOk(){
-        return nbtotalLines - nbLinesWithErrorsInBestPPNSearch - nbLinesWithInputDataErrors;
+    public ExecutionReport() {
+        nbBestPpnFind = new AtomicInteger(0);
+        nbLinesWithInputDataErrors = new AtomicInteger(0);
+        nbLinesWithErrorsInBestPPNSearch = new AtomicInteger(0);
+        nbtotalLines = 0;
     }
 
-    public void addNbBestPpnFind(){
-        nbBestPpnFind++;
+    public int getNbtotalLines() {
+        return this.nbtotalLines;
     }
 
-    public void addNbLinesWithInputDataErrors(){
-        nbLinesWithInputDataErrors++;
+    public void setNbtotalLines(int nbtotalLines) {
+        this.nbtotalLines = nbtotalLines;
     }
 
-    public void addNbLinesWithErrorsInBestPPNSearch(){
-        nbLinesWithErrorsInBestPPNSearch++;
+    public int getNbBestPpnFind() {
+        return nbBestPpnFind.get();
+    }
+
+    public int getNbLinesWithInputDataErrors() {
+        return nbLinesWithInputDataErrors.get();
+    }
+
+    public int getNbLinesWithErrorsInBestPPNSearch() {
+        return nbLinesWithErrorsInBestPPNSearch.get();
+    }
+
+    public int getNbLinesOk(){
+        return nbtotalLines - nbLinesWithErrorsInBestPPNSearch.get() - nbLinesWithInputDataErrors.get();
+    }
+
+    public void incrementNbBestPpnFind() {
+        nbBestPpnFind.incrementAndGet();
+    }
+
+    public void incrementNbLinesWithInputDataErrors() {
+        nbLinesWithInputDataErrors.incrementAndGet();
     }
 
     public void clear(){
         nbtotalLines = 0;
-        nbBestPpnFind = 0;
-        nbLinesWithInputDataErrors = 0;
-        nbLinesWithErrorsInBestPPNSearch = 0;
+        nbBestPpnFind.set(0);
+        nbLinesWithInputDataErrors.set(0);
+        nbLinesWithErrorsInBestPPNSearch.set(0);
+    }
+
+    public void incrementNbLinesWithErrorsInBestPPNSearch() {
+        nbLinesWithErrorsInBestPPNSearch.incrementAndGet();
     }
 }

src/main/java/fr/abes/bestppn/exception/ExceptionControllerHandler.java

@@ -30,12 +30,6 @@ protected ResponseEntity<Object> handleIllegalArgumentException(IllegalArgumentE
         return buildResponseEntity(new ApiReturnError(HttpStatus.BAD_REQUEST, error, ex));
     }
 
-    @ExceptionHandler(IllegalPpnException.class)
-    protected ResponseEntity<Object> handleIllegalPpnException(IllegalPpnException ex) {
-        String error = "Erreur dans la récupération d'une notice";
-        log.debug(ex.getLocalizedMessage());
-        return buildResponseEntity(new ApiReturnError(HttpStatus.SERVICE_UNAVAILABLE, error, ex));
-    }
 
     @ExceptionHandler(BestPpnException.class)
     protected ResponseEntity<Object> handleBestPpnException(BestPpnException ex) {

src/main/java/fr/abes/bestppn/exception/IllegalDoiException.java

@@ -0,0 +1,7 @@
+package fr.abes.bestppn.exception;
+
+public class IllegalDoiException extends Exception {
+    public IllegalDoiException(String message) {
+        super(message);
+    }
+}