This commit is contained in:
parent
e809cbc3c0
commit
700e2442cc
7 changed files with 106 additions and 8 deletions
|
|
@ -26,6 +26,7 @@ public class FileService {
|
||||||
|
|
||||||
public void moveSuccess(Path file) {
|
public void moveSuccess(Path file) {
|
||||||
try {
|
try {
|
||||||
|
log.info("moving {} to {}", file, outDir);
|
||||||
Files.move(file, outDir.resolve(file.getFileName()), REPLACE_EXISTING);
|
Files.move(file, outDir.resolve(file.getFileName()), REPLACE_EXISTING);
|
||||||
} catch (IOException e) {
|
} catch (IOException e) {
|
||||||
log.error("failed to move to {}, trying to move to {}: {}", outDir, errDir, ExceptionUtils.getStackTrace(e));
|
log.error("failed to move to {}, trying to move to {}: {}", outDir, errDir, ExceptionUtils.getStackTrace(e));
|
||||||
|
|
@ -36,6 +37,7 @@ public class FileService {
|
||||||
|
|
||||||
public void moveError(Path file) {
|
public void moveError(Path file) {
|
||||||
try {
|
try {
|
||||||
|
log.info("moving {} to {}", file, errDir);
|
||||||
Files.move(file, errDir.resolve(file.getFileName()), REPLACE_EXISTING);
|
Files.move(file, errDir.resolve(file.getFileName()), REPLACE_EXISTING);
|
||||||
} catch (IOException e) {
|
} catch (IOException e) {
|
||||||
log.error("failed to move to {}: {}", errDir, ExceptionUtils.getStackTrace(e));
|
log.error("failed to move to {}: {}", errDir, ExceptionUtils.getStackTrace(e));
|
||||||
|
|
|
||||||
|
|
@ -6,6 +6,7 @@ import java.util.List;
|
||||||
import org.slf4j.Logger;
|
import org.slf4j.Logger;
|
||||||
import org.slf4j.LoggerFactory;
|
import org.slf4j.LoggerFactory;
|
||||||
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.scheduling.annotation.EnableScheduling;
|
||||||
import org.springframework.scheduling.annotation.Scheduled;
|
import org.springframework.scheduling.annotation.Scheduled;
|
||||||
import org.springframework.stereotype.Component;
|
import org.springframework.stereotype.Component;
|
||||||
import ru.clearing.classes.statics.data.misc.SOrders;
|
import ru.clearing.classes.statics.data.misc.SOrders;
|
||||||
|
|
@ -20,6 +21,7 @@ import ru.spcex.platform.imdg.api.ImdgProvider;
|
||||||
import ru.spcex.platform.utils.log.ExceptionUtils;
|
import ru.spcex.platform.utils.log.ExceptionUtils;
|
||||||
|
|
||||||
@Component
|
@Component
|
||||||
|
@EnableScheduling
|
||||||
public class SOrdersScheduler {
|
public class SOrdersScheduler {
|
||||||
private final static Logger log = LoggerFactory.getLogger(SOrdersScheduler.class);
|
private final static Logger log = LoggerFactory.getLogger(SOrdersScheduler.class);
|
||||||
|
|
||||||
|
|
@ -32,7 +34,7 @@ public class SOrdersScheduler {
|
||||||
public SOrdersScheduler(CsvImporterSettings settings, FileService fileService, TroProcessorProvider provider, ImdgProvider imdgProvider) {
|
public SOrdersScheduler(CsvImporterSettings settings, FileService fileService, TroProcessorProvider provider, ImdgProvider imdgProvider) {
|
||||||
this.importDir = Path.of(settings.getSrcDir());
|
this.importDir = Path.of(settings.getSrcDir());
|
||||||
this.fileService = fileService;
|
this.fileService = fileService;
|
||||||
processor = provider.createProcessor(TroParserType.SOrder, SOrder.class);
|
this.processor = provider.createProcessor(TroParserType.SOrder, SOrder.class);
|
||||||
this.sOrderImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_SOrders, SOrders.class);
|
this.sOrderImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_SOrders, SOrders.class);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -51,7 +53,9 @@ public class SOrdersScheduler {
|
||||||
|
|
||||||
private void processFile(Path path) {
|
private void processFile(Path path) {
|
||||||
try {
|
try {
|
||||||
|
log.info("processing file {}", path);
|
||||||
List<SOrder> sOrders = processor.readFile(path);
|
List<SOrder> sOrders = processor.readFile(path);
|
||||||
|
log.info("file {}: read {} orders", path, sOrders.size());
|
||||||
for (SOrder orderFromFile : sOrders) {
|
for (SOrder orderFromFile : sOrders) {
|
||||||
Long transId = orderFromFile.getTransId();
|
Long transId = orderFromFile.getTransId();
|
||||||
SOrders order = sOrderImdg.getSingleObjectByID(transId);
|
SOrders order = sOrderImdg.getSingleObjectByID(transId);
|
||||||
|
|
@ -63,8 +67,10 @@ public class SOrdersScheduler {
|
||||||
order.setState(orderFromFile.getStatus());
|
order.setState(orderFromFile.getStatus());
|
||||||
order.setOrderNum(orderFromFile.getOrderNumber());
|
order.setOrderNum(orderFromFile.getOrderNumber());
|
||||||
if (wasFound) {
|
if (wasFound) {
|
||||||
|
log.trace("updating order transId[{}]: state {}, orderNum {}", transId, orderFromFile.getStatus(), orderFromFile.getOrderNumber());
|
||||||
sOrderImdg.update(order);
|
sOrderImdg.update(order);
|
||||||
} else {
|
} else {
|
||||||
|
log.trace("creating order transId[{}]: state {}, orderNum {}", transId, orderFromFile.getStatus(), orderFromFile.getOrderNumber());
|
||||||
sOrderImdg.insert(order);
|
sOrderImdg.insert(order);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -2,6 +2,7 @@ package ru.spcex.clearing.csv.importer.component.tro.processor;
|
||||||
|
|
||||||
import java.io.BufferedReader;
|
import java.io.BufferedReader;
|
||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
|
import java.nio.charset.Charset;
|
||||||
import java.nio.file.Files;
|
import java.nio.file.Files;
|
||||||
import java.nio.file.Path;
|
import java.nio.file.Path;
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
|
|
@ -17,8 +18,7 @@ public class TroProcessor<T> {
|
||||||
|
|
||||||
public List<T> readFile(Path path) {
|
public List<T> readFile(Path path) {
|
||||||
List<T> result = new ArrayList<>();
|
List<T> result = new ArrayList<>();
|
||||||
|
try (BufferedReader reader = Files.newBufferedReader(path, Charset.forName("windows-1251"))) {
|
||||||
try (BufferedReader reader = Files.newBufferedReader(path)) {
|
|
||||||
String line;
|
String line;
|
||||||
int lineNumber = 0;
|
int lineNumber = 0;
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -4,11 +4,11 @@ import java.util.HashMap;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
import org.springframework.context.annotation.Configuration;
|
import org.springframework.stereotype.Component;
|
||||||
import ru.spcex.clearing.csv.importer.component.tro.parser.TroParser;
|
import ru.spcex.clearing.csv.importer.component.tro.parser.TroParser;
|
||||||
import ru.spcex.clearing.csv.importer.component.tro.parser.TroParserType;
|
import ru.spcex.clearing.csv.importer.component.tro.parser.TroParserType;
|
||||||
|
|
||||||
@Configuration
|
@Component
|
||||||
public class TroProcessorProvider {
|
public class TroProcessorProvider {
|
||||||
private final Map<TroParserType, TroParser<?>> m;
|
private final Map<TroParserType, TroParser<?>> m;
|
||||||
|
|
||||||
|
|
@ -23,13 +23,12 @@ public class TroProcessorProvider {
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
//@SuppressWarnings("unchecked")
|
@SuppressWarnings("unchecked")
|
||||||
public <T> TroProcessor<T> createProcessor(TroParserType type, Class<T> clazz) {
|
public <T> TroProcessor<T> createProcessor(TroParserType type, Class<T> clazz) {
|
||||||
TroParser<?> parser = m.get(type);
|
TroParser<?> parser = m.get(type);
|
||||||
if (parser == null) {
|
if (parser == null) {
|
||||||
throw new IllegalArgumentException("Unknown parser type: " + type);
|
throw new IllegalArgumentException("Unknown parser type: " + type);
|
||||||
}
|
}
|
||||||
return new TroProcessor<>((TroParser<T>) parser);
|
return new TroProcessor<>((TroParser<T>) parser);
|
||||||
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -17,4 +17,7 @@ csv-importer.kafka-producer.retries=0
|
||||||
csv-importer.kafka-producer.batch-size=16384
|
csv-importer.kafka-producer.batch-size=16384
|
||||||
csv-importer.kafka-producer.linger-ms=1
|
csv-importer.kafka-producer.linger-ms=1
|
||||||
csv-importer.kafka-producer.buffer-memory=33554432
|
csv-importer.kafka-producer.buffer-memory=33554432
|
||||||
csv-importer.poll-period=5000
|
csv-importer.poll-period=5000
|
||||||
|
csv-importer.src-dir=c:\\storage\\projects\\bin\\clearing\\files\\csv-importer\\
|
||||||
|
csv-importer.out-dir=c:\\storage\\projects\\bin\\clearing\\files\\csv-importer-out\\
|
||||||
|
csv-importer.err-dir=c:\\storage\\projects\\bin\\clearing\\files\\csv-importer-out-err\\
|
||||||
56
clearing-parent/csv-importer/src/main/resources/logback.xml
Normal file
56
clearing-parent/csv-importer/src/main/resources/logback.xml
Normal file
|
|
@ -0,0 +1,56 @@
|
||||||
|
<?xml version="1.0" encoding="UTF-8"?>
|
||||||
|
<configuration>
|
||||||
|
<property name="LOG_PATH" value="./log" />
|
||||||
|
<property name="FILE_NAME" value="csv-importer" />
|
||||||
|
<property name="CONSOLE_LOG_PATTERN" value="%date{HH:mm:ss.SSS} [%thread] %-5level %class{0}:%line - %message%n" />
|
||||||
|
<property name="FILE_LOG_PATTERN" value="%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %class{0}:%msg%n" />
|
||||||
|
<appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
|
||||||
|
<encoder>
|
||||||
|
<!-- |%X{ru.nbch.scoring.web.logging.mdc_key}-->
|
||||||
|
<Pattern>${CONSOLE_LOG_PATTERN}</Pattern>
|
||||||
|
<charset>utf-8</charset>
|
||||||
|
</encoder>
|
||||||
|
</appender>
|
||||||
|
<!-- first FILE TEXT appender -->
|
||||||
|
<appender name="TEXT_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
|
||||||
|
<file>${LOG_PATH}/${FILE_NAME}-text.log</file>
|
||||||
|
<encoder>
|
||||||
|
<!-- |%X{ru.nbch.scoring.web.logging.mdc_key}-->
|
||||||
|
<Pattern>${FILE_LOG_PATTERN}</Pattern>
|
||||||
|
<charset>utf8</charset>
|
||||||
|
</encoder>
|
||||||
|
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
|
||||||
|
<fileNamePattern>${LOG_PATH}/${FILE_NAME}-text.%d{yyyy-MM-dd}.%i.gz
|
||||||
|
</fileNamePattern>
|
||||||
|
<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
|
||||||
|
<maxFileSize>100MB</maxFileSize>
|
||||||
|
</timeBasedFileNamingAndTriggeringPolicy>
|
||||||
|
<maxHistory>10</maxHistory>
|
||||||
|
</rollingPolicy>
|
||||||
|
</appender>
|
||||||
|
<!-- second FILE JSON appender -->
|
||||||
|
<appender name="JSON_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
|
||||||
|
<file>${LOG_PATH}/${FILE_NAME}-json.log</file>
|
||||||
|
<encoder class="net.logstash.logback.encoder.LogstashEncoder" />
|
||||||
|
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
|
||||||
|
<fileNamePattern>${LOG_PATH}/${FILE_NAME}-json.%d{yyyy-MM-dd}.%i.gz
|
||||||
|
</fileNamePattern>
|
||||||
|
<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
|
||||||
|
<maxFileSize>100MB</maxFileSize>
|
||||||
|
</timeBasedFileNamingAndTriggeringPolicy>
|
||||||
|
<maxHistory>10</maxHistory>
|
||||||
|
</rollingPolicy>
|
||||||
|
</appender>
|
||||||
|
|
||||||
|
<root level="info">
|
||||||
|
<!-- <appender-ref ref="CONSOLE"/>-->
|
||||||
|
<appender-ref ref="TEXT_FILE"/>
|
||||||
|
<appender-ref ref="JSON_FILE" />
|
||||||
|
</root>
|
||||||
|
|
||||||
|
<logger name="ru.spcex" level="trace" additivity="false">
|
||||||
|
<appender-ref ref="TEXT_FILE"/>
|
||||||
|
<appender-ref ref="JSON_FILE" />
|
||||||
|
<!-- <appender-ref ref="CONSOLE"/>-->
|
||||||
|
</logger>
|
||||||
|
</configuration>
|
||||||
|
|
@ -0,0 +1,32 @@
|
||||||
|
package ru.spcex.clearing.csv.importer.component.tro.processor;
|
||||||
|
|
||||||
|
import java.net.URISyntaxException;
|
||||||
|
import java.nio.file.Path;
|
||||||
|
import java.util.List;
|
||||||
|
import org.junit.jupiter.api.Assertions;
|
||||||
|
import org.junit.jupiter.api.BeforeEach;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import ru.spcex.clearing.csv.importer.component.tro.data.SOrder;
|
||||||
|
import ru.spcex.clearing.csv.importer.component.tro.parser.TroParserType;
|
||||||
|
import ru.spcex.clearing.csv.importer.component.tro.parser.impl.SOrderParser;
|
||||||
|
|
||||||
|
class TroProcessorTest {
|
||||||
|
private TroProcessorProvider provider;
|
||||||
|
|
||||||
|
@BeforeEach
|
||||||
|
void setUp() {
|
||||||
|
this.provider = new TroProcessorProvider(List.of(
|
||||||
|
new SOrderParser()
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testSOrderTroParsing() throws URISyntaxException {
|
||||||
|
Path path = Path.of(
|
||||||
|
getClass().getClassLoader().getResource("tro/TVWK.tro").toURI()
|
||||||
|
);
|
||||||
|
TroProcessor<SOrder> processor = provider.createProcessor(TroParserType.SOrder, SOrder.class);
|
||||||
|
List<SOrder> sOrders = processor.readFile(path);
|
||||||
|
Assertions.assertTrue(sOrders.size() > 0, "read SOrder from tro file");
|
||||||
|
}
|
||||||
|
}
|
||||||
Loading…
Add table
Reference in a new issue