This commit is contained in:
@@ -161,6 +161,87 @@ public class FileUploadServiceImpl implements FileUploadService {
|
||||
return processChunk(session, chunkNumber, chunkFile);
|
||||
}
|
||||
|
||||
@Override
|
||||
@Transactional
|
||||
public void handleExpiredSession(FileUploadSession session) {
|
||||
session.setStatus(UploadStatus.FAILED);
|
||||
session.setLastError("Upload session expired");
|
||||
sessionRepository.save(session);
|
||||
|
||||
CompletableFuture.runAsync(() -> cleanupSessionFiles(session));
|
||||
}
|
||||
|
||||
@Async("fileUploadTaskExecutor")
|
||||
@Transactional
|
||||
public void assembleFileAsync(FileUploadSession session) {
|
||||
try {
|
||||
assembleFile(session);
|
||||
log.info("File assembly completed successfully for session: {}",
|
||||
session.getUploadId());
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("Failed to assemble file for session {}: {}",
|
||||
session.getUploadId(), e.getMessage(), e);
|
||||
|
||||
handleAssemblyFailure(session, e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
@Transactional
|
||||
public void cancelUpload(String uploadId) {
|
||||
FileUploadSession session = sessionRepository.findById(uploadId)
|
||||
.orElseThrow(() -> new UploadSessionNotFoundException(uploadId));
|
||||
|
||||
if (session.getStatus() == UploadStatus.COMPLETED || session.getStatus() == UploadStatus.CANCELLED) {
|
||||
throw new FileUploadException("Cannot cancel completed or cancelled upload");
|
||||
}
|
||||
|
||||
session.setStatus(UploadStatus.CANCELLED);
|
||||
sessionRepository.save(session);
|
||||
|
||||
CompletableFuture.runAsync(() -> cleanupSessionFiles(session));
|
||||
|
||||
log.info("Upload cancelled: {}", uploadId);
|
||||
}
|
||||
|
||||
@Override
|
||||
@Transactional
|
||||
public void retryFailedUpload(String uploadId) {
|
||||
FileUploadSession session = sessionRepository.findById(uploadId)
|
||||
.orElseThrow(() -> new UploadSessionNotFoundException(uploadId));
|
||||
|
||||
if (session.getStatus() != UploadStatus.FAILED) {
|
||||
throw new FileUploadException("Only failed uploads can be retried");
|
||||
}
|
||||
|
||||
if (session.getRetryCount() >= maxRetryAttempts) {
|
||||
throw new FileUploadException("Max retry attempts exceeded");
|
||||
}
|
||||
|
||||
session.setStatus(UploadStatus.UPLOADING);
|
||||
session.setRetryCount(session.getRetryCount() + 1);
|
||||
session.setLastError(null);
|
||||
|
||||
sessionRepository.save(session);
|
||||
|
||||
if (session.getChunksUploaded().equals(session.getTotalChunks())) {
|
||||
log.info("Retrying file assembly for session: {}", uploadId);
|
||||
assembleFileAsync(session);
|
||||
}
|
||||
|
||||
log.info("Upload retry initiated for session: {} (attempt {})",
|
||||
uploadId, session.getRetryCount());
|
||||
}
|
||||
|
||||
@Override
|
||||
public UploadProgressResponse getUploadProgress(String uploadId) {
|
||||
FileUploadSession session = sessionRepository.findById(uploadId)
|
||||
.orElseThrow(() -> new UploadSessionNotFoundException(uploadId));
|
||||
|
||||
return UploadProgressResponse.fromSession(session);
|
||||
}
|
||||
|
||||
private void validateSession(FileUploadSession session) {
|
||||
UploadStatus status = session.getStatus();
|
||||
|
||||
@@ -190,14 +271,6 @@ public class FileUploadServiceImpl implements FileUploadService {
|
||||
}
|
||||
}
|
||||
|
||||
private void handleExpiredSession(FileUploadSession session) {
|
||||
session.setStatus(UploadStatus.FAILED);
|
||||
session.setLastError("Upload session expired");
|
||||
sessionRepository.save(session);
|
||||
|
||||
CompletableFuture.runAsync(() -> cleanupSessionFiles(session));
|
||||
}
|
||||
|
||||
private UploadProgressResponse processChunk(FileUploadSession session,
|
||||
Integer chunkNumber,
|
||||
MultipartFile chunkFile) {
|
||||
@@ -213,6 +286,7 @@ public class FileUploadServiceImpl implements FileUploadService {
|
||||
session.getChunkPaths().put(chunkNumber, chunkPath);
|
||||
session.setChunksUploaded(session.getChunksUploaded() + 1);
|
||||
session.setStatus(UploadStatus.UPLOADING);
|
||||
session.setExpiresAt(LocalDateTime.now().plusMinutes(1));
|
||||
|
||||
log.debug("Chunk {} saved successfully. Uploaded: {}/{}",
|
||||
chunkNumber, session.getChunksUploaded(), session.getTotalChunks());
|
||||
@@ -323,22 +397,6 @@ public class FileUploadServiceImpl implements FileUploadService {
|
||||
}
|
||||
}
|
||||
|
||||
@Async("fileUploadTaskExecutor")
|
||||
@Transactional
|
||||
public void assembleFileAsync(FileUploadSession session) {
|
||||
try {
|
||||
assembleFile(session);
|
||||
log.info("File assembly completed successfully for session: {}",
|
||||
session.getUploadId());
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("Failed to assemble file for session {}: {}",
|
||||
session.getUploadId(), e.getMessage(), e);
|
||||
|
||||
handleAssemblyFailure(session, e);
|
||||
}
|
||||
}
|
||||
|
||||
private void assembleFile(FileUploadSession session) throws IOException {
|
||||
log.info("Starting file assembly for session: {} ({})",
|
||||
session.getUploadId(), session.getFileName());
|
||||
@@ -503,14 +561,6 @@ public class FileUploadServiceImpl implements FileUploadService {
|
||||
log.debug("Final file checksum: {}", checksum);
|
||||
}
|
||||
|
||||
private void updateSessionOnSuccess(FileUploadSession session, Path finalPath) {
|
||||
session.setFilePath(finalPath.toString());
|
||||
session.setChecksum(calculateChecksum(finalPath));
|
||||
session.setStatus(UploadStatus.COMPLETED);
|
||||
session.setCompletedAt(LocalDateTime.now());
|
||||
sessionRepository.save(session);
|
||||
}
|
||||
|
||||
private void cleanupSessionFiles(FileUploadSession session) {
|
||||
try {
|
||||
Path chunkDir = getChunkDirectory(session.getUserId(), session.getUploadId());
|
||||
@@ -547,90 +597,37 @@ public class FileUploadServiceImpl implements FileUploadService {
|
||||
session.getUploadId(), e.getMessage());
|
||||
}
|
||||
|
||||
@Override
|
||||
@Transactional
|
||||
public void cancelUpload(String uploadId) {
|
||||
FileUploadSession session = sessionRepository.findById(uploadId)
|
||||
.orElseThrow(() -> new UploadSessionNotFoundException(uploadId));
|
||||
|
||||
if (session.getStatus() == UploadStatus.COMPLETED || session.getStatus() == UploadStatus.CANCELLED) {
|
||||
throw new FileUploadException("Cannot cancel completed or cancelled upload");
|
||||
}
|
||||
|
||||
session.setStatus(UploadStatus.CANCELLED);
|
||||
sessionRepository.save(session);
|
||||
|
||||
CompletableFuture.runAsync(() -> cleanupSessionFiles(session));
|
||||
|
||||
log.info("Upload cancelled: {}", uploadId);
|
||||
}
|
||||
|
||||
@Override
|
||||
@Transactional
|
||||
public void retryFailedUpload(String uploadId) {
|
||||
FileUploadSession session = sessionRepository.findById(uploadId)
|
||||
.orElseThrow(() -> new UploadSessionNotFoundException(uploadId));
|
||||
|
||||
if (session.getStatus() != UploadStatus.FAILED) {
|
||||
throw new FileUploadException("Only failed uploads can be retried");
|
||||
}
|
||||
|
||||
if (session.getRetryCount() >= maxRetryAttempts) {
|
||||
throw new FileUploadException("Max retry attempts exceeded");
|
||||
}
|
||||
|
||||
session.setStatus(UploadStatus.UPLOADING);
|
||||
session.setRetryCount(session.getRetryCount() + 1);
|
||||
session.setLastError(null);
|
||||
|
||||
sessionRepository.save(session);
|
||||
|
||||
if (session.getChunksUploaded().equals(session.getTotalChunks())) {
|
||||
log.info("Retrying file assembly for session: {}", uploadId);
|
||||
assembleFileAsync(session);
|
||||
}
|
||||
|
||||
log.info("Upload retry initiated for session: {} (attempt {})",
|
||||
uploadId, session.getRetryCount());
|
||||
}
|
||||
|
||||
@Override
|
||||
public UploadProgressResponse getUploadProgress(String uploadId) {
|
||||
FileUploadSession session = sessionRepository.findById(uploadId)
|
||||
.orElseThrow(() -> new UploadSessionNotFoundException(uploadId));
|
||||
|
||||
return UploadProgressResponse.fromSession(session);
|
||||
}
|
||||
|
||||
@Override
|
||||
@Transactional
|
||||
public void cleanupExpiredSessions() {
|
||||
log.info("Starting cleanup of expired upload sessions");
|
||||
|
||||
LocalDateTime expiryThreshold = LocalDateTime.now().minusHours(sessionExpiryHours);
|
||||
|
||||
try {
|
||||
var expiredSessions = sessionRepository.findByStatusInAndCreatedAtBefore(
|
||||
Set.of(UploadStatus.INITIATED, UploadStatus.UPLOADING, UploadStatus.FAILED),
|
||||
expiryThreshold
|
||||
);
|
||||
|
||||
int cleanedCount = 0;
|
||||
for (FileUploadSession session : expiredSessions) {
|
||||
session.setStatus(UploadStatus.FAILED);
|
||||
session.setLastError("Session expired during cleanup");
|
||||
sessionRepository.save(session);
|
||||
|
||||
CompletableFuture.runAsync(() -> cleanupSessionFiles(session));
|
||||
cleanedCount++;
|
||||
}
|
||||
|
||||
log.info("Cleaned up {} expired upload sessions", cleanedCount);
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("Error during session cleanup", e);
|
||||
}
|
||||
}
|
||||
// @Override
|
||||
// @Transactional
|
||||
// public void cleanupExpiredSessions() {
|
||||
// log.info("Starting cleanup of expired upload sessions");
|
||||
//
|
||||
// LocalDateTime expiryThreshold = LocalDateTime.now().minusHours(sessionExpiryHours);
|
||||
//
|
||||
// try {
|
||||
// var expiredSessions = sessionRepository.findByStatusInAndCreatedAtBefore(
|
||||
// Set.of(UploadStatus.INITIATED, UploadStatus.UPLOADING, UploadStatus.FAILED),
|
||||
// expiryThreshold
|
||||
// );
|
||||
//
|
||||
// int cleanedCount = 0;
|
||||
// for (FileUploadSession session : expiredSessions) {
|
||||
// session.setStatus(UploadStatus.FAILED);
|
||||
// session.setLastError("Session expired during cleanup");
|
||||
// sessionRepository.save(session);
|
||||
//
|
||||
// CompletableFuture.runAsync(() -> cleanupSessionFiles(session));
|
||||
// cleanedCount++;
|
||||
// }
|
||||
//
|
||||
// log.info("Cleaned up {} expired upload sessions", cleanedCount);
|
||||
//
|
||||
// } catch (Exception e) {
|
||||
// log.error("Error during session cleanup", e);
|
||||
// }
|
||||
// }
|
||||
|
||||
private Path getChunkDirectory(Long userId, String uploadId) {
|
||||
return storageRoot.resolve("temp")
|
||||
|
||||
Reference in New Issue
Block a user