AbstractFileSystem implements IBackupFileSystem, EventGenerator<BackupEvent> { @Override public void uploadFile( final Path localPath, final Path remotePath, final AbstractBackupPath path, final int retry, final boolean deleteAfterSuccessfulUpload) throws FileNotFoundException, BackupRestoreException { if (localPath == null || remotePath == null || !localPath.toFile().exists() || localPath.toFile().isDirectory()) throw new FileNotFoundException( "File do not exist or is a directory. localPath: " + localPath + ", remotePath: " + remotePath); if (tasksQueued.add(localPath)) { logger.info("Uploading file: {} to location: {}", localPath, remotePath); try { long uploadedFileSize; if (path.getType() != BackupFileType.SST_V2 || !checkObjectExists(remotePath)) { notifyEventStart(new BackupEvent(path)); uploadedFileSize = new BoundedExponentialRetryCallable<Long>(500, 10000, retry) { @Override public Long retriableCall() throws Exception { return uploadFileImpl(localPath, remotePath); } }.call(); if (path.getType() == BackupFileType.SST_V2) addObjectCache(remotePath); backupMetrics.recordUploadRate(uploadedFileSize); backupMetrics.incrementValidUploads(); path.setCompressedFileSize(uploadedFileSize); notifyEventSuccess(new BackupEvent(path)); } else { logger.info("File: {} already present on remoteFileSystem.", remotePath); } logger.info( "Successfully uploaded file: {} to location: {}", localPath, remotePath); if (deleteAfterSuccessfulUpload && !FileUtils.deleteQuietly(localPath.toFile())) logger.warn( String.format( "Failed to delete local file %s.", localPath.toFile().getAbsolutePath())); } catch (Exception e) { backupMetrics.incrementInvalidUploads(); notifyEventFailure(new BackupEvent(path)); logger.error( "Error while uploading file: {} to location: {}. Exception: Msg: [{}], Trace: {}", localPath, remotePath, e.getMessage(), e.getStackTrace()); throw new BackupRestoreException(e.getMessage()); } finally { tasksQueued.remove(localPath); } } else logger.info("Already in queue, no-op. File: {}", localPath); } @Inject  AbstractFileSystem(
            IConfiguration configuration,
            BackupMetrics backupMetrics,
            BackupNotificationMgr backupNotificationMgr,
            Provider<AbstractBackupPath> pathProvider); @Override Future<Path> asyncDownloadFile(
            final Path remotePath, final Path localPath, final int retry); @Override void downloadFile(final Path remotePath, final Path localPath, final int retry); @Override Future<Path> asyncUploadFile(
            final Path localPath,
            final Path remotePath,
            final AbstractBackupPath path,
            final int retry,
            final boolean deleteAfterSuccessfulUpload); @Override void uploadFile(
            final Path localPath,
            final Path remotePath,
            final AbstractBackupPath path,
            final int retry,
            final boolean deleteAfterSuccessfulUpload); @Override boolean checkObjectExists(Path remotePath); @Override void deleteRemoteFiles(List<Path> remotePaths); @Override String getShard(); @Override Path getPrefix(); @Override Iterator<AbstractBackupPath> listPrefixes(Date date); @Override Iterator<AbstractBackupPath> list(String path, Date start, Date till); @Override final void addObserver(EventObserver<BackupEvent> observer); @Override void removeObserver(EventObserver<BackupEvent> observer); @Override void notifyEventStart(BackupEvent event); @Override void notifyEventSuccess(BackupEvent event); @Override void notifyEventFailure(BackupEvent event); @Override void notifyEventStop(BackupEvent event); @Override int getUploadTasksQueued(); @Override int getDownloadTasksQueued(); @Override void clearCache();  }
@Test public void testFailedRetriesUpload() throws Exception { try { Collection<File> files = generateFiles(1, 1, 1); for (File file : files) { failureFileSystem.uploadFile( file.toPath(), file.toPath(), getDummyPath(file.toPath()), 2, true); } } catch (BackupRestoreException e) { Assert.assertEquals(1, (int) backupMetrics.getInvalidUploads().count()); } }
@Test public void testUpload() throws Exception { Collection<File> files = generateFiles(1, 1, 1); for (File file : files) { myFileSystem.uploadFile( file.toPath(), Paths.get(file.toString() + ".tmp"), getDummyPath(file.toPath()), 2, true); Assert.assertEquals(1, (int) backupMetrics.getValidUploads().actualCount()); Assert.assertFalse(file.exists()); break; } }
@Test public void testUploadDedup() throws Exception { Collection<File> files = generateFiles(1, 1, 1); File file = files.iterator().next(); AbstractBackupPath abstractBackupPath = getDummyPath(file.toPath()); int size = 10; ExecutorService threads = Executors.newFixedThreadPool(size); List<Callable<Boolean>> torun = new ArrayList<>(size); for (int i = 0; i < size; i++) { torun.add( () -> { myFileSystem.uploadFile( file.toPath(), file.toPath(), abstractBackupPath, 2, true); return Boolean.TRUE; }); } List<Future<Boolean>> futures = threads.invokeAll(torun); threads.shutdown(); for (Future future : futures) { try { future.get(); } catch (InterruptedException | ExecutionException e) { } } Assert.assertNotEquals(size, (int) backupMetrics.getValidUploads().actualCount()); }