Java 类org.apache.commons.io.input.CountingInputStream 实例源码

@Test
public void testReadCloseReleaseEntity() throws Exception {
    final Host host = new Host(new SwiftProtocol(), "identity.api.rackspacecloud.com", new Credentials(
            System.getProperties().getProperty("rackspace.key"), System.getProperties().getProperty("rackspace.secret")
    ));
    final SwiftSession session = new SwiftSession(host);
    session.open(new DisabledHostKeyCallback(), new DisabledLoginCallback());
    session.login(new DisabledPasswordStore(), new DisabledLoginCallback(), new DisabledCancelCallback());
    final TransferStatus status = new TransferStatus();
    final Path container = new Path(".ACCESS_LOGS", EnumSet.of(Path.Type.directory, Path.Type.volume));
    container.attributes().setRegion("DFW");
    final SwiftRegionService regionService = new SwiftRegionService(session);
    final CountingInputStream in = new CountingInputStream(new SwiftReadFeature(session, regionService).read(new Path(container,
            "/cdn.cyberduck.ch/2015/03/01/10/3b1d6998c430d58dace0c16e58aaf925.log.gz",
            EnumSet.of(Path.Type.file)), status, new DisabledConnectionCallback()));
    in.close();
    assertEquals(0L, in.getByteCount(), 0L);
    session.close();
}

@Test
public void testReadCloseReleaseEntity() throws Exception {
    final Host host = new Host(new SDSProtocol(), "duck.ssp-europe.eu", new Credentials(
        System.getProperties().getProperty("sds.user"), System.getProperties().getProperty("sds.key")
    ));
    final SDSSession session = new SDSSession(host, new DisabledX509TrustManager(), new DefaultX509KeyManager());
    session.open(new DisabledHostKeyCallback(), new DisabledLoginCallback());
    session.login(new DisabledPasswordStore(), new DisabledLoginCallback(), new DisabledCancelCallback());
    final TransferStatus status = new TransferStatus();
    final byte[] content = RandomUtils.nextBytes(32769);
    final TransferStatus writeStatus = new TransferStatus();
    writeStatus.setLength(content.length);
    final Path room = new SDSDirectoryFeature(session).mkdir(
        new Path(new AlphanumericRandomStringService().random(), EnumSet.of(Path.Type.directory, Path.Type.volume)), null, new TransferStatus());
    final Path test = new Path(room, UUID.randomUUID().toString(), EnumSet.of(Path.Type.file));
    final SDSWriteFeature writer = new SDSWriteFeature(session);
    final HttpResponseOutputStream<VersionId> out = writer.write(test, writeStatus, new DisabledConnectionCallback());
    assertNotNull(out);
    new StreamCopier(writeStatus, writeStatus).transfer(new ByteArrayInputStream(content), out);
    final CountingInputStream in = new CountingInputStream(new SDSReadFeature(session).read(test, status, new DisabledConnectionCallback()));
    in.close();
    assertEquals(0L, in.getByteCount(), 0L);
    new SDSDeleteFeature(session).delete(Collections.singletonList(room), new DisabledLoginCallback(), new Delete.DisabledCallback());
    session.close();
}

@Test
public void testReadCloseReleaseEntity() throws Exception {
    final Host host = new Host(new S3Protocol(), new S3Protocol().getDefaultHostname(), new Credentials(
            System.getProperties().getProperty("s3.key"), System.getProperties().getProperty("s3.secret")
    ));
    final S3Session session = new S3Session(host);
    session.open(new DisabledHostKeyCallback(), new DisabledLoginCallback());
    session.login(new DisabledPasswordStore(), new DisabledLoginCallback(), new DisabledCancelCallback());
    final Path container = new Path("test-us-east-1-cyberduck", EnumSet.of(Path.Type.directory, Path.Type.volume));
    final Path file = new Path(container, UUID.randomUUID().toString(), EnumSet.of(Path.Type.file));
    final int length = 2048;
    final byte[] content = RandomUtils.nextBytes(length);
    final TransferStatus status = new TransferStatus().length(content.length);
    status.setChecksum(new SHA256ChecksumCompute().compute(new ByteArrayInputStream(content), status));
    final OutputStream out = new S3WriteFeature(session).write(file, status, new DisabledConnectionCallback());
    new StreamCopier(new TransferStatus(), new TransferStatus()).transfer(new ByteArrayInputStream(content), out);
    out.close();
    final CountingInputStream in = new CountingInputStream(new S3ReadFeature(session).read(file, status, new DisabledConnectionCallback()));
    in.close();
    assertEquals(0L, in.getByteCount(), 0L);
    new S3DefaultDeleteFeature(session).delete(Collections.singletonList(file), new DisabledLoginCallback(), new Delete.DisabledCallback());
    session.close();
}

private static ArscData readTable(File file , CountingInputStream countIn, ExtDataInput in )
        throws IOException {
    ArscData arscData = new ArscData();
    arscData.mFile = file;
    arscData.mHeader = Header.read(in);
    int packageCount = in.readInt();
    if (packageCount != 1) {
        throw new UnsupportedOperationException("not support more then 1 package");
    }
    arscData.mTableStrings = StringBlock.read(in);
    arscData.mPkgHeaderIndex = (int) countIn.getByteCount();
    arscData.mPkgHeader = PackageHeader.read(in);
    arscData.mTypeStrStart = (int) countIn.getByteCount();
    arscData.mTypeNames = StringBlock.read(in);
    arscData.mTypeStrEnd =  (int) countIn.getByteCount();
    arscData.mSpecNames = StringBlock.read(in);
    arscData.mResIndex =  (int) countIn.getByteCount();
    return arscData;
}

public MonitoredFileReader(String fileName, InputStream is, long length, String encoding, double reportProgressAfter) throws IOException {
    InputStream in = countingIn = new CountingInputStream(is);
    if (fileName.endsWith(".gz")) {
        try {
            @SuppressWarnings("resource")
            InputStream gzIn = new GZIPInputStream(in);
            log.info("[" + fileName + "] GZipped file detected. Reading using decompressor.");
            in = gzIn;
        } catch (ZipException e) {
            // proceed like nothing happened (gzIn has not been assigned to in)
            log.error("[" + fileName + "] Warning: Unsuccessfully tried top uncompress file ending with .gz, reading file without decompression.", e);
        }
    }
    inReader = new InputStreamReader(in, encoding);
    monitor = new ProgressMonitor(fileName, "bytes", length, reportProgressAfter);
}

private RsrcSection readResourceSection(CountingInputStream executableInputStream, SectionHeader[] sectionHeaders)
        throws IOException {
    SectionHeader rsrcSectionHeader = null;
    for (SectionHeader sectionHeader : sectionHeaders) {
        if (".rsrc\u0000\u0000\u0000".equals(new String(sectionHeader.name))) {
            rsrcSectionHeader = sectionHeader;
        }
    }

    if (rsrcSectionHeader == null) {
        return null;
    }

    long numberToSkip = rsrcSectionHeader.pointerToRawData.getUnsignedValue() - executableInputStream.getCount();
    executableInputStream.skip(numberToSkip);
    byte[] rsrcSection = new byte[(int) rsrcSectionHeader.sizeOfRawData.getUnsignedValue()];
    executableInputStream.read(rsrcSection);

    return new RsrcSection(rsrcSection);
}

/**
 * Establish socket connection with client
 */
private void create_socket_and_listen() throws Exception {
    sock = new ServerSocket(EstimateNConfig.socketPort);            // create socket and bind to port
    System.out.println("waiting for client to connect");
    clientSocket = sock.accept();                   // wait for client to connect
    System.out.println("client has connected");

    CountingOutputStream cos = new CountingOutputStream(clientSocket.getOutputStream());
    CountingInputStream cis = new CountingInputStream(clientSocket.getInputStream());

    ProgCommon.oos = new ObjectOutputStream(cos);
    ProgCommon.ois = new ObjectInputStream(cis);

    StopWatch.cos = cos;
    StopWatch.cis = cis;
}

private boolean processSha1SyncResponse(Response response) throws IOException {
    int expected = m_server.level() + 1;
    CountingInputStream counter = new CountingInputStream(response.getResultStream());
    InputStreamReader reader = new InputStreamReader(new BufferedInputStream(counter), UTF8.UTF8);
    try {
        m_server = new Gson().fromJson(reader, Sha1SyncJson.class);
        if (expected != m_server.level()) {
            throw new IllegalStateException("Level warp! expected("+expected+"), actual("+m_server.level()+")");
        }
        if (!versionFeatures.getToken().equals(m_server.version())) {
            throw new IllegalStateException("Version warp! expected("+versionFeatures.getToken()+"), actual("+m_server.version()+")");
        }
        if (isServerEmpty()) {
            clearLocal();
            return true;
        }
        if (isServerHashesEmpty()) {
            return true;
        }
        return false;
    } finally {
        m_rxBytes += counter.getByteCount();
        reader.close();
    }
}

public Void invoke(File dir, VirtualChannel channel) throws IOException, InterruptedException {
    URLConnection con = archive.openConnection();
    // Jira Bug JENKINS-21033: Changing the User-Agent from "Java/<Java version #>" to "Jenkins/<Jenkins version #>"
    con.setRequestProperty("User-Agent", "Jenkins/" + Jenkins.getVersion().toString());
    InputStream in = con.getInputStream();
    try {
        CountingInputStream cis = new CountingInputStream(in);
        try {
            LOGGER.log(Level.INFO, "Invoke called for Unpack class to unpack to " + dir.getAbsolutePath());
            if (archive.toExternalForm().endsWith(".zip")) {
              LOGGER.log(Level.INFO, "Archive unzipped as it ends with '.zip'.  Starting unzip.");
              unzip(dir, cis);
            }
        } catch (IOException x) {
            throw new IOException(String.format("Failed to unpack %s (%d bytes read)", archive, cis.getByteCount()), x);
        }
    } finally {
        in.close();
    }
    return null;
}

private ARSCDecoder(InputStream arscStream, ResTable resTable, boolean storeFlagsOffsets, boolean keepBroken) {
    arscStream = mCountIn = new CountingInputStream(arscStream);
    if (storeFlagsOffsets) {
        mFlagsOffsets = new ArrayList<FlagsOffset>();
    } else {
        mFlagsOffsets = null;
    }
    mIn = new ExtDataInput(new LittleEndianDataInputStream(arscStream));
    mResTable = resTable;
    mKeepBroken = keepBroken;
}

public static Header read(ExtDataInput in, CountingInputStream countIn) throws IOException {
    short type;
    int start = countIn.getCount();
    try {
        type = in.readShort();
    } catch (EOFException ex) {
        return new Header(TYPE_NONE, 0, 0, countIn.getCount());
    }
    return new Header(type, in.readShort(), in.readInt(), start);
}

@Test
public void testReadCloseReleaseEntity() throws Exception {
    final Host host = new Host(new DAVSSLProtocol(), "svn.cyberduck.ch", new Credentials(
            PreferencesFactory.get().getProperty("connection.login.anon.name"), null
    ));
    final DAVSession session = new DAVSession(host);
    session.open(new DisabledHostKeyCallback(), new DisabledLoginCallback());
    session.login(new DisabledPasswordStore(), new DisabledLoginCallback(), new DisabledCancelCallback());
    final TransferStatus status = new TransferStatus();
    final Path test = new Path("/trunk/LICENSE.txt", EnumSet.of(Path.Type.file));
    final CountingInputStream in = new CountingInputStream(new DAVReadFeature(session).read(test, status, new DisabledConnectionCallback()));
    in.close();
    assertEquals(0L, in.getByteCount(), 0L);
    session.close();
}

public FixedLengthInputStream(InputStream stream, long maxLen) {
  super(new CountingInputStream(new CloseShieldInputStream(stream)));

  // Save a correctly-typed reference to the underlying stream.
  this.countingIn = (CountingInputStream) this.in;
  this.maxBytes = maxLen;
}

/**
 * Creates a new {@link SessionInputBuffer} bounded to a given maximum length.
 *
 * @param buffer the buffer to wrap
 * @param length the maximum number of bytes to read (from the buffered stream).
 */
public BoundSessionInputBuffer(final SessionInputBuffer buffer, final long length) {
    super(new HttpTransportMetricsImpl(), BUFFER_SIZE, 0, null, null);
    this.bounded = new ContentLengthInputStream(buffer, length);
    this.input = new CountingInputStream(this.bounded);
    super.bind(this.input);
    this.length = length;
}

private void saveDocument(DownloadableDocument document, Path target) throws IOException {
    Page downloadPage = browser.getPage(document.getDownloadLink());
    WebResponse response = downloadPage.getWebResponse();
    document.setSize(response.getContentLength());
    try (CountingInputStream in = new CountingInputStream(response.getContentAsStream())) {
        copy(in, target, document);
    }
}

private void copyWithNotifyProgress(CountingInputStream source, OutputStream sink, DownloadableDocument document) throws IOException {
    int n;
    byte[] buffer = new byte[BUFFER_SIZE];
    while((n = source.read(buffer)) > 0) {
        sink.write(buffer, 0, n);
        double progress = (double) source.getByteCount() / document.getSize();
        notifyObserversProgress(document, progress);
    }
}

private boolean handleResponseBody(InputStream inputStream, TickBuilder tickBuilder) {
  CountingInputStream countingStream = new CountingInputStream(inputStream);
  DataInputStream input = new DataInputStream(countingStream);
  RemoteMessage message;

  try {
    while ((message = mapper.decode(input)) != null) {
      if (message instanceof TrackStartResponseMessage) {
        handleTrackStartResponse((TrackStartResponseMessage) message);
      } else if (message instanceof TrackFrameDataMessage) {
        handleTrackFrameData((TrackFrameDataMessage) message);
      } else if (message instanceof TrackExceptionMessage) {
        handleTrackException((TrackExceptionMessage) message);
      } else if (message instanceof NodeStatisticsMessage) {
        handleNodeStatistics((NodeStatisticsMessage) message);
      }
    }
  } catch (InterruptedException interruption) {
    log.error("Node {} processing thread was interrupted.", nodeAddress);
    Thread.currentThread().interrupt();
    return false;
  } catch (Throwable e) {
    log.error("Error when processing response from node {}.", nodeAddress, e);
    ExceptionTools.rethrowErrors(e);
  } finally {
    tickBuilder.responseSize = countingStream.getCount();
  }

  return true;
}

/**
 *
 * @param container
 * @param inputStream closed on exit
 * @param store
 * @throws IOException
 * @throws ArithmeticException on input streams over 2 Gb.
 * @throws IllegalArgumentException on non 0x01 chunk keys
 */
public void apply(StorageHostChunkList container, InputStream inputStream, ChunkStore store) throws IOException {
    logger.trace("<< apply() - input: {}", inputStream);
    // Ensure our chunk offsets are sequentially ordered.
    List<ChunkInfo> list = container.getChunkInfoList()
            .stream()
            .sorted(CHUNK_OFFSET_COMPARATOR)
            .collect(toList());

    try (CountingInputStream countingInputStream = new CountingInputStream(inputStream)) {
        streamChunks(list, countingInputStream, store);
    } catch (UncheckedIOException ex) {
        throw ex.getCause();
    }

    if (logger.isDebugEnabled()) {
        // Sanity check. Has a minor IO cost with a disk based chunk store.
        String missingChunks = list.stream()
                .map(ci -> ci.getChunkChecksum().toByteArray())
                .filter(c -> !store.contains(c))
                .map(c -> "0x" + Hex.toHexString(c))
                .collect(joining(" "));
        if (missingChunks.isEmpty()) {
            logger.debug("-- apply() - all chunks have been stored");
        } else {
            logger.warn("-- apply() - missing chunks: {}", missingChunks);
        }
    }
    logger.trace(">> apply()");
}

void streamChunks(List<ChunkInfo> chunkInfos, CountingInputStream inputStream, ChunkStore store) {
    logger.debug("-- streamChunks() - chunk count: {}", chunkInfos.size());
    chunkInfos.stream()
            .peek(ci -> logger.debug("-- streamChunks() - chunk info: {}", ci))
            .filter(u -> isChunkMissing(u, store))
            .forEach(u -> streamChunk(inputStream, inputStream.getCount(), u, store));
}

private ARSCDecoder(InputStream arscStream, ResTable resTable,
        boolean storeFlagsOffsets, boolean keepBroken) {
    if (storeFlagsOffsets) {
        arscStream = mCountIn = new CountingInputStream(arscStream);
        mFlagsOffsets = new ArrayList<FlagsOffset>();
    } else {
        mCountIn = null;
        mFlagsOffsets = null;
    }
    mIn = new ExtDataInput(new LEDataInputStream(arscStream));
    mResTable = resTable;
    mKeepBroken = keepBroken;
}

private static List<Object> readObjectFile(File objectFile) throws IOException
{
  long fileLength = objectFile.length();

  FileInputStream fis = new FileInputStream(objectFile);
  CountingInputStream cis = new CountingInputStream(fis);
  ObjectInputStream ois = new ObjectInputStream(cis);

  List<Object> objects = new ArrayList<Object>();

  try
  {
    while (cis.getByteCount() < fileLength)
    {
      objects.add(ois.readObject());
    }
  }
  catch (ClassNotFoundException e)
  {
    throw new IOException(e);
  }
  finally
  {
    ois.close();
  }

  return objects;
}

CodeClass parse(InputStream is, Model model) throws IOException {
    final CountingInputStream counter = new CountingInputStream(is);
    in = new DataInputStream(counter);

    parseMagic();
    parseMinorVersion();
    parseMajorVersion();

    constantPool = ConstantPool.fromData(in);

    final int flags = parseAccessFlags();

    final String className = parseClassName();
    final String superClassName = parseSuperClassName();
    final List<String> interfaceNames = parseInterfaces();
    final List<MemberInfo> fields = parseMembers();
    final List<MemberInfo> methods = parseMembers();
    final List<AttributeInfo> attributes = parseAttributes();

    return new CodeClassBuilder(className, model, constantPool)
            .addClassConstantReferences()
            .addFlags(flags)
            .addSuperClass(superClassName)
            .addInterfaces(interfaceNames)
            .addFieldRefs(fields)
            .addMethodRefs(methods)
            .addAttributeRefs(attributes)
            .addPackageInfo(model, className)
            .addCodeSizes(counter.getCount(), methods)
            .clazz;
}

public PEFile parseExecutable(InputStream inputStream) throws IOException {
    try (CountingInputStream executableInputStream = new CountingInputStream(inputStream)) {
        final ImageDOSHeader imageDOSHeader = readDosHeader(executableInputStream);
        final byte[] realModeStubProgram = readRealModeStubProgram(executableInputStream, imageDOSHeader);
        final ImageNTHeaders imageNTHeaders = readImageNTHeaders(executableInputStream);
        final SectionHeader[] sectionHeaders = readSectionHeaders(executableInputStream, imageNTHeaders);
        final RsrcSection resourceSection = readResourceSection(executableInputStream, sectionHeaders);
        return new PEFile(imageDOSHeader, realModeStubProgram, imageNTHeaders, sectionHeaders, resourceSection);
    }
}

private SectionHeader[] readSectionHeaders(CountingInputStream executableInputStream, ImageNTHeaders imageNTHeaders)
        throws IOException {
    final int numberOfSectionHeaders = imageNTHeaders.fileHeader.numberOfSections.getUnsignedValue();
    final SectionHeader[] sectionHeaders = new SectionHeader[numberOfSectionHeaders];
    for (int i = 0; i < numberOfSectionHeaders; i++) {
        byte[] sectionHeaderBytes = new byte[SectionHeader.SECTION_HEADER_SIZE];
        executableInputStream.read(sectionHeaderBytes);
        sectionHeaders[i] = new SectionHeader(sectionHeaderBytes);
    }
    return sectionHeaders;
}

private ARSCDecoder(InputStream arscStream, ResTable resTable, boolean storeFlagsOffsets, boolean keepBroken) {
    arscStream = mCountIn = new CountingInputStream(arscStream);
    if (storeFlagsOffsets) {
        mFlagsOffsets = new ArrayList<FlagsOffset>();
    } else {
        mFlagsOffsets = null;
    }
    mIn = new ExtDataInput(new LittleEndianDataInputStream(arscStream));
    mResTable = resTable;
    mKeepBroken = keepBroken;
}

public static Header read(ExtDataInput in, CountingInputStream countIn) throws IOException {
    short type;
    int start = countIn.getCount();
    try {
        type = in.readShort();
    } catch (EOFException ex) {
        return new Header(TYPE_NONE, 0, 0, countIn.getCount());
    }
    return new Header(type, in.readShort(), in.readInt(), start);
}

public FixedLengthInputStream(InputStream stream, long maxLen) {
  super(new CountingInputStream(new CloseShieldInputStream(stream)));

  // Save a correctly-typed reference to the underlying stream.
  this.countingIn = (CountingInputStream) this.in;
  this.maxBytes = maxLen;
}

void processGmlResponse(Response response) throws IOException, SAXException, ParserConfigurationException {
    FeatureCollection<?, ?> features;
    if (response instanceof ResponseFeatureCollection) {
        ResponseFeatureCollection responseFeatures = (ResponseFeatureCollection) response;
        features = responseFeatures.getFeatureCollection();
    } else {
        CountingInputStream counter = new CountingInputStream(response.getResultStream());
        long s = System.currentTimeMillis();
        features = (FeatureCollection<?, ?>) parseWfs(counter);
        long e = System.currentTimeMillis();
        m_parseMillis = e - s;
        m_rxGml += counter.getByteCount();
    }

    FeatureIterator<?> it = features.features();
    try {
        while (it.hasNext()) {
            Feature feature = it.next();
            FeatureId fid = feature.getIdentifier();
            m_potentialDeletes.remove(fid);
            if (!m_features.containsKey(fid)) {
                m_listener.featureCreate(fid, feature);
                m_numCreates++;
            } else {
                m_listener.featureUpdate(fid, feature);
                m_numUpdates++;
            }
        }
    } finally {
        it.close();
    }
}

@Override
protected InputStream doSeekLoad(long offset, int maxLength)
        throws IOException {

    URL theUrl = new URL(url);      

       URLConnection connection = theUrl.openConnection();

       httpUrlConn = (HttpURLConnection)connection;

       connection.setConnectTimeout(connTimeout);
       connection.setReadTimeout(readTimeout);

       String rangeHeader = makeRangeHeader(offset, maxLength);

       if (rangeHeader != null) {
        httpUrlConn.addRequestProperty("Range", rangeHeader);
       }

    if (this.isNoKeepAlive()) {
        httpUrlConn.addRequestProperty("Connection", "close");
    }

    if (this.getCookie() != null) {
        httpUrlConn.addRequestProperty("Cookie", cookie);
    }

    httpUrlConn.connect();

    int code = httpUrlConn.getResponseCode();
    connectedUrl =  httpUrlConn.getURL().toString();

    if ((code != 206) && (code != 200)) {
        throw new BadHttpStatusException(code, connectedUrl + " " + rangeHeader);
    }

    InputStream is = httpUrlConn.getInputStream();
    cin = new CountingInputStream(is);
    return cin;
}

public FixedLengthInputStream(InputStream stream, long maxLen) {
  super(new CountingInputStream(new CloseShieldInputStream(stream)));

  // Save a correctly-typed reference to the underlying stream.
  this.countingIn = (CountingInputStream) this.in;
  this.maxBytes = maxLen;
}

/**
 * Creates a new instance of this object with given configuration.
 *
 * @param config {@link #config}
 * @throws IOException In case file doesn't contain actual data
 */
protected ImportAdapterCSV(ImportConfigurationCSV config) throws IOException {

    super(config);
    this.config = config;
    this.bytesTotal = new File(config.getFileLocation()).length();

    /* Used to keep track of progress */
    cin = new CountingInputStream(new FileInputStream(new File(config.getFileLocation())));

    /* Get CSV iterator */
    in = new CSVDataInput(cin, config.getCharset(), config.getDelimiter(), config.getQuote(), config.getEscape(), config.getLinebreak());
    it = in.iterator();

    /* Check whether there is actual data within the CSV file */
    if (it.hasNext()) {
        row = it.next();
        if (config.getContainsHeader()) {
            if (!it.hasNext()) {
                throw new IOException("CSV contains nothing but header");
            }
        }
    } else {
        throw new IOException("CSV file contains no data");
    }

    // Create header
    header = createHeader();
}

@Test
public void testAppendSecondPart() 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 container = new Path("test-us-east-1-cyberduck", EnumSet.of(Path.Type.directory, Path.Type.volume));
    final String name = UUID.randomUUID().toString();
    final Path test = new Path(container, name, EnumSet.of(Path.Type.file));
    final int length = 12 * 1024 * 1024;
    final byte[] content = RandomUtils.nextBytes(length);
    Local local = new Local(System.getProperty("java.io.tmpdir"), name);
    IOUtils.write(content, local.getOutputStream(false));
    final AtomicBoolean started = new AtomicBoolean();
    final TransferStatus status = new TransferStatus() {
        @Override
        public void progress(long bytes) {
            super.progress(bytes);
            started.set(true);
        }
    };
    status.setLength(content.length);
    final AtomicBoolean interrupt = new AtomicBoolean();
    try {
        new S3MultipartUploadService(session, new S3WriteFeature(session, new S3DisabledMultipartService()), 10L * 1024L * 1024L, 1).upload(test, new Local(System.getProperty("java.io.tmpdir"), name) {
                    @Override
                    public InputStream getInputStream() throws AccessDeniedException {
                        return new CountingInputStream(super.getInputStream()) {
                            @Override
                            protected void beforeRead(int n) throws IOException {
                                if(started.get()) {
                                    if(this.getByteCount() >= 11L * 1024L * 1024L) {
                                        throw new IOException();
                                    }
                                }
                            }
                        };
                    }
                },
                new BandwidthThrottle(BandwidthThrottle.UNLIMITED), new DisabledStreamListener(), status,
                new DisabledLoginCallback());
    }
    catch(BackgroundException e) {
        // Expected
        interrupt.set(true);
    }
    assertTrue(interrupt.get());
    assertEquals(10L * 1024L * 1024L, status.getOffset(), 0L);
    assertFalse(status.isComplete());
    assertFalse(new S3FindFeature(session).find(test));

    final TransferStatus append = new TransferStatus().append(true).length(2L * 1024L * 1024L).skip(10L * 1024L * 1024L);
    new S3MultipartUploadService(session, new S3WriteFeature(session, new S3DisabledMultipartService()), 10L * 1024L * 1024L, 1).upload(test, local,
            new BandwidthThrottle(BandwidthThrottle.UNLIMITED), new DisabledStreamListener(), append,
            new DisabledConnectionCallback());
    assertEquals(12L * 1024L * 1024L, append.getOffset(), 0L);
    assertTrue(append.isComplete());
    assertTrue(new S3FindFeature(session).find(test));
    assertEquals(12L * 1024L * 1024L, new S3AttributesFinderFeature(session).find(test).getSize(), 0L);
    final byte[] buffer = new byte[content.length];
    final InputStream in = new S3ReadFeature(session).read(test, new TransferStatus(), new DisabledConnectionCallback());
    IOUtils.readFully(in, buffer);
    in.close();
    assertArrayEquals(content, buffer);
    new S3DefaultDeleteFeature(session).delete(Collections.singletonList(test), new DisabledLoginCallback(), new Delete.DisabledCallback());
    local.delete();
    session.close();
}

@Test
public void testAppendNoPartCompleted() 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 container = new Path("test-us-east-1-cyberduck", EnumSet.of(Path.Type.directory, Path.Type.volume));
    String name = UUID.randomUUID().toString();
    final Path test = new Path(container, name, EnumSet.of(Path.Type.file));
    final Local local = new Local(System.getProperty("java.io.tmpdir"), name);
    final int length = 32769;
    final byte[] content = RandomUtils.nextBytes(length);
    IOUtils.write(content, local.getOutputStream(false));
    final AtomicBoolean started = new AtomicBoolean();
    final TransferStatus status = new TransferStatus() {
        @Override
        public void progress(long bytes) {
            super.progress(bytes);
            started.set(true);
        }
    };
    status.setLength(content.length);
    final AtomicBoolean interrupt = new AtomicBoolean();
    try {
        new S3MultipartUploadService(session, new S3WriteFeature(session, new S3DisabledMultipartService()), 10485760L, 1).upload(test, new Local(System.getProperty("java.io.tmpdir"), name) {
                    @Override
                    public InputStream getInputStream() throws AccessDeniedException {
                        return new CountingInputStream(super.getInputStream()) {
                            @Override
                            protected void beforeRead(int n) throws IOException {
                                if(started.get()) {
                                    if(this.getByteCount() >= 32768) {
                                        throw new IOException();
                                    }
                                }
                            }
                        };
                    }
                }, new BandwidthThrottle(BandwidthThrottle.UNLIMITED), new DisabledStreamListener(), status,
                new DisabledConnectionCallback());
    }
    catch(BackgroundException e) {
        // Expected
        interrupt.set(true);
    }
    assertTrue(interrupt.get());
    assertEquals(0L, status.getOffset(), 0L);
    assertFalse(status.isComplete());

    final TransferStatus append = new TransferStatus().append(true).length(content.length);
    new S3MultipartUploadService(session, new S3WriteFeature(session, new S3DisabledMultipartService()), 10485760L, 1).upload(
            test, local, new BandwidthThrottle(BandwidthThrottle.UNLIMITED),
            new DisabledStreamListener(), append,
            new DisabledConnectionCallback());
    assertEquals(32769L, append.getOffset(), 0L);
    assertTrue(append.isComplete());
    assertTrue(new S3FindFeature(session).find(test));
    assertEquals(content.length, new S3AttributesFinderFeature(session).find(test).getSize());
    final byte[] buffer = new byte[content.length];
    final InputStream in = new S3ReadFeature(session).read(test, new TransferStatus(), new DisabledConnectionCallback());
    IOUtils.readFully(in, buffer);
    in.close();
    assertArrayEquals(content, buffer);
    new S3DefaultDeleteFeature(session).delete(Collections.singletonList(test), new DisabledLoginCallback(), new Delete.DisabledCallback());
    local.delete();
    session.close();
}

private void copy(CountingInputStream in, Path target, DownloadableDocument document) throws IOException {
    try (OutputStream sink = Files.newOutputStream(target)) {
        notifyObserversStart(document);
        copyWithNotifyProgress(in, sink, document);
    }
}

/**
 * @param inputStream Input stream to read from.
 */
public MessageInput(InputStream inputStream) {
  this.countingInputStream = new CountingInputStream(inputStream);
  this.dataInputStream = new DataInputStream(inputStream);
}

public CountingInputStreamTransportMetrics(InputStream in) {
     this.in = new CountingInputStream(in);
}

public IndexDatabaseScanner(@Nonnull InputStream is) {
    this.countingInputStream = new CountingInputStream(is);
    this.input = new DataInputStream(this.countingInputStream);
}

protected InputStream doSeekLoad(long offset, int maxLength) throws IOException {
    if (activeMethod != null) {
        doClose();
    }

    br = null;

    try {

        activeMethod = new GetMethod(url);

        String rangeHeader = makeRangeHeader(offset, maxLength);

        if (rangeHeader != null) {
            activeMethod.setRequestHeader("Range", rangeHeader);
        }

        if (this.isNoKeepAlive()) {
            activeMethod.setRequestHeader("Connection", "close");
        }

        if (this.getCookie() != null) {
            activeMethod.getParams().setCookiePolicy(CookiePolicy.IGNORE_COOKIES);
            activeMethod.setRequestHeader("Cookie", this.getCookie());
        }

        int code = http.executeMethod(activeMethod);

        connectedUrl = activeMethod.getURI().toString();

        if ((code != 206) && (code != 200)) {
            throw new BadHttpStatusException(code, connectedUrl + " " + rangeHeader);
        }

        InputStream is = activeMethod.getResponseBodyAsStream();
        cin = new CountingInputStream(is);
        return cin;

    } catch (IOException io) {
        if (saveErrHeader != null) {
            errHeader = getHeaderValue(saveErrHeader);  
        }

        connectedUrl = activeMethod.getURI().toString();
        doClose();
        throw io;
    }
}

private CountingInputStream lzmaInputStream() throws IOException {
    in.position(bundle.header().headerSize());
    return new CountingInputStream(new LzmaInputStream(in.stream()));
}