diff --git a/pom.xml b/pom.xml index c57d7bc..1e84518 100644 --- a/pom.xml +++ b/pom.xml @@ -10,12 +10,12 @@ ru.nbch credit-tracker - 1.15 + 2.0 credit_tracker Credit Tracker Application - 17 + 21 git_info git.properties rh8-vm3.nbki.msk:5000 diff --git a/src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilderV2.java b/src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilderV2.java index 0e49669..393d03a 100644 --- a/src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilderV2.java +++ b/src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilderV2.java @@ -1,8 +1,7 @@ package ru.nbch.credit_tracker.builder; -import java.text.DateFormat; -import java.text.SimpleDateFormat; 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; @@ -13,7 +12,7 @@ import ru.nbch.credit_tracker.utils.CTUtil; @Slf4j public class PersonFactoryBuilderV2 { - private final static DateFormat DATE_FORMAT = new SimpleDateFormat("dd.MM.yyyy"); + private final static DateTimeFormatter DATE_FORMAT = DateTimeFormatter.ofPattern("dd.MM.yyyy"); private final static String DEFAULT_DATE = "01.01.2000"; private final static String IP = "0"; @@ -58,10 +57,10 @@ public class PersonFactoryBuilderV2 { person.setS(nameData.getLastName()); person.setN(nameData.getFirstName()); person.setM(StringUtils.isNotBlank(nameData.getMiddleName()) ? nameData.getMiddleName() : null); - person.setD(DATE_FORMAT.format(CTUtil.localDatetoDate(nameData.getBirthDate()))); + person.setD(DATE_FORMAT.format(nameData.getBirthDate())); person.setPn(getPn(passportData.getSerNum(), passportData.getIdNum())); - person.setPd(DATE_FORMAT.format(CTUtil.localDatetoDate(LocalDate.now()))); // Do not use passport iss_date. Signal service could reject the package + person.setPd(DATE_FORMAT.format(LocalDate.now())); // Do not use passport iss_date. Signal service could reject the package person.setDt(getDt(passportData.getIdType(), passportData.getIdType2())); if (person.getDt() != null && person.getDt().equals("999")) { person.setOdt(ODT); @@ -70,12 +69,12 @@ public class PersonFactoryBuilderV2 { person.setIp(IP); person.setUid(uid); person.setOksm(OKSM); - person.setConsentDate(DATE_FORMAT.format(CTUtil.localDatetoDate(LocalDate.now()))); + person.setConsentDate(DATE_FORMAT.format(LocalDate.now())); person.setConsentPurpose(CONSENT_PURPOSE); person.setReportUser(REPORT_USER); person.setLiability(LIABILITY); person.setConsentPeriod(CONSENT_PERIOD); - person.setAgrDate(DATE_FORMAT.format(CTUtil.localDatetoDate(LocalDate.now()))); + person.setAgrDate(DATE_FORMAT.format(LocalDate.now())); person.setReportUserRegNum(REPORT_USER_REG_NUM); person.setReportUserTaxID(REPORT_USER_TAX_ID); person.setConsentHash(CONSENT_HASH); diff --git a/src/main/java/ru/nbch/credit_tracker/config/datasource/DatabaseConfig.java b/src/main/java/ru/nbch/credit_tracker/config/datasource/DatabaseConfig.java index dafc1a7..5a57985 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/datasource/DatabaseConfig.java +++ b/src/main/java/ru/nbch/credit_tracker/config/datasource/DatabaseConfig.java @@ -1,10 +1,9 @@ package ru.nbch.credit_tracker.config.datasource; +import javax.sql.DataSource; import org.springframework.boot.jdbc.DataSourceBuilder; import ru.nbch.credit_tracker.config.settings.DataSourceSettings; -import javax.sql.DataSource; - public class DatabaseConfig { protected DataSource createDataSource(DataSourceSettings settings) { diff --git a/src/main/java/ru/nbch/credit_tracker/config/datasource/IndicDataSourceConfig.java b/src/main/java/ru/nbch/credit_tracker/config/datasource/IndicDataSourceConfig.java index b802514..503a97b 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/datasource/IndicDataSourceConfig.java +++ b/src/main/java/ru/nbch/credit_tracker/config/datasource/IndicDataSourceConfig.java @@ -1,5 +1,6 @@ package ru.nbch.credit_tracker.config.datasource; +import com.zaxxer.hikari.HikariDataSource; import lombok.extern.slf4j.Slf4j; import org.apache.ibatis.session.SqlSessionFactory; import org.mybatis.spring.SqlSessionFactoryBean; @@ -29,7 +30,8 @@ public class IndicDataSourceConfig extends DatabaseConfig { @Bean(name = "indicDataSource") public DataSource dataSource(CreditTrackerSettings config) throws SQLException { - DataSource dataSource = createDataSource(config.getIndicDataSourceSettings()); + HikariDataSource dataSource = (HikariDataSource) createDataSource(config.getIndicDataSourceSettings()); + dataSource.setMaximumPoolSize(50); try (Connection connection = dataSource.getConnection()) { connection.isValid(3); } catch (SQLException e) { diff --git a/src/main/java/ru/nbch/credit_tracker/config/datasource/ScoringDataSourceConfig.java b/src/main/java/ru/nbch/credit_tracker/config/datasource/ScoringDataSourceConfig.java index 8d9bbde..788d3e9 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/datasource/ScoringDataSourceConfig.java +++ b/src/main/java/ru/nbch/credit_tracker/config/datasource/ScoringDataSourceConfig.java @@ -1,5 +1,9 @@ package ru.nbch.credit_tracker.config.datasource; +import com.zaxxer.hikari.HikariDataSource; +import java.sql.Connection; +import java.sql.SQLException; +import javax.sql.DataSource; import lombok.extern.slf4j.Slf4j; import org.apache.ibatis.session.SqlSessionFactory; import org.mybatis.spring.SqlSessionFactoryBean; @@ -11,10 +15,6 @@ import org.springframework.context.annotation.Configuration; import org.springframework.core.io.Resource; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; -import javax.sql.DataSource; -import java.sql.Connection; -import java.sql.SQLException; - @Slf4j @Configuration @MapperScan( @@ -28,7 +28,8 @@ public class ScoringDataSourceConfig extends DatabaseConfig { @Bean(name = "scoringDataSource") public DataSource dataSource(CreditTrackerSettings config) throws SQLException { - DataSource dataSource = createDataSource(config.getScoringDataSourceSettings()); + HikariDataSource dataSource = (HikariDataSource) createDataSource(config.getScoringDataSourceSettings()); + dataSource.setMaximumPoolSize(10); try (Connection connection = dataSource.getConnection()) { connection.isValid(3); } catch (SQLException e) { diff --git a/src/main/java/ru/nbch/credit_tracker/config/datasource/SignalDataSourceConfig.java b/src/main/java/ru/nbch/credit_tracker/config/datasource/SignalDataSourceConfig.java index c2e94c3..4436560 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/datasource/SignalDataSourceConfig.java +++ b/src/main/java/ru/nbch/credit_tracker/config/datasource/SignalDataSourceConfig.java @@ -1,7 +1,13 @@ package ru.nbch.credit_tracker.config.datasource; +import com.zaxxer.hikari.HikariDataSource; +import java.sql.Connection; +import java.sql.SQLException; +import java.util.UUID; +import javax.sql.DataSource; import lombok.extern.slf4j.Slf4j; import org.apache.ibatis.session.SqlSessionFactory; +import org.apache.ibatis.type.TypeHandlerRegistry; import org.mybatis.spring.SqlSessionFactoryBean; import org.mybatis.spring.annotation.MapperScan; import org.springframework.beans.factory.annotation.Qualifier; @@ -9,12 +15,9 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.io.Resource; +import ru.nbch.credit_tracker.config.datasource.handlers.UUIDTypeHandler; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; -import javax.sql.DataSource; -import java.sql.Connection; -import java.sql.SQLException; - @Slf4j @Configuration @MapperScan( @@ -27,7 +30,8 @@ public class SignalDataSourceConfig extends DatabaseConfig { @Bean(name = "signalDataSource") public DataSource dataSource(CreditTrackerSettings config) throws SQLException { - DataSource dataSource = createDataSource(config.getSignalDataSourceSettings()); + HikariDataSource dataSource = (HikariDataSource) createDataSource(config.getSignalDataSourceSettings()); + dataSource.setMaximumPoolSize(100); try (Connection connection = dataSource.getConnection()) { connection.isValid(3); } catch (SQLException e) { @@ -43,6 +47,11 @@ public class SignalDataSourceConfig extends DatabaseConfig { SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean(); sqlSessionFactoryBean.setDataSource(dataSource); sqlSessionFactoryBean.setMapperLocations(resources); + + SqlSessionFactory sqlSessionFactory = sqlSessionFactoryBean.getObject(); + TypeHandlerRegistry typeHandlerRegistry = sqlSessionFactory.getConfiguration().getTypeHandlerRegistry(); + typeHandlerRegistry.register(UUID.class, UUIDTypeHandler.class); + return sqlSessionFactoryBean.getObject(); } } diff --git a/src/main/java/ru/nbch/credit_tracker/config/datasource/handlers/UUIDTypeHandler.java b/src/main/java/ru/nbch/credit_tracker/config/datasource/handlers/UUIDTypeHandler.java new file mode 100644 index 0000000..77f33fc --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/config/datasource/handlers/UUIDTypeHandler.java @@ -0,0 +1,53 @@ +package ru.nbch.credit_tracker.config.datasource.handlers; + +import java.sql.CallableStatement; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Types; +import java.util.UUID; +import org.apache.ibatis.type.JdbcType; +import org.apache.ibatis.type.MappedTypes; +import org.apache.ibatis.type.TypeHandler; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +@MappedTypes({UUID.class}) +public class UUIDTypeHandler implements TypeHandler { + private static final Logger log = LoggerFactory.getLogger(UUIDTypeHandler.class); + + @Override + public void setParameter(PreparedStatement ps, int i, UUID parameter, JdbcType jdbcType) throws SQLException { + if (parameter == null) { + ps.setObject(i, null, Types.OTHER); + } else { + ps.setObject(i, parameter.toString(), Types.OTHER); + } + } + + @Override + public UUID getResult(ResultSet rs, String columnName) throws SQLException { + return toUUID(rs.getString(columnName)); + } + + @Override + public UUID getResult(ResultSet rs, int columnIndex) throws SQLException { + return toUUID(rs.getString(columnIndex)); + } + + @Override + public UUID getResult(CallableStatement cs, int columnIndex) throws SQLException { + return toUUID(cs.getString(columnIndex)); + } + + private static UUID toUUID(String val) { + if (val != null && !val.isEmpty() && !val.isBlank()) { + try { + return UUID.fromString(val); + } catch (IllegalArgumentException e) { + log.warn("Bad UUID found: {}", val); + } + } + return null; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/config/task/ConsumerConfigV2.java b/src/main/java/ru/nbch/credit_tracker/config/task/ConsumerConfigV2.java new file mode 100644 index 0000000..2b8a357 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/config/task/ConsumerConfigV2.java @@ -0,0 +1,55 @@ +package ru.nbch.credit_tracker.config.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 { + + @Bean + public BlockingQueue savingRequest2() { + return new LinkedBlockingQueue<>(); + } + @Bean + public BlockingQueue enricherQueue2() { + return new LinkedBlockingQueue<>(); + } + + @Bean + public BlockingQueue enricherIdQueue2() { + return new LinkedBlockingQueue<>(); + } + + @Bean + public TaskPayloadV2 POISON() { + return () -> true; + } + + @SuppressWarnings("unchecked") + @Bean + public Integer consumerConfigurer(BlockingQueue savingRequest2, + BlockingQueue enricherQueue2, + BlockingQueue enricherIdQueue2, + FidNameEnricherV2 fidNameEnricherV2, + FidIdEnricherV2 fidIdEnricherV2) { + + SubjectsProfilePayloadV2 POISON = new SubjectsProfilePayloadV2(true); + SendSignalsPayload FID_ID_ENRICHER_POISON = new SendSignalsPayload(true); + TaskConsumerV2 fidNameEnricher = new TaskConsumerV2(fidNameEnricherV2, savingRequest2, enricherQueue2, POISON); + TaskConsumerV2 fidIdEnricher = new TaskConsumerV2(fidIdEnricherV2, enricherQueue2, enricherIdQueue2, FID_ID_ENRICHER_POISON); + Thread thread1 = new Thread(fidNameEnricher::start); + Thread thread2 = new Thread(fidIdEnricher::start); + thread1.start(); + thread2.start(); + return 0; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/entities/dto/PhoneFidMap.java b/src/main/java/ru/nbch/credit_tracker/entities/dto/PhoneFidMap.java index f2cd37c..456b6f6 100644 --- a/src/main/java/ru/nbch/credit_tracker/entities/dto/PhoneFidMap.java +++ b/src/main/java/ru/nbch/credit_tracker/entities/dto/PhoneFidMap.java @@ -1,5 +1,6 @@ package ru.nbch.credit_tracker.entities.dto; +import java.util.UUID; import lombok.*; import java.time.LocalDate; @@ -12,6 +13,7 @@ import java.time.LocalDateTime; @Builder public class PhoneFidMap { private Integer id; + private UUID recUuid; private Integer packageId; private String subjectId; private String phone; diff --git a/src/main/java/ru/nbch/credit_tracker/model/PackageToSavePayloadV2.java b/src/main/java/ru/nbch/credit_tracker/model/PackageToSavePayloadV2.java new file mode 100644 index 0000000..2ec322e --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/model/PackageToSavePayloadV2.java @@ -0,0 +1,33 @@ +package ru.nbch.credit_tracker.model; + +import ru.nbch.credit_tracker.service.task.TaskPayloadV2; + +public class PackageToSavePayloadV2 implements TaskPayloadV2 { + private final Integer packageId; + private final String filePath; + private final boolean poison; + + public PackageToSavePayloadV2(boolean poison) { + this.packageId = null; + this.filePath = null; + this.poison = true; + } + public PackageToSavePayloadV2(Integer packageId, String filePath) { + this.packageId = packageId; + this.filePath = filePath; + this.poison = false; + } + + public Integer getPackageId() { + return packageId; + } + + public String getFilePath() { + return filePath; + } + + @Override + public boolean isPoison() { + return poison; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/model/PhoneFidPayloadV2.java b/src/main/java/ru/nbch/credit_tracker/model/PhoneFidPayloadV2.java new file mode 100644 index 0000000..3a79ca7 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/model/PhoneFidPayloadV2.java @@ -0,0 +1,36 @@ +package ru.nbch.credit_tracker.model; + +import java.util.List; +import ru.nbch.credit_tracker.entities.dto.PhoneFidMap; +import ru.nbch.credit_tracker.service.task.TaskPayloadV2; + +public class PhoneFidPayloadV2 implements TaskPayloadV2 { + private final Integer packageId; + private final List phoneFidsBatch; + private final boolean poison; + + public PhoneFidPayloadV2(boolean poison) { + this.packageId = null; + this.phoneFidsBatch = null; + this.poison = poison; + } + + public PhoneFidPayloadV2(Integer packageId, List phoneFidsBatch) { + this.packageId = packageId; + this.phoneFidsBatch = phoneFidsBatch; + this.poison = false; + } + + public Integer getPackageId() { + return packageId; + } + + public List getPhoneFidsBatch() { + return phoneFidsBatch; + } + + @Override + public boolean isPoison() { + return poison; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/model/SendSignalsPayload.java b/src/main/java/ru/nbch/credit_tracker/model/SendSignalsPayload.java new file mode 100644 index 0000000..3f72c73 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/model/SendSignalsPayload.java @@ -0,0 +1,17 @@ +package ru.nbch.credit_tracker.model; + +import ru.nbch.credit_tracker.service.task.TaskPayloadV2; + +public class SendSignalsPayload implements TaskPayloadV2 { + private String signalXmlPath; + private boolean poison; + + public SendSignalsPayload(boolean poison) { + this.poison = poison; + } + + @Override + public boolean isPoison() { + return poison; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/model/SubjectsProfilePayloadV2.java b/src/main/java/ru/nbch/credit_tracker/model/SubjectsProfilePayloadV2.java new file mode 100644 index 0000000..73338e8 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/model/SubjectsProfilePayloadV2.java @@ -0,0 +1,36 @@ +package ru.nbch.credit_tracker.model; + +import java.util.List; +import ru.nbch.credit_tracker.entities.dto.SubjectProfile; +import ru.nbch.credit_tracker.service.task.TaskPayloadV2; + +public class SubjectsProfilePayloadV2 implements TaskPayloadV2 { + private final Integer packageId; + private final List subjectProfiles; + private final boolean poison; + + public SubjectsProfilePayloadV2(boolean poison) { + this.packageId = null; + this.subjectProfiles = null; + this.poison = poison; + } + + public SubjectsProfilePayloadV2(Integer packageId, List subjectProfiles) { + this.packageId = packageId; + this.subjectProfiles = subjectProfiles; + this.poison = false; + } + + public Integer getPackageId() { + return packageId; + } + + public List getSubjectProfiles() { + return subjectProfiles; + } + + @Override + public boolean isPoison() { + return poison; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java b/src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java index b3615ee..c63c754 100644 --- a/src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java +++ b/src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java @@ -1,5 +1,9 @@ package ru.nbch.credit_tracker.repository; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.time.LocalDate; +import java.util.List; import lombok.AllArgsConstructor; import lombok.Getter; import lombok.NoArgsConstructor; @@ -9,48 +13,45 @@ import org.springframework.jdbc.core.namedparam.MapSqlParameterSource; import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate; import org.springframework.stereotype.Repository; -import java.time.LocalDate; -import java.util.List; - @Repository public class PhoneRepository { private static final String SELCT_FIDS = """ - SELECT - pn.number, - pn.fid, - pn.file_since_dt - FROM PHONE_NORM pn - JOIN (SELECT DISTINCT fid FROM NAME) nf - ON nf.fid = pn.fid - WHERE number in (:phones) - ORDER BY - pn.number ASC, - pn.file_since_dt DESC, - pn.fid DESC - WITH UR - """; + SELECT + pn.number, + pn.fid, + pn.file_since_dt + FROM PHONE_NORM pn + JOIN (SELECT DISTINCT fid FROM NAME) nf + ON nf.fid = pn.fid + WHERE number in (:phones) + ORDER BY + pn.number ASC, + pn.file_since_dt DESC, + pn.fid DESC + WITH UR + """; private static final String SELCT_FIDS_WITH_NAME = """ - SELECT - pn.number, - pn.fid, - pn.file_since_dt, - nf.first, - nf.middle, - nf.name_1, - nf.birth_dt, - nf.file_since_dt as file_since_dt_name - FROM PHONE_NORM pn - JOIN NAME nf - ON nf.fid = pn.fid - WHERE number in (:phones) - ORDER BY - pn.number ASC, - pn.file_since_dt DESC, - pn.fid DESC - WITH UR - """; + SELECT + pn.number, + pn.fid, + pn.file_since_dt, + nf.first, + nf.middle, + nf.name_1, + nf.birth_dt, + nf.file_since_dt as file_since_dt_name + FROM PHONE_NORM pn + JOIN NAME nf + ON nf.fid = pn.fid + WHERE number in (:phones) + ORDER BY + pn.number ASC, + pn.file_since_dt DESC, + pn.fid DESC + WITH UR + """; private final NamedParameterJdbcTemplate jdbcTemplate; @@ -64,7 +65,7 @@ public class PhoneRepository { params.addValue("phones", phones); return jdbcTemplate.query(SELCT_FIDS, params, (rs, rowNum) -> - new PhoneRepository.FidsData(rs.getInt("fid"), rs.getString("number"), rs.getObject("file_since_dt", LocalDate.class)) + new PhoneRepository.FidsData(rs.getInt("fid"), rs.getString("number"), rs.getObject("file_since_dt", LocalDate.class)) ); } @@ -73,16 +74,21 @@ public class PhoneRepository { params.addValue("phones", phones); return jdbcTemplate.query(SELCT_FIDS_WITH_NAME, params, (rs, rowNum) -> - new PhoneRepository.FidsDataWithName( - rs.getInt("FID"), - rs.getString("NUMBER"), - rs.getObject("FILE_SINCE_DT", LocalDate.class), - rs.getString("FIRST"), - rs.getString("NAME_1"), - rs.getString("MIDDLE"), - rs.getObject("BIRTH_DT", LocalDate.class), - rs.getObject("FILE_SINCE_DT_NAME", LocalDate.class) - )); + new PhoneRepository.FidsDataWithName( + rs.getInt("FID"), + rs.getString("NUMBER"), + getLocalDate(rs, "FILE_SINCE_DT"), + rs.getString("FIRST"), + rs.getString("NAME_1"), + rs.getString("MIDDLE"), + getLocalDate(rs, "BIRTH_DT"), + getLocalDate(rs, "FILE_SINCE_DT_NAME") + )); + } + + private static LocalDate getLocalDate(ResultSet rs, String col) throws SQLException { + java.sql.Date d = rs.getDate(col); + return d != null ? d.toLocalDate() : null; } public List findFidsLong(List phones) { @@ -90,7 +96,7 @@ public class PhoneRepository { params.addValue("phones", phones); return jdbcTemplate.query(SELCT_FIDS, params, (rs, rowNum) -> - new PhoneRepository.FidsDataLong(rs.getInt("fid"), Long.parseLong(rs.getString("number")), rs.getObject("file_since_dt", LocalDate.class)) + new PhoneRepository.FidsDataLong(rs.getInt("fid"), Long.parseLong(rs.getString("number")), rs.getObject("file_since_dt", LocalDate.class)) ); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/DispatcherV2.java b/src/main/java/ru/nbch/credit_tracker/service/task/DispatcherV2.java new file mode 100644 index 0000000..063c515 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/DispatcherV2.java @@ -0,0 +1,5 @@ +package ru.nbch.credit_tracker.service.task; + +public interface DispatcherV2 { + Z handle(X taskPayload); +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/FidIdEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/FidIdEnricher.java index d2f81e8..1df5eaf 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/FidIdEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/FidIdEnricher.java @@ -42,13 +42,14 @@ public class FidIdEnricher implements Dispatcher{ @Override public void handle(SubjectsProfilePayload taskPayload) { - log.debug("FidIdEnricher handle by packageId: {}", taskPayload.getPackageId()); List subjectsProfile = taskPayload.getSubjectProfiles(); + log.debug("FidIdEnricher handle by packageId: {}; size: {}", taskPayload.getPackageId(), subjectsProfile.size()); personService.filledPassportsData(subjectsProfile); List personOnProcessing = subjectsProfile.stream() .filter(SubjectProfile::isFilled) .map(subjectProfile -> PersonFactoryBuilderV2.builder().personData(subjectProfile).build()) .toList(); + log.debug("personOnProcessing size: {}", personOnProcessing.size()); writeSignals(personOnProcessing, taskPayload.getPackageId()); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/FidIdEnricherV2.java b/src/main/java/ru/nbch/credit_tracker/service/task/FidIdEnricherV2.java new file mode 100644 index 0000000..0dce400 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/FidIdEnricherV2.java @@ -0,0 +1,89 @@ +package ru.nbch.credit_tracker.service.task; + +import java.io.IOException; +import java.io.OutputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.nio.file.StandardCopyOption; +import java.nio.file.StandardOpenOption; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.BlockingQueue; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import ru.nbch.credit_tracker.builder.PersonFactoryBuilderV2; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.dto.SubjectProfile; +import ru.nbch.credit_tracker.entities.external_request.signals.Person; +import ru.nbch.credit_tracker.model.PartFilePayload; +import ru.nbch.credit_tracker.model.SendSignalsPayload; +import ru.nbch.credit_tracker.model.SubjectsProfilePayloadV2; +import ru.nbch.credit_tracker.service.SignalService; +import ru.nbch.credit_tracker.service.indic.PersonService; +import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriterV2; + +@Service +@Slf4j +public class FidIdEnricherV2 implements DispatcherV2{ + private final PersonService personService; + private final SignalService signalService; + private final CreditTrackerSettings config; + private final BlockingQueue mergerQueue; + public FidIdEnricherV2(PersonService personService, + SignalService signalService, + CreditTrackerSettings config, + BlockingQueue mergerQueue) { + this.personService = personService; + this.signalService = signalService; + this.config = config; + this.mergerQueue = mergerQueue; + } + + @Override + public SendSignalsPayload handle(SubjectsProfilePayloadV2 taskPayload) { + List subjectsProfile = taskPayload.getSubjectProfiles(); + log.debug("FidIdEnricher handle by packageId: {}; size: {}", taskPayload.getPackageId(), subjectsProfile.size()); + personService.filledPassportsData(subjectsProfile); + List personOnProcessing = subjectsProfile.stream() + .filter(SubjectProfile::isFilled) + .map(subjectProfile -> PersonFactoryBuilderV2.builder().personData(subjectProfile).build()) + .toList(); + log.debug("personOnProcessing size: {}", personOnProcessing.size()); + writeSignals(personOnProcessing, taskPayload.getPackageId()); + return new SendSignalsPayload(false); + } + + private void writeSignals(List persons, Integer packageId){ + Path dir = Paths.get(config.getLoadDir(), "tmp_folder"); + try { + Files.createDirectories(dir); + } catch (IOException e) { + throw new RuntimeException(e); + } + + String base = UUID.randomUUID() + "_output.xml"; + log.debug("Write tmp xml by packageId: {}", packageId); + Path tmp = dir.resolve(base + ".part"); + Path done = dir.resolve(base); + +// Signals signals = signalService.buildSignals(packageId); + + try (OutputStream os = Files.newOutputStream(tmp, StandardOpenOption.CREATE_NEW)) { + SignalsStaxXmlWriterV2 writer = new SignalsStaxXmlWriterV2(persons, null, config.getEncoding()); + writer.writeWithStax(os); // перегрузи на OutputStream, если нужно + } catch (IOException e) { + throw new RuntimeException(e); + } + try { + Files.move(tmp, done, StandardCopyOption.ATOMIC_MOVE); + } catch (IOException e) { + throw new RuntimeException(e); + } + try { + mergerQueue.put(new PartFilePayload(done, packageId, null)); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/FidNameEnricherV2.java b/src/main/java/ru/nbch/credit_tracker/service/task/FidNameEnricherV2.java new file mode 100644 index 0000000..79c3c17 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/FidNameEnricherV2.java @@ -0,0 +1,56 @@ +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.PhoneFidPayloadV2; +import ru.nbch.credit_tracker.model.SubjectsProfilePayloadV2; +import ru.nbch.credit_tracker.service.indic.PhoneService2; +import ru.nbch.credit_tracker.service.signal.PhoneFidMapService; + +@Service +@Slf4j +public class FidNameEnricherV2 implements DispatcherV2 { + private final PhoneService2 phoneService2; + private final PhoneFidMapService phoneFidMapService; + private final PhoneFidMapMapper mapper; + + public FidNameEnricherV2(PhoneService2 phoneService2, + PhoneFidMapService phoneFidMapService, + PhoneFidMapMapper mapper) { + this.phoneService2 = phoneService2; + this.phoneFidMapService = phoneFidMapService; + this.mapper = mapper; + } + + @Override + public SubjectsProfilePayloadV2 handle(PhoneFidPayloadV2 taskPayload) { + List phoneFidMap = taskPayload.getPhoneFidsBatch(); + log.debug("FidNameEnricher handle by packageId: {}, size: {}", taskPayload.getPackageId(), phoneFidMap.size()); +// phoneFidMapService.register(phoneFidMap, taskPayload.getPackageId()); + Map fidsMapByBatch = phoneService2.defineFidsAndNameByPhone(phoneFidMap); + log.debug("Found {} fids by batch", fidsMapByBatch.size()); + //todo по phoneFidMap может быть найден не полный набор фидов, поэтому стоит это контролировать +// mapper.updateFid(fidsMapByBatch.values()); + + List subjectProfiles = fidsMapByBatch.values().stream().map(pf -> { + SubjectProfile subjectProfile = new SubjectProfile(); + subjectProfile.setFid(pf.getFid()); + subjectProfile.setNameProfile(NameProfile.builder() + .firstName(pf.getFirstName()) + .middleName(pf.getMiddleName()) + .lastName(pf.getLastName()) + .birthDate(pf.getBirthDate()) + .fileSinceDate(pf.getFileSinceDate()) + .build()); + return subjectProfile; + }).toList(); + + return new SubjectsProfilePayloadV2(taskPayload.getPackageId(), subjectProfiles); + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PackageSaver.java b/src/main/java/ru/nbch/credit_tracker/service/task/PackageSaver.java index 9a6c7ce..57a04b9 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PackageSaver.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PackageSaver.java @@ -7,6 +7,8 @@ import java.nio.file.Path; import java.nio.file.Paths; import java.util.List; import java.util.Queue; +import java.util.UUID; +import java.util.concurrent.BlockingQueue; import java.util.stream.Collectors; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; @@ -14,6 +16,7 @@ import ru.nbch.credit_tracker.entities.dto.PhoneFidMap; import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest; import ru.nbch.credit_tracker.model.PhoneFidPayload; import ru.nbch.credit_tracker.model.PackageToSavePayload; +import ru.nbch.credit_tracker.model.PhoneFidPayloadV2; import ru.nbch.credit_tracker.service.signal.PhoneFidMapService; import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; @@ -23,14 +26,17 @@ public class PackageSaver implements Dispatcher { private final StaxXmlProcessor xmlProcessor; private final PhoneFidMapService phoneFidMapService; private final Queue fidNameEnricherQueue; + private final BlockingQueue savingRequest2; private final static int BATCH_SIZE = 10_000; public PackageSaver(StaxXmlProcessor xmlProcessor, PhoneFidMapService phoneFidMapService, - Queue fidNameEnricherQueue) { + Queue fidNameEnricherQueue, + BlockingQueue savingRequest2) { this.xmlProcessor = xmlProcessor; this.phoneFidMapService = phoneFidMapService; this.fidNameEnricherQueue = fidNameEnricherQueue; + this.savingRequest2 = savingRequest2; } @Override public void handle(PackageToSavePayload taskPayload) { @@ -43,11 +49,12 @@ public class PackageSaver implements Dispatcher { if (!request.getSubjects().getSubject().isEmpty()) { List phoneFidMap = request.getSubjects().getSubject().stream().map(subject -> PhoneFidMap.builder() .subjectId(subject.getId()) + .recUuid(UUID.randomUUID()) .phone(subject.getClientId()) .packageId(packageId) .build()).collect(Collectors.toList()); log.trace("Reading subject's batch. Total count: {}. PackageId: {}", request.getSubjects().getSubject().size(), packageId); - fidNameEnricherQueue.add(new PhoneFidPayload(packageId, phoneFidMap)); + savingRequest2.add(new PhoneFidPayloadV2(packageId, phoneFidMap)); request.getSubjects().getSubject().clear(); } }, @@ -56,6 +63,8 @@ public class PackageSaver implements Dispatcher { BATCH_SIZE, "/MonitoringRequest/Subjects/Subject"); } catch (IOException e) { throw new RuntimeException("Error while parsing monitoring request", e); + } finally { + savingRequest2.add(new PhoneFidPayloadV2(true)); } log.info("Subjects were registered. PackageId: {}", packageId); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/TaskConsumer.java b/src/main/java/ru/nbch/credit_tracker/service/task/TaskConsumer.java index 7a49112..66fe227 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/TaskConsumer.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/TaskConsumer.java @@ -30,7 +30,13 @@ public class TaskConsumer implements SmartLifecycle { while (running && !Thread.currentThread().isInterrupted()) { try { X job = taskQueue.take(); - worker.submit(() -> dispatcher.handle(job)); + worker.submit(() -> { + try { + dispatcher.handle(job); + } catch (Exception e) { + log.error("Handle Error: {}", ExceptionUtils.getStackTrace(e)); + } + }); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } catch (Exception e) { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/TaskConsumerV2.java b/src/main/java/ru/nbch/credit_tracker/service/task/TaskConsumerV2.java new file mode 100644 index 0000000..6ca9505 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/TaskConsumerV2.java @@ -0,0 +1,131 @@ +package ru.nbch.credit_tracker.service.task; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.exception.ExceptionUtils; + +@Slf4j +public class TaskConsumerV2 +// implements SmartLifecycle +{ + private final DispatcherV2 dispatcher; + private final BlockingQueue in; + private final BlockingQueue out; + private Z POISON; + private final ExecutorService ioPool; + + public TaskConsumerV2(DispatcherV2 dispatcher, + BlockingQueue in, + BlockingQueue out, + Z POISON) { + this.dispatcher = dispatcher; + this.in = in; + this.out = out; + this.ioPool = newIoExecutor(); + this.POISON = POISON; + } + +// @Override + public void start() { + List> inflight = new ArrayList<>(); + AtomicInteger inFlightCount = new AtomicInteger(); + + try { + while (true) { + X b = takeQuiet(in, Duration.ofSeconds(5)); + if (b == null) continue; + if (b.isPoison()) { + // дождаться всех in-flight задач + for (Future f : inflight) { + putQuiet(out, f.get()); + } + putQuiet(out, POISON); + return; + } + Future f = ioPool.submit(() -> dispatcher.handle(b)); + inflight.add(f); + + // чтобы не накапливать слишком много future в памяти, периодически сливаем + if (inFlightCount.incrementAndGet() % 8 == 0) { + drainDone(inflight, out); + } + } + } catch (Exception e) { + log.error("Error: {}", ExceptionUtils.getStackTrace(e)); + putQuiet(out, POISON); + } finally { + drainAll(inflight, out); + } + } + + void drainDone(List> fs, BlockingQueue out) throws Exception { + Iterator> it = fs.iterator(); + while (it.hasNext()) { + Future f = it.next(); + if (f.isDone()) { + putQuiet(out, f.get()); + it.remove(); + } + } + } + + void drainAll(List> fs, BlockingQueue out) { + for (Future f : fs) { + try { + putQuiet(out, f.get()); + } catch (Exception e) { + log.error("Error: {}", ExceptionUtils.getStackTrace(e)); + } + } + fs.clear(); + } + + private ExecutorService newIoExecutor() { + // один carrier на ядро, задачи — виртуальные + return Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name("vt-io-", 0).factory()); + } + + void putQuiet(BlockingQueue q, Z b) { + try { + q.put(b); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + X takeQuiet(BlockingQueue q, Duration timeout) { + try { + return q.poll(timeout.toMillis(), TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return null; + } + } + + static void sleep(long ms) { + try { + Thread.sleep(ms); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + +// @Override + public void stop() { + + } + +// @Override + public boolean isRunning() { + return true; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/TaskPayloadV2.java b/src/main/java/ru/nbch/credit_tracker/service/task/TaskPayloadV2.java new file mode 100644 index 0000000..ed6e5c8 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/TaskPayloadV2.java @@ -0,0 +1,5 @@ +package ru.nbch.credit_tracker.service.task; + +public interface TaskPayloadV2 { + boolean isPoison(); +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/XmlMergeCoordinator.java b/src/main/java/ru/nbch/credit_tracker/service/task/XmlMergeCoordinator.java index b8464bb..82ac282 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/XmlMergeCoordinator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/XmlMergeCoordinator.java @@ -11,7 +11,6 @@ import java.nio.file.StandardCopyOption; import java.nio.file.StandardOpenOption; import java.util.ArrayList; import java.util.List; -import java.util.UUID; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -91,7 +90,7 @@ public final class XmlMergeCoordinator implements InitializingBean { private void mergeBatch(List inputs) throws Exception { Integer packageId = inputs.stream().findFirst().get().getPackageId(); - String outName = "merged_" + UUID.randomUUID() + ".xml"; + String outName = "merged_" + packageId + ".xml"; Path tmp = mergedDir.resolve(outName + ".part"); Path out = mergedDir.resolve(outName); @@ -102,6 +101,7 @@ public final class XmlMergeCoordinator implements InitializingBean { w.writeStartElement("Persons"); for (PartFilePayload in : inputs) { + log.debug("Copy person to file {}", in.getPartPath()); copyPersonsChildren(in.getPartPath(), w); } diff --git a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml index 06b2538..6b97082 100644 --- a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml @@ -25,16 +25,12 @@ - - INSERT INTO signals_phone_fid_map (subject_id, phone, package_id) + + INSERT INTO signals_phone_fid_map (rec_uuid, subject_id, phone, package_id) values - (#{phoneFid.subjectId}, #{phoneFid.phone}, #{packageId}) + (#{phoneFid.recUuid}, #{phoneFid.subjectId}, #{phoneFid.phone}, #{packageId}) - RETURNING id @@ -73,7 +69,7 @@ UPDATE signals_phone_fid_map SET fid = #{subject.fid} - WHERE id = #{subject.id} + WHERE rec_uuid = #{subject.id} @@ -82,7 +78,7 @@ UPDATE signals_phone_fid_map SET error_text = #{subject.errorText} - WHERE id = #{subject.id} + WHERE id = #{subject.recUuid}