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

dbf-importer swt-importer platform-enum http://jira.mfd.msk:8088/browse/CLS-587 Нотификейшены по sDF загрузке

Browse source
This commit is contained in:
AKurakin 2024-04-15 19:03:28 +03:00
parent 9d0e092d70
commit cbad15aab9
9 changed files with 104 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/service/ClearingAccountService.java
View file

@ -550,7 +550,7 @@ public class ClearingAccountService extends QueueConsumer implements Initializin
final String destination = Consts.NOTIFICATION_NEW;
NotificationNewRequest request = new NotificationNewRequest();
//request.setObjectId(account.getId());
request.setObjectType(ObjectType.registry.getKey());
request.setObjectType(ObjectType.rgst.getKey());
request.setPriority(Priority.HIGH.getKey());
request.setComment(message);
log.debug("Send message to kafka \"{}\": {}", destination, LogFormatter.toStringWrapper(request));

2
clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/service/DepoAccountService.java
View file

@ -254,7 +254,7 @@ public class DepoAccountService extends QueueConsumer implements InitializingBea
final String destination = Consts.NOTIFICATION_NEW;
NotificationNewRequest request = new NotificationNewRequest();
//request.setObjectId(account.getId());
request.setObjectType(ObjectType.registry.getKey());
request.setObjectType(ObjectType.rgst.getKey());
request.setPriority(Priority.HIGH.getKey());
request.setComment(message);
log.debug("Send message to kafka \"{}\": {}", destination, LogFormatter.toStringWrapper(request));

40
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/DbfImportKafkaMessenger.java
View file

@ -4,11 +4,17 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;
import ru.spcex.clearing.dbf.importer.logic.data.ResultContainer;
import ru.spcex.clearing.dbf.importer.logic.data.enums.ETable;
import ru.spcex.clearing.dbf.importer.logic.data.enums.StageResult;
import ru.spcex.clearing.platform.messaging.domain.Consts;
import ru.spcex.clearing.platform.messaging.domain.cud.balance.StatementRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.clearing.Sdf04Request;
import ru.spcex.clearing.platform.messaging.domain.cud.utilities.NotificationNewRequest;
import ru.spcex.clearing.platform.messaging.serialization.LogFormatter;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.platform.enumeration.ObjectType;
import ru.spcex.platform.enumeration.Priority;
import ru.spcex.platform.enumeration.SdfTable;
import java.util.HashMap;
@ -50,6 +56,40 @@ public class DbfImportKafkaMessenger implements InitializingBean {
}
}
public void notifyUserAboutErrorParsing(Throwable error, ResultContainer resultContainer) {
if (resultContainer == null || resultContainer.getDbfFile() == null || resultContainer.getDbfFile() == null)
return;
ETable currTable = resultContainer.getDbfTable();
String fileName = resultContainer.getDbfFile().getName();
log.debug("Notify user about error {} \"{}\"",
resultContainer.getDbfTable(), fileName);
sendUserNotification(ObjectType.rgst, String.format("Файл \"%s\" не сохранен", fileName), Priority.HIGH);
}
public void notifyUserAboutSuccessLoad(ResultContainer resultContainer) {
if (ETable.DF_04 == resultContainer.getDbfTable() || ETable.DF_09 == resultContainer.getDbfTable()) {
if (!(StageResult.OK == resultContainer.getLastStageStatus() || StageResult.COMPLETE == resultContainer.getLastStageStatus())) {
return;
}
String fileName = resultContainer.getDbfFile().getName();
log.debug("Notify user about success {} \"{}\"",
resultContainer.getDbfTable(), fileName);
sendUserNotification(ObjectType.rgst, String.format("Загружен \"%s\" - успешно", fileName), Priority.LOW);
}
}
private void sendUserNotification(ObjectType objectType, String comment, Priority priority) {
final String destination = Consts.NOTIFICATION_NEW;
NotificationNewRequest request = new NotificationNewRequest();
//request.setObjectId();
request.setObjectType(objectType.getKey());
request.setPriority(priority.getKey());
request.setComment(comment);
log.debug("Send message to kafka \"{}\": {}", destination, LogFormatter.toStringWrapper(request));
Long rKey = kafka.get().sendRequestToQueue(destination, request);
log.trace("For user send message, request id={}", rKey);
}
private void messageStatement(Long groupId, SdfTable table) {
messageStatement(groupId, table, Consts.STATEMENT_PROCESS);
}

3
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/ImportToDB.java
View file

@ -59,11 +59,14 @@ public class ImportToDB extends Stage {
table.injectEntity(table.getEntity(entity));
}
kafkaMessenger.notifySystemIfNeeded(currTable, fileId);
kafkaMessenger.notifyUserAboutSuccessLoad(resultContainer);
} catch (IOException exception) {
log.warn(exception.getMessage());
kafkaMessenger.notifyUserAboutErrorParsing(exception, resultContainer);
return StageResult.ERROR;
} catch (Throwable e) {
log.error(ExceptionUtils.getStackTrace(e));
kafkaMessenger.notifyUserAboutErrorParsing(e, resultContainer);
return StageResult.ERROR;
}

10
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/LoadFileFromDisk.java
View file

@ -1,5 +1,6 @@
package ru.spcex.clearing.dbf.importer.logic.stages;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import ru.spcex.clearing.dbf.importer.config.settings.ImportDBFServiceSettings;
import ru.spcex.clearing.dbf.importer.logic.data.ResultContainer;
@ -16,8 +17,12 @@ import java.util.UUID;
public class LoadFileFromDisk extends Stage {
private final ImportDBFServiceSettings settings;
public LoadFileFromDisk(ImportDBFServiceSettings settings) {
private final DbfImportKafkaMessenger kafkaMessenger;
@Autowired
public LoadFileFromDisk(ImportDBFServiceSettings settings, DbfImportKafkaMessenger kafkaMessenger) {
this.settings = settings;
this.kafkaMessenger = kafkaMessenger;
}
@Override
@ -34,6 +39,9 @@ public class LoadFileFromDisk extends Stage {
if (settings.getStore().isDeleteSrcFiles()) Files.delete(dbfFile.toPath());
} catch (IOException e) {
log.error(String.format("uuid %s. Can't read file %s", taskUuid, resultContainer.getDbfFile().getName()), e);
if (kafkaMessenger != null) {
kafkaMessenger.notifyUserAboutErrorParsing(e, resultContainer);
}
return StageResult.ERROR;
}
log.debug("uuid {}. Read all bytes from source file complete", taskUuid);

3
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/ImportToDB.java
View file

@ -74,11 +74,14 @@ public class ImportToDB extends Stage {
});
} while (swtReader.hasNextRecord());
kafkaMessenger.notifySystemIfNeeded(currTable, fileId);
kafkaMessenger.notifyUserAboutSuccessLoad(resultContainer);
} catch (IOException exception) {
log.warn(exception.getMessage());
kafkaMessenger.notifyUserAboutErrorParsing(exception, resultContainer);
return StageResult.ERROR;
} catch (Throwable e) {
log.error(ExceptionUtils.getStackTrace(e));
kafkaMessenger.notifyUserAboutErrorParsing(e, resultContainer);
return StageResult.ERROR;
}

7
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/LoadFileFromDisk.java
View file

@ -15,9 +15,11 @@ import java.util.UUID;
@Component
public class LoadFileFromDisk extends Stage {
private final ImportSWTServiceSettings settings;
private final SWTImportKafkaMessenger kafkaMessenger;
public LoadFileFromDisk(ImportSWTServiceSettings settings) {
public LoadFileFromDisk(ImportSWTServiceSettings settings, SWTImportKafkaMessenger kafkaMessenger) {
this.settings = settings;
this.kafkaMessenger = kafkaMessenger;
}
@Override
@ -34,6 +36,9 @@ public class LoadFileFromDisk extends Stage {
if (settings.getStore().isDeleteSrcFiles()) Files.delete(swtFile.toPath());
} catch (IOException e) {
log.error(String.format("uuid %s. Can't read file %s", taskUuid, resultContainer.getSwtFile().getName()), e);
if (kafkaMessenger != null) {
kafkaMessenger.notifyUserAboutErrorParsing(e, resultContainer);
}
return StageResult.ERROR;
}
log.debug("uuid {}. Read all bytes from source file complete", taskUuid);

40
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/SWTImportKafkaMessenger.java
View file

@ -6,8 +6,14 @@ import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;
import ru.spcex.clearing.platform.messaging.domain.Consts;
import ru.spcex.clearing.platform.messaging.domain.cud.balance.StatementRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.utilities.NotificationNewRequest;
import ru.spcex.clearing.platform.messaging.serialization.LogFormatter;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.clearing.swt.importer.logic.data.ResultContainer;
import ru.spcex.clearing.swt.importer.logic.data.enums.ETable;
import ru.spcex.clearing.swt.importer.logic.data.enums.StageResult;
import ru.spcex.platform.enumeration.ObjectType;
import ru.spcex.platform.enumeration.Priority;
import ru.spcex.platform.enumeration.SdfTable;
import java.util.HashMap;
@ -56,4 +62,38 @@ public class SWTImportKafkaMessenger implements InitializingBean {
log.debug("Send message id={}; groupId={}, table={} to kafka \"{}\"", rid, groupId, table, Consts.STATEMENT_PROCESS);
}
public void notifyUserAboutErrorParsing(Throwable error, ResultContainer resultContainer) {
if (resultContainer == null || resultContainer.getSwtFile() == null || resultContainer.getSwtTable() == null)
return;
ETable currTable = resultContainer.getSwtTable();
String fileName = resultContainer.getSwtFile().getName();
log.debug("Notify user about error {} \"{}\"",
resultContainer.getSwtTable(), fileName);
sendUserNotification(ObjectType.rgst, String.format("Файл \"%s\" не сохранен", fileName), Priority.HIGH);
}
public void notifyUserAboutSuccessLoad(ResultContainer resultContainer) {
// if (ETable.S_DF_02 == resultContainer.getSwtTable()) {
// if (!(StageResult.OK == resultContainer.getLastStageStatus() || StageResult.COMPLETE == resultContainer.getLastStageStatus())) {
// return;
// }
// String fileName = resultContainer.getSwtFile().getName();
// log.debug("Notify user about success {} \"{}\"",
// resultContainer.getSwtTable(), fileName);
// sendUserNotification(ObjectType.rgst, String.format("Загружен \"%s\" - успешно", fileName), Priority.LOW);
// }
}
private void sendUserNotification(ObjectType objectType, String comment, Priority priority) {
final String destination = Consts.NOTIFICATION_NEW;
NotificationNewRequest request = new NotificationNewRequest();
//request.setObjectId();
request.setObjectType(objectType.getKey());
request.setPriority(priority.getKey());
request.setComment(comment);
log.debug("Send message to kafka \"{}\": {}", destination, LogFormatter.toStringWrapper(request));
Long rKey = kafka.get().sendRequestToQueue(destination, request);
log.trace("For user send message, request id={}", rKey);
}
}

3
platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/ObjectType.java
View file

@ -4,8 +4,7 @@ import ru.spcex.platform.utils.enumeration.IEnumKey;
public enum ObjectType implements IEnumKey {
statement("STMT"), vfrs("VFRS"), rgst("RGST"), gateway("GTWY"), session("SESN"),
account_block("ACCB"), account_active("ACCA"), diff("DIFF"),
registry("RGST");
account_block("ACCB"), account_active("ACCA"), diff("DIFF");
private final String key;