treshenkov/clearing
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

work with files changed, new task added, sending file info to kafka in SWT exporter added, pom changed, application.properties changed

Browse source
This commit is contained in:
Ivan Nikolaev-Axenov 2024-07-08 12:37:16 +03:00
parent 95f7741111
commit d3a6456d33
13 changed files with 219 additions and 146 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
clearing-parent/snapshot-maker/pom.xml
View file

@ -18,17 +18,13 @@
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<hikaricp.version>5.1.0</hikaricp.version>
<springdoc-openapi.version>2.5.0</springdoc-openapi.version>
<commons-io.version>2.16.1</commons-io.version>
</properties>
<dependencies>
<!-- Spring dependencies -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<!-- Kafka dependencies -->
@ -38,11 +34,6 @@
</dependency>
<!-- Misc dependencies -->
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<version>${commons-io.version}</version>
</dependency>
<dependency>
<groupId>ru.spcex.clearing</groupId>
<artifactId>classes</artifactId>

67
clearing-parent/snapshot-maker/src/main/java/ru/spcex/clearing/snapshot/maker/config/settings/LocationSettings.java
View file

@ -1,67 +0,0 @@
package ru.spcex.clearing.snapshot.maker.config.settings;
public class LocationSettings {
private String localStore;
private String dbfImporterFolder;
private String dbfExporterFolder;
private String swtImporterFolder;
private String swtExporterFolder;
private String xmlImporterFolder;
private String xmlExporterFolder;
public String getLocalStore() {
return localStore;
}
public void setLocalStore(String localStore) {
this.localStore = localStore;
}
public String getDbfImporterFolder() {
return dbfImporterFolder;
}
public void setDbfImporterFolder(String dbfImporterFolder) {
this.dbfImporterFolder = dbfImporterFolder;
}
public String getDbfExporterFolder() {
return dbfExporterFolder;
}
public void setDbfExporterFolder(String dbfExporterFolder) {
this.dbfExporterFolder = dbfExporterFolder;
}
public String getSwtImporterFolder() {
return swtImporterFolder;
}
public void setSwtImporterFolder(String swtImporterFolder) {
this.swtImporterFolder = swtImporterFolder;
}
public String getSwtExporterFolder() {
return swtExporterFolder;
}
public void setSwtExporterFolder(String swtExporterFolder) {
this.swtExporterFolder = swtExporterFolder;
}
public String getXmlImporterFolder() {
return xmlImporterFolder;
}
public void setXmlImporterFolder(String xmlImporterFolder) {
this.xmlImporterFolder = xmlImporterFolder;
}
public String getXmlExporterFolder() {
return xmlExporterFolder;
}
public void setXmlExporterFolder(String xmlExporterFolder) {
this.xmlExporterFolder = xmlExporterFolder;
}
}

10
clearing-parent/snapshot-maker/src/main/java/ru/spcex/clearing/snapshot/maker/config/settings/SnapshotMakerSettings.java
View file

@ -13,7 +13,7 @@ public class SnapshotMakerSettings {
private DatabaseSettings database;
private KafkaConsumerSettings kafkaConsumer;
private KafkaProducerSettings kafkaProducer;
private LocationSettings location;
private String storeLocation;
public DatabaseSettings getDatabase() {
return database;
@ -39,11 +39,11 @@ public class SnapshotMakerSettings {
this.kafkaProducer = kafkaProducer;
}
public LocationSettings getLocation() {
return location;
public String getStoreLocation() {
return storeLocation;
}
public void setLocation(LocationSettings location) {
this.location = location;
public void setStoreLocation(String storeLocation) {
this.storeLocation = storeLocation;
}
}

18
clearing-parent/snapshot-maker/src/main/java/ru/spcex/clearing/snapshot/maker/service/CommandService.java
View file

@ -9,20 +9,25 @@ import org.springframework.stereotype.Service;
import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
import ru.spcex.clearing.platform.messaging.domain.Consts;
import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.utilities.ExportedSDFFile;
import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
import ru.spcex.clearing.platform.messaging.service.RequestInfoUpdate;
import ru.spcex.clearing.platform.messaging.service.Status;
import ru.spcex.clearing.snapshot.maker.utils.FileStore;
@Service
public class CommandService extends QueueConsumer implements InitializingBean {
private final Logger log = LoggerFactory.getLogger(getClass());
private final SessionService sessionService;
private final FileStore fileStore;
public CommandService(Consumer<String, Object> kafkaQueue,
Producer<String, Object> kafkaResponseQueue,
SessionService sessionService) {
SessionService sessionService,
FileStore fileStore) {
super(kafkaQueue, kafkaResponseQueue);
this.sessionService = sessionService;
this.fileStore = fileStore;
}
@Override
@ -30,7 +35,16 @@ public class CommandService extends QueueConsumer implements InitializingBean {
callback(LauncherCommandRequest.class)
.setFunction(this::process)
.forDestination(Consts.LAUNCHER_NEW, callbacks::put);
callback(ExportedSDFFile.class)
.setConsumer(r -> {
if (fileStore.isSessionStarted()) {
log.info("Adding {} to files list", r.getRequestPayload());
} else {
log.info("Session hasn't been started, ignoring {}", r.getRequestPayload());
}
fileStore.addFile(r.getRequestPayload());
})
.forDestination(Consts.FILE_CREATED, callbacks::put);
init();
}

91
clearing-parent/snapshot-maker/src/main/java/ru/spcex/clearing/snapshot/maker/service/SessionService.java
View file

@ -1,6 +1,7 @@
package ru.spcex.clearing.snapshot.maker.service;
import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.file.Files;
@ -11,30 +12,35 @@ import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import ru.spcex.clearing.platform.messaging.domain.cud.utilities.ExportedSDFFile;
import ru.spcex.clearing.platform.messaging.service.FileType;
import ru.spcex.clearing.platform.messaging.service.RequestInfoUpdate;
import ru.spcex.clearing.platform.messaging.service.Status;
import ru.spcex.clearing.snapshot.maker.config.settings.SnapshotMakerSettings;
import ru.spcex.clearing.snapshot.maker.utils.FileStore;
@Service
public class SessionService {
private final static Pattern JDBC_URL_PATTERN = Pattern.compile("(?<=//)(.+?)(?=:):(.+?)(?=/)/(.+?)(?=\\?|$)");
private final Logger log = LoggerFactory.getLogger(getClass());
private final SnapshotMakerSettings settings;
private final DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH-mm-ss");
private final static Pattern JDBC_URL_PATTERN = Pattern.compile("(?<=//)(.+?)(?=:):(.+?)(?=/)/(.+?)(?=\\?|$)");
private final FileStore fileStore;
private Path currentPath;
public SessionService(SnapshotMakerSettings settings) {
public SessionService(SnapshotMakerSettings settings,
FileStore fileStore) {
this.settings = settings;
this.fileStore = fileStore;
}
public RequestInfoUpdate startSession() {
RequestInfoUpdate requestInfoUpdate = new RequestInfoUpdate();
if (currentPath != null) {
if (currentPath != null || fileStore.isSessionStarted()) {
log.info("You need to stop session first!");
requestInfoUpdate.setStatus(Status.Error);
requestInfoUpdate.setMessage("You need to stop session first!");
@ -44,7 +50,7 @@ public class SessionService {
log.info("Starting making snapshot");
try {
this.currentPath = Files.createDirectories(Paths.get(settings.getLocation().getLocalStore()).resolve(LocalDateTime.now().format(dtf)));
this.currentPath = Files.createDirectories(Paths.get(settings.getStoreLocation()).resolve(LocalDateTime.now().format(dtf)));
log.info("Session folder created: {}", this.currentPath);
} catch (IOException e) {
log.info("Can't create directory, error={}", e.getMessage());
@ -58,6 +64,8 @@ public class SessionService {
return requestInfoUpdate;
}
this.fileStore.setSessionStarted(true);
requestInfoUpdate.setStatus(Status.Success);
requestInfoUpdate.setMessage("Snapshot created successfully in folder " + this.currentPath.toAbsolutePath());
return requestInfoUpdate;
@ -66,7 +74,7 @@ public class SessionService {
public RequestInfoUpdate stopSession() {
RequestInfoUpdate requestInfoUpdate = new RequestInfoUpdate();
if (currentPath == null) {
if (currentPath == null || !fileStore.isSessionStarted()) {
log.info("You need to start session first!");
requestInfoUpdate.setStatus(Status.Error);
requestInfoUpdate.setMessage("You need to start session first!");
@ -75,8 +83,10 @@ public class SessionService {
log.info("Stopping session");
this.fileStore.setSessionStarted(false);
copyDirectories(requestInfoUpdate);
this.currentPath = null;
this.fileStore.clearFiles();
if (requestInfoUpdate.getStatus() == Status.Error) {
return requestInfoUpdate;
@ -117,7 +127,11 @@ public class SessionService {
out.append(line).append(System.lineSeparator());
}
int statusCode = process.waitFor();
log.info("Status code={}, output {}", statusCode, out);
if (statusCode != 0) {
log.info("Can't dump database, status code = {}. Make sure that pg-dump is installed or PostgreSQL version and pg-dump version is the same.", statusCode);
requestInfoUpdate.setStatus(Status.Error);
requestInfoUpdate.setMessage("Can't dump database, status code = " + statusCode);
}
} catch (IOException | InterruptedException e) {
log.info("Can't dump database, error={}", e.getMessage());
requestInfoUpdate.setStatus(Status.Error);
@ -126,40 +140,45 @@ public class SessionService {
}
private void copyDirectories(RequestInfoUpdate requestInfoUpdate) {
List<Path> directoriesToCopy = List.of(
Paths.get(settings.getLocation().getDbfImporterFolder()),
Paths.get(settings.getLocation().getDbfExporterFolder()),
Paths.get(settings.getLocation().getSwtImporterFolder()),
Paths.get(settings.getLocation().getSwtExporterFolder()),
Paths.get(settings.getLocation().getXmlImporterFolder()),
Paths.get(settings.getLocation().getXmlExporterFolder())
);
Path dbfFolder = this.currentPath.resolve("dbf");
Path xmlFolder = this.currentPath.resolve("xml");
Path swtFolder = this.currentPath.resolve("swt");
try {
Files.createDirectories(dbfFolder);
Files.createDirectories(xmlFolder);
Files.createDirectories(swtFolder);
} catch (IOException e) {
log.info("Can't create directory, error={}", e.getMessage());
requestInfoUpdate.setStatus(Status.Error);
requestInfoUpdate.setMessage("Can't create directory, error = " + e.getMessage());
return;
}
for (Path directory : directoriesToCopy) {
if (!Files.exists(directory) || !Files.isDirectory(directory)) {
log.info("{} is not a directory or does not exist.", directory.toAbsolutePath());
requestInfoUpdate.setStatus(Status.Error);
requestInfoUpdate.setMessage(directory.toAbsolutePath() + " is not a directory or does not exist.");
return;
}
List<ExportedSDFFile> exportedFiles = this.fileStore.getFiles().stream()
.filter(f -> f.getStatus().equals(Status.Success))
.toList();
Path target;
for (ExportedSDFFile file : exportedFiles) {
try {
target = Files.createDirectories(this.currentPath.resolve(directory.getParent().getFileName().toString()).resolve(directory.getFileName().toString()));
if (file.getFileType().equals(FileType.DBF) && new File(file.getPath()).exists()) {
log.info("Copying from {} to {}", file.getPath(), dbfFolder.resolve(file.getFilename()));
Files.copy(Paths.get(file.getPath()), dbfFolder.resolve(file.getFilename()));
} else if (file.getFileType().equals(FileType.XML) && new File(file.getPath()).exists()) {
log.info("Copying from {} to {}", file.getPath(), xmlFolder.resolve(file.getFilename()));
Files.copy(Paths.get(file.getPath()), xmlFolder.resolve(file.getFilename()));
} else if (file.getFileType().equals(FileType.SWT) && new File(file.getPath()).exists()) {
log.info("Copying from {} to {}", file.getPath(), swtFolder.resolve(file.getFilename()));
Files.copy(Paths.get(file.getPath()), swtFolder.resolve(file.getFilename()));
} else {
log.info("File {} have wrong type {}", file.getPath(), file.getFileType());
requestInfoUpdate.setStatus(Status.Error);
requestInfoUpdate.setMessage("File " + file.getPath() + "have wrong type " + file.getFileType());
return;
}
} catch (IOException e) {
log.info("Can't create directory, error={}", e.getMessage());
log.info("Can't copy file {}, error={}", file.getPath(), e.getMessage());
requestInfoUpdate.setStatus(Status.Error);
requestInfoUpdate.setMessage("Can't create directory, error = " + e.getMessage());
return;
}
try {
log.info("Copying {} to {}", directory.toAbsolutePath(), target.toAbsolutePath());
FileUtils.copyDirectory(directory.toFile(), target.toFile());
} catch (IOException e) {
log.info("Can't copy directory {}, error={}", directory, e.getMessage());
requestInfoUpdate.setStatus(Status.Error);
requestInfoUpdate.setMessage("Can't copy directory " + directory + ", error = " + e.getMessage());
requestInfoUpdate.setMessage("Can't copy file " + file.getPath() + ", error = " + e.getMessage());
return;
}
}

39
clearing-parent/snapshot-maker/src/main/java/ru/spcex/clearing/snapshot/maker/utils/FileStore.java Normal file
View file

@ -0,0 +1,39 @@
package ru.spcex.clearing.snapshot.maker.utils;
import java.util.ArrayList;
import java.util.List;
import org.springframework.stereotype.Component;
import ru.spcex.clearing.platform.messaging.domain.cud.utilities.ExportedSDFFile;
@Component
public class FileStore {
private final List<ExportedSDFFile> files;
private boolean isSessionStarted;
public FileStore() {
this.files = new ArrayList<>();
this.isSessionStarted = false;
}
public List<ExportedSDFFile> getFiles() {
return files;
}
public void addFile(ExportedSDFFile file) {
if (isSessionStarted) {
files.add(file);
}
}
public void clearFiles() {
files.clear();
}
public boolean isSessionStarted() {
return isSessionStarted;
}
public void setSessionStarted(boolean sessionStarted) {
isSessionStarted = sessionStarted;
}
}

8
clearing-parent/snapshot-maker/src/main/resources/application.properties
View file

@ -4,13 +4,7 @@ snapshot-maker.database.username=clearing
snapshot-maker.database.password=Aa111111
# Locations setting
snapshot-maker.location.local-store=/mnt/c/Users/ivan/Desktop/snapshot-test/out
snapshot-maker.location.dbf-importer-folder=/mnt/c/Users/ivan/Desktop/snapshot-test/in/dbf/importer
snapshot-maker.location.dbf-exporter-folder=/mnt/c/Users/ivan/Desktop/snapshot-test/in/dbf/exporter
snapshot-maker.location.swt-importer-folder=/mnt/c/Users/ivan/Desktop/snapshot-test/in/swt/importer
snapshot-maker.location.swt-exporter-folder=/mnt/c/Users/ivan/Desktop/snapshot-test/in/swt/exporter
snapshot-maker.location.xml-importer-folder=/mnt/c/Users/ivan/Desktop/snapshot-test/in/xml/importer
snapshot-maker.location.xml-exporter-folder=/mnt/c/Users/ivan/Desktop/snapshot-test/in/xml/exporter
snapshot-maker.store-location=/mnt/c/Users/ivan/Desktop/snapshot-test/out
# Kafka producer settings
snapshot-maker.kafka-producer.bootstrap-servers=localhost:9092

51
clearing-parent/swt-exporter/src/main/java/ru/spcex/clearing/swt/exporter/services/AbstractExporterService.java
View file

@ -1,22 +1,10 @@
package ru.spcex.clearing.swt.exporter.services;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import ru.clearing.classes.statics.data.misc.Session;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
import ru.spcex.clearing.platform.messaging.domain.cud.importexport.SwtExporterRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.utilities.JournalEventExportedRequest;
import ru.spcex.clearing.platform.messaging.serialization.LogFormatter;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.clearing.swt.exporter.util.ConvertionContext;
import ru.spcex.clearing.swt.exporter.util.SWTHeaderData;
import ru.spcex.platform.classes.base.SpcexObjectBase;
import ru.spcex.platform.enumeration.ResultStatuses;
import ru.spcex.platform.enumeration.SwtTable;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgProvider;
import static ru.spcex.clearing.platform.messaging.domain.Consts.FILE_CREATED;
import static ru.spcex.clearing.platform.messaging.domain.Consts.JOURNAL_SERVICE;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.nio.charset.Charset;
@ -25,8 +13,24 @@ import java.time.format.DateTimeFormatter;
import java.util.Collection;
import java.util.Objects;
import java.util.Optional;
import static ru.spcex.clearing.platform.messaging.domain.Consts.JOURNAL_SERVICE;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import ru.clearing.classes.statics.data.misc.Session;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
import ru.spcex.clearing.platform.messaging.domain.cud.importexport.SwtExporterRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.utilities.ExportedSDFFile;
import ru.spcex.clearing.platform.messaging.domain.cud.utilities.JournalEventExportedRequest;
import ru.spcex.clearing.platform.messaging.serialization.LogFormatter;
import ru.spcex.clearing.platform.messaging.service.FileType;
import ru.spcex.clearing.platform.messaging.service.Status;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.clearing.swt.exporter.util.ConvertionContext;
import ru.spcex.clearing.swt.exporter.util.SWTHeaderData;
import ru.spcex.platform.classes.base.SpcexObjectBase;
import ru.spcex.platform.enumeration.ResultStatuses;
import ru.spcex.platform.enumeration.SwtTable;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgProvider;
public abstract class AbstractExporterService<T extends SpcexObjectBase> {
protected final Logger log = LoggerFactory.getLogger(getClass());
@ -83,7 +87,8 @@ public abstract class AbstractExporterService<T extends SpcexObjectBase> {
makeSWTData(swtHeaderData, records, outBuffer);
data = outBuffer.toByteArray();
}
fileStorage.saveFile(fileName, data);
File createdFile = fileStorage.saveFile(fileName, data);
sendFileCreatedNotification(createdFile);
} catch (Exception e) { // IOException, ...
log.error("Failed export {} file", fileName);
sendSwtExportedNotification(exportAt, null, ResultStatuses.notSuccess);
@ -105,6 +110,16 @@ public abstract class AbstractExporterService<T extends SpcexObjectBase> {
protected abstract String getDocumentNameForJournal();
void sendFileCreatedNotification(File file) {
ExportedSDFFile exportedSDFFile = new ExportedSDFFile();
exportedSDFFile.setStatus(Status.Success);
exportedSDFFile.setFileType(FileType.SWT);
exportedSDFFile.setFilename(file.getName());
exportedSDFFile.setPath(file.getAbsolutePath());
log.debug("Send message to kafka \"{}\": {}", FILE_CREATED, LogFormatter.toStringWrapper(exportedSDFFile));
kafkaSender.sendRequestToQueue(FILE_CREATED, exportedSDFFile);
}
void sendSwtExportedNotification(LocalDateTime registrationAt, Long registrationNumber, ResultStatuses resultStatus) {
JournalEventExportedRequest exportedRequest = new JournalEventExportedRequest();
exportedRequest.setRegistratoinDate(registrationAt.toLocalDate());

3
clearing-parent/swt-exporter/src/main/java/ru/spcex/clearing/swt/exporter/services/FileStorage.java
View file

@ -37,7 +37,7 @@ public class FileStorage {
this.gateway = gateway;
}
public void saveFile(String fileName, byte[] data) throws IOException {
public File saveFile(String fileName, byte[] data) throws IOException {
File toFile = new File(outPath, fileName);
FileUtils.writeByteArrayToFile(toFile, data);
if (gateway == null) {
@ -46,5 +46,6 @@ public class FileStorage {
log.debug("sftp is enabled. sending {}", toFile.getAbsolutePath());
gateway.sendToSftp(toFile);
}
return toFile;
}
}

1
platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/Task.java
View file

@ -61,6 +61,7 @@ public enum Task implements IEnumKey {
makeFiles_MTCR("MTCR"), // Формирование файлов с МТКР
startSessionSnapshot("STSS"), // Старт формирования снэпшота сессии
stopSessionSnapshot("SPSS"), // Завершение формирования снэпшота сессии
fileCreated("FCRD"), // SDF файл создан
;
private final String key;

1
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java
View file

@ -161,6 +161,7 @@ public interface Consts {
String BALANCE_ACCOUNT_UPDATE = "balance-account-update";
String CONTINUE_CLEARING = "continue-clearing";
String LAUNCHER_NEW = "launcher-new";
String FILE_CREATED = "file-created";
String DESTINATION_DEPO_ACCOUNT_SYMBOLS_NEW = "depo-accounts-symbols-new";
String DESTINATION_DEPO_ACCOUNT_SYMBOLS_DELETE = "depo-accounts-symbols-delete";

58
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/utilities/ExportedSDFFile.java Normal file
View file

@ -0,0 +1,58 @@
package ru.spcex.clearing.platform.messaging.domain.cud.utilities;
import com.fasterxml.jackson.annotation.JsonProperty;
import ru.spcex.clearing.platform.messaging.service.FileType;
import ru.spcex.clearing.platform.messaging.service.Status;
public class ExportedSDFFile {
@JsonProperty
private Status status;
@JsonProperty
private FileType fileType;
@JsonProperty
private String filename;
@JsonProperty
private String path;
public Status getStatus() {
return status;
}
public void setStatus(Status status) {
this.status = status;
}
public FileType getFileType() {
return fileType;
}
public void setFileType(FileType fileType) {
this.fileType = fileType;
}
public String getFilename() {
return filename;
}
public void setFilename(String filename) {
this.filename = filename;
}
public String getPath() {
return path;
}
public void setPath(String path) {
this.path = path;
}
@Override
public String toString() {
return "ExportedSDFFile{" +
"status=" + status +
", fileType=" + fileType +
", filename='" + filename + '\'' +
", path='" + path + '\'' +
'}';
}
}

7
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/FileType.java Normal file
View file

@ -0,0 +1,7 @@
package ru.spcex.clearing.platform.messaging.service;
public enum FileType {
DBF,
SWT,
XML
}