Java 类com.amazonaws.services.s3.transfer.Transfer 实例源码

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);
}

@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;
}

@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;
}

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;
            }
        }
    };
}

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);
}

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);
  }
}

@Override
public synchronized boolean isDone() {
    for ( Transfer subTransfer : subTransfers ) {
        if ( !subTransfer.isDone() )
            return false;
    }
    return true;
}

@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;
}

@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);
}

@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);
}

@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);
}

@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;
}

@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;
}

@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;
}

/**
 * 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());
    }
}

public int getProgress() 
{
    Transfer transfer = getTransfer();
    if(transfer != null) 
    {
        int ret = (int)transfer.getProgress().getPercentTransferred();
        return ret;
    }
    return 0;
}

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$
}

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();
        }            
    };
}

Transfer uploadDirectory(String bucketName, String virtualDirectoryKeyPrefix,
File targetDirectory, boolean includeSubdirectories);

@Override
public Transfer copy(CopyObjectRequest copyObjectRequest) {
  return transferManager.copy(copyObjectRequest);
}

@Override
public Transfer download(String bucketName, String keyName, File target) {
  return transferManager.download(bucketName, keyName, target);
}

@Override
public Transfer upload(PutObjectRequest putObjectRequest) {
  return transferManager.upload(putObjectRequest);
}

@Override
public Transfer upload(String bucketName, String keyName, File target) {
  return transferManager.upload(bucketName, keyName, target);
}

@Override
public Transfer uploadDirectory(String bucketName, String virtualDirectoryKeyPrefix,
    File targetDirectory, boolean includeSubdirectories) {
  return transferManager.uploadDirectory(bucketName, virtualDirectoryKeyPrefix, targetDirectory,
      includeSubdirectories);
}

@Test
public void testRefreshPageCallsTheLifecycleManagerCorrectly() throws Exception {

  // refreshPage should query the lifecycle manager for the lifecycle
  // state.

  // ARRANGE
  // Expect method to query the lifecycle manager for the lifecycle
  // state. N.B. Name mock as it's replacing the default lifecycleManager
  // mock
  mockLifecycleManager = mockery.mock(ILifecycleManager.class, "replacementLifecycleManagerMock");
  mockery.checking(new Expectations() {
    {
      // One call for creating the page, and one for creating the json data.
      exactly(2).of(mockLifecycleManager).getLifecycleState();
      will(returnValue(activeLifecycleState.get()));
    }
  });
  initialisePageManager();

  // Set up S3 expectations - these are incidental to current test:
  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
      allowing(mockTransferManager).upload(with(any(PutObjectRequest.class)));
      will(returnValue(mockTransfer));
    }
  });
  pageManager.setS3TransferManager(mockTransferManager);

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

@Test
public void testRefreshPageWithDuplicateCorrectlyCallsS3() 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 happen when we ask for it.

  initialisePageManager();

  // Set up S3 expectations for 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_ have the copy in this case - for the page, but not for the
      // cached data
      oneOf(mockTransferManager).copy(with(any(CopyObjectRequest.class)));
      will(returnValue(mockTransfer));
    }
  });
  pageManager.setS3TransferManager(mockTransferManager);

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

@Test
public void testRefreshAllPagesCorrectlyCallsS3() throws Exception {

  initialisePageManager();

  // Set up S3 expectations for upload (without copy) for each valid date:
  // 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.
  final Sequence refreshSequence = mockery.sequence("refresh");
  mockery.checking(new Expectations() {
    {
      // 2 uploads for each date + 3 uploads for the index pages + 1 upload
      // for the validdates json + 1 upload for the famous players json.
      exactly(2 * validDates.size() + 5).of(mockTransferManager).upload(
          with(any(PutObjectRequest.class)));
      will(returnValue(mockTransfer));
      inSequence(refreshSequence);

      // We do _not_ have the copy in this case
      never(mockTransferManager).copy(with(anything()));
    }
  });
  // Delete previous day's bookings and cached data at end
  mockS3Client = mockery.mock(AmazonS3.class);
  mockery.checking(new Expectations() {
    {
      exactly(2).of(mockS3Client).deleteObject(with(aNonNull(DeleteObjectRequest.class)));
      // Ensures this delete occurs after uploads of new pages and cached data
      inSequence(refreshSequence);

      exactly(1).of(mockTransferManager).getAmazonS3Client();
      will(returnValue(mockS3Client));
    }
  });
  pageManager.setS3TransferManager(mockTransferManager);

  // ACT
  pageManager.refreshAllPages(validDates, apiGatewayBaseUrl, revvingSuffix);
}

@Test
public void testRefreshAllPagesNotifiesTheSnsTopicWhenItThrows() throws Exception {
  // It is useful for the admin user to be notified whenever the refreshing
  // of booking pages does not succeed - so that they can update the pages
  // manually instead. This tests that whenever the page manager catches an
  // exception while refreshing pages, it notifies the admin SNS topic.

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

  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);

  // Set up mock SNS client to expect a notification
  mockSNSClient = mockery.mock(AmazonSNS.class);
  String partialMessage = "Apologies - but there was an error refreshing the booking pages in S3";
  mockery.checking(new Expectations() {
    {
      oneOf(mockSNSClient).publish(with(equal(adminSnsTopicArn)),
          with(startsWith(partialMessage)),
          with(equal("Sqawsh booking pages in S3 failed to refresh")));
    }
  });
  pageManager.setSNSClient(mockSNSClient);

  // ACT - this should throw - and notify the SNS topic
  pageManager.refreshAllPages(validDates, apiGatewayBaseUrl, revvingSuffix);
}

public void doTestBackupBookingsAndBookingRulesCorrectlyCallsS3(
    Boolean backupAllBookingsAndBookingRules, Boolean backupSingleBookingRule,
    Boolean isNotDeletion) throws Exception {

  // Prevent this method being called with invalid combination of parameters
  assertTrue("Either backup all bookingsAndRules or backup a single booking/rule",
      !(backupAllBookingsAndBookingRules && backupSingleBookingRule));

  // Set up mock transfer manager
  Transfer mockTransfer = mockery.mock(Transfer.class);
  mockery.checking(new Expectations() {
    {
      allowing(mockTransfer).isDone();
      will(returnValue(true));
      allowing(mockTransfer).waitForCompletion();
    }
  });
  mockTransferManager = mockery.mock(IS3TransferManager.class);

  // Set up mock booking and rule managers
  mockBookingManager = mockery.mock(IBookingManager.class);
  mockRuleManager = mockery.mock(IRuleManager.class);
  mockery.checking(new Expectations() {
    {
      if (backupAllBookingsAndBookingRules) {
        // When backing everything up, we call through to the managers:
        oneOf(mockBookingManager).getAllBookings(with.booleanIs(anything()));
        will(returnValue(bookings));
        oneOf(mockRuleManager).getRules(with.booleanIs(anything()));
        will(returnValue(bookingRules));

        // Each type of backup uploads to a different S3 key:
        oneOf(mockTransferManager).upload(
            with(allOf(any(PutObjectRequest.class),
                hasProperty("key", equal("AllBookingsAndBookingRules")))));
        will(returnValue(mockTransfer));
      } else {
        ignoring(mockBookingManager);
        ignoring(mockRuleManager);

        if (backupSingleBookingRule) {
          oneOf(mockTransferManager).upload(
              with(allOf(any(PutObjectRequest.class),
                  hasProperty("key", equal("LatestBookingRule")))));
          will(returnValue(mockTransfer));
        } else {
          oneOf(mockTransferManager)
              .upload(
                  with(allOf(any(PutObjectRequest.class),
                      hasProperty("key", equal("LatestBooking")))));
          will(returnValue(mockTransfer));
        }
      }
    }
  });
  backupManager.initialise(mockBookingManager, mockRuleManager, mockLogger);
  backupManager.setS3TransferManager(mockTransferManager);

  // Set up mock SNS client
  mockSNSClient = mockery.mock(AmazonSNS.class);
  mockery.checking(new Expectations() {
    {
      ignoring(mockSNSClient);
    }
  });
  backupManager.setSNSClient(mockSNSClient);

  // ACT
  if (backupAllBookingsAndBookingRules) {
    backupManager.backupAllBookingsAndBookingRules();
  } else {
    if (backupSingleBookingRule) {
      backupManager.backupSingleBookingRule(bookingRule, isNotDeletion);
    } else {
      backupManager.backupSingleBooking(booking, isNotDeletion);
    }
  }
}

public void doTestBackupSingleBookingOrRuleCorrectlyCallsSNS(Boolean isBooking, Boolean isCreation)
    throws Exception {

  // Not interested in manager calls in this test
  mockBookingManager = mockery.mock(IBookingManager.class);
  mockRuleManager = mockery.mock(IRuleManager.class);
  mockery.checking(new Expectations() {
    {
      ignoring(mockBookingManager);
      ignoring(mockRuleManager);
    }
  });
  backupManager.initialise(mockBookingManager, mockRuleManager, mockLogger);

  // Not interested in S3 calls in this test
  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() {
    {
      allowing(mockTransferManager).upload(with(any(PutObjectRequest.class)));
      will(returnValue(mockTransfer));
    }
  });
  backupManager.setS3TransferManager(mockTransferManager);

  // Set up expectation to publish to our SNS topic with expected message and
  // subject - both of which differ between bookings and booking rules.
  String backupMessage;
  String backupSubject;
  if (!isBooking) {
    // Encode booking rule as JSON
    backupMessage = (isCreation ? "Booking rule updated: " : "Booking rule deleted: ")
        + System.getProperty("line.separator") + getExpectedBookingRuleJson(bookingRule);
    backupSubject = "Sqawsh single booking rule backup";
  } else {
    // Encode booking as JSON
    backupMessage = (isCreation ? "Booking created: " : "Booking deleted: ")
        + System.getProperty("line.separator") + getExpectedBookingJson(booking);
    backupSubject = "Sqawsh single booking backup";
  }

  // Set up mock SNS client
  mockSNSClient = mockery.mock(AmazonSNS.class);
  mockery.checking(new Expectations() {
    {
      oneOf(mockSNSClient).publish(with(equal(adminSnsTopicArn)), with(equal(backupMessage)),
          with(equal(backupSubject)));
    }
  });
  backupManager.setSNSClient(mockSNSClient);

  // ACT
  if (isBooking) {
    backupManager.backupSingleBooking(booking, isCreation);
  } else {
    backupManager.backupSingleBookingRule(bookingRule, isCreation);
  }
}

@Test
public void testBackupAllBookingsAndBookingRulesCorrectlyCallsSNS() throws Exception {

  // Add extra booking and booking rule to verify ALL bookings and rules are
  // backed up.
  Booking booking2 = new Booking(booking);
  // Tweak booking2 so it's different to booking
  booking2.setCourt(booking2.getCourt() + 1);
  bookings.add(booking2);
  BookingRule bookingRule2 = new BookingRule(bookingRule);
  // Tweak bookingRule2 so it's different to bookingRule
  bookingRule2.setIsRecurring(!bookingRule2.getIsRecurring());
  bookingRules.add(bookingRule2);

  // Set up mock managers
  mockBookingManager = mockery.mock(IBookingManager.class);
  mockRuleManager = mockery.mock(IRuleManager.class);
  mockery.checking(new Expectations() {
    {
      // When backing everything up, we call through to the managers:
      allowing(mockBookingManager).getAllBookings(with.booleanIs(anything()));
      will(returnValue(bookings));
      allowing(mockRuleManager).getRules(with.booleanIs(anything()));
      will(returnValue(bookingRules));
    }
  });
  backupManager.initialise(mockBookingManager, mockRuleManager, mockLogger);

  // Not interested in S3 calls in this test
  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() {
    {
      allowing(mockTransferManager).upload(with(any(PutObjectRequest.class)));
      will(returnValue(mockTransfer));
    }
  });
  backupManager.setS3TransferManager(mockTransferManager);

  // Set up mock SNS client
  String backupMessage = "{\"bookings\":[" + getExpectedBookingJson(booking) + ","
      + getExpectedBookingJson(booking2) + "],\"bookingRules\":["
      + getExpectedBookingRuleJson(bookingRule) + "," + getExpectedBookingRuleJson(bookingRule2)
      + "],\"clearBeforeRestore\":true}";
  mockSNSClient = mockery.mock(AmazonSNS.class);
  mockery.checking(new Expectations() {
    {
      oneOf(mockSNSClient).publish(with(equal(adminSnsTopicArn)), with(equal(backupMessage)),
          with(equal("Sqawsh all-bookings and booking rules backup")));
    }
  });
  backupManager.setSNSClient(mockSNSClient);

  // ACT
  backupManager.backupAllBookingsAndBookingRules();
}

@Test
public void testBackupAllBookingsAndBookingRulesCorrectlyCallsTheManagers() throws Exception {

  // Checks BackupManager correctly calls the booking and rule managers when
  // backing up everything. In particular they should say it's not making
  // service user calls - so calls get allowed even when not in ACTIVE
  // lifecycle state.

  // Set up mock managers
  mockBookingManager = mockery.mock(IBookingManager.class);
  mockRuleManager = mockery.mock(IRuleManager.class);
  mockery.checking(new Expectations() {
    {
      // When backing everything up, we call through to the managers:
      oneOf(mockBookingManager).getAllBookings(false);
      will(returnValue(bookings));
      oneOf(mockRuleManager).getRules(false);
      will(returnValue(bookingRules));
    }
  });
  backupManager.initialise(mockBookingManager, mockRuleManager, mockLogger);

  // Not interested in S3 calls in this test
  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() {
    {
      allowing(mockTransferManager).upload(with(any(PutObjectRequest.class)));
      will(returnValue(mockTransfer));
    }
  });
  backupManager.setS3TransferManager(mockTransferManager);

  // Not interested in SNS calls in this test
  mockSNSClient = mockery.mock(AmazonSNS.class);
  mockery.checking(new Expectations() {
    {
      ignoring(mockSNSClient);
    }
  });
  backupManager.setSNSClient(mockSNSClient);

  // ACT
  backupManager.backupAllBookingsAndBookingRules();
}

@Test
public void testBackupAllBookingsAndBookingRulesReturnsCorrectBookingsAndBookingRules()
    throws Exception {

  // Add extra booking and booking rule to verify ALL bookings and rules are
  // returned.
  Booking booking2 = new Booking(booking);
  // Tweak booking2 so it's different to booking
  booking2.setCourt(booking2.getCourt() + 1);
  bookings.add(booking2);
  BookingRule bookingRule2 = new BookingRule(bookingRule);
  // Tweak bookingRule2 so it's different to bookingRule
  bookingRule2.setIsRecurring(!bookingRule2.getIsRecurring());
  bookingRules.add(bookingRule2);

  // Set up mock managers
  mockBookingManager = mockery.mock(IBookingManager.class);
  mockRuleManager = mockery.mock(IRuleManager.class);
  mockery.checking(new Expectations() {
    {
      // When backing everything up, we call through to the managers:
      allowing(mockBookingManager).getAllBookings(with.booleanIs(anything()));
      will(returnValue(bookings));
      allowing(mockRuleManager).getRules(with.booleanIs(anything()));
      will(returnValue(bookingRules));
    }
  });
  backupManager.initialise(mockBookingManager, mockRuleManager, mockLogger);

  // Not interested in S3 calls in this test
  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() {
    {
      allowing(mockTransferManager).upload(with(any(PutObjectRequest.class)));
      will(returnValue(mockTransfer));
    }
  });
  backupManager.setS3TransferManager(mockTransferManager);

  // Not interested in SNS calls in this test
  mockSNSClient = mockery.mock(AmazonSNS.class);
  mockery.checking(new Expectations() {
    {
      ignoring(mockSNSClient);
    }
  });
  backupManager.setSNSClient(mockSNSClient);

  // ACT
  ImmutablePair<List<Booking>, List<BookingRule>> bookingsAndBookingRules = backupManager
      .backupAllBookingsAndBookingRules();

  // ASSERT
  // Verify we've got the correct bookings and booking rules.
  assertEquals("Unexpected bookings returned", bookingsAndBookingRules.left, bookings);
  assertEquals("Unexpected booking rules returned", bookingsAndBookingRules.right, bookingRules);
}

@Override
public S3FileTransferResultsDto copyFile(final S3FileCopyRequestParamsDto params) throws InterruptedException
{
    LOGGER
        .info("Copying S3 object... sourceS3Key=\"{}\" sourceS3BucketName=\"{}\" targetS3Key=\"{}\" targetS3BucketName=\"{}\"", params.getSourceObjectKey(),
            params.getSourceBucketName(), params.getTargetObjectKey(), params.getTargetBucketName());

    // Perform the copy.
    S3FileTransferResultsDto results = performTransfer(params, new Transferer()
    {
        @Override
        public Transfer performTransfer(TransferManager transferManager)
        {
            // Create a copy request.
            CopyObjectRequest copyObjectRequest =
                new CopyObjectRequest(params.getSourceBucketName(), params.getSourceObjectKey(), params.getTargetBucketName(), params.getTargetObjectKey());

            // If KMS Key ID is specified, set the AWS Key Management System parameters to be used to encrypt the object.
            if (StringUtils.isNotBlank(params.getKmsKeyId()))
            {
                copyObjectRequest.withSSEAwsKeyManagementParams(new SSEAwsKeyManagementParams(params.getKmsKeyId()));
            }
            // Otherwise, specify the server-side encryption algorithm for encrypting the object using AWS-managed keys.
            else
            {
                ObjectMetadata metadata = new ObjectMetadata();
                metadata.setSSEAlgorithm(ObjectMetadata.AES_256_SERVER_SIDE_ENCRYPTION);
                copyObjectRequest.setNewObjectMetadata(metadata);
            }

            return s3Operations.copyFile(copyObjectRequest, transferManager);
        }
    });

    LOGGER.info("Copied S3 object. sourceS3Key=\"{}\" sourceS3BucketName=\"{}\" targetS3Key=\"{}\" targetS3BucketName=\"{}\" " +
        "totalBytesTransferred={} transferDuration=\"{}\"", params.getSourceObjectKey(), params.getSourceBucketName(), params.getTargetObjectKey(),
        params.getTargetBucketName(), results.getTotalBytesTransferred(), HerdDateUtils.formatDuration(results.getDurationMillis()));

    logOverallTransferRate(results);

    return results;
}

@Override
public boolean uploadDirectoryOrFileAndListenProgress(final String bucketName, final File source,
        final String virtualDirectoryKeyPrefix)
        throws AmazonClientException, AmazonServiceException, FileNotFoundException {
    LOGGER.info("uploadDirectoryOrFileAndWaitForCompletion invoked, bucketName: {} , Source: {}", bucketName,
            source.getAbsolutePath());
    Transfer transfer = null;
    final TransferManager transferMgr = new TransferManager(s3client);
    if (source.isFile()) {
        transfer = transferMgr.upload(bucketName,source.getPath(),source);
    } else if (source.isDirectory()) {
        //upload recursively
        transfer = transferMgr.uploadDirectory(bucketName, virtualDirectoryKeyPrefix, source, true);
    } else {
        throw new FileNotFoundException("File is neither a regular file nor a directory " + source);
    }

    // You can poll your transfer's status to check its progress
    if (transfer.isDone()) {
        LOGGER.info("Start: {}  , State: {} and Progress (%): {}", transfer.getDescription(),
                transfer.getState(), transfer.getProgress().getPercentTransferred());

    }

    // Add progressListener to listen asynchronous notifications about your transfer's progress
    // Uncomment below code snippet during development
    /*transfer.addProgressListener(new ProgressListener() {
        public void progressChanged(ProgressEvent event) {
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("Transferred bytes: " + (long) event.getBytesTransferred());
            }
            }
    });*/

    try {
        //Block the current thread and wait for completion
        //If the transfer fails AmazonClientException will be thrown
        transfer.waitForCompletion();
    } catch (AmazonClientException | InterruptedException excp) {
        LOGGER.error("Exception occured while waiting for transfer: ",excp);
    }

    LOGGER.info("End: {} , State: {} , Progress (%): {}", transfer.getDescription(),
            transfer.getState(), transfer.getProgress().getPercentTransferred());
    return transfer.isDone();
}

@Override
public Transfer getTransfer() 
{
    return mUpload; 
}

@Override
public Transfer getTransfer() 
{
    return mUpload; 
}

/**
 * Perform a transfer using the specified transfer manager.
 *
 * @param transferManager the transfer manager.
 *
 * @return the transfer information for the transfer. This will typically be returned from an operation on the transfer manager (e.g. upload).
 */
public Transfer performTransfer(TransferManager transferManager);

/**
 * Upload directory or file.<br/>
 * S3 will overwrite any existing objects that happen to have the same key,
 * just as when uploading individual files, so use with caution.<br/>
 * This method will upload the files or directory to S3 asynchronously<br/>
 * You can use {@link com.amazonaws.services.s3.transfer.Transfer} returning object to check for progress,
 * Usage: <code>transfer.getProgress().getPercentTransferred()</code><br/>
 * It is implemented via {@link com.amazonaws.services.s3.transfer.TransferManager} <br/>
 * TransferManager provides a simple API for uploading content to Amazon S3,
 * and makes extensive use of Amazon S3 multipart uploads to achieve
 * enhanced throughput, performance and reliability. <br/>When possible,
 * TransferManager attempts to use multiple threads to upload multiple parts
 * of a single upload at once. <br/> When dealing with large content sizes and
 * high bandwidth, this can have a significant increase on throughput.
 *
 * @param bucketName the bucket name
 * @param source the source fir or directory
 * @param virtualDirectoryKeyPrefix the key prefix of the virtual directory to upload to. Use the null or empty string to upload files to the root of the bucket.
 * @return the transfer
 * @throws AmazonClientException the amazon client exception
 * @throws AmazonServiceException the amazon service exception
 * @throws IOException Signals that an I/O exception has occurred.
 */
Transfer uploadDirectoryOrFile(final String bucketName, final File source, final String virtualDirectoryKeyPrefix)
        throws AmazonClientException, AmazonServiceException, IOException;

public void transferStateChanged(Transfer transfer, TransferState state);