Merge pull request #3 from DiSSCo/bug-fix/elastic-instant

Fix Elastic Instant and add message on no change

src/main/java/eu/dissco/core/digitalspecimenprocessor/configuration/ElasticSearchConfiguration.java

@@ -4,6 +4,7 @@
 import co.elastic.clients.json.jackson.JacksonJsonpMapper;
 import co.elastic.clients.transport.ElasticsearchTransport;
 import co.elastic.clients.transport.rest_client.RestClientTransport;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import eu.dissco.core.digitalspecimenprocessor.property.ElasticSearchProperties;
 import lombok.RequiredArgsConstructor;
 import org.apache.http.HttpHost;
@@ -15,14 +16,15 @@
 @RequiredArgsConstructor
 public class ElasticSearchConfiguration {
 
+  private final ObjectMapper mapper;
   private final ElasticSearchProperties properties;
 
   @Bean
   public ElasticsearchClient elasticsearchClient() {
     RestClient restClient = RestClient.builder(new HttpHost(properties.getHostname(),
         properties.getPort())).build();
     ElasticsearchTransport transport = new RestClientTransport(restClient,
-        new JacksonJsonpMapper());
+        new JacksonJsonpMapper(mapper));
     return new ElasticsearchClient(transport);
   }
 

src/main/java/eu/dissco/core/digitalspecimenprocessor/controller/DigitalSpecimenController.java

@@ -3,14 +3,17 @@
 import eu.dissco.core.digitalspecimenprocessor.Profiles;
 import eu.dissco.core.digitalspecimenprocessor.domain.DigitalSpecimenEvent;
 import eu.dissco.core.digitalspecimenprocessor.domain.DigitalSpecimenRecord;
+import eu.dissco.core.digitalspecimenprocessor.exception.NoChangesFoundException;
 import eu.dissco.core.digitalspecimenprocessor.service.ProcessingService;
 import javax.xml.transform.TransformerException;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.context.annotation.Profile;
+import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
 import org.springframework.security.access.prepost.PreAuthorize;
+import org.springframework.web.bind.annotation.ExceptionHandler;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -28,9 +31,14 @@ public class DigitalSpecimenController {
   @PreAuthorize("isAuthenticated()")
   @PostMapping(consumes = MediaType.APPLICATION_JSON_VALUE, produces = MediaType.APPLICATION_JSON_VALUE)
   public ResponseEntity<DigitalSpecimenRecord> upsertDigitalSpecimen(@RequestBody
-      DigitalSpecimenEvent event) throws TransformerException {
+      DigitalSpecimenEvent event) throws TransformerException, NoChangesFoundException {
     log.info("Received digitalSpecimen upsert: {}", event);
     var result = processingService.handleMessages(event);
     return ResponseEntity.ok(result);
   }
+
+  @ExceptionHandler(NoChangesFoundException.class)
+  public ResponseEntity<String> handleException(NoChangesFoundException e) {
+    return ResponseEntity.status(HttpStatus.OK).body(e.getMessage());
+  }
 }

src/main/java/eu/dissco/core/digitalspecimenprocessor/exception/NoChangesFoundException.java

@@ -0,0 +1,8 @@
+package eu.dissco.core.digitalspecimenprocessor.exception;
+
+public class NoChangesFoundException extends Exception{
+
+  public NoChangesFoundException(String s) {
+    super(s);
+  }
+}

src/main/java/eu/dissco/core/digitalspecimenprocessor/service/KafkaConsumerService.java

@@ -4,6 +4,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import eu.dissco.core.digitalspecimenprocessor.Profiles;
 import eu.dissco.core.digitalspecimenprocessor.domain.DigitalSpecimenEvent;
+import eu.dissco.core.digitalspecimenprocessor.exception.NoChangesFoundException;
 import javax.xml.transform.TransformerException;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -25,7 +26,11 @@ public class KafkaConsumerService {
   public void getMessages(@Payload String message)
       throws JsonProcessingException, TransformerException {
     var event = mapper.readValue(message, DigitalSpecimenEvent.class);
-    processingService.handleMessages(event);
+    try {
+      processingService.handleMessages(event);
+    } catch (NoChangesFoundException e) {
+      log.info(e.getMessage());
+    }
   }
 
 }

src/main/java/eu/dissco/core/digitalspecimenprocessor/service/ProcessingService.java

@@ -3,6 +3,7 @@
 import eu.dissco.core.digitalspecimenprocessor.domain.DigitalSpecimen;
 import eu.dissco.core.digitalspecimenprocessor.domain.DigitalSpecimenEvent;
 import eu.dissco.core.digitalspecimenprocessor.domain.DigitalSpecimenRecord;
+import eu.dissco.core.digitalspecimenprocessor.exception.NoChangesFoundException;
 import eu.dissco.core.digitalspecimenprocessor.repository.DigitalSpecimenRepository;
 import eu.dissco.core.digitalspecimenprocessor.repository.ElasticSearchRepository;
 import java.time.Instant;
@@ -25,7 +26,7 @@ public class ProcessingService {
   private final KafkaPublisherService kafkaService;
 
   public DigitalSpecimenRecord handleMessages(DigitalSpecimenEvent event)
-      throws TransformerException {
+      throws TransformerException, NoChangesFoundException {
     var digitalSpecimen = event.digitalSpecimen();
     log.info("ds: {}", digitalSpecimen);
     var currentDigitalSpecimenOptional = repository.getDigitalSpecimen(
@@ -39,7 +40,7 @@ public DigitalSpecimenRecord handleMessages(DigitalSpecimenEvent event)
         log.info("Received digital specimen is equal to digital specimen: {}",
             currentDigitalSpecimen.id());
         processEqualDigitalSpecimen(currentDigitalSpecimen);
-        return null;
+        throw new NoChangesFoundException("No changes were necessary to specimen with id: " + currentDigitalSpecimen.id());
       } else {
         log.info("Specimen with id: {} has received an update", currentDigitalSpecimen.id());
         return updateExistingDigitalSpecimen(currentDigitalSpecimen, digitalSpecimen);