Merge pull request #19 from redlink-gmbh/fix-jdbc-connection-exception

MORE-Platform#237: Fixed a DB leak

Author: janoliver20

studymanager-services/src/main/java/io/redlink/more/studymanager/scheduling/ValidateActiveObservationDataCron.java

@@ -4,7 +4,10 @@
 import io.redlink.more.studymanager.core.factory.ObservationFactory;
 import io.redlink.more.studymanager.core.io.TimeRange;
 import io.redlink.more.studymanager.core.io.Timeframe;
-import io.redlink.more.studymanager.model.*;
+import io.redlink.more.studymanager.model.Observation;
+import io.redlink.more.studymanager.model.OccurredObservation;
+import io.redlink.more.studymanager.model.Participant;
+import io.redlink.more.studymanager.model.Study;
 import io.redlink.more.studymanager.sdk.MoreSDK;
 import io.redlink.more.studymanager.service.ObservationService;
 import io.redlink.more.studymanager.service.OccurredObservationService;
@@ -41,7 +44,7 @@ public ValidateActiveObservationDataCron(
             OccurredObservationService occurredObservationService,
             ObservationService observationService,
             MoreSDK sdk
-    ){
+    ) {
         this.studyService = studyService;
         this.participantService = participantService;
         this.occurredObservationService = occurredObservationService;
@@ -51,8 +54,8 @@ public ValidateActiveObservationDataCron(
 
 
     //TODO: Make schedules configureable
-    @Scheduled(cron="30 */5 * * * ?") //every 5min, 30sec after updating active observation data
-    public void validateActiveObservationData(){
+    @Scheduled(cron = "30 */5 * * * ?") //every 5min, 30sec after updating active observation data
+    public void validateActiveObservationData() {
         final Instant now = Instant.now();
         final LocalTime currentTime = LocalTime.ofInstant(now, ZoneId.systemDefault());
         //once every hour validate all occurred observations
@@ -61,25 +64,27 @@ public void validateActiveObservationData(){
         final Set<ObservationDataState> validateStates = validateAll
                 ? EnumSet.allOf(ObservationDataState.class)
                 : EnumSet.complementOf(EnumSet.of(ObservationDataState.COMPLETE));
-        Map<String, ObservationFactory>  observationFactories = new HashMap<>();
-        studyService.getStudiesByStates(Study.Status.ACTIVE_STATES).forEach(study -> {
-            //NOTE: OccurredObservations will refer the same observations and participants several times, A single lookup
-            //      per run is sufficient, so we keep them in a cache
-            Map<Integer, Observation> observationCache = new HashMap<>();
-            Map<Integer, Participant> participantCache = new HashMap<>();
-            Map<Integer, io.redlink.more.studymanager.core.component.Observation> observationComponentCache = new HashMap<>();
-            try (var ooStrem = occurredObservationService.streamOccurredObservations(study.getStudyId(), null, null, validateAll, validateStates)) {
-                ooStrem
-                    .filter(oo -> validateAll || oo.end().plus(1, ChronoUnit.DAYS).isAfter(now))
-                    .forEach(occurredObservation -> validateOccurrence(
-                        study,
-                        occurredObservation,
-                        observationCache,
-                        observationFactories,
-                        observationComponentCache,
-                        participantCache));
-            }
-        });
+        Map<String, ObservationFactory> observationFactories = new HashMap<>();
+        try (var stream = studyService.getStudiesByStates(Study.Status.ACTIVE_STATES)) {
+            stream.forEach(study -> {
+                //NOTE: OccurredObservations will refer the same observations and participants several times, A single lookup
+                //      per run is sufficient, so we keep them in a cache
+                Map<Integer, Observation> observationCache = new HashMap<>();
+                Map<Integer, Participant> participantCache = new HashMap<>();
+                Map<Integer, io.redlink.more.studymanager.core.component.Observation> observationComponentCache = new HashMap<>();
+                try (var ooStrem = occurredObservationService.streamOccurredObservations(study.getStudyId(), null, null, validateAll, validateStates)) {
+                    ooStrem
+                            .filter(oo -> validateAll || oo.end().plus(1, ChronoUnit.DAYS).isAfter(now))
+                            .forEach(occurredObservation -> validateOccurrence(
+                                    study,
+                                    occurredObservation,
+                                    observationCache,
+                                    observationFactories,
+                                    observationComponentCache,
+                                    participantCache));
+                }
+            });
+        }
     }
 
     private void validateOccurrence(Study study, OccurredObservation occuredObservation, Map<Integer, Observation> observationCache, Map<String, ObservationFactory> observationFactories, Map<Integer, io.redlink.more.studymanager.core.component.Observation> observationComponentCache, Map<Integer, Participant> participantCache) {
@@ -112,17 +117,17 @@ private void validateOccurrence(Study study, OccurredObservation occuredObservat
 
 
     private void validate(ValidateionContext ctx) {
-        if(Instant.now().isBefore(ctx.occurrence.start())) {
+        if (Instant.now().isBefore(ctx.occurrence.start())) {
             //do not try to validate occurrences that are in the future
             log.warn("Unexpected call to validate {} with a future start timestamp", ctx.occurrence);
             return;
         }
         var validationResults = sdk.validateData(
-            ctx.getStudyId(), ctx.getStudyGroupId(),ctx.getObservationId(),
-            ctx.participant.getParticipantId(), ctx.getTimeRage(),
-            ctx.observationFactory.getMeasurementSet());
+                ctx.getStudyId(), ctx.getStudyGroupId(), ctx.getObservationId(),
+                ctx.participant.getParticipantId(), ctx.getTimeRage(),
+                ctx.observationFactory.getMeasurementSet());
         final OccurredObservation updatedOccurredObservation;
-        if(validationResults != null) {
+        if (validationResults != null) {
             var validationResult = ctx.observationComponent.validateData(ctx.occurrence.start(), ctx.occurrence.end(), validationResults);
             log.debug("validated {}: results: {}, state: {}", ctx.occurrence, validationResults, validationResult);
             updatedOccurredObservation = new OccurredObservation(
@@ -153,27 +158,27 @@ private void validate(ValidateionContext ctx) {
 
 
     private record ValidateionContext(
-        Study study,
-        OccurredObservation occurrence,
-        Participant participant,
-        Observation observation,
-        ObservationFactory observationFactory,
-        io.redlink.more.studymanager.core.component.Observation observationComponent
+            Study study,
+            OccurredObservation occurrence,
+            Participant participant,
+            Observation observation,
+            ObservationFactory observationFactory,
+            io.redlink.more.studymanager.core.component.Observation observationComponent
     ) {
 
-        public Long getStudyId(){
+        public Long getStudyId() {
             return study.getStudyId();
         }
 
-        public Integer getStudyGroupId(){
+        public Integer getStudyGroupId() {
             return observation.getStudyGroupId();
         }
 
-        public Integer getObservationId(){
+        public Integer getObservationId() {
             return observation.getObservationId();
         }
 
-        public Integer getParticipantId(){
+        public Integer getParticipantId() {
             return participant.getParticipantId();
         }