import com.amazonaws.event.ProgressEventType; //导入依赖的package包/类
/**
 * Performs the copy of an Amazon S3 object from source bucket to
 * destination bucket as multiple copy part requests. The information about
 * the part to be copied is specified in the request as a byte range
 * (first-last)
 *
 * @throws Exception
 *             Any Exception that occurs while carrying out the request.
 */
private void copyInParts() throws Exception {
    multipartUploadId = initiateMultipartUpload(copyObjectRequest);

    long optimalPartSize = getOptimalPartSize(metadata.getContentLength());

    try {
        CopyPartRequestFactory requestFactory = new CopyPartRequestFactory(
                copyObjectRequest, multipartUploadId, optimalPartSize,
                metadata.getContentLength());
        copyPartsInParallel(requestFactory);
    } catch (Exception e) {
        publishProgress(listenerChain, ProgressEventType.TRANSFER_FAILED_EVENT);
        abortMultipartCopy();
        throw new RuntimeException("Unable to perform multipart copy", e);
    }
}
 

import com.amazonaws.event.ProgressEventType; //导入依赖的package包/类
/**
 * This method is also responsible for firing COMPLETED signal to the
 * listeners.
 */
@Override
public void setState(TransferState state) {
    super.setState(state);

    switch (state) {
        case Completed :
            fireProgressEvent(ProgressEventType.TRANSFER_COMPLETED_EVENT);
            break;
        case Canceled:
            fireProgressEvent(ProgressEventType.TRANSFER_CANCELED_EVENT);
            break;
        case Failed:
            fireProgressEvent(ProgressEventType.TRANSFER_FAILED_EVENT);
            break;
        default:
            break;
    }
}
 

import com.amazonaws.event.ProgressEventType; //导入依赖的package包/类
public void progressChanged(ProgressEvent progressEvent) {
  if (progress != null) {
    progress.progress();
  }

  // There are 3 http ops here, but this should be close enough for now
  ProgressEventType pet = progressEvent.getEventType();
  if (pet == TRANSFER_PART_STARTED_EVENT ||
      pet == TRANSFER_COMPLETED_EVENT) {
    statistics.incrementWriteOps(1);
  }

  long transferred = upload.getProgress().getBytesTransferred();
  long delta = transferred - lastBytesTransferred;
  if (statistics != null && delta != 0) {
    statistics.incrementBytesWritten(delta);
  }

  lastBytesTransferred = transferred;
}
 

import com.amazonaws.event.ProgressEventType; //导入依赖的package包/类
@Override
public void progressChanged(ProgressEvent event)
{
	if ( event.getEventType() == ProgressEventType.REQUEST_CONTENT_LENGTH_EVENT )
	{
		contentLength = event.getBytes();
		getLog().info("Content size: " + contentLength + " bytes");
	}
	else if ( event.getEventType() == ProgressEventType.REQUEST_BYTE_TRANSFER_EVENT )
	{
		contentSent += event.getBytesTransferred();
		double div = (double) (((double)contentSent/(double)contentLength));
		double mul = div*(double)100.0;
		int mod = (int)mul / 10;
		if ( mod > lastTenPct )
		{
			lastTenPct = mod;
			getLog().info("Uploaded " + (mod*10) + "% of " + (contentLength/(1024*1024)) + " MB");
		}
	}
}
 

import com.amazonaws.event.ProgressEventType; //导入依赖的package包/类
@Bean
public S3ProgressListener s3ProgressListener() {
	return new S3ProgressListener() {

		@Override
		public void onPersistableTransfer(PersistableTransfer persistableTransfer) {

		}

		@Override
		public void progressChanged(ProgressEvent progressEvent) {
			if (ProgressEventType.TRANSFER_COMPLETED_EVENT.equals(progressEvent.getEventType())) {
				transferCompletedLatch().countDown();
			}
		}

	};
}
 

import com.amazonaws.event.ProgressEventType; //导入依赖的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.event.ProgressEventType; //导入依赖的package包/类
@Override
public void progressChanged(ProgressEvent progressEvent) {
    ProgressEventType type = progressEvent.getEventType();
    if (type.equals(TRANSFER_COMPLETED_EVENT) || type.equals(TRANSFER_STARTED_EVENT)) {
        out.println();
    }

    if (type.isByteCountEvent()) {
        long timeLeft = getTimeLeft();
        if (lastTimeLeft < 1 && timeLeft > 0) {
            // prime this value with a sane starting point
            lastTimeLeft = timeLeft;
        }

        // use an exponential moving average to smooth the estimate
        lastTimeLeft += 0.90 * (timeLeft - lastTimeLeft);

        out.print(String.format("\r%1$s  %2$s / %3$s  %4$s      ",
                generate(saturatedCast(round(completed + (progress.getPercentTransferred() * multiplier)))),
                humanReadableByteCount(progress.getBytesTransferred(), true),
                humanReadableByteCount(progress.getTotalBytesToTransfer(), true), fromSeconds(lastTimeLeft)));
        out.flush();
    }
}
 

import com.amazonaws.event.ProgressEventType; //导入依赖的package包/类
public MessageDigest copyAndHash(InputStream input, long totalBytes, Progress progress)
        throws IOException, CloneNotSupportedException {

    // clone the current digest, such that it remains unchanged in this method
    MessageDigest computedDigest = (MessageDigest) currentDigest.clone();
    byte[] buffer = new byte[DEFAULT_BUF_SIZE];

    long count = 0;
    int n;
    while (-1 != (n = input.read(buffer))) {
        output.write(buffer, 0, n);
        if (progressListener != null) {
            progress.updateProgress(n);
            progressListener.progressChanged(new ProgressEvent(ProgressEventType.RESPONSE_BYTE_TRANSFER_EVENT, n));
        }
        computedDigest.update(buffer, 0, n);
        count += n;
    }

    // verify that at least this many bytes were read
    if (totalBytes != count) {
        throw new IOException(String.format("%d bytes downloaded instead of expected %d bytes", count, totalBytes));
    }
    return computedDigest;
}
 

import com.amazonaws.event.ProgressEventType; //导入依赖的package包/类
public MessageDigest copyAndHash(InputStream input, long totalBytes, Progress progress)
        throws IOException, CloneNotSupportedException {

    // clone the current digest, such that it remains unchanged in this method
    MessageDigest computedDigest = (MessageDigest) currentDigest.clone();
    byte[] buffer = new byte[DEFAULT_BUF_SIZE];

    long count = 0;
    int n;
    while (-1 != (n = input.read(buffer))) {
        output.write(buffer, 0, n);
        if (progressListener != null) {
            progress.updateProgress(n);
            progressListener.progressChanged(new ProgressEvent(ProgressEventType.REQUEST_BYTE_TRANSFER_EVENT, n));
        }
        computedDigest.update(buffer, 0, n);
        count += n;
    }

    // verify that at least this many bytes were read
    if (totalBytes != count) {
        throw new IOException(String.format("%d bytes downloaded instead of expected %d bytes", count, totalBytes));
    }
    return computedDigest;
}
 

import com.amazonaws.event.ProgressEventType; //导入依赖的package包/类
@Override
public void progressChanged(ProgressEvent progressEvent) {

    if (progressEvent.getEventType() == ProgressEventType.REQUEST_CONTENT_LENGTH_EVENT) {
        partSize = progressEvent.getBytes();
        ArchiveUploadHighLevel.this.log.info("Part size: " + partSize);
    }

    if (progressEvent.getEventType() == ProgressEventType.CLIENT_REQUEST_SUCCESS_EVENT) {
        counter += partSize;
        int percentage = (int)(counter * 100.0 / total);
        ArchiveUploadHighLevel.this.log.info("Successfully transferred: " + counter + " / " + total + " (" + percentage + "%)");
    }
}
 

import com.amazonaws.event.ProgressEventType; //导入依赖的package包/类
private Response<Output> doExecute() throws InterruptedException {
    runBeforeRequestHandlers();
    setSdkTransactionId(request);
    setUserAgent(request);

    ProgressListener listener = requestConfig.getProgressListener();
    // add custom headers
    request.getHeaders().putAll(config.getHeaders());
    request.getHeaders().putAll(requestConfig.getCustomRequestHeaders());
    // add custom query parameters
    mergeQueryParameters(requestConfig.getCustomQueryParameters());
    Response<Output> response = null;
    final InputStream origContent = request.getContent();
    final InputStream toBeClosed = beforeRequest(); // for progress tracking
    // make "notCloseable", so reset would work with retries
    final InputStream notCloseable = (toBeClosed == null) ? null
            : ReleasableInputStream.wrap(toBeClosed).disableClose();
    request.setContent(notCloseable);
    try {
        publishProgress(listener, ProgressEventType.CLIENT_REQUEST_STARTED_EVENT);
        response = executeHelper();
        publishProgress(listener, ProgressEventType.CLIENT_REQUEST_SUCCESS_EVENT);
        awsRequestMetrics.getTimingInfo().endTiming();
        afterResponse(response);
        return response;
    } catch (AmazonClientException e) {
        publishProgress(listener, ProgressEventType.CLIENT_REQUEST_FAILED_EVENT);
        afterError(response, e);
        throw e;
    } finally {
        // Always close so any progress tracking would get the final events propagated.
        closeQuietly(toBeClosed, log);
        request.setContent(origContent); // restore the original content
    }
}
 

import com.amazonaws.event.ProgressEventType; //导入依赖的package包/类
/**
 * Pause before the next retry and record metrics around retry behavior.
 */
private void pauseBeforeRetry(ExecOneRequestParams execOneParams,
                              final ProgressListener listener) throws InterruptedException {
    publishProgress(listener, ProgressEventType.CLIENT_REQUEST_RETRY_EVENT);
    // Notify the progress listener of the retry
    awsRequestMetrics.startEvent(Field.RetryPauseTime);
    try {
        doPauseBeforeRetry(execOneParams);
    } finally {
        awsRequestMetrics.endEvent(Field.RetryPauseTime);
    }
}
 

import com.amazonaws.event.ProgressEventType; //导入依赖的package包/类
@Override
public void progressChanged(ProgressEvent progressEvent) {
    ProgressEventType type = progressEvent.getEventType();
    if (type.isByteCountEvent())
        return;
    if (type != types[count]) {
        throw new AssertionError("Expect event type "
                + types[count] + " but got "
                + progressEvent.getEventType());
    }
    count++;
}
 

import com.amazonaws.event.ProgressEventType; //导入依赖的package包/类
@Override
public ProgressEvent filter(ProgressEvent progressEvent) {
    // Block COMPLETE events from the low-level GetObject operation,
    // but we still want to keep the BytesTransferred
    return progressEvent.getEventType() == ProgressEventType.TRANSFER_COMPLETED_EVENT
         ? null // discard this event
         : progressEvent
         ;
}
 

import com.amazonaws.event.ProgressEventType; //导入依赖的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.event.ProgressEventType; //导入依赖的package包/类
/**
 * Override this method so that TransferState updates are also sent out to the
 * progress listener chain in forms of ProgressEvent.
 */
@Override
public void setState(TransferState state) {
    super.setState(state);

    switch (state) {
    case Waiting:
        fireProgressEvent(ProgressEventType.TRANSFER_PREPARING_EVENT);
        break;
    case InProgress:
        if ( subTransferStarted.compareAndSet(false, true) ) {
            /* The first InProgress signal */
            fireProgressEvent(ProgressEventType.TRANSFER_STARTED_EVENT);
        }
        /* Don't need any event code update for subsequent InProgress signals */
        break;
    case Completed:
        fireProgressEvent(ProgressEventType.TRANSFER_COMPLETED_EVENT);
        break;
    case Canceled:
        fireProgressEvent(ProgressEventType.TRANSFER_CANCELED_EVENT);
        break;
    case Failed:
        fireProgressEvent(ProgressEventType.TRANSFER_FAILED_EVENT);
        break;
    default:
        break;
    }
}
 

import com.amazonaws.event.ProgressEventType; //导入依赖的package包/类
@Override
public UploadResult call() throws Exception {
    CompleteMultipartUploadResult res;

    try {
        CompleteMultipartUploadRequest req = new CompleteMultipartUploadRequest(
                origReq.getBucketName(), origReq.getKey(), uploadId,
                collectPartETags())
                .withRequesterPays(origReq.isRequesterPays())
            .withGeneralProgressListener(origReq.getGeneralProgressListener())
            .withRequestMetricCollector(origReq.getRequestMetricCollector())
            ;
        res = s3.completeMultipartUpload(req);
    } catch (Exception e) {
        publishProgress(listener, ProgressEventType.TRANSFER_FAILED_EVENT);
        throw e;
    }

    UploadResult uploadResult = new UploadResult();
    uploadResult.setBucketName(origReq
            .getBucketName());
    uploadResult.setKey(origReq.getKey());
    uploadResult.setETag(res.getETag());
    uploadResult.setVersionId(res.getVersionId());

    monitor.uploadComplete();

    return uploadResult;
}
 

import com.amazonaws.event.ProgressEventType; //导入依赖的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.event.ProgressEventType; //导入依赖的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.event.ProgressEventType; //导入依赖的package包/类
public UploadResult call() throws Exception {
    upload.setState(TransferState.InProgress);
    if ( isMultipartUpload() ) {
        publishProgress(listener, ProgressEventType.TRANSFER_STARTED_EVENT);
        return uploadInParts();
    } else {
        return uploadInOneChunk();
    }
}