import org.jets3t.service.model.MultipartPart; //导入依赖的package包/类
@Override
public HttpResponseOutputStream<List<MultipartPart>> write(final Path file, final TransferStatus status, final ConnectionCallback callback) throws BackgroundException {
    final S3Object object = new S3WriteFeature(session, new S3DisabledMultipartService())
        .getDetails(file, status);
    // ID for the initiated multipart upload.
    final MultipartUpload multipart;
    try {
        multipart = session.getClient().multipartStartUpload(
            containerService.getContainer(file).getName(), object);
        if(log.isDebugEnabled()) {
            log.debug(String.format("Multipart upload started for %s with ID %s",
                multipart.getObjectKey(), multipart.getUploadId()));
        }
    }
    catch(ServiceException e) {
        throw new S3ExceptionMappingService().map("Upload {0} failed", e, file);
    }
    final MultipartOutputStream proxy = new MultipartOutputStream(multipart, file, status);
    return new HttpResponseOutputStream<List<MultipartPart>>(new MemorySegementingOutputStream(proxy,
        preferences.getInteger("s3.upload.multipart.partsize.minimum"))) {
        @Override
        public List<MultipartPart> getStatus() throws BackgroundException {
            return proxy.getCompleted();
        }
    };
}
 

import org.jets3t.service.model.MultipartPart; //导入依赖的package包/类
/**
 * @return No Content-Range support
 */
@Override
public Append append(final Path file, final Long length, final Cache<Path> cache) throws BackgroundException {
    if(length >= preferences.getLong("s3.upload.multipart.threshold")) {
        if(preferences.getBoolean("s3.upload.multipart")) {
            try {
                final List<MultipartUpload> upload = multipartService.find(file);
                if(!upload.isEmpty()) {
                    Long size = 0L;
                    for(MultipartPart completed : multipartService.list(upload.iterator().next())) {
                        size += completed.getSize();
                    }
                    return new Append(size);
                }
            }
            catch(AccessDeniedException | InteroperabilityException e) {
                log.warn(String.format("Ignore failure listing incomplete multipart uploads. %s", e.getDetail()));
            }
        }
    }
    if(finder.withCache(cache).find(file)) {
        final PathAttributes attributes = this.attributes.withCache(cache).find(file);
        return new Append(false, true).withSize(attributes.getSize()).withChecksum(attributes.getChecksum());
    }
    return Write.notfound;
}
 

import org.jets3t.service.model.MultipartPart; //导入依赖的package包/类
@Override
public List<MultipartPart> list(final MultipartUpload multipart) throws BackgroundException {
    if(log.isInfoEnabled()) {
        log.info(String.format("List completed parts of %s", multipart.getUploadId()));
    }
    // This operation lists the parts that have been uploaded for a specific multipart upload.
    try {
        return session.getClient().multipartListParts(multipart);
    }
    catch(S3ServiceException e) {
        throw new S3ExceptionMappingService().map(MessageFormat.format("Upload {0} failed", multipart.getObjectKey()), e);
    }
}
 

import org.jets3t.service.model.MultipartPart; //导入依赖的package包/类
private Future<MultipartPart> submit(final Path source,
                                     final MultipartUpload multipart,
                                     final int partNumber, final long offset, final long length) throws BackgroundException {
    if(log.isInfoEnabled()) {
        log.info(String.format("Submit part %d of %s to queue with offset %d and length %d", partNumber, source, offset, length));
    }
    return pool.execute(new Callable<MultipartPart>() {
        @Override
        public MultipartPart call() throws BackgroundException {
            try {
                final HttpRange range = HttpRange.byLength(offset, length);
                final MultipartPart part = session.getClient().multipartUploadPartCopy(multipart, partNumber,
                        containerService.getContainer(source).getName(), containerService.getKey(source),
                        null, null, null, null, range.getStart(), range.getEnd(), source.attributes().getVersionId());
                if(log.isInfoEnabled()) {
                    log.info(String.format("Received response %s for part number %d", part, partNumber));
                }
                // Populate part with response data that is accessible via the object's metadata
                return new MultipartPart(partNumber,
                        null == part.getLastModified() ? new Date(System.currentTimeMillis()) : part.getLastModified(),
                        null == part.getEtag() ? StringUtils.EMPTY : part.getEtag(),
                        part.getSize());
            }
            catch(S3ServiceException e) {
                throw new S3ExceptionMappingService().map("Cannot copy {0}", e, source);
            }
        }
    });
}
 

import org.jets3t.service.model.MultipartPart; //导入依赖的package包/类
@Test
public void testWrite() throws Exception {
    final S3Session session = new S3Session(
            new Host(new S3Protocol(), new S3Protocol().getDefaultHostname(),
                    new Credentials(
                            System.getProperties().getProperty("s3.key"), System.getProperties().getProperty("s3.secret")
                    )));
    session.open(new DisabledHostKeyCallback(), new DisabledLoginCallback());
    session.login(new DisabledPasswordStore(), new DisabledLoginCallback(), new DisabledCancelCallback());
    final Path home = new Path("test-eu-central-1-cyberduck", EnumSet.of(Path.Type.volume, Path.Type.directory));
    final Path vault = new Path(home, new AlphanumericRandomStringService().random(), EnumSet.of(Path.Type.directory));
    final Path test = new Path(vault, new AlphanumericRandomStringService().random(), EnumSet.of(Path.Type.file));
    final CryptoVault cryptomator = new CryptoVault(vault, new DisabledPasswordStore());
    cryptomator.create(session, null, new VaultCredentials("test"));
    session.withRegistry(new DefaultVaultRegistry(new DisabledPasswordStore(), new DisabledPasswordCallback(), cryptomator));
    final CryptoWriteFeature feature = new CryptoWriteFeature<List<MultipartPart>>(session, new S3MultipartWriteFeature(session), cryptomator);
    final byte[] content = RandomUtils.nextBytes(6 * 1024 * 1024);
    final TransferStatus writeStatus = new TransferStatus();
    final Cryptor cryptor = cryptomator.getCryptor();
    final FileHeader header = cryptor.fileHeaderCryptor().create();
    writeStatus.setHeader(cryptor.fileHeaderCryptor().encryptHeader(header));
    writeStatus.setNonces(new RandomNonceGenerator());
    writeStatus.setLength(-1L);
    final StatusOutputStream out = feature.write(test, writeStatus, new DisabledConnectionCallback());
    final ByteArrayInputStream in = new ByteArrayInputStream(content);
    final TransferStatus progress = new TransferStatus();
    new StreamCopier(new TransferStatus(), progress).transfer(in, out);
    assertEquals(content.length, progress.getOffset());
    assertNotNull(out.getStatus());
    assertTrue(new CryptoFindFeature(session, new S3FindFeature(session), cryptomator).find(test));
    final byte[] compare = new byte[content.length];
    final InputStream stream = new CryptoReadFeature(session, new S3ReadFeature(session), cryptomator).read(test, new TransferStatus().length(content.length), new DisabledConnectionCallback());
    IOUtils.readFully(stream, compare);
    stream.close();
    assertArrayEquals(content, compare);
    new CryptoDeleteFeature(session, new S3DefaultDeleteFeature(session), cryptomator).delete(Arrays.asList(test, vault), new DisabledLoginCallback(), new Delete.DisabledCallback());
    session.close();
}
 

import org.jets3t.service.model.MultipartPart; //导入依赖的package包/类
@Test
public void testWrite() throws Exception {
    final S3Session session = new S3Session(
            new Host(new S3Protocol(), new S3Protocol().getDefaultHostname(),
                    new Credentials(
                            System.getProperties().getProperty("s3.key"), System.getProperties().getProperty("s3.secret")
                    )));
    session.open(new DisabledHostKeyCallback(), new DisabledLoginCallback());
    session.login(new DisabledPasswordStore(), new DisabledLoginCallback(), new DisabledCancelCallback());
    final S3MultipartWriteFeature feature = new S3MultipartWriteFeature(session);
    final Path container = new Path("test-eu-central-1-cyberduck", EnumSet.of(Path.Type.volume));
    final TransferStatus status = new TransferStatus();
    status.setLength(-1L);
    final Path file = new Path(container, UUID.randomUUID().toString(), EnumSet.of(Path.Type.file));
    final HttpResponseOutputStream<List<MultipartPart>> out = feature.write(file, status, new DisabledConnectionCallback());
    final byte[] content = RandomUtils.nextBytes(6 * 1024 * 1024);
    final ByteArrayInputStream in = new ByteArrayInputStream(content);
    final TransferStatus progress = new TransferStatus();
    new StreamCopier(new TransferStatus(), progress).transfer(in, out);
    assertEquals(content.length, progress.getOffset());
    in.close();
    out.close();
    assertNotNull(out.getStatus());
    assertTrue(new S3FindFeature(session).find(file));
    final byte[] compare = new byte[content.length];
    final InputStream stream = new S3ReadFeature(session).read(file, new TransferStatus().length(content.length), new DisabledConnectionCallback());
    IOUtils.readFully(stream, compare);
    stream.close();
    assertArrayEquals(content, compare);
    new S3DefaultDeleteFeature(session).delete(Collections.singletonList(file), new DisabledLoginCallback(), new Delete.DisabledCallback());
    session.close();
}
 

import org.jets3t.service.model.MultipartPart; //导入依赖的package包/类
@Test
public void testWriteZeroLength() throws Exception {
    final S3Session session = new S3Session(
            new Host(new S3Protocol(), new S3Protocol().getDefaultHostname(),
                    new Credentials(
                            System.getProperties().getProperty("s3.key"), System.getProperties().getProperty("s3.secret")
                    )));
    session.open(new DisabledHostKeyCallback(), new DisabledLoginCallback());
    session.login(new DisabledPasswordStore(), new DisabledLoginCallback(), new DisabledCancelCallback());
    final S3MultipartWriteFeature feature = new S3MultipartWriteFeature(session);
    final Path container = new Path("test-eu-central-1-cyberduck", EnumSet.of(Path.Type.volume));
    final byte[] content = RandomUtils.nextBytes(0);
    final TransferStatus status = new TransferStatus();
    status.setLength(-1L);
    final Path file = new Path(container, new AlphanumericRandomStringService().random(), EnumSet.of(Path.Type.file));
    final HttpResponseOutputStream<List<MultipartPart>> out = feature.write(file, status, new DisabledConnectionCallback());
    final ByteArrayInputStream in = new ByteArrayInputStream(content);
    assertEquals(content.length, IOUtils.copyLarge(in, out));
    in.close();
    out.close();
    assertNotNull(out.getStatus());
    assertTrue(new DefaultFindFeature(session).find(file));
    final byte[] compare = new byte[content.length];
    final InputStream stream = new S3ReadFeature(session).read(file, new TransferStatus().length(content.length), new DisabledConnectionCallback());
    IOUtils.readFully(stream, compare);
    stream.close();
    assertArrayEquals(content, compare);
    new S3DefaultDeleteFeature(session).delete(Collections.singletonList(file), new DisabledLoginCallback(), new Delete.DisabledCallback());
}
 

import org.jets3t.service.model.MultipartPart; //导入依赖的package包/类
public void copyLargeFile(S3Object srcObject, String dstKey) throws IOException {
  try {
    long partCount = srcObject.getContentLength() / multipartCopyBlockSize +
        (srcObject.getContentLength() % multipartCopyBlockSize > 0 ? 1 : 0);

    MultipartUpload multipartUpload = s3Service.multipartStartUpload
        (bucket.getName(), dstKey, srcObject.getMetadataMap());

    List<MultipartPart> listedParts = new ArrayList<MultipartPart>();
    for (int i = 0; i < partCount; i++) {
      long byteRangeStart = i * multipartCopyBlockSize;
      long byteLength;
      if (i < partCount - 1) {
        byteLength = multipartCopyBlockSize;
      } else {
        byteLength = srcObject.getContentLength() % multipartCopyBlockSize;
        if (byteLength == 0) {
          byteLength = multipartCopyBlockSize;
        }
      }

      MultipartPart copiedPart = s3Service.multipartUploadPartCopy
          (multipartUpload, i + 1, bucket.getName(), srcObject.getKey(),
           null, null, null, null, byteRangeStart,
           byteRangeStart + byteLength - 1, null);
      listedParts.add(copiedPart);
    }
    
    Collections.reverse(listedParts);
    s3Service.multipartCompleteUpload(multipartUpload, listedParts);
  } catch (ServiceException e) {
    handleException(e, srcObject.getKey());
  }
}
 

import org.jets3t.service.model.MultipartPart; //导入依赖的package包/类
protected abstract MultipartPart multipartUploadPartCopyImpl(String uploadId,
String targetBucketName, String targetObjectKey, Integer partNumber,
String sourceBucketName, String sourceObjectKey,
Calendar ifModifiedSince, Calendar ifUnmodifiedSince,
String[] ifMatchTags, String[] ifNoneMatchTags,
Long byteRangeStart, Long byteRangeEnd,
String versionId) throws S3ServiceException;
 

import org.jets3t.service.model.MultipartPart; //导入依赖的package包/类
@Override
protected List<MultipartPart> multipartListPartsImpl(String uploadId,
    String bucketName, String objectKey) throws S3ServiceException
{
    Map<String, String> requestParameters = new HashMap<String, String>();
    requestParameters.put("uploadId", uploadId);
    requestParameters.put("max-parts", "1000");

    try {
        List<MultipartPart> parts = new ArrayList<MultipartPart>();
        String nextPartNumberMarker = null;
        boolean incompleteListing = true;
        do {
            if (nextPartNumberMarker != null) {
                requestParameters.put("part-number-marker", nextPartNumberMarker);
            } else {
                requestParameters.remove("part-number-marker");
            }

            HttpResponse httpResponse = performRestGet(bucketName, objectKey, requestParameters, null);
            ListMultipartPartsResultHandler handler = getXmlResponseSaxParser()
                .parseListMultipartPartsResult(
                    new HttpMethodReleaseInputStream(httpResponse));
            parts.addAll(handler.getMultipartPartList());

            incompleteListing = handler.isTruncated();
            nextPartNumberMarker = handler.getNextPartNumberMarker();

            // Sanity check for valid pagination values.
            if (incompleteListing && nextPartNumberMarker == null)
            {
                throw new ServiceException("Unable to retrieve paginated "
                    + "ListMultipartPartsResult without valid NextKeyMarker value.");
            }
        } while (incompleteListing);
        return parts;
    } catch (ServiceException se) {
        throw new S3ServiceException(se);
    }
}
 

import org.jets3t.service.model.MultipartPart; //导入依赖的package包/类
public MultipartPart parseMultipartUploadPartCopyResult(InputStream inputStream)
    throws ServiceException
{
    MultipartPartResultHandler handler = new MultipartPartResultHandler(xr);
    parseXmlInputStream(handler, sanitizeXmlDocument(handler, inputStream));
    return handler.getMultipartPart();
}
 

import org.jets3t.service.model.MultipartPart; //导入依赖的package包/类
public void copyLargeFile(S3Object srcObject, String dstKey) throws IOException {
  try {
    long partCount = srcObject.getContentLength() / multipartCopyBlockSize +
        (srcObject.getContentLength() % multipartCopyBlockSize > 0 ? 1 : 0);

    MultipartUpload multipartUpload = s3Service.multipartStartUpload
        (bucket.getName(), dstKey, srcObject.getMetadataMap());

    List<MultipartPart> listedParts = new ArrayList<MultipartPart>();
    for (int i = 0; i < partCount; i++) {
      long byteRangeStart = i * multipartCopyBlockSize;
      long byteLength;
      if (i < partCount - 1) {
        byteLength = multipartCopyBlockSize;
      } else {
        byteLength = srcObject.getContentLength() % multipartCopyBlockSize;
        if (byteLength == 0) {
          byteLength = multipartCopyBlockSize;
        }
      }

      MultipartPart copiedPart = s3Service.multipartUploadPartCopy
          (multipartUpload, i + 1, bucket.getName(), srcObject.getKey(),
           null, null, null, null, byteRangeStart,
           byteRangeStart + byteLength - 1, null);
      listedParts.add(copiedPart);
    }
    
    Collections.reverse(listedParts);
    s3Service.multipartCompleteUpload(multipartUpload, listedParts);
  } catch (ServiceException e) {
    handleServiceException(e);
  }
}
 

import org.jets3t.service.model.MultipartPart; //导入依赖的package包/类
public List<MultipartPart> getCompleted() {
    return completed;
}
 

import org.jets3t.service.model.MultipartPart; //导入依赖的package包/类
private Future<MultipartPart> submit(final ThreadPool pool, final Path file, final Local local,
                                     final BandwidthThrottle throttle, final StreamListener listener,
                                     final TransferStatus overall, final MultipartUpload multipart,
                                     final int partNumber, final long offset, final long length, final ConnectionCallback callback) throws BackgroundException {
    if(log.isInfoEnabled()) {
        log.info(String.format("Submit part %d of %s to queue with offset %d and length %d", partNumber, file, offset, length));
    }
    return pool.execute(new DefaultRetryCallable<MultipartPart>(new BackgroundExceptionCallable<MultipartPart>() {
        @Override
        public MultipartPart call() throws BackgroundException {
            if(overall.isCanceled()) {
                throw new ConnectionCanceledException();
            }
            final Map<String, String> requestParameters = new HashMap<String, String>();
            requestParameters.put("uploadId", multipart.getUploadId());
            requestParameters.put("partNumber", String.valueOf(partNumber));
            final TransferStatus status = new TransferStatus()
                    .length(length)
                    .skip(offset)
                    .withParameters(requestParameters);
            status.setHeader(overall.getHeader());
            status.setNonces(overall.getNonces());
            switch(session.getSignatureVersion()) {
                case AWS4HMACSHA256:
                    status.setChecksum(writer.checksum(file)
                            .compute(StreamCopier.skip(new BoundedInputStream(local.getInputStream(), offset + length), offset), status)
                    );
                    break;
            }
            status.setSegment(true);
            final StorageObject part = S3MultipartUploadService.super.upload(
                    file, local, throttle, listener, status, overall, new StreamProgress() {
                        @Override
                        public void progress(final long bytes) {
                            status.progress(bytes);
                            // Discard sent bytes in overall progress if there is an error reply for segment.
                            overall.progress(bytes);
                        }

                        @Override
                        public void setComplete() {
                            status.setComplete();
                        }
                    }, callback);
            if(log.isInfoEnabled()) {
                log.info(String.format("Received response %s for part number %d", part, partNumber));
            }
            // Populate part with response data that is accessible via the object's metadata
            return new MultipartPart(partNumber,
                    null == part.getLastModifiedDate() ? new Date(System.currentTimeMillis()) : part.getLastModifiedDate(),
                    null == part.getETag() ? StringUtils.EMPTY : part.getETag(),
                    part.getContentLength());

        }
    }, overall));
}
 

import org.jets3t.service.model.MultipartPart; //导入依赖的package包/类
@Override
public List<MultipartPart> list(final MultipartUpload multipart) throws BackgroundException {
    return Collections.emptyList();
}
 

import org.jets3t.service.model.MultipartPart; //导入依赖的package包/类
/**
 * From an existing object, copy an individual part that will comprise a piece of a
 * multipart upload object.
 *
 * @param upload
 * the multipart upload to which this part will be added.
 * @param partNumber
 * the part's number; must be between 1 and 10,000 and must uniquely identify a given
 * part and represent its order compared to all other parts. Part numbers need not
 * be sequential.
 * @param sourceBucketName
 * the name of the bucket that contains the original object.
 * @param sourceObjectKey
 * the key name of the original object.
 * @param ifModifiedSince
 * a precondition specifying a date after which the source object must have been
 * modified, ignored if null.
 * @param ifUnmodifiedSince
 * a precondition specifying a date after which the source object must not have
 * been modified, ignored if null.
 * @param ifMatchTags
 * a precondition specifying an MD5 hash the source object must match, ignored if
 * null.
 * @param ifNoneMatchTags
 * a precondition specifying an MD5 hash the source object must not match, ignored
 * if null.
 * @param byteRangeStart
 * include only a portion of the source object's data - starting at this point, ignored if null.
 * Byte ranges may only be used for source objects larger than 5 GB.
 * @param byteRangeEnd
 * include only a portion of the source object's data - ending at this point, ignored if null.
 * Byte ranges may only be used for source objects larger than 5 GB.
 * @param versionId
 * identifier matching an existing source object version that will be retrieved.
 *
 * @return
 * information about the uploaded copy part, retain this information to eventually complete
 * the object with {@link #multipartCompleteUpload(MultipartUpload, List)}.
 * @throws S3ServiceException
 */
public MultipartPart multipartUploadPartCopy(MultipartUpload upload, Integer partNumber,
    String sourceBucketName, String sourceObjectKey,
    Calendar ifModifiedSince, Calendar ifUnmodifiedSince,
    String[] ifMatchTags, String[] ifNoneMatchTags,
    Long byteRangeStart, Long byteRangeEnd,
    String versionId) throws S3ServiceException
{
    MultipartPart part = multipartUploadPartCopyImpl(upload.getUploadId(),
        upload.getBucketName(), upload.getObjectKey(), partNumber,
        sourceBucketName, sourceObjectKey,
        ifModifiedSince, ifUnmodifiedSince,
        ifMatchTags, ifNoneMatchTags,
        byteRangeStart, byteRangeEnd, versionId);
    upload.addMultipartPartToUploadedList(part);
    return part;
}
 

import org.jets3t.service.model.MultipartPart; //导入依赖的package包/类
protected abstract List<MultipartPart> multipartListPartsImpl(String uploadId,
String bucketName, String objectKey) throws S3ServiceException;
 

import org.jets3t.service.model.MultipartPart; //导入依赖的package包/类
protected abstract MultipartCompleted multipartCompleteUploadImpl(String uploadId, String bucketName,
String objectKey, List<MultipartPart> parts) throws S3ServiceException;
 

import org.jets3t.service.model.MultipartPart; //导入依赖的package包/类
protected abstract MultipartPart multipartUploadPartImpl(String uploadId, String bucketName,
Integer partNumber, S3Object object) throws S3ServiceException;
 

import org.jets3t.service.model.MultipartPart; //导入依赖的package包/类
public MultipartPart getMultipartPart() {
    return new MultipartPart(partNumber, lastModified, etag, size);
}