Added FlagsEnricher; Added id to SubjectProfile to use it as uid in PersonFactoryBuilderV2

credit-tracker.properties

@@ -8,9 +8,9 @@ server.tomcat.accesslog.pattern=%h %l %u %t "%r" %s %b
 # datasources
 # indic
 ct.indic-data-source-settings.driver=com.ibm.db2.jcc.DB2Driver
-ct.indic-data-source-settings.url=${DB_URL_INDIC:jdbc:db2://10.230.227.101:2668/cprosd22:currentSchema=INDIC;}
-ct.indic-data-source-settings.username=${USERNAME_INDIC:tester}
-ct.indic-data-source-settings.password=${PASSWORD_INDIC:tester}
+ct.indic-data-source-settings.url=${DB_URL_INDIC:jdbc:db2://10.230.227.100:2668/cprosd22:currentSchema=INDIC;}
+ct.indic-data-source-settings.username=${USERNAME_INDIC:ros9appl}
+ct.indic-data-source-settings.password=${PASSWORD_INDIC:R0s9Appl}
 ct.indic-data-source-settings.maximum-pool-size=${MAXIMUM_POOL_SIZE_INDIC:10}
 ct.indic-data-source-settings.minimum-idle=${MINIMUM_IDLE_INDIC:10}
 ct.indic-data-source-settings.connection-timeout=${CONNECTION_TIMEOUT_INDIC:10000}
@@ -60,8 +60,8 @@ ct.signals-root-uri=${SIGNALS_ROOT_URI:http://10.230.227.102:11094/signalRUalfa/
 ct.signals-token=${SIGNALS_TOKEN:gttcHxwcGojoAvZK}
 ct.signals-user=${SIGNALS_USER:0001ZZ000001}
 
-ct.load-dir=/opt/app/reports
-#ct.load-dir=D:\\folder_name\\signals
+#ct.load-dir=/opt/app/reports
+ct.load-dir=D:\\folder_name\\signals
 
 # min
 ct.monitoring-interval=${MONITORING_INTERVAL:15}

src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilderV2.java

@@ -1,7 +1,5 @@
 package ru.nbch.credit_tracker.builder;
 
-import java.time.LocalDate;
-import java.time.format.DateTimeFormatter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import ru.nbch.credit_tracker.entities.dto.IdProfile;
@@ -10,6 +8,9 @@ import ru.nbch.credit_tracker.entities.dto.SubjectProfile;
 import ru.nbch.credit_tracker.entities.external_request.signals.Person;
 import ru.nbch.credit_tracker.utils.CTUtil;
 
+import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
+
 @Slf4j
 public class PersonFactoryBuilderV2 {
     private final static DateTimeFormatter DATE_FORMAT = DateTimeFormatter.ofPattern("dd.MM.yyyy");
@@ -39,11 +40,7 @@ public class PersonFactoryBuilderV2 {
 
     public PersonFactoryBuilderV2 personData(SubjectProfile personData) {
         this.personData = personData;
-        return this;
-    }
-
-    public PersonFactoryBuilderV2 uid(Integer uid) {
-        this.uid = uid;
+        this.uid = personData.getId();
         return this;
     }
 

src/main/java/ru/nbch/credit_tracker/config/task/ConsumerConfigV2.java

@@ -1,16 +1,12 @@
 package ru.nbch.credit_tracker.config.task;
 
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import ru.nbch.credit_tracker.model.*;
+import ru.nbch.credit_tracker.service.task.*;
+
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import ru.nbch.credit_tracker.model.PhoneFidPayloadV2;
-import ru.nbch.credit_tracker.model.SendSignalsPayload;
-import ru.nbch.credit_tracker.model.SubjectsProfilePayloadV2;
-import ru.nbch.credit_tracker.service.task.FidIdEnricherV2;
-import ru.nbch.credit_tracker.service.task.FidNameEnricherV2;
-import ru.nbch.credit_tracker.service.task.TaskConsumerV2;
-import ru.nbch.credit_tracker.service.task.TaskPayloadV2;
 
 @Configuration
 public class ConsumerConfigV2 {
@@ -19,6 +15,12 @@ public class ConsumerConfigV2 {
     public BlockingQueue<PhoneFidPayloadV2> savingRequest2() {
         return new LinkedBlockingQueue<>();
     }
+
+    @Bean
+    public BlockingQueue<FlagsPayload> flagsPayload() {
+        return new LinkedBlockingQueue<>();
+    }
+
     @Bean
     public BlockingQueue<SubjectsProfilePayloadV2> enricherQueue2() {
         return new LinkedBlockingQueue<>();
@@ -39,17 +41,23 @@ public class ConsumerConfigV2 {
     public Integer consumerConfigurer(BlockingQueue<PhoneFidPayloadV2> savingRequest2,
                                       BlockingQueue<SubjectsProfilePayloadV2> enricherQueue2,
                                       BlockingQueue<SendSignalsPayload> enricherIdQueue2,
+                                      BlockingQueue<FlagsPayload> flagsPayloadQueue,
                                       FidNameEnricherV2 fidNameEnricherV2,
-                                     FidIdEnricherV2 fidIdEnricherV2) {
+                                      FidIdEnricherV2 fidIdEnricherV2,
+                                      FlagsEnricher flagsEnricher) {
 
         SubjectsProfilePayloadV2 POISON = new SubjectsProfilePayloadV2(true);
         SendSignalsPayload FID_ID_ENRICHER_POISON = new SendSignalsPayload(true);
+        DummyPayload FLAGS_POISON = new DummyPayload(true);
         TaskConsumerV2 fidNameEnricher = new TaskConsumerV2(fidNameEnricherV2, savingRequest2, enricherQueue2, POISON);
         TaskConsumerV2 fidIdEnricher = new TaskConsumerV2(fidIdEnricherV2, enricherQueue2, enricherIdQueue2, FID_ID_ENRICHER_POISON);
+        TaskConsumerV2 flagsEnricherConsumer = new TaskConsumerV2(flagsEnricher, flagsPayloadQueue, new LinkedBlockingQueue(), FLAGS_POISON);
         Thread thread1 = new Thread(fidNameEnricher::start);
         Thread thread2 = new Thread(fidIdEnricher::start);
+        Thread thread3 = new Thread(flagsEnricherConsumer::start);
         thread1.start();
         thread2.start();
+        thread3.start();
         return 0;
     }
 }

src/main/java/ru/nbch/credit_tracker/entities/dto/SubjectProfile.java

@@ -6,6 +6,7 @@ import lombok.Setter;
 @Getter
 @Setter
 public class SubjectProfile {
+    private Integer id;
     private Integer fid;
     private IdProfile idProfile;
     private NameProfile nameProfile;

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

@@ -1,17 +1,19 @@
 package ru.nbch.credit_tracker.mapper.signal;
 
-import java.util.Collection;
-import java.util.List;
 import org.apache.ibatis.annotations.Mapper;
 import org.apache.ibatis.annotations.Param;
 import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
 import ru.nbch.credit_tracker.entities.user_request.monitoring_request.Subject;
 
+import java.util.Collection;
+import java.util.List;
+
 @Mapper
 public interface PhoneFidMapMapper {
 
     void registerSubjects(@Param("subjects") List<Subject> subjects, @Param("packageId") Integer packageId);
     void register(@Param("phoneFidMap") List<PhoneFidMap> phoneFidMap, @Param("packageId") Integer packageId);
+    void registerWithId(@Param("phoneFidMap") List<PhoneFidMap> phoneFidMap);
 
     List<PhoneFidMap> findFullPhoneFidMap(@Param("packageId") Integer packageId, @Param("offset") int offset, @Param("limit") int limit);
 

src/main/java/ru/nbch/credit_tracker/model/DummyPayload.java

@@ -0,0 +1,17 @@
+package ru.nbch.credit_tracker.model;
+
+import ru.nbch.credit_tracker.service.task.TaskPayloadV2;
+
+public class DummyPayload implements TaskPayloadV2 {
+
+    private final boolean poison;
+
+    public DummyPayload(boolean poison) {
+        this.poison = poison;
+    }
+
+    @Override
+    public boolean isPoison() {
+        return poison;
+    }
+}

src/main/java/ru/nbch/credit_tracker/model/FlagsPayload.java

@@ -0,0 +1,38 @@
+package ru.nbch.credit_tracker.model;
+
+import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
+import ru.nbch.credit_tracker.service.task.TaskPayloadV2;
+
+import java.util.Map;
+
+public class FlagsPayload implements TaskPayloadV2 {
+    private final Integer packageId;
+    private final Map<Integer, PhoneFidMap> fidsMapByBatch;
+    private final boolean poison;
+
+    public FlagsPayload(Integer packageId, Map<Integer, PhoneFidMap> fidsMapByBatch) {
+        this.packageId = packageId;
+        this.fidsMapByBatch = fidsMapByBatch;
+        this.poison = false;
+    }
+
+    public FlagsPayload(boolean poison) {
+        this.packageId = null;
+        this.fidsMapByBatch = null;
+        this.poison = poison;
+    }
+
+    public Integer getPackageId() {
+        return packageId;
+    }
+
+    public Map<Integer, PhoneFidMap> getFidsMapByBatch() {
+        return fidsMapByBatch;
+    }
+
+    @Override
+    public boolean isPoison() {
+        return poison;
+    }
+
+}

src/main/java/ru/nbch/credit_tracker/service/indic/FlagsService2.java

@@ -5,12 +5,14 @@ import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
 import org.springframework.stereotype.Service;
+import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
 
 import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.time.LocalDate;
+import java.time.LocalDateTime;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
@@ -40,6 +42,18 @@ public class FlagsService2 {
         this.jdbcTemplate = jdbcTemplate;
     }
 
+    public void fillFlagsData(Map<Integer, PhoneFidMap> fidsMapByBatch) {
+        List<Integer> fids = fidsMapByBatch.keySet().stream().toList();
+        Map<Integer, FlagsService.FlagsData> fidsflagsDataMap = findFlags(fids);
+
+        fidsflagsDataMap.forEach((fid, data) -> {
+            PhoneFidMap phoneFidMap = fidsMapByBatch.get(fid);
+            phoneFidMap.setFlagDebt5000(data.getFlagDebt5000());
+            phoneFidMap.setFlag90days(data.getFlag90days());
+            phoneFidMap.setCalculatedAt(LocalDateTime.now());
+        });
+    }
+
     public Map<Integer, FlagsService.FlagsData> findFlags(List<Integer> fids) {
         if (fids.isEmpty()) {
             return new HashMap<>();

src/main/java/ru/nbch/credit_tracker/service/task/FidNameEnricherV2.java

@@ -1,45 +1,48 @@
 package ru.nbch.credit_tracker.service.task;
 
-import java.util.List;
-import java.util.Map;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.entities.dto.NameProfile;
 import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
 import ru.nbch.credit_tracker.entities.dto.SubjectProfile;
 import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
+import ru.nbch.credit_tracker.model.FlagsPayload;
 import ru.nbch.credit_tracker.model.PhoneFidPayloadV2;
 import ru.nbch.credit_tracker.model.SubjectsProfilePayloadV2;
 import ru.nbch.credit_tracker.service.indic.PhoneService2;
-import ru.nbch.credit_tracker.service.signal.PhoneFidMapService;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
 
 @Service
 @Slf4j
 public class FidNameEnricherV2 implements DispatcherV2<PhoneFidPayloadV2, SubjectsProfilePayloadV2> {
     private final PhoneService2 phoneService2;
-    private final PhoneFidMapService phoneFidMapService;
     private final PhoneFidMapMapper mapper;
+    private final BlockingQueue<FlagsPayload> flagsPayloadQueue;
 
     public FidNameEnricherV2(PhoneService2 phoneService2,
-                             PhoneFidMapService phoneFidMapService,
-                             PhoneFidMapMapper mapper) {
+                             PhoneFidMapMapper mapper,
+                             BlockingQueue<FlagsPayload> flagsPayloadQueue) {
         this.phoneService2 = phoneService2;
-        this.phoneFidMapService = phoneFidMapService;
         this.mapper = mapper;
+        this.flagsPayloadQueue = flagsPayloadQueue;
     }
 
     @Override
     public SubjectsProfilePayloadV2 handle(PhoneFidPayloadV2 taskPayload) {
         List<PhoneFidMap> phoneFidMap = taskPayload.getPhoneFidsBatch();
         log.debug("FidNameEnricher handle by packageId: {}, size: {}", taskPayload.getPackageId(), phoneFidMap.size());
-//        phoneFidMapService.register(phoneFidMap, taskPayload.getPackageId());
+        mapper.registerWithId(phoneFidMap); // надо зарегистрировать phoneFidMap'ы т.к далее id участвует в формированиии запроса в оф. сервис uid = id
         Map<Integer, PhoneFidMap> fidsMapByBatch = phoneService2.defineFidsAndNameByPhone(phoneFidMap);
         log.debug("Found {} fids by batch", fidsMapByBatch.size());
+        flagsPayloadQueue.add(new FlagsPayload(taskPayload.getPackageId(), fidsMapByBatch)); // TODO find out how to use it in other way
         //todo по phoneFidMap может быть найден не полный набор фидов, поэтому стоит это контролировать
-//        mapper.updateFid(fidsMapByBatch.values());
 
         List<SubjectProfile> subjectProfiles = fidsMapByBatch.values().stream().map(pf -> {
             SubjectProfile subjectProfile = new SubjectProfile();
+            subjectProfile.setId(pf.getId());
             subjectProfile.setFid(pf.getFid());
             subjectProfile.setNameProfile(NameProfile.builder()
                 .firstName(pf.getFirstName())

src/main/java/ru/nbch/credit_tracker/service/task/FlagsEnricher.java

@@ -0,0 +1,48 @@
+package ru.nbch.credit_tracker.service.task;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
+import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
+import ru.nbch.credit_tracker.model.DummyPayload;
+import ru.nbch.credit_tracker.model.FlagsPayload;
+import ru.nbch.credit_tracker.service.indic.FlagsService2;
+
+import java.util.List;
+import java.util.Map;
+
+@Service
+@Slf4j
+public class FlagsEnricher implements DispatcherV2<FlagsPayload, DummyPayload>{
+
+    private final FlagsService2 flagsService;
+    private final PhoneFidMapMapper mapper;
+
+    public FlagsEnricher(FlagsService2 flagsService,
+                         PhoneFidMapMapper mapper) {
+        this.flagsService = flagsService;
+        this.mapper = mapper;
+    }
+
+    @Override
+    public DummyPayload handle(FlagsPayload taskPayload) {
+        Map<Integer, PhoneFidMap> fidsMapByBatch = taskPayload.getFidsMapByBatch();
+        log.debug("FlagsEnricher handle by packageId: {}, size: {}", taskPayload.getPackageId(), fidsMapByBatch.size());
+        flagsService.fillFlagsData(fidsMapByBatch);
+
+        List<PhoneFidMap> subjects = fidsMapByBatch.values().stream().toList();
+
+        log.debug("Start updating subject's batch");
+        if (!subjects.isEmpty()) {
+            int updateBatchSize = 7000;
+            for (int j = 0; j < subjects.size(); j += updateBatchSize) {
+                int endIndex = Math.min(j + updateBatchSize, subjects.size());
+                mapper.updatePhoneFidMaps(subjects.subList(j, endIndex));
+                log.trace("Updated fids and flags for subject's batch from: {} to: {}. Total count: {}. PackageId: {}", j, endIndex, subjects.size(), taskPayload.getPackageId());
+            }
+        }
+        log.debug("Complete updating subject's batch");
+
+        return new DummyPayload(false);
+    }
+}

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

@@ -26,10 +26,18 @@
     </insert>
 
     <insert id="register" parameterType="java.util.List">
-        INSERT INTO signals_phone_fid_map (rec_uuid, subject_id, phone, package_id)
+        INSERT INTO signals_phone_fid_map (subject_id, phone, package_id)
         values
         <foreach collection="phoneFidMap" item="phoneFid" separator=",">
-            (#{phoneFid.recUuid}, #{phoneFid.subjectId}, #{phoneFid.phone}, #{packageId})
+            (#{phoneFid.subjectId}, #{phoneFid.phone}, #{packageId})
+        </foreach>
+    </insert>
+
+    <insert id="registerWithId" parameterType="java.util.List" useGeneratedKeys="true" keyColumn="id" keyProperty="id">
+        INSERT INTO signals_phone_fid_map (subject_id, phone, package_id)
+        values
+        <foreach collection="phoneFidMap" item="phoneFid" separator=",">
+                (#{phoneFid.subjectId}, #{phoneFid.phone}, #{phoneFid.packageId})
         </foreach>
     </insert>