flags new scheduller

credit-tracker.properties

@@ -67,6 +67,7 @@ ct.load-dir=D:\\folder_name\\signals
 ct.monitoring-interval=${MONITORING_INTERVAL:15}
 # days
 ct.update-flags-cron-expression=${UPDATE_FLAGS_CRON_EXPRESSION:0 25 18 * * *}
+ct.new-flags-cron-expression=${UPDATE_FLAGS_CRON_EXPRESSION:0 25 18 * * *}
 ct.days-to-update-flags=${DAYS_TO_UPDATE_FLAGS:30}
 ct.update-fid-cron-expression=${UPDATE_FLAGS_CRON_EXPRESSION:0 10 12 * * *}
 ct.days-to-update-fids=${DAYS_TO_UPDATE_FIDS:30}

src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java

@@ -12,6 +12,8 @@ public interface PhonesMapMapper {
     PhonesMap findByPhone(@Param("phone") Long phone);
     List<PhonesMap> findEligibleForFlagsUpdate(@Param("lastId") Long lastId,
                                                @Param("batchSize") Integer batchSize);
+    List<PhonesMap> findEligibleForFlagsInsert(@Param("lastId") Long lastId,
+                                               @Param("batchSize") Integer batchSize);
     void updateIdCachedData(@Param("json") String json);
     void updateNameCachedData(@Param("json") String json);
     void updateFlagCachedData(@Param("json") String json);

src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java

@@ -1,7 +1,10 @@
 package ru.nbch.credit_tracker.service;
 
 import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
 import lombok.extern.slf4j.Slf4j;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
@@ -17,9 +20,12 @@ import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher;
 @Service
 @EnableScheduling
 public class UpdatingFlagsService {
+    private final Logger log = LoggerFactory.getLogger(this.getClass());
     private final static int BATCH_SIZE = 1000;
     private final PhonesMapMapper phonesMapMapper;
     private final FlagsEnricher flagsEnricherV2;
+    private final AtomicBoolean insertIsRunning = new AtomicBoolean(false);
+    private final AtomicBoolean updateIsRunning = new AtomicBoolean(false);
 
     public UpdatingFlagsService(PhonesMapMapper phonesMapMapper, FlagsEnricher flagsEnricherV2) {
         this.phonesMapMapper = phonesMapMapper;
@@ -27,21 +33,59 @@ public class UpdatingFlagsService {
     }
 
     @Scheduled(cron = "${ct.update-flags-cron-expression}")
-    private void update() {
+    private void updateFlags() {
-        log.info("Start filling flags");
+        if (!updateIsRunning.compareAndSet(false, true)) {
+            log.error("update flags processor already running... skipping scheduled update");
+            return;
+        }
+        try {
+            log.info("start update flags processor");
             long lastId = 0L;
             int batchNumber = 1;
             List<PhonesMap> phonesMaps;
             while (true) {
                 phonesMaps = phonesMapMapper.findEligibleForFlagsUpdate(lastId, BATCH_SIZE);
-            log.trace("batch {}, signals_phones_map records loaded: {}", batchNumber, phonesMaps.size());
+                log.trace("update batch {}, signals_phones_map records loaded: {}", batchNumber, phonesMaps.size());
                 if (phonesMaps.isEmpty()) {
                     break;
                 }
 
                 flagsEnricherV2.accept(phonesMaps);
                 lastId = phonesMaps.getLast().getId();
-            log.trace("batch {}, lastId {}", batchNumber++, lastId);
+                log.trace("update batch {}, lastId {}", batchNumber++, lastId);
             }
+            log.info("end update flags processor");
+        } finally {
+            updateIsRunning.set(false);
+        }
+    }
+
+    @Scheduled(cron = "${ct.new-flags-cron-expression}")
+    private void newFlags() {
+        if (!insertIsRunning.compareAndSet(false, true)) {
+            log.warn("new flags processor already running... skipping scheduled insert");
+            return;
+        }
+        try {
+            log.info("start processing new flags");
+            long lastId = 0L;
+            int batchNumber = 1;
+            List<PhonesMap> phonesMaps;
+            while (true) {
+                phonesMaps = phonesMapMapper.findEligibleForFlagsInsert(lastId, BATCH_SIZE);
+                log.trace("new batch {}, signals_phones_map records loaded: {}", batchNumber, phonesMaps.size());
+                if (phonesMaps.isEmpty()) {
+                    break;
+                }
+
+                flagsEnricherV2.accept(phonesMaps);
+                lastId = phonesMaps.getLast().getId();
+                log.trace("new batch {}, lastId {}", batchNumber++, lastId);
+            }
+            log.info("end new flags processor");
+        } finally {
+            insertIsRunning.set(false);
+        }
+
     }
 }

src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml

@@ -120,7 +120,17 @@
         select id, fid, phone
         from signals_phones_map
         where fid is not null
-          and (calculated_at is null  or calculated_at < searched_at)
+          and (calculated_at is not null and calculated_at < searched_at)
+          and id > #{lastId}
+        order by id
+        limit #{batchSize}
+    </select>
+
+    <select id="findEligibleForFlagsInsert" resultMap="PhoneMap">
+        select id, fid, phone
+        from signals_phones_map
+        where fid is not null
+          and calculated_at is null
           and id > #{lastId}
         order by id
         limit #{batchSize}