BFLOWS/bflows-bandi-be
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added async loading for document uploading to external system.

Browse Source
This commit is contained in:
piyushkag
2024-12-10 12:47:11 +05:30
parent 05f64af404
commit 2a5f344ea0
9 changed files with 100 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

117
src/main/java/net/gepafin/tendermanagement/dao/AppointmentDao.java
View File

@@ -5,9 +5,11 @@ import com.amazonaws.services.s3.model.GetObjectRequest;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import feign.FeignException;
import io.jsonwebtoken.Claims;
import jakarta.servlet.http.HttpServletRequest;
import lombok.extern.slf4j.Slf4j;
import net.gepafin.tendermanagement.config.Translator;
import net.gepafin.tendermanagement.config.jwt.TokenProvider;
import net.gepafin.tendermanagement.constants.AppointmentApiConstant;
import net.gepafin.tendermanagement.constants.GepafinConstant;
import net.gepafin.tendermanagement.entities.ApplicationEntity;
@@ -125,8 +127,15 @@ public class AppointmentDao {
@Autowired
private LoggingUtil loggingUtil;
@Autowired
private TokenProvider tokenProvider;
private final Map<Long, ExecutorService> executorMap = new ConcurrentHashMap<>();
private final ConcurrentHashMap<Long, ExecutorService> threadForDocumentMap = new ConcurrentHashMap<>();
private static final ThreadLocal<Long> threadLocalHubId = new ThreadLocal<>();
public NdgResponse checkNdgForAppointment(Long applicationId) {
ApplicationEntity application = applicationService.validateApplication(applicationId);
@@ -719,87 +728,101 @@ public class AppointmentDao {
return appointmentCreationRequest;
}
public DocumentUploadResponse uploadDocumentToExternalSystem(Long documentId, UploadDocToExternalSystemRequest docToExternalSystemRequest, Long applicationId) {
DocumentUploadResponse response = new DocumentUploadResponse();
public DocumentUploadResponse uploadDocumentToExternalSystem(Long documentId, UploadDocToExternalSystemRequest docToExternalSystemRequest) {
// Check if the document is already being processed
DocumentEntity systemDoc = documentDao.validateDocument(documentId);
ApplicationEntity application = getApplicationEntityForDocument(applicationId, systemDoc);
//cloned for old document data
DocumentEntity oldDocumentEntity = Utils.getClonedEntityForData(systemDoc);
if (!docToExternalSystemRequest.getInput().getAttributes().getNdg().equalsIgnoreCase(application.getNdg())) {
throw new CustomValidationException(Status.BAD_REQUEST, Translator.toLocale(GepafinConstant.NDG_NOT_MATCHED_OR_NOT_FOUND));
Claims claims = tokenProvider.getClaimsFromToken(tokenProvider.extractTokenFromRequest(request));
Long hubId = Utils.extractHubIdFromPayload(claims.getSubject());
if (systemDoc.getDocumentAttachmentId() != null) {
// If the documentAttachmentId is already set, return the response
log.info("Document already uploaded with documentAttachmentId: {}", systemDoc.getDocumentAttachmentId());
DocumentUploadResponse response = new DocumentUploadResponse();
response.setDocumentAttachmentId(systemDoc.getDocumentAttachmentId());
return response;
}
// Check if a thread is already running for this document upload
if (threadForDocumentMap.containsKey(documentId)) {
log.warn("Document upload already running for documentId: {}", documentId);
throw new CustomValidationException(Status.SUCCESS, Translator.toLocale(GepafinConstant.DOCUMENT_UPLOADING_IN_PROGRESS));
}
// Start the upload process in the background
ExecutorService executor = Executors.newSingleThreadExecutor(runnable -> {
Thread thread = new Thread(runnable);
thread.setName(GepafinConstant.ASYNC_DOCUMENT_UPLOAD_NAME + documentId);
return thread;
});
threadForDocumentMap.put(documentId, executor);
Long hubId = application.getHubId();
executor.submit(() -> {
threadLocalHubId.set(hubId);
try {
log.info("Starting async document upload for documentId: {}", documentId);
uploadDocumentToExternalSystemSync(documentId, docToExternalSystemRequest);
} catch (Exception e) {
log.error("Error in async document upload for documentId: {}", documentId, e);
} finally {
// Cleanup resources
ExecutorService executorToShutdown = threadForDocumentMap.remove(documentId);
if (executorToShutdown != null) {
executorToShutdown.shutdown();
threadLocalHubId.remove();
}
log.info("Async document upload completed for documentId: {}", documentId);
}
});
// Return an immediate response indicating the process is in progress
throw new CustomValidationException(Status.SUCCESS, Translator.toLocale(GepafinConstant.DOCUMENT_UPLOADING_IN_PROGRESS));
}
private void uploadDocumentToExternalSystemSync(Long documentId, UploadDocToExternalSystemRequest docToExternalSystemRequest) {
// Synchronous upload logic
DocumentEntity systemDoc = documentDao.validateDocument(documentId);
Long hubId = threadLocalHubId.get();
HubEntity hub = hubRepository.findByHubId(hubId);
if (!hub.getUniqueUuid().equals(defaultHubUuid)) {
log.info("Document cannot be uploaded for another Hub, it is default for Gepafin.");
throw new CustomValidationException(Status.BAD_REQUEST, Translator.toLocale(GepafinConstant.NO_DOCUMENT_UPLOAD_FOR_ANOTHER_HUB));
}
log.info("Got Document in system {}", systemDoc);
log.info("Got Document in system: {}", systemDoc);
String oldUrl = systemDoc.getFilePath();
log.info("Processing {}", oldUrl);
String authorizationToken = getBearerToken(hub);
try {
File localFile = downloadFileFromS3(oldUrl);
MultipartFile multipartFile = convertFileToMultipartFile(localFile);
UploadDocToExternalSystemRequest externalSystemRequest = new UploadDocToExternalSystemRequest();
UploadDocToExternalSystemRequest.Input input = getUploadDocumentInput(docToExternalSystemRequest);
externalSystemRequest.setInput(input);
externalSystemRequest.setInput(getUploadDocumentInput(docToExternalSystemRequest));
String uploadDocRequest = Utils.convertObjectToJson(externalSystemRequest);
ResponseEntity<Object> uploadedDocumentData = appointmentApiService.uploadDocumentToExternalSystemForAppointment(authorizationToken, context, uploadDocRequest,
multipartFile);
String responseData = Utils.convertObjectToJson(uploadedDocumentData.getBody());
DocumentUploadResponse parsedDocumentUploadResponse = parseDocumentUploadResponse(responseData);
if (parsedDocumentUploadResponse == null) {
String responseData = Utils.convertObjectToJson(uploadedDocumentData.getBody());
DocumentUploadResponse parsedResponse = parseDocumentUploadResponse(responseData);
if (parsedResponse == null) {
throw new CustomValidationException(Status.BAD_REQUEST, Translator.toLocale(GepafinConstant.ERROR_UPLOADING_DOCUMENT));
}
systemDoc.setDocumentAttachmentId(parsedDocumentUploadResponse.getDocumentAttachmentId());
// Save the documentAttachmentId to the database
systemDoc.setDocumentAttachmentId(parsedResponse.getDocumentAttachmentId());
documentRepository.save(systemDoc);
/** This code is responsible for adding a version history log for the "Update document with document attachment id" operation. **/
loggingUtil.addVersionHistory(
VersionHistoryRequest.builder().request(request).actionType(VersionActionTypeEnum.UPDATE).oldData(oldDocumentEntity).newData(systemDoc).build());
log.info("Document uploaded successfully to external system : {}", parsedDocumentUploadResponse);
response.setDocumentAttachmentId(systemDoc.getDocumentAttachmentId());
return response;
log.info("Document uploaded successfully to external system: {}", parsedResponse);
} catch (FeignException.Forbidden forbiddenException) {
log.error("403 Forbidden received while uploading document to external system. Regenerating token...");
log.error("403 Forbidden received while uploading document. Regenerating token...");
regenerateTokenAndSave(hub);
return uploadDocumentToExternalSystem(systemDoc.getSourceId(), docToExternalSystemRequest, applicationId);
uploadDocumentToExternalSystemSync(documentId, docToExternalSystemRequest);
} catch (Exception e) {
log.error("Exception in uploading document to external system {}", e.getMessage());
log.error("Exception during document upload: {}", e.getMessage(), e);
throw new CustomValidationException(Status.BAD_REQUEST, Translator.toLocale(GepafinConstant.EXTERNAL_DOCUMENT_UPLOAD_FAILURE_MSG));
}
}
private ApplicationEntity getApplicationEntityForDocument(Long applicationId, DocumentEntity systemDoc) {
if (systemDoc.getDocumentAttachmentId() != null) {
throw new CustomValidationException(Status.BAD_REQUEST, Translator.toLocale(GepafinConstant.DOCUMENT_ALREADY_UPLOADED));
}
ApplicationEntity application;
if (systemDoc.getSource().equalsIgnoreCase(DocumentSourceTypeEnum.APPLICATION.getValue()) && Objects.equals(systemDoc.getSourceId(), applicationId)) {
application = applicationService.validateApplication(systemDoc.getSourceId());
} else {
throw new CustomValidationException(Status.BAD_REQUEST, Translator.toLocale(GepafinConstant.PROVIDE_VALID_APPLICATION_DOC_ID));
}
return application;
}
private UploadDocToExternalSystemRequest.Input getUploadDocumentInput(UploadDocToExternalSystemRequest docToExternalSystemRequest) {