diff --git a/clearing-parent/pom.xml b/clearing-parent/pom.xml index 9ff3b6aab..f879cdf1e 100644 --- a/clearing-parent/pom.xml +++ b/clearing-parent/pom.xml @@ -36,6 +36,7 @@ registry-service test-clearing cleaning-builders + trade-importer diff --git a/clearing-parent/trade-importer/pom.xml b/clearing-parent/trade-importer/pom.xml new file mode 100644 index 000000000..7ea41af25 --- /dev/null +++ b/clearing-parent/trade-importer/pom.xml @@ -0,0 +1,127 @@ + + 4.0.0 + + ru.spcex.clearing + clearing-parent + SPCEX-1.0.0.0 + + trade-importer + trade-importer + Trade importer module + SPCEX-1.0.0.0 + jar + + + 17 + 17 + UTF-8 + + + + + org.springframework.boot + spring-boot-starter + + + org.springframework.boot + spring-boot-configuration-processor + true + + + com.fasterxml.jackson.core + jackson-databind + + + + + + org.springframework + spring-jdbc + + + com.microsoft.sqlserver + mssql-jdbc + + + com.mchange + c3p0 + + + + ru.spcex.clearing + classes + SPCEX-1.0.0.0 + compile + + + ru.spcex.platform + platform-messaging + + + ru.spcex.platform + platform-imdg-api-hazelcast-impl + + + ru.spcex.platform + platform-enum + + + + + org.springframework.boot + spring-boot-starter-test + test + + + ru.spcex.clearing + test-clearing + test + + + + + + + src/main/resources + + application.properties + + false + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + repackage + + + + + ${project.artifactId} + + + + org.apache.maven.plugins + maven-surefire-plugin + 2.21.0 + + + org.junit.platform + junit-platform-surefire-provider + 1.2.0-M1 + + + org.junit.jupiter + junit-jupiter-engine + 5.2.0-M1 + + + + + + diff --git a/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/TradeImporterApplication.java b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/TradeImporterApplication.java new file mode 100644 index 000000000..fe1e53e01 --- /dev/null +++ b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/TradeImporterApplication.java @@ -0,0 +1,12 @@ +package ru.spcex.clearing.trade.importer; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class TradeImporterApplication { + public static void main(String[] args) { + SpringApplication app = new SpringApplication(TradeImporterApplication.class); + app.run(args); + } +} diff --git a/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/DbConnectionConfig.java b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/DbConnectionConfig.java new file mode 100644 index 000000000..3afddae6b --- /dev/null +++ b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/DbConnectionConfig.java @@ -0,0 +1,62 @@ +package ru.spcex.clearing.trade.importer.config; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.SingleConnectionDataSource; +import ru.spcex.clearing.trade.importer.config.settings.DatabaseSettings; +import ru.spcex.clearing.trade.importer.config.settings.ImportTradeServiceSettings; +import ru.spcex.clearing.trade.importer.error.ModuleInitializeException; + +import javax.sql.DataSource; +import java.sql.Connection; + +@SuppressWarnings("UnnecessaryLocalVariable") +@Configuration +public class DbConnectionConfig { + private final Logger log = LoggerFactory.getLogger(this.getClass()); + + private final DatabaseSettings settings; + + public DbConnectionConfig(ImportTradeServiceSettings settings) { + this.settings = settings.getDatabase(); + } + + @Bean(destroyMethod = "destroy") + public SingleConnectionDataSource dataSource() { + String login = settings.getLogin(); + String password = settings.getPassword(); + String dbUrl = settings.getUrl(); + + SingleConnectionDataSource cpds = new SingleConnectionDataSource(); + try { + cpds.setDriverClassName("com.microsoft.sqlserver.jdbc.SQLServerDriver"); + } catch (Exception ue) { + throw new RuntimeException(ue); + } + cpds.setUrl(dbUrl); + cpds.setUsername(login); + cpds.setPassword(password); + + String OPERATION_DATABASE_CONNECTION_CHECK = String.format("Database [%s] connection check", dbUrl); + try { + Connection conn = cpds.getConnection(); +// conn.close(); + log.info("{}: success", OPERATION_DATABASE_CONNECTION_CHECK); + return cpds; + } catch (Throwable e) { + String msg = String.format("%s: failed: %s -> %s", + OPERATION_DATABASE_CONNECTION_CHECK, e.getClass().getSimpleName(), e.getMessage()); + log.error(msg); + throw new ModuleInitializeException(msg, e); + } + } + + @Bean + public JdbcTemplate jdbcTemplate(DataSource dataSource) { + JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource); + return jdbcTemplate; + } +} diff --git a/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/ImdgConfig.java b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/ImdgConfig.java new file mode 100644 index 000000000..b5910ee3f --- /dev/null +++ b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/ImdgConfig.java @@ -0,0 +1,47 @@ +package ru.spcex.clearing.trade.importer.config; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import ru.spcex.clearing.trade.importer.config.settings.ImportTradeServiceSettings; +import ru.spcex.platform.imdg.api.ImdgProvider; +import ru.spcex.platform.imdg.iml.hazelcast.service.HazelcastService; + +@Configuration +public class ImdgConfig { + private static ThreadPoolTaskExecutor createThreadPoolTaskExecutor(int maxPoolSz, boolean waitForCompletion) { + ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor(); + if (maxPoolSz > 2) { + pool.setKeepAliveSeconds(60); + pool.setAllowCoreThreadTimeOut(true); + } + pool.setCorePoolSize(maxPoolSz); + pool.setWaitForTasksToCompleteOnShutdown(waitForCompletion); + return pool; + } + + @Bean(name = "taskExecutorHazelcastClientInitializer") + public ThreadPoolTaskExecutor taskExecutorHazelcastClientInitializer() { + return createThreadPoolTaskExecutor(1, true); + } + + @Bean(name = "taskExecutorIdGeneratorAwaiter") + public ThreadPoolTaskExecutor taskExecutorIdGeneratorAwaiter() { + return createThreadPoolTaskExecutor(1, false); + } + + @Autowired + @Bean + public ImdgProvider imdgProvider( + @Qualifier("taskExecutorHazelcastClientInitializer") ThreadPoolTaskExecutor taskExecutorHazelcastClientInitializer, + @Qualifier("taskExecutorIdGeneratorAwaiter") ThreadPoolTaskExecutor taskExecutorIdGeneratorAwaiter, + ImportTradeServiceSettings settings + ) { + return new HazelcastService(taskExecutorHazelcastClientInitializer, + taskExecutorIdGeneratorAwaiter, + settings.getHazelcast()); + } + +} diff --git a/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/KafkaConfig.java b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/KafkaConfig.java new file mode 100644 index 000000000..0708fa158 --- /dev/null +++ b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/KafkaConfig.java @@ -0,0 +1,29 @@ +package ru.spcex.clearing.trade.importer.config; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.producer.Producer; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.config.ConfigurableBeanFactory; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Scope; +import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory; +import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory; +import ru.spcex.clearing.trade.importer.config.settings.ImportTradeServiceSettings; + +@Configuration +public class KafkaConfig { + + @Autowired + @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE) + @Bean + public Consumer createConsumer(ImportTradeServiceSettings settings) { + return KafkaConsumerFactory.consumer(settings.getKafkaConsumer()); + } + + @Autowired + @Bean + public Producer createProducer(ImportTradeServiceSettings settings) { + return KafkaProducerFactory.producer(settings.getKafkaProducer()); + } +} diff --git a/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/TradeImporterConfig.java b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/TradeImporterConfig.java new file mode 100644 index 000000000..f6ae24622 --- /dev/null +++ b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/TradeImporterConfig.java @@ -0,0 +1,20 @@ +package ru.spcex.clearing.trade.importer.config; + +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.ComponentScan; +import org.springframework.context.annotation.Configuration; +import ru.spcex.clearing.trade.importer.config.settings.ImportTradeServiceSettings; + +@Configuration +@EnableConfigurationProperties +@ComponentScan(basePackages = {"ru.spcex.clearing.trade.importer"}) +public class TradeImporterConfig { + private final ImportTradeServiceSettings settings; + private final ApplicationContext context; + + public TradeImporterConfig(ImportTradeServiceSettings settings, ApplicationContext context) { + this.settings = settings; + this.context = context; + } +} diff --git a/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/settings/Cron.java b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/settings/Cron.java new file mode 100644 index 000000000..08284dba5 --- /dev/null +++ b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/settings/Cron.java @@ -0,0 +1,14 @@ +package ru.spcex.clearing.trade.importer.config.settings; + +public class Cron { + + private String checkSrcDirCron; + + public String getCheckSrcDirCron() { + return checkSrcDirCron; + } + + public void setCheckSrcDirCron(String checkSrcDirCron) { + this.checkSrcDirCron = checkSrcDirCron; + } +} diff --git a/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/settings/DatabaseSettings.java b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/settings/DatabaseSettings.java new file mode 100644 index 000000000..c019613e1 --- /dev/null +++ b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/settings/DatabaseSettings.java @@ -0,0 +1,31 @@ +package ru.spcex.clearing.trade.importer.config.settings; + +public class DatabaseSettings { + private String login; + private String password; + private String url; + + public String getLogin() { + return login; + } + + public void setLogin(String login) { + this.login = login; + } + + public String getPassword() { + return password; + } + + public void setPassword(String password) { + this.password = password; + } + + public String getUrl() { + return url; + } + + public void setUrl(String url) { + this.url = url; + } +} diff --git a/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/settings/ImportTradeServiceSettings.java b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/settings/ImportTradeServiceSettings.java new file mode 100644 index 000000000..4a864d328 --- /dev/null +++ b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/config/settings/ImportTradeServiceSettings.java @@ -0,0 +1,60 @@ +package ru.spcex.clearing.trade.importer.config.settings; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.context.annotation.PropertySource; +import org.springframework.stereotype.Component; +import ru.spcex.clearing.platform.messaging.config.element.KafkaConsumerSettings; +import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings; +import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams; + +@Component +@PropertySource("file:${spring.config.location}/application.properties") +@ConfigurationProperties("trade-importer") +public class ImportTradeServiceSettings { + private HazelcastClientParams hazelcast; + private KafkaProducerSettings kafkaProducer; + private KafkaConsumerSettings kafkaConsumer; + private DatabaseSettings database; + private Cron cron; + + public HazelcastClientParams getHazelcast() { + return hazelcast; + } + + public void setHazelcast(HazelcastClientParams hazelcast) { + this.hazelcast = hazelcast; + } + + public KafkaProducerSettings getKafkaProducer() { + return kafkaProducer; + } + + public void setKafkaProducer(KafkaProducerSettings kafkaProducer) { + this.kafkaProducer = kafkaProducer; + } + + public KafkaConsumerSettings getKafkaConsumer() { + return kafkaConsumer; + } + + public void setKafkaConsumer(KafkaConsumerSettings kafkaConsumer) { + this.kafkaConsumer = kafkaConsumer; + } + + + public DatabaseSettings getDatabase() { + return database; + } + + public void setDatabase(DatabaseSettings database) { + this.database = database; + } + + public Cron getCron() { + return cron; + } + + public void setCron(Cron cron) { + this.cron = cron; + } +} diff --git a/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/error/ModuleInitializeException.java b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/error/ModuleInitializeException.java new file mode 100644 index 000000000..e6b7f5cc5 --- /dev/null +++ b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/error/ModuleInitializeException.java @@ -0,0 +1,19 @@ +package ru.spcex.clearing.trade.importer.error; + +public class ModuleInitializeException extends RuntimeException { + public ModuleInitializeException() { + } + + public ModuleInitializeException(String message) { + super(message); + } + + public ModuleInitializeException(String message, Throwable cause) { + super(message, cause); + } + + public ModuleInitializeException(Throwable cause) { + super(cause); + } + +} diff --git a/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/services/LauncherCommandReceiver.java b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/services/LauncherCommandReceiver.java new file mode 100644 index 000000000..074343102 --- /dev/null +++ b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/services/LauncherCommandReceiver.java @@ -0,0 +1,32 @@ +package ru.spcex.clearing.trade.importer.services; + +import org.apache.kafka.clients.consumer.Consumer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Service; +import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest; +import ru.spcex.clearing.platform.messaging.service.QueueConsumer; +import ru.spcex.platform.enumeration.Task; + +@Service +public class LauncherCommandReceiver extends QueueConsumer implements InitializingBean { + private final Logger log = LoggerFactory.getLogger(getClass()); + private final TradeImporterService importer; + + @Autowired + public LauncherCommandReceiver(Consumer kafkaQueue, + TradeImporterService importer) { + super(kafkaQueue); + this.importer = importer; + } + + @Override + public void afterPropertiesSet() { + callback(LauncherCommandRequest.class) + .setConsumer(action -> importer.process()) + .forDestination(Task.getOfTrades.topic(), callbacks::put); // GTRD + init(); + } +} diff --git a/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/services/TradeImporterService.java b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/services/TradeImporterService.java new file mode 100644 index 000000000..22a023f7f --- /dev/null +++ b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/services/TradeImporterService.java @@ -0,0 +1,72 @@ +package ru.spcex.clearing.trade.importer.services; + +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.jdbc.core.BeanPropertyRowMapper; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.core.RowMapper; +import org.springframework.scheduling.annotation.EnableScheduling; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Service; +import org.springframework.util.StringUtils; +import ru.clearing.classes.statics.data.misc.STrades; +import ru.spcex.clearing.imdg.IMDGDistributedNames; +import ru.spcex.clearing.platform.messaging.domain.cud.utilities.STradesImportedRequest; +import ru.spcex.platform.imdg.api.Imdg; +import ru.spcex.platform.imdg.api.ImdgProvider; + +import java.util.Collection; +import java.util.Map; + +import static ru.spcex.clearing.platform.messaging.domain.Consts.S_TRADES_IMPORTED; + +@Service +@EnableScheduling +public class TradeImporterService { + private final Logger log = LoggerFactory.getLogger(getClass()); + private final Imdg sTradesImdg; + private final JdbcTemplate jdbcTemplate; + private final Producer producer; + private static final RowMapper ROW_MAPPER = BeanPropertyRowMapper.newInstance(STrades.class); + + + public TradeImporterService(ImdgProvider imdgProvider, JdbcTemplate jdbcTemplate, Producer producer) { + this.sTradesImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_STrades, STrades.class); + this.jdbcTemplate = jdbcTemplate; + this.producer = producer; + } + + @Scheduled(cron = "${trade-importer.cron.load-from-db-cron}") + public void run() { + process(); + } + + public void process() { + Collection tradesFromDB = jdbcTemplate.query("SELECT * FROM Trades", ROW_MAPPER); + + for (STrades tradesDb : tradesFromDB) { + if (isValidTrades(tradesDb)) { + STrades sTrades = sTradesImdg.getSingleObjectByFieldValues(Map.of("tradeDate", tradesDb.getTradeDate(), + "tradeNum", tradesDb.getTradeNum(), + "operation", tradesDb.getOperation(), + "classCode", tradesDb.getClassCode())); + if (sTrades != null) tradesDb.setId(sTrades.getId()); + sTradesImdg.insert(tradesDb); + } else { + log.warn(String.format("(10001) \"Сделка с номером в ТС = %s некорректна\"", tradesDb.getTradeNum())); + } + } + + STradesImportedRequest sTradesImportedRequest = new STradesImportedRequest(); + producer.send(new ProducerRecord<>(S_TRADES_IMPORTED, sTradesImportedRequest)); + } + + private boolean isValidTrades(STrades trades) { + return trades.getTradeDate() != null + && trades.getTradeNum() != null + && StringUtils.hasText(trades.getOperation()) + && trades.getTradeNum() != null; + } +} diff --git a/clearing-parent/trade-importer/src/main/resources/application.properties b/clearing-parent/trade-importer/src/main/resources/application.properties new file mode 100644 index 000000000..fb9ca6cef --- /dev/null +++ b/clearing-parent/trade-importer/src/main/resources/application.properties @@ -0,0 +1,28 @@ +spring.main.web-application-type=none + +#trade-importer.cron.load-from-db-cron=0 0/5 * * * ? - каждые 5 минут +trade-importer.cron.load-from-db-cron=0 0/5 * * * ? + +trade-importer.database.login=sa +trade-importer.database.password=Aa123456 +trade-importer.database.url=jdbc:sqlserver://localhost:1433;database=SPVB_TS;schema=dbo + +trade-importer.hazelcast.cluster-members=127.0.0.1:5701 +trade-importer.hazelcast.login=dev +trade-importer.hazelcast.password=dev-pass + +trade-importer.kafka-consumer.bootstrap-servers=localhost:9092 +trade-importer.kafka-consumer.group-id=dev-group- trade-importer +trade-importer.kafka-consumer.enable-auto-commit=false +trade-importer.kafka-consumer.session-timeout-ms=30000 +trade-importer.kafka-consumer.auto-offset-reset=latest +trade-importer.kafka-consumer.linger-ms=1 +trade-importer.kafka-consumer.buffer-memory=33554432 + +trade-importer.kafka-producer.bootstrap-servers=localhost:9092 +trade-importer.kafka-producer.acks=all +trade-importer.kafka-producer.retries=0 +trade-importer.kafka-producer.batch-size=16384 +trade-importer.kafka-producer.linger-ms=1 +trade-importer.kafka-producer.buffer-memory=33554432 + diff --git a/clearing-parent/trade-importer/src/test/java/ru/spcex/clearing/trade/importer/AbstractServiceTest.java b/clearing-parent/trade-importer/src/test/java/ru/spcex/clearing/trade/importer/AbstractServiceTest.java new file mode 100644 index 000000000..3e1e33c3d --- /dev/null +++ b/clearing-parent/trade-importer/src/test/java/ru/spcex/clearing/trade/importer/AbstractServiceTest.java @@ -0,0 +1,61 @@ +package ru.spcex.clearing.trade.importer; + +import org.apache.kafka.clients.producer.MockProducer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Captor; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.test.mock.mockito.MockBean; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit.jupiter.SpringExtension; +import ru.clearing.classes.statics.data.misc.STrades; +import ru.clearing.classes.statics.data.scheduler.PlannerAllToday; +import ru.spcex.clearing.imdg.IMDGDistributedNames; +import ru.spcex.clearing.test.MatcherFactory; +import ru.spcex.clearing.test.TestUtils; +import ru.spcex.clearing.test.config.ImdgTestConfig; +import ru.spcex.clearing.test.config.KafkaTestConfig; +import ru.spcex.clearing.trade.importer.config.config.DbTestConnectionConfig; +import ru.spcex.clearing.trade.importer.config.settings.ImportTradeServiceSettings; +import ru.spcex.clearing.trade.importer.services.LauncherCommandReceiver; +import ru.spcex.clearing.trade.importer.services.TradeImporterService; +import ru.spcex.platform.imdg.api.Imdg; +import ru.spcex.platform.imdg.api.ImdgProvider; + +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.spy; +import static ru.spcex.clearing.test.MatcherFactory.usingIgnoringFieldsComparator; +import static ru.spcex.clearing.test.config.ImdgTestConfig.currentID; +import static ru.spcex.clearing.test.config.ImdgTestConfig.waitAvailableImdgProviderAndAddAdminWithDefaultId; + +@ExtendWith(SpringExtension.class) +@ContextConfiguration(classes = { + DbTestConnectionConfig.class, + ImportTradeServiceSettings.class, + LauncherCommandReceiver.class, + TradeImporterService.class, + ImdgTestConfig.class, + KafkaTestConfig.class}) +public abstract class AbstractServiceTest { + protected static final MatcherFactory.Matcher PLANNER_ALL_TODAY_MATCHER = usingIgnoringFieldsComparator("created", "updated"); + protected static final long id = currentID.getAndIncrement(); + protected Imdg sTradesImdg; + + @Captor + protected ArgumentCaptor producerRecord; + @MockBean + protected MockProducer mockProducer; + @Autowired + @Qualifier("hazelcastServiceTest") + protected ImdgProvider imdgProvider; + + protected void init() { + waitAvailableImdgProviderAndAddAdminWithDefaultId(); + this.sTradesImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_STrades, STrades.class); + + TestUtils.FutureRecordMetadata future = spy(new TestUtils.FutureRecordMetadata()); + doReturn(future).when(mockProducer).send(producerRecord.capture()); + } +} diff --git a/clearing-parent/trade-importer/src/test/java/ru/spcex/clearing/trade/importer/config/config/DbTestConnectionConfig.java b/clearing-parent/trade-importer/src/test/java/ru/spcex/clearing/trade/importer/config/config/DbTestConnectionConfig.java new file mode 100644 index 000000000..c5a0b68cb --- /dev/null +++ b/clearing-parent/trade-importer/src/test/java/ru/spcex/clearing/trade/importer/config/config/DbTestConnectionConfig.java @@ -0,0 +1,55 @@ +package ru.spcex.clearing.trade.importer.config.config; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.SingleConnectionDataSource; +import ru.spcex.clearing.trade.importer.error.ModuleInitializeException; + +import javax.sql.DataSource; +import java.sql.Connection; + +@SuppressWarnings("UnnecessaryLocalVariable") +@Configuration +public class DbTestConnectionConfig { + private final Logger log = LoggerFactory.getLogger(this.getClass()); + + @Bean(destroyMethod = "destroy") + public SingleConnectionDataSource dataSource() { + String login = "sa"; + String password = "Aa123456"; + String dbUrl = "jdbc:sqlserver://localhost:1433;database=SPVB_TS;schema=dbo"; + + + SingleConnectionDataSource cpds = new SingleConnectionDataSource(); + try { + cpds.setDriverClassName("com.microsoft.sqlserver.jdbc.SQLServerDriver"); + } catch (Exception ue) { + throw new RuntimeException(ue); + } + cpds.setUrl(dbUrl); + cpds.setUsername(login); + cpds.setPassword(password); + + String OPERATION_DATABASE_CONNECTION_CHECK = String.format("Database [%s] connection check", dbUrl); + try { + Connection conn = cpds.getConnection(); +// conn.close(); + log.info("{}: success", OPERATION_DATABASE_CONNECTION_CHECK); + return cpds; + } catch (Throwable e) { + String msg = String.format("%s: failed: %s -> %s", + OPERATION_DATABASE_CONNECTION_CHECK, e.getClass().getSimpleName(), e.getMessage()); + log.error(msg); + throw new ModuleInitializeException(msg, e); + } + } + + @Bean + public JdbcTemplate jdbcTemplate(DataSource dataSource) { + JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource); + return jdbcTemplate; + } +} diff --git a/clearing-parent/trade-importer/src/test/java/ru/spcex/clearing/trade/importer/services/TradeImporterServiceTest.java b/clearing-parent/trade-importer/src/test/java/ru/spcex/clearing/trade/importer/services/TradeImporterServiceTest.java new file mode 100644 index 000000000..615339e94 --- /dev/null +++ b/clearing-parent/trade-importer/src/test/java/ru/spcex/clearing/trade/importer/services/TradeImporterServiceTest.java @@ -0,0 +1,68 @@ +package ru.spcex.clearing.trade.importer.services; + +import org.apache.kafka.clients.consumer.MockConsumer; +import org.junit.jupiter.api.BeforeAll; +import org.springframework.beans.factory.annotation.Autowired; +import ru.clearing.classes.statics.data.misc.STrades; +import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest; +import ru.spcex.clearing.trade.importer.AbstractServiceTest; +import ru.spcex.platform.enumeration.Task; + +import javax.annotation.PostConstruct; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.time.LocalDate; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.verify; +import static ru.spcex.clearing.test.TestUtils.addRecordToKafka; +import static ru.spcex.clearing.test.TestUtils.getJsonStringForNew; + +class TradeImporterServiceTest extends AbstractServiceTest { + @Autowired + LauncherCommandReceiver launcherCommandReceiver; + + @PostConstruct + public void init() { + super.init(); + } + + @BeforeAll + static void setProperty() { + Path path = Paths.get("src", "main", "resources"); + String currentPath = path.toAbsolutePath().toString(); + System.setProperty("spring.config.location", currentPath); +// Hazelcast.shutdownAll(); + } + + /** + * {@link TradeImporterService#process()}
+ * Тест проверяет обновление сущности {@link STrades}.
+ * Входной запрос {@link LauncherCommandRequest}:
+ */ +// @Test для работы теста нужна тестовая база Microsoft SQL с данными + void process() { + STrades trades = new STrades(); + trades.setId(22L); + trades.setTradeDate(LocalDate.of(2023,4,19)); + trades.setTradeNum(661486L); + trades.setOperation("operation20"); + trades.setClassCode("UESC"); + Long id = sTradesImdg.insert(trades); + + addRecordToKafka((MockConsumer) launcherCommandReceiver.getConsumer(), Task.getOfTrades.topic(), 0, 1, getJsonStringForNew(new LauncherCommandRequest(),0)); + + //waiting for kafka producer send message + verify(mockProducer, timeout(30_000L).times(1)) + .send(producerRecord.capture()); + + STrades sTrades = sTradesImdg.getSingleObjectByFieldValues(Map.of("tradeDate", trades.getTradeDate(), + "tradeNum", trades.getTradeNum(), + "operation", trades.getOperation(), + "classCode", trades.getClassCode())); + + assertEquals(sTrades.getId(), id); + } +} \ No newline at end of file diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java index f07a3033e..04ec6af5c 100644 --- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java @@ -86,6 +86,7 @@ public interface Consts { String SDF03_PROCESS = "sdf03-process"; String SDF11_PROCESS = "sdf11-process"; String EXPORT_PROCESS = "export-process"; + String S_TRADES_IMPORTED = "s_trades-imported"; String ACCOUNT_NEW = "account-new"; String ACCOUNT_TERMINATION = "account-termination"; String BALANCE_ACCOUNT_NEW = "balance-account-new"; diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/utilities/STradesImportedRequest.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/utilities/STradesImportedRequest.java new file mode 100644 index 000000000..348527aae --- /dev/null +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/utilities/STradesImportedRequest.java @@ -0,0 +1,4 @@ +package ru.spcex.clearing.platform.messaging.domain.cud.utilities; + +public class STradesImportedRequest { +}