import com.amazonaws.services.s3.transfer.Transfer; //导入依赖的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; //导入依赖的package包/类
@Override
public S3FileTransferResultsDto downloadFile(final S3FileTransferRequestParamsDto params) throws InterruptedException
{
    LOGGER.info("Downloading S3 file... s3Key=\"{}\" s3BucketName=\"{}\" localPath=\"{}\"", params.getS3KeyPrefix(), params.getS3BucketName(),
        params.getLocalPath());

    // Perform the transfer.
    S3FileTransferResultsDto results = performTransfer(params, new Transferer()
    {
        @Override
        public Transfer performTransfer(TransferManager transferManager)
        {
            return s3Operations.download(params.getS3BucketName(), params.getS3KeyPrefix(), new File(params.getLocalPath()), transferManager);
        }
    });

    LOGGER
        .info("Downloaded S3 file to the local system. s3Key=\"{}\" s3BucketName=\"{}\" localPath=\"{}\" totalBytesTransferred={} transferDuration=\"{}\"",
            params.getS3KeyPrefix(), params.getS3BucketName(), params.getLocalPath(), results.getTotalBytesTransferred(),
            HerdDateUtils.formatDuration(results.getDurationMillis()));

    logOverallTransferRate(results);

    return results;
}
 

import com.amazonaws.services.s3.transfer.Transfer; //导入依赖的package包/类
@Override
public Transfer uploadDirectoryOrFile(final String bucketName, final File source,
		final String virtualDirectoryKeyPrefix) throws AmazonClientException, AmazonServiceException, IOException {
	LOGGER.info("uploadDirectoryOrFile invoked, bucketName: {} , Source: {}", bucketName,
			source.getAbsolutePath());
	Transfer transfer = null;
	final TransferManager trMgr = new TransferManager(s3client);
	if (source.isFile()) {
		transfer = trMgr.upload(bucketName,source.getPath(),source);
	} else if (source.isDirectory()) {
		//Upload recursively
		//virtualDirectoryKeyPrefix could be virtual directory name inside the bucket
		transfer = trMgr.uploadDirectory(bucketName, virtualDirectoryKeyPrefix, source, true);
	} else {
		throw new FileNotFoundException("Source is neither a regular file nor a directory " + source);
	}
	return transfer;
}
 

import com.amazonaws.services.s3.transfer.Transfer; //导入依赖的package包/类
private ProgressListener createProgressListener(Transfer transfer)
{
    return new ProgressListener()
    {
        private ProgressEventType previousType;
        private double previousTransferred;

        @Override
        public synchronized void progressChanged(ProgressEvent progressEvent)
        {
            ProgressEventType eventType = progressEvent.getEventType();
            if (previousType != eventType) {
                log.debug("Upload progress event (%s/%s): %s", host, key, eventType);
                previousType = eventType;
            }

            double transferred = transfer.getProgress().getPercentTransferred();
            if (transferred >= (previousTransferred + 10.0)) {
                log.debug("Upload percentage (%s/%s): %.0f%%", host, key, transferred);
                previousTransferred = transferred;
            }
        }
    };
}
 

import com.amazonaws.services.s3.transfer.Transfer; //导入依赖的package包/类
public SwingProgressListener(Transfer transfer) {
	this.transfer = transfer;
	
	JFrame f = new JFrame("Transfer Progress");
	f.setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);
	Container content = f.getContentPane();
	
	progressBar = new JProgressBar();
	progressBar.setValue(0);
	progressBar.setStringPainted(true);
	
	Border border = BorderFactory.createTitledBorder("Transferring...");
	progressBar.setBorder(border);
	
	content.add(progressBar, BorderLayout.NORTH);
	f.setSize(350, 200);
	f.setLocationRelativeTo(null);
	f.setVisible(true);
}
 

import com.amazonaws.services.s3.transfer.Transfer; //导入依赖的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; //导入依赖的package包/类
@Override
public synchronized boolean isDone() {
    for ( Transfer subTransfer : subTransfers ) {
        if ( !subTransfer.isDone() )
            return false;
    }
    return true;
}
 

import com.amazonaws.services.s3.transfer.Transfer; //导入依赖的package包/类
@Override
public File call() throws Exception {
    for (Future<File> file : partFiles) {
        ServiceUtils.appendFile(file.get(), destinationFile);
        download.updatePersistableTransfer(currentPartNumber++);
    }

    download.setState(Transfer.TransferState.Completed);
    return destinationFile;
}
 

import com.amazonaws.services.s3.transfer.Transfer; //导入依赖的package包/类
@Test
public void testRefreshPageWithoutDuplicateCorrectlyCallsS3() throws Exception {

  // Refresh page when Cloudformation creates the stack should just
  // upload pages to S3 but not duplicate them. Duplicating is done as
  // bookings are later mutated - and is a workaround to ensure
  // ReadAfterWrite consistency. This tests that this duplication
  // does not happen when we do not ask for it (i.e. on stack creation).

  initialisePageManager();

  // Set up S3 expectations for no copy:
  // Transfer interface is implemented by Uploads, Downloads, and Copies
  Transfer mockTransfer = mockery.mock(Transfer.class);
  mockery.checking(new Expectations() {
    {
      allowing(mockTransfer).isDone();
      will(returnValue(true));
      allowing(mockTransfer).waitForCompletion();
    }
  });
  mockTransferManager = mockery.mock(IS3TransferManager.class);
  // Just check S3 methods called correct number of times - don't bother
  // checking argument details.
  mockery.checking(new Expectations() {
    {
      // We have one upload for the page and one for the cached data
      exactly(2).of(mockTransferManager).upload(with(any(PutObjectRequest.class)));
      will(returnValue(mockTransfer));
      // We do _not_ have the copy in this case
      never(mockTransferManager).copy(with(anything()));
    }
  });
  pageManager.setS3TransferManager(mockTransferManager);

  // ACT
  pageManager.refreshPage(fakeCurrentDateString, validDates, apiGatewayBaseUrl, false, bookings,
      revvingSuffix);
}
 

import com.amazonaws.services.s3.transfer.Transfer; //导入依赖的package包/类
@Test
public void testRefreshPageThrowsWhenS3Throws() throws Exception {

  // ARRANGE
  thrown.expect(Exception.class);
  thrown.expectMessage("Exception caught while copying booking page to S3");

  initialisePageManager();

  // Make S3 throw:
  // Transfer interface is implemented by Uploads, Downloads, and Copies
  Transfer mockTransfer = mockery.mock(Transfer.class);
  mockery.checking(new Expectations() {
    {
      allowing(mockTransfer).isDone();
      will(returnValue(true));
      allowing(mockTransfer).waitForCompletion();
    }
  });
  mockTransferManager = mockery.mock(IS3TransferManager.class);
  mockery.checking(new Expectations() {
    {
      oneOf(mockTransferManager).upload(with(any(PutObjectRequest.class)));
      will(throwException(new AmazonServiceException("Grrr...")));
      // Should throw before copy is called
      never(mockTransferManager).copy(with(any(CopyObjectRequest.class)));
    }
  });
  pageManager.setS3TransferManager(mockTransferManager);

  // ACT - this should throw
  pageManager.refreshPage(fakeCurrentDateString, validDates, apiGatewayBaseUrl, false, bookings,
      revvingSuffix);
}
 

import com.amazonaws.services.s3.transfer.Transfer; //导入依赖的package包/类
@Test
public void testRefreshAllPagesThrowsWhenS3Throws() throws Exception {
  // ARRANGE
  thrown.expect(Exception.class);
  thrown.expectMessage("Exception caught while copying booking page to S3");

  initialisePageManager();

  // Make S3 throw:
  // Transfer interface is implemented by Uploads, Downloads, and Copies
  Transfer mockTransfer = mockery.mock(Transfer.class);
  mockery.checking(new Expectations() {
    {
      allowing(mockTransfer).isDone();
      will(returnValue(true));
      allowing(mockTransfer).waitForCompletion();
    }
  });
  mockTransferManager = mockery.mock(IS3TransferManager.class);
  mockery.checking(new Expectations() {
    {
      oneOf(mockTransferManager).upload(with(any(PutObjectRequest.class)));
      will(throwException(new AmazonServiceException("Grrr...")));
      // Should throw before copy is called
      never(mockTransferManager).copy(with(any(CopyObjectRequest.class)));
    }
  });
  pageManager.setS3TransferManager(mockTransferManager);

  mockSNSClient = mockery.mock(AmazonSNS.class);
  mockery.checking(new Expectations() {
    {
      ignoring(mockSNSClient);
    }
  });
  pageManager.setSNSClient(mockSNSClient);

  // ACT - this should throw
  pageManager.refreshAllPages(validDates, apiGatewayBaseUrl, revvingSuffix);
}
 

import com.amazonaws.services.s3.transfer.Transfer; //导入依赖的package包/类
@Override
public S3FileTransferResultsDto downloadDirectory(final S3FileTransferRequestParamsDto params) throws InterruptedException
{
    LOGGER.info("Downloading S3 directory to the local system... s3KeyPrefix=\"{}\" s3BucketName=\"{}\" localDirectory=\"{}\"", params.getS3KeyPrefix(),
        params.getS3BucketName(), params.getLocalPath());

    // Note that the directory download always recursively copies sub-directories.
    // To not recurse, we would have to list the files on S3 (AmazonS3Client.html#listObjects) and manually copy them one at a time.

    // Perform the transfer.
    S3FileTransferResultsDto results = performTransfer(params, new Transferer()
    {
        @Override
        public Transfer performTransfer(TransferManager transferManager)
        {
            return s3Operations.downloadDirectory(params.getS3BucketName(), params.getS3KeyPrefix(), new File(params.getLocalPath()), transferManager);
        }
    });

    LOGGER.info("Downloaded S3 directory to the local system. " +
        "s3KeyPrefix=\"{}\" s3BucketName=\"{}\" localDirectory=\"{}\" s3KeyCount={} totalBytesTransferred={} transferDuration=\"{}\"",
        params.getS3KeyPrefix(), params.getS3BucketName(), params.getLocalPath(), results.getTotalFilesTransferred(), results.getTotalBytesTransferred(),
        HerdDateUtils.formatDuration(results.getDurationMillis()));

    logOverallTransferRate(results);

    return results;
}
 

import com.amazonaws.services.s3.transfer.Transfer; //导入依赖的package包/类
@Override
public S3FileTransferResultsDto uploadDirectory(final S3FileTransferRequestParamsDto params) throws InterruptedException
{
    LOGGER.info("Uploading local directory to S3... localDirectory=\"{}\" s3KeyPrefix=\"{}\" s3BucketName=\"{}\"", params.getLocalPath(),
        params.getS3KeyPrefix(), params.getS3BucketName());

    // Perform the transfer.
    S3FileTransferResultsDto results = performTransfer(params, new Transferer()
    {
        @Override
        public Transfer performTransfer(TransferManager transferManager)
        {
            return s3Operations.uploadDirectory(params.getS3BucketName(), params.getS3KeyPrefix(), new File(params.getLocalPath()), params.isRecursive(),
                new ObjectMetadataProvider()
                {
                    @Override
                    public void provideObjectMetadata(File file, ObjectMetadata metadata)
                    {
                        prepareMetadata(params, metadata);
                    }
                }, transferManager);
        }
    });

    LOGGER.info("Uploaded local directory to S3. " +
        "localDirectory=\"{}\" s3KeyPrefix=\"{}\" s3BucketName=\"{}\" s3KeyCount={} totalBytesTransferred={} transferDuration=\"{}\"",
        params.getLocalPath(), params.getS3KeyPrefix(), params.getS3BucketName(), results.getTotalFilesTransferred(), results.getTotalBytesTransferred(),
        HerdDateUtils.formatDuration(results.getDurationMillis()));

    logOverallTransferRate(results);

    return results;
}
 

import com.amazonaws.services.s3.transfer.Transfer; //导入依赖的package包/类
@Override
public S3FileTransferResultsDto uploadFile(final S3FileTransferRequestParamsDto params) throws InterruptedException
{
    LOGGER.info("Uploading local file to S3... localPath=\"{}\" s3Key=\"{}\" s3BucketName=\"{}\"", params.getLocalPath(), params.getS3KeyPrefix(),
        params.getS3BucketName());

    // Perform the transfer.
    S3FileTransferResultsDto results = performTransfer(params, new Transferer()
    {
        @Override
        public Transfer performTransfer(TransferManager transferManager)
        {
            // Get a handle to the local file.
            File localFile = new File(params.getLocalPath());

            // Create and prepare the metadata.
            ObjectMetadata metadata = new ObjectMetadata();
            prepareMetadata(params, metadata);

            // Create a put request and a transfer manager with the parameters and the metadata.
            PutObjectRequest putObjectRequest = new PutObjectRequest(params.getS3BucketName(), params.getS3KeyPrefix(), localFile);
            putObjectRequest.setMetadata(metadata);

            return s3Operations.upload(putObjectRequest, transferManager);
        }
    });

    LOGGER.info("Uploaded local file to the S3. localPath=\"{}\" s3Key=\"{}\" s3BucketName=\"{}\" totalBytesTransferred={} transferDuration=\"{}\"",
        params.getLocalPath(), params.getS3KeyPrefix(), params.getS3BucketName(), results.getTotalBytesTransferred(),
        HerdDateUtils.formatDuration(results.getDurationMillis()));

    logOverallTransferRate(results);

    return results;
}
 

import com.amazonaws.services.s3.transfer.Transfer; //导入依赖的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; //导入依赖的package包/类
public int getProgress() 
{
    Transfer transfer = getTransfer();
    if(transfer != null) 
    {
        int ret = (int)transfer.getProgress().getPercentTransferred();
        return ret;
    }
    return 0;
}
 

import com.amazonaws.services.s3.transfer.Transfer; //导入依赖的package包/类
private void waitForUpload(Transfer upload, TransferManager tm) throws AmazonServiceException,
    AmazonClientException, InterruptedException {
  long bytes = upload.getProgress().getTotalBytesToTransfer();

  console.write(new Size(bytes) + " to upload"); //$NON-NLS-1$

  long fraction = bytes / 50;

  upload.addProgressListener(new SharpProgressListener(fraction));
  upload.waitForCompletion();

  tm.shutdownNow();
  console.write("\nDone"); //$NON-NLS-1$
}
 

import com.amazonaws.services.s3.transfer.Transfer; //导入依赖的package包/类
public MultipleFileTransferMonitor(AbstractTransfer transfer, Collection<? extends AbstractTransfer> subTransfers) {
    this.subTransfers = subTransfers;
    this.transfer = transfer;

    /*
     * The future object is not publicly exposed, so we only need to worry
     * about implementing get(). The other methods are implemented badly,
     * just to meet the interface contract.
     */
    this.future = new Future<Object>() {

        @Override
        public boolean cancel(boolean mayInterruptIfRunning) {
            return true;
        }

        @Override
        public Object get() throws InterruptedException, ExecutionException {
            Object result = null;
            for ( AbstractTransfer download : MultipleFileTransferMonitor.this.subTransfers ) {
                result = download.getMonitor().getFuture().get();
            }
            return result;
        }

        @Override
        public Object get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException,
                TimeoutException {
            Object result = null;
            for ( AbstractTransfer subTransfer : MultipleFileTransferMonitor.this.subTransfers ) {
                result = subTransfer.getMonitor().getFuture().get(timeout, unit);
            }
            return result;
        }

        @Override
        public boolean isCancelled() {
            return MultipleFileTransferMonitor.this.transfer.getState() == Transfer.TransferState.Canceled;
        }

        @Override
        public boolean isDone() {
            return MultipleFileTransferMonitor.this.isDone();
        }            
    };
}
 

import com.amazonaws.services.s3.transfer.Transfer; //导入依赖的package包/类
Transfer uploadDirectory(String bucketName, String virtualDirectoryKeyPrefix,
File targetDirectory, boolean includeSubdirectories);
 

import com.amazonaws.services.s3.transfer.Transfer; //导入依赖的package包/类
@Override
public Transfer copy(CopyObjectRequest copyObjectRequest) {
  return transferManager.copy(copyObjectRequest);
}