agave-files/files-core/src/main/java/org/iplantc/service/io/queue/UploadJob2.java
/**
	 * Perform the actual upload.
	 *
	 * @param dataMap
	 * @throws JobExecutionException
	 */
Perform the actual upload.
@param dataMap
@throws JobExecutionException
Perform the actual upload . @param dataMap @throws JobExecutionException
public void doUpload(JobDataMap dataMap) throws JobExecutionException
	{
		RemoteDataClient remoteDataClient = null;
		String owner = null;
		String createdBy = null;
		String cachedFile = null;
		try {
			try {
				Long logicalFileId = dataMap.getLong("logicalFileId");
				cachedFile = dataMap.getString("cachedFile");
				owner = dataMap.getString("owner");
				createdBy = dataMap.getString("createdBy");
				String tenantCode = dataMap.getString("tenantId");
				String uploadSource = dataMap.getString("sourceUrl");
				String uploadDest = dataMap.getString("destUrl");
				boolean isRangeCopyOperation = dataMap.getBooleanFromString("isRangeCopyOperation");
				long rangeIndex = dataMap.getLong("rangeIndex");
				long rangeSize = dataMap.getLong("rangeSize");
				TenancyHelper.setCurrentTenantId(tenantCode);
				TenancyHelper.setCurrentEndUser(createdBy);
				logicalFile = LogicalFileDao.findById(logicalFileId);
				if (logicalFile == null) {
					log.info("Aborting upload task from " + uploadSource + " to " + uploadDest + " because logicalFile has been deleted.");
					return;
				}
				remoteDataClient = logicalFile.getSystem().getRemoteDataClient(logicalFile.getInternalUsername());
				remoteDataClient.authenticate();
				String agavePath = logicalFile.getAgaveRelativePathFromAbsolutePath();
				TransferTask transferTask = new TransferTask(
						uploadSource,
						uploadDest,
						createdBy,
						null,
						null);
				TransferTaskDao.persist(transferTask);
				urlCopy = new URLCopy(new Local(null, "/", "/"), remoteDataClient);
				if (isRangeCopyOperation) {
					transferTask = urlCopy.copyRange(cachedFile, 0, new File(cachedFile).length(), agavePath, rangeIndex, transferTask);
				} else {
					transferTask = urlCopy.copy(cachedFile, agavePath, transferTask);
				}
				if (logicalFile.getOwner() != null && remoteDataClient.isPermissionMirroringRequired()) {
					try {
						remoteDataClient.setOwnerPermission(remoteDataClient.getUsername(), logicalFile.getAgaveRelativePathFromAbsolutePath(), true);
						remoteDataClient.setOwnerPermission(owner, logicalFile.getAgaveRelativePathFromAbsolutePath(), true);
					} catch (RemoteDataException e) {
						if (! (e.getCause() instanceof org.irods.jargon.core.exception.InvalidUserException)) {
							throw new PermissionException(e);
						}
					}
				}
				FileEventProcessor eventProcessor = new FileEventProcessor();
				eventProcessor.processContentEvent(logicalFile, new FileEvent(FileEventType.UPLOAD,
						FileEventType.UPLOAD.getDescription(),
						createdBy));
				log.info("Completed staging file " + logicalFile.getAgaveRelativePathFromAbsolutePath() + " for user " + owner);
				logicalFile.setStatus(StagingTaskStatus.STAGING_COMPLETED.name());
				logicalFile.setNativeFormat("raw");
				logicalFile.addContentEvent(new FileEvent(FileEventType.STAGING_COMPLETED,
						"Your scheduled transfer of " + uploadSource +
								" completed staging. You can access the raw file on " + logicalFile.getSystem().getName() + " at " +
								logicalFile.getPath() + " or via the API at " +
								logicalFile.getPublicLink() + ".",
						createdBy));
				LogicalFileDao.persist(logicalFile);
			}
			catch (ClosedByInterruptException e) {
				if (logicalFile != null) {
					LogicalFileDao.updateTransferStatus(logicalFile,
							StagingTaskStatus.STAGING_FAILED.name(),
							"Failed to transfer uploaded file to the remote system. "
									+ "The server was stopped prior to the transfer completing "
									+ "and the file was no longer available. No further action "
									+ "can be taken to recover. Please attempt the upload again.",
							createdBy);
				}
				log.error("Failed to transfer uploaded file "
						+ dataMap.getString("cachedFile") + " to remote destination for "
						+ " when the worker was interrupted.");
				Thread.currentThread().interrupt();
			}
			catch (StaleObjectStateException e) {
				log.debug("Just avoided a file upload race condition from worker");
			}
			catch (RemoteDataException e) {
				if (logicalFile != null) {
					LogicalFileDao.updateTransferStatus(logicalFile, StagingTaskStatus.STAGING_FAILED, createdBy);
					try {
						if (logicalFile.getOwner() != null && remoteDataClient.isPermissionMirroringRequired()) {
							remoteDataClient.setOwnerPermission(remoteDataClient.getUsername(), logicalFile.getAgaveRelativePathFromAbsolutePath(), true);
							remoteDataClient.setOwnerPermission(owner, logicalFile.getAgaveRelativePathFromAbsolutePath(), true);
						}
					} catch (Exception e1) {}
				}
				log.error("File upload worker failed to transfer the cached file " +
						dataMap.getString("cachedFile") + " to remote destination.", e);
			}
			catch (Throwable e) {
				if (logicalFile != null) {
					LogicalFileDao.updateTransferStatus(logicalFile, StagingTaskStatus.STAGING_FAILED, createdBy);
					try {
						if (logicalFile.getOwner() != null && remoteDataClient.isPermissionMirroringRequired()) {
							remoteDataClient.setOwnerPermission(remoteDataClient.getUsername(), logicalFile.getAgaveRelativePathFromAbsolutePath(), true);
							remoteDataClient.setOwnerPermission(owner, logicalFile.getAgaveRelativePathFromAbsolutePath(), true);
						}
					} catch (Exception e1) {}
				}
				log.error("File upload worker failed unexpectedly while attempting to transfer cached file " +
						dataMap.getString("cachedFile") + " to remote destination.", e);
			}
		} catch (Throwable e) {
			log.error("File upload worker failed unexpectedly while attempting to transfer cached "
					+ "file to remote destination.", e);
		}
		finally {
			try { remoteDataClient.disconnect(); } catch (Exception e) {}
			if (StringUtils.isNotEmpty(cachedFile)) {
				FileUtils.deleteQuietly(new File(cachedFile).getParentFile());
			}
		}
	}
public void doUpload ( JobDataMap dataMap ) throws JobExecutionException { RemoteDataClient remoteDataClient = null ; String owner = null ; String createdBy = null ; String cachedFile = null ; try { try { Long logicalFileId = dataMap . getLong ( "logicalFileId" ) ; cachedFile = dataMap . getString ( "cachedFile" ) ; owner = dataMap . getString ( "owner" ) ; createdBy = dataMap . getString ( "createdBy" ) ; String tenantCode = dataMap . getString ( "tenantId" ) ; String uploadSource = dataMap . getString ( "sourceUrl" ) ; String uploadDest = dataMap . getString ( "destUrl" ) ; boolean isRangeCopyOperation = dataMap . getBooleanFromString ( "isRangeCopyOperation" ) ; long rangeIndex = dataMap . getLong ( "rangeIndex" ) ; long rangeSize = dataMap . getLong ( "rangeSize" ) ; TenancyHelper . setCurrentTenantId ( tenantCode ) ; TenancyHelper . setCurrentEndUser ( createdBy ) ; logicalFile = LogicalFileDao . findById ( logicalFileId ) ; if ( logicalFile == null ) { log . info ( "Aborting upload task from " + uploadSource + " to " + uploadDest + " because logicalFile has been deleted." ) ; return ; } remoteDataClient = logicalFile . getSystem ( ) . getRemoteDataClient ( logicalFile . getInternalUsername ( ) ) ; remoteDataClient . authenticate ( ) ; String agavePath = logicalFile . getAgaveRelativePathFromAbsolutePath ( ) ; TransferTask transferTask = new TransferTask ( uploadSource , uploadDest , createdBy , null , null ) ; TransferTaskDao . persist ( transferTask ) ; urlCopy = new URLCopy ( new Local ( null , "/" , "/" ) , remoteDataClient ) ; if ( isRangeCopyOperation ) { transferTask = urlCopy . copyRange ( cachedFile , 0 , new File ( cachedFile ) . length ( ) , agavePath , rangeIndex , transferTask ) ; } else { transferTask = urlCopy . copy ( cachedFile , agavePath , transferTask ) ; } if ( logicalFile . getOwner ( ) != null && remoteDataClient . isPermissionMirroringRequired ( ) ) { try { remoteDataClient . setOwnerPermission ( remoteDataClient . getUsername ( ) , logicalFile . getAgaveRelativePathFromAbsolutePath ( ) , true ) ; remoteDataClient . setOwnerPermission ( owner , logicalFile . getAgaveRelativePathFromAbsolutePath ( ) , true ) ; } catch ( RemoteDataException e ) { if ( ! ( e . getCause ( ) instanceof org . irods . jargon . core . exception . InvalidUserException ) ) { throw new PermissionException ( e ) ; } } } FileEventProcessor eventProcessor = new FileEventProcessor ( ) ; eventProcessor . processContentEvent ( logicalFile , new FileEvent ( FileEventType . UPLOAD , FileEventType . UPLOAD . getDescription ( ) , createdBy ) ) ; log . info ( "Completed staging file " + logicalFile . getAgaveRelativePathFromAbsolutePath ( ) + " for user " + owner ) ; logicalFile . setStatus ( StagingTaskStatus . STAGING_COMPLETED . name ( ) ) ; logicalFile . setNativeFormat ( "raw" ) ; logicalFile . addContentEvent ( new FileEvent ( FileEventType . STAGING_COMPLETED , "Your scheduled transfer of " + uploadSource + " completed staging. You can access the raw file on " + logicalFile . getSystem ( ) . getName ( ) + " at " + logicalFile . getPath ( ) + " or via the API at " + logicalFile . getPublicLink ( ) + "." , createdBy ) ) ; LogicalFileDao . persist ( logicalFile ) ; } catch ( ClosedByInterruptException e ) { if ( logicalFile != null ) { LogicalFileDao . updateTransferStatus ( logicalFile , StagingTaskStatus . STAGING_FAILED . name ( ) , "Failed to transfer uploaded file to the remote system. " + "The server was stopped prior to the transfer completing " + "and the file was no longer available. No further action " + "can be taken to recover. Please attempt the upload again." , createdBy ) ; } log . error ( "Failed to transfer uploaded file " + dataMap . getString ( "cachedFile" ) + " to remote destination for " + " when the worker was interrupted." ) ; Thread . currentThread ( ) . interrupt ( ) ; } catch ( StaleObjectStateException e ) { log . debug ( "Just avoided a file upload race condition from worker" ) ; } catch ( RemoteDataException e ) { if ( logicalFile != null ) { LogicalFileDao . updateTransferStatus ( logicalFile , StagingTaskStatus . STAGING_FAILED , createdBy ) ; try { if ( logicalFile . getOwner ( ) != null && remoteDataClient . isPermissionMirroringRequired ( ) ) { remoteDataClient . setOwnerPermission ( remoteDataClient . getUsername ( ) , logicalFile . getAgaveRelativePathFromAbsolutePath ( ) , true ) ; remoteDataClient . setOwnerPermission ( owner , logicalFile . getAgaveRelativePathFromAbsolutePath ( ) , true ) ; } } catch ( Exception e1 ) { } } log . error ( "File upload worker failed to transfer the cached file " + dataMap . getString ( "cachedFile" ) + " to remote destination." , e ) ; } catch ( Throwable e ) { if ( logicalFile != null ) { LogicalFileDao . updateTransferStatus ( logicalFile , StagingTaskStatus . STAGING_FAILED , createdBy ) ; try { if ( logicalFile . getOwner ( ) != null && remoteDataClient . isPermissionMirroringRequired ( ) ) { remoteDataClient . setOwnerPermission ( remoteDataClient . getUsername ( ) , logicalFile . getAgaveRelativePathFromAbsolutePath ( ) , true ) ; remoteDataClient . setOwnerPermission ( owner , logicalFile . getAgaveRelativePathFromAbsolutePath ( ) , true ) ; } } catch ( Exception e1 ) { } } log . error ( "File upload worker failed unexpectedly while attempting to transfer cached file " + dataMap . getString ( "cachedFile" ) + " to remote destination." , e ) ; } } catch ( Throwable e ) { log . error ( "File upload worker failed unexpectedly while attempting to transfer cached " + "file to remote destination." , e ) ; } finally { try { remoteDataClient . disconnect ( ) ; } catch ( Exception e ) { } if ( StringUtils . isNotEmpty ( cachedFile ) ) { FileUtils . deleteQuietly ( new File ( cachedFile ) . getParentFile ( ) ) ; } } }
Perform the actual upload.
Perform the actual upload .
// file will be untouched after staging, so just mark as completed // update the file task //Agave will treat all files as "raw". The functionality to transform (encode) files has been decommissioned.