import com.amazonaws.services.s3.transfer.Transfer.TransferState; //导入依赖的package包/类
/**
 * Determines the pause status based on the current state of transfer.
 */
public static PauseStatus determinePauseStatus(TransferState transferState,
        boolean forceCancel) {

    if (forceCancel) {
        if (transferState == TransferState.Waiting) {
            return PauseStatus.CANCELLED_BEFORE_START;
        } else if (transferState == TransferState.InProgress) {
            return PauseStatus.CANCELLED;
        }
    }
    if (transferState == TransferState.Waiting) {
        return PauseStatus.NOT_STARTED;
    }
    return PauseStatus.NO_EFFECT;
}
 

import com.amazonaws.services.s3.transfer.Transfer.TransferState; //导入依赖的package包/类
public static void showTransferProgress(Transfer xfer)
{
    // print the transfer's human-readable description
    System.out.println(xfer.getDescription());
    // print an empty progress bar...
    printProgressBar(0.0);
    // update the progress bar while the xfer is ongoing.
    do {
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            return;
        }
        // Note: so_far and total aren't used, they're just for
        // documentation purposes.
        TransferProgress progress = xfer.getProgress();
        long so_far = progress.getBytesTransferred();
        long total = progress.getTotalBytesToTransfer();
        double pct = progress.getPercentTransferred();
        eraseProgressBar();
        printProgressBar(pct);
    } while (xfer.isDone() == false);
    // print the final state of the transfer.
    TransferState xfer_state = xfer.getState();
    System.out.println(": " + xfer_state);
}
 

import com.amazonaws.services.s3.transfer.Transfer.TransferState; //导入依赖的package包/类
protected void transferFile(boolean deleteSource, String bucket, String filename, String localDirectory) {
	File source = new File(localDirectory + BaseESReducer.DIR_SEPARATOR + filename);
	Preconditions.checkArgument(source.exists(), "Could not find source file: " + source.getAbsolutePath());
	logger.info("Transfering + " + source + " to " + bucket + " with key " + filename);
	FileInputStream fis;
	try {
		fis = new FileInputStream(source);
		ObjectMetadata objectMetadata = new ObjectMetadata();
		objectMetadata.setSSEAlgorithm("AES256");
		objectMetadata.setContentLength(source.length());
		Upload upload = tx.upload(bucket, filename, fis, objectMetadata);
		
		while(!upload.isDone());
		Preconditions.checkState(upload.getState().equals(TransferState.Completed), "File " + filename + " failed to upload with state: " + upload.getState());
		if(deleteSource) {
			source.delete();
		}
	} catch (FileNotFoundException e) {
		// Exception should never be thrown because the precondition above has already validated existence of file
		logger.error("Filename could not be found " + filename, e);
	}
}
 

import com.amazonaws.services.s3.transfer.Transfer.TransferState; //导入依赖的package包/类
private void submitCopyJobsFromListing(
    AmazonS3URI sourceS3Uri,
    final AmazonS3URI targetS3Uri,
    ListObjectsRequest request,
    ObjectListing listing) {
  LOG.debug("Found objects to copy {}, for request {}/{}", listing.getObjectSummaries(), request.getBucketName(),
      request.getPrefix());
  List<S3ObjectSummary> objectSummaries = listing.getObjectSummaries();
  for (final S3ObjectSummary s3ObjectSummary : objectSummaries) {
    String fileName = StringUtils.removeStart(s3ObjectSummary.getKey(), sourceS3Uri.getKey());
    final String targetKey = Strings.nullToEmpty(targetS3Uri.getKey()) + fileName;
    LOG.info("copying object from '{}/{}' to '{}/{}'", s3ObjectSummary.getBucketName(), s3ObjectSummary.getKey(),
        targetS3Uri.getBucket(), targetKey);

    CopyObjectRequest copyObjectRequest = new CopyObjectRequest(s3ObjectSummary.getBucketName(),
        s3ObjectSummary.getKey(), targetS3Uri.getBucket(), targetKey);

    TransferStateChangeListener stateChangeListener = new TransferStateChangeListener() {

      @Override
      public void transferStateChanged(Transfer transfer, TransferState state) {
        if (state == TransferState.Completed) {
          // NOTE: running progress doesn't seem to be reported correctly.
          // transfer.getProgress().getBytesTransferred() is always 0. Unsure what is the cause of this at this moment
          // so just printing total bytes when completed.
          LOG.debug("copied object from '{}/{}' to '{}/{}': {} bytes transferred", s3ObjectSummary.getBucketName(),
              s3ObjectSummary.getKey(), targetS3Uri.getBucket(), targetKey,
              transfer.getProgress().getTotalBytesToTransfer());
        }
      }
    };
    Copy copy = transferManager.copy(copyObjectRequest, srcClient, stateChangeListener);
    totalBytesToReplicate += copy.getProgress().getTotalBytesToTransfer();
    copyJobs.add(copy);
  }
}
 

import com.amazonaws.services.s3.transfer.Transfer.TransferState; //导入依赖的package包/类
/**
 * Takes the result from serial download,
 * updates the transfer state and monitor in downloadImpl object
 * based on the result.
 */
private void updateDownloadStatus(S3Object result) {
    if (result == null) {
        download.setState(TransferState.Canceled);
        download.setMonitor(new DownloadMonitor(download, null));
    } else {
        download.setState(TransferState.Completed);
    }
}
 

import com.amazonaws.services.s3.transfer.Transfer.TransferState; //导入依赖的package包/类
@Override
public void transferStateChanged(Transfer upload, TransferState state) {
    // There's a race here: we can't start monitoring the state of
    // individual transfers until we have added all the transfers to the
    // list, or we may incorrectly report completion.
    try {
        latch.await();
    } catch ( InterruptedException e ) {
        throw new SdkClientException("Couldn't wait for all downloads to be queued");
    }

    synchronized (multipleFileTransfer) {
        if ( multipleFileTransfer.getState() == state || multipleFileTransfer.isDone() )
            return;

        /*
         * If we're not already in a terminal state, allow a transition
         * to a non-waiting state. Mark completed if this download is
         * completed and the monitor says all of the rest are as well.
         */
        if ( state == TransferState.InProgress ) {
            multipleFileTransfer.setState(state);
        } else if ( multipleFileTransfer.getMonitor().isDone() ) {
            multipleFileTransfer.collateFinalState();
        } else {
            multipleFileTransfer.setState(TransferState.InProgress);
        }
    }
}
 

import com.amazonaws.services.s3.transfer.Transfer.TransferState; //导入依赖的package包/类
public CopyResult call() throws Exception {
    copy.setState(TransferState.InProgress);
    if (isMultipartCopy()) {
        publishProgress(listenerChain, ProgressEventType.TRANSFER_STARTED_EVENT);
        copyInParts();
        return null;
    } else {
        return copyInOneChunk();
    }
}
 

import com.amazonaws.services.s3.transfer.Transfer.TransferState; //导入依赖的package包/类
void copyComplete() {
    markAllDone();
    transfer.setState(TransferState.Completed);
    // AmazonS3Client takes care of all the events for single part uploads,
    // so we only need to send a completed event for multipart uploads.
    if (multipartCopyCallable.isMultipartCopy()) {
        publishProgress(listener, ProgressEventType.TRANSFER_COMPLETED_EVENT);
    }
}
 

import com.amazonaws.services.s3.transfer.Transfer.TransferState; //导入依赖的package包/类
void uploadComplete() {
    markAllDone();
    transfer.setState(TransferState.Completed);

    // AmazonS3Client takes care of all the events for single part uploads,
    // so we only need to send a completed event for multipart uploads.
    if (multipartUploadCallable.isMultipartUpload()) {
        publishProgress(listener, ProgressEventType.TRANSFER_COMPLETED_EVENT);
    }
}
 

import com.amazonaws.services.s3.transfer.Transfer.TransferState; //导入依赖的package包/类
public UploadResult call() throws Exception {
    upload.setState(TransferState.InProgress);
    if ( isMultipartUpload() ) {
        publishProgress(listener, ProgressEventType.TRANSFER_STARTED_EVENT);
        return uploadInParts();
    } else {
        return uploadInOneChunk();
    }
}
 

import com.amazonaws.services.s3.transfer.Transfer.TransferState; //导入依赖的package包/类
/**
 * Test S3 file copy with an invalid KMS Id that will result in a cancelled transfer.
 */
@Test
public void testCopyFileInvalidKmsIdCancelled() throws InterruptedException
{
    // Put a 1 byte file in S3.
    s3Operations
        .putObject(new PutObjectRequest(storageDaoTestHelper.getS3LoadingDockBucketName(), TARGET_S3_KEY, new ByteArrayInputStream(new byte[1]), null),
            null);

    try
    {
        S3FileCopyRequestParamsDto transferDto = new S3FileCopyRequestParamsDto();
        transferDto.setSourceBucketName(storageDaoTestHelper.getS3LoadingDockBucketName());
        transferDto.setTargetBucketName(storageDaoTestHelper.getS3ExternalBucketName());
        transferDto.setSourceObjectKey(TARGET_S3_KEY);
        transferDto.setTargetObjectKey(TARGET_S3_KEY);
        transferDto.setKmsKeyId(MockS3OperationsImpl.MOCK_KMS_ID_CANCELED_TRANSFER);
        s3Dao.copyFile(transferDto);
        fail("An IllegalStateException was expected but not thrown.");
    }
    catch (IllegalStateException ex)
    {
        assertEquals("Invalid IllegalStateException message returned.",
            "The transfer operation \"" + MockS3OperationsImpl.MOCK_TRANSFER_DESCRIPTION + "\" did not complete successfully. " + "Current state: \"" +
                Transfer.TransferState.Canceled + "\".", ex.getMessage());
    }
}
 

import com.amazonaws.services.s3.transfer.Transfer.TransferState; //导入依赖的package包/类
/**
 * This method must return a non-null object, or else the existing
 * implementation in {@link AbstractTransfer#waitForCompletion()}
 * would block forever.
 *
 * @return the downloaded file
 */
@Override
public File call() throws Exception {
    try {
        latch.await();

        if (isTimeoutEnabled()) {
            timedExecutor.schedule(new Runnable() {
                public void run() {
                    try {
                        if (download.getState() != TransferState.Completed) {
                            download.abort();
                        }
                    } catch(Exception e) {
                        throw new SdkClientException(
                                "Unable to abort download after timeout", e);
                    }
                }
            }, timeout, TimeUnit.MILLISECONDS);
        }

        download.setState(TransferState.InProgress);
        ServiceUtils.createParentDirectoryIfNecessary(dstfile);

        if (isDownloadParallel) {
            downloadInParallel(ServiceUtils.getPartCount(req, s3));
        } else {
            S3Object s3Object = retryableDownloadS3ObjectToFile(dstfile,
                    new DownloadTaskImpl(s3, download, req));
            updateDownloadStatus(s3Object);
        }
        return dstfile;
    } catch (Throwable t) {
        // Cancel all the futures
        for (Future<File> f : futureFiles) {
            f.cancel(true);
        }
        // Downloads aren't allowed to move from canceled to failed
        if (download.getState() != TransferState.Canceled) {
            download.setState(TransferState.Failed);
        }
        if (t instanceof Exception)
            throw (Exception) t;
        else
            throw (Error) t;
    }
}
 

import com.amazonaws.services.s3.transfer.Transfer.TransferState; //导入依赖的package包/类
@Test
public void testPerformTransferAssertErrorWhenTransferBytesMismatch() throws Exception
{
    S3Operations originalS3Operations = (S3Operations) ReflectionTestUtils.getField(s3Dao, "s3Operations");
    S3Operations mockS3Operations = mock(S3Operations.class);
    ReflectionTestUtils.setField(s3Dao, "s3Operations", mockS3Operations);

    // Shorten the sleep interval for faster tests
    long originalSleepIntervalsMillis = (long) ReflectionTestUtils.getField(s3Dao, "sleepIntervalsMillis");
    ReflectionTestUtils.setField(s3Dao, "sleepIntervalsMillis", 1l);

    try
    {
        S3FileTransferRequestParamsDto s3FileTransferRequestParamsDto = new S3FileTransferRequestParamsDto();
        s3FileTransferRequestParamsDto.setLocalPath("localPath");

        when(mockS3Operations.upload(any(), any())).then(new Answer<Upload>()
        {
            @Override
            public Upload answer(InvocationOnMock invocation) throws Throwable
            {
                Upload mockedUpload = mock(Upload.class);
                TransferProgress transferProgress = new TransferProgress();
                // bytesTransferred < totalBytesToTransfer should cause error
                ReflectionTestUtils.setField(transferProgress, "bytesTransferred", 0l);
                ReflectionTestUtils.setField(transferProgress, "totalBytesToTransfer", 1l);
                when(mockedUpload.getProgress()).thenReturn(transferProgress);
                when(mockedUpload.isDone()).thenReturn(true);
                when(mockedUpload.getState()).thenReturn(TransferState.Completed);
                return mockedUpload;
            }
        });

        try
        {
            s3Dao.uploadFile(s3FileTransferRequestParamsDto);
            fail();
        }
        catch (Exception e)
        {
            assertEquals(IllegalArgumentException.class, e.getClass());
            assertEquals("Actual number of bytes transferred is less than expected (actual: 0 bytes; expected: 1 bytes).", e.getMessage());
        }
    }
    finally
    {
        ReflectionTestUtils.setField(s3Dao, "s3Operations", originalS3Operations);
        ReflectionTestUtils.setField(s3Dao, "sleepIntervalsMillis", originalSleepIntervalsMillis);
    }
}
 

import com.amazonaws.services.s3.transfer.Transfer.TransferState; //导入依赖的package包/类
@Test
public void testPerformTransferAssertHandleFailedWithAmazonClientException() throws Exception
{
    S3Operations originalS3Operations = (S3Operations) ReflectionTestUtils.getField(s3Dao, "s3Operations");
    S3Operations mockS3Operations = mock(S3Operations.class);
    ReflectionTestUtils.setField(s3Dao, "s3Operations", mockS3Operations);

    // Shorten the sleep interval for faster tests
    long originalSleepIntervalsMillis = (long) ReflectionTestUtils.getField(s3Dao, "sleepIntervalsMillis");
    ReflectionTestUtils.setField(s3Dao, "sleepIntervalsMillis", 1l);

    try
    {
        S3FileCopyRequestParamsDto s3FileCopyRequestParamsDto = new S3FileCopyRequestParamsDto();
        s3FileCopyRequestParamsDto.setSourceBucketName("sourceBucketName");
        s3FileCopyRequestParamsDto.setSourceObjectKey("sourceObjectKey");
        s3FileCopyRequestParamsDto.setTargetBucketName("targetBucketName");
        s3FileCopyRequestParamsDto.setTargetObjectKey("targetObjectKey");
        s3FileCopyRequestParamsDto.setKmsKeyId("kmsKeyId");

        when(mockS3Operations.copyFile(any(), any())).then(new Answer<Copy>()
        {
            @Override
            public Copy answer(InvocationOnMock invocation) throws Throwable
            {
                Copy mockTransfer = mock(Copy.class);

                when(mockTransfer.getProgress()).thenReturn(new TransferProgress());
                when(mockTransfer.getState()).thenReturn(TransferState.Failed);
                when(mockTransfer.isDone()).thenReturn(true);
                when(mockTransfer.waitForException()).thenReturn(new AmazonClientException("message"));
                return mockTransfer;
            }
        });

        try
        {
            s3Dao.copyFile(s3FileCopyRequestParamsDto);
            fail();
        }
        catch (Exception e)
        {
            assertEquals(AmazonClientException.class, e.getClass());
            assertEquals("message", e.getMessage());
        }
    }
    finally
    {
        ReflectionTestUtils.setField(s3Dao, "s3Operations", originalS3Operations);
        ReflectionTestUtils.setField(s3Dao, "sleepIntervalsMillis", originalSleepIntervalsMillis);
    }
}
 

import com.amazonaws.services.s3.transfer.Transfer.TransferState; //导入依赖的package包/类
/**
 * {@inheritDoc}
 * <p/>
 * This implementation creates any directory that does not exist in the path to the destination directory.
 */
@Override
public MultipleFileDownload downloadDirectory(String bucketName, String keyPrefix, File destinationDirectory, TransferManager transferManager)
{
    LOGGER.debug("downloadDirectory(): bucketName = " + bucketName + ", keyPrefix = " + keyPrefix + ", destinationDirectory = " + destinationDirectory);

    MockS3Bucket mockS3Bucket = mockS3Buckets.get(bucketName);

    List<Download> downloads = new ArrayList<>();
    long totalBytes = 0;

    if (mockS3Bucket != null)
    {
        for (MockS3Object mockS3Object : mockS3Bucket.getObjects().values())
        {
            if (mockS3Object.getKey().startsWith(keyPrefix))
            {
                String filePath = destinationDirectory.getAbsolutePath() + "/" + mockS3Object.getKey();
                File file = new File(filePath);
                file.getParentFile().mkdirs(); // Create any directory in the path that does not exist.
                try (FileOutputStream fileOutputStream = new FileOutputStream(file))
                {
                    LOGGER.debug("downloadDirectory(): Writing file " + file);
                    fileOutputStream.write(mockS3Object.getData());
                    totalBytes += mockS3Object.getData().length;
                    downloads.add(new DownloadImpl(null, null, null, null, null, new GetObjectRequest(bucketName, mockS3Object.getKey()), file,
                        mockS3Object.getObjectMetadata(), false));
                }
                catch (IOException e)
                {
                    throw new RuntimeException("Error writing to file " + file, e);
                }
            }
        }
    }

    TransferProgress progress = new TransferProgress();
    progress.setTotalBytesToTransfer(totalBytes);
    progress.updateProgress(totalBytes);

    MultipleFileDownloadImpl multipleFileDownload = new MultipleFileDownloadImpl(null, progress, null, keyPrefix, bucketName, downloads);
    multipleFileDownload.setState(TransferState.Completed);
    return multipleFileDownload;
}
 

import com.amazonaws.services.s3.transfer.Transfer.TransferState; //导入依赖的package包/类
@Override
public MultipleFileUpload uploadFileList(String bucketName, String virtualDirectoryKeyPrefix, File directory, List<File> files,
    ObjectMetadataProvider metadataProvider, TransferManager transferManager)
{
    LOGGER.debug(
        "uploadFileList(): bucketName = " + bucketName + ", virtualDirectoryKeyPrefix = " + virtualDirectoryKeyPrefix + ", directory = " + directory +
            ", files = " + files);

    String directoryPath = directory.getAbsolutePath();

    long totalFileLength = 0;
    List<Upload> subTransfers = new ArrayList<>();
    for (File file : files)
    {
        // Get path to file relative to the specified directory
        String relativeFilePath = file.getAbsolutePath().substring(directoryPath.length());

        // Replace any backslashes (i.e. Windows separator) with a forward slash.
        relativeFilePath = relativeFilePath.replace("\\", "/");

        // Remove any leading slashes
        relativeFilePath = relativeFilePath.replaceAll("^/+", "");

        long fileLength = file.length();

        // Remove any trailing slashes
        virtualDirectoryKeyPrefix = virtualDirectoryKeyPrefix.replaceAll("/+$", "");

        String s3ObjectKey = virtualDirectoryKeyPrefix + "/" + relativeFilePath;
        totalFileLength += fileLength;

        PutObjectRequest putObjectRequest = new PutObjectRequest(bucketName, s3ObjectKey, file);

        ObjectMetadata objectMetadata = new ObjectMetadata();
        metadataProvider.provideObjectMetadata(null, objectMetadata);
        putObjectRequest.setMetadata(objectMetadata);

        putObject(putObjectRequest, transferManager.getAmazonS3Client());

        subTransfers.add(new UploadImpl(null, null, null, null));
    }

    TransferProgress progress = new TransferProgress();
    progress.setTotalBytesToTransfer(totalFileLength);
    progress.updateProgress(totalFileLength);

    MultipleFileUploadImpl multipleFileUpload = new MultipleFileUploadImpl(null, progress, null, virtualDirectoryKeyPrefix, bucketName, subTransfers);
    multipleFileUpload.setState(TransferState.Completed);
    return multipleFileUpload;
}
 

import com.amazonaws.services.s3.transfer.Transfer.TransferState; //导入依赖的package包/类
public void transferStateChanged(Transfer transfer, TransferState state);