Java 类java.nio.channels.FileChannel 实例源码

private FileChannel openReader(long generationId) throws IOException {
    ensureOpen();
    if (readChannels.containsKey(generationId)) {
        return readChannels.get(generationId);
    }
    try {
        Path translogFilePath = this.translogPath.resolve(getFileNameFromId(tmpTranslogGeneration.get()));
        if (!Files.exists(translogFilePath)) {
            return null;
        }
        // maybe a lot of readers try to open reader and put it to readChannel cache, because read lock is shared
        FileChannel readChannel = FileChannel.open(translogFilePath, StandardOpenOption.READ);
        FileChannel originReadChannel = readChannels.putIfAbsent(generationId, readChannel);
        if (originReadChannel != null) {
            IOUtils.close(readChannel);
            return originReadChannel;
        } else {
            return readChannel;
        }
    } catch (Throwable e) {
        throw e;
    }
}

@Override
public long truncateFileAtURL(LocalFilesystemURL inputURL, long size) throws IOException {
       File file = new File(filesystemPathForURL(inputURL));

       if (!file.exists()) {
           throw new FileNotFoundException("File at " + inputURL.uri + " does not exist.");
       }

       RandomAccessFile raf = new RandomAccessFile(filesystemPathForURL(inputURL), "rw");
       try {
           if (raf.length() >= size) {
               FileChannel channel = raf.getChannel();
               channel.truncate(size);
               return size;
           }

           return raf.length();
       } finally {
           raf.close();
       }


}

/**
 * Cria a cópia de um arquivo para o local passado como parâmetro
 *
 * @param origem caminho completo do arquivo a ser copiado
 * @param destino caminho completo (com nome e extensão) do local de
 * gravação
 * @return {@code true} ou {@code false}
 * @throws java.io.IOException
 *
 */
public static boolean copiarArquivos(File origem, File destino) throws IOException {
    if (destino.exists()) {
        destino.delete();
    }
    FileChannel sourceChannel = null;
    FileChannel destinationChannel = null;

    try {
        sourceChannel = new FileInputStream(origem).getChannel();
        destinationChannel = new FileOutputStream(destino).getChannel();
        sourceChannel.transferTo(0, sourceChannel.size(),
                destinationChannel);
    } catch (IOException ioe) {
        return false;
    } finally {
        if (sourceChannel != null && sourceChannel.isOpen()) {
            sourceChannel.close();
        }
        if (destinationChannel != null && destinationChannel.isOpen()) {
            destinationChannel.close();
        }
    }
    return true;
}

/**
 * Is ID3 tag
 *
 * @param fc
 * @return
 * @throws IOException
 */
public static boolean isId3Tag(FileChannel fc) throws IOException
{
    if (!isID3V2Header(fc))
    {
        return false;
    }
    //So we have a tag
    ByteBuffer bb = ByteBuffer.allocateDirect(FIELD_TAG_SIZE_LENGTH);
    fc.position(fc.position() + FIELD_TAGID_LENGTH + FIELD_TAG_MAJOR_VERSION_LENGTH + FIELD_TAG_MINOR_VERSION_LENGTH + FIELD_TAG_FLAG_LENGTH);
    fc.read(bb);
    bb.flip();
    int size = ID3SyncSafeInteger.bufferToValue(bb);
    fc.position(size + TAG_HEADER_LENGTH);
    return true;
}

/**
 * 将字节数组写入文件
 *
 * @param file    文件
 * @param bytes   字节数组
 * @param append  是否追加在文件末
 * @param isForce 是否写入文件
 * @return {@code true}: 写入成功<br>{@code false}: 写入失败
 */
public static boolean writeFileFromBytesByChannel(final File file, final byte[] bytes, final boolean append, final boolean isForce) {
    if (bytes == null) return false;
    FileChannel fc = null;
    try {
        fc = new FileOutputStream(file, append).getChannel();
        fc.position(fc.size());
        fc.write(ByteBuffer.wrap(bytes));
        if (isForce) fc.force(true);
        return true;
    } catch (IOException e) {
        e.printStackTrace();
        return false;
    } finally {
        CloseUtils.closeIO(fc);
    }
}

/**
 * 将字节数组写入文件
 *
 * @param file    文件
 * @param bytes   字节数组
 * @param append  是否追加在文件末
 * @param isForce 是否写入文件
 * @return {@code true}: 写入成功<br>{@code false}: 写入失败
 */
public static boolean writeFileFromBytesByMap(final File file, final byte[] bytes, final boolean append, final boolean isForce) {
    if (bytes == null || !createOrExistsFile(file)) return false;
    FileChannel fc = null;
    try {
        fc = new FileOutputStream(file, append).getChannel();
        MappedByteBuffer mbb = fc.map(FileChannel.MapMode.READ_WRITE, fc.size(), bytes.length);
        mbb.put(bytes);
        if (isForce) mbb.force();
        return true;
    } catch (IOException e) {
        e.printStackTrace();
        return false;
    } finally {
        CloseUtils.closeIO(fc);
    }
}

/**
 * At this point, we are still muxing, and trying to read beyond muxed data.
 *
 * We park here and wait until it is available.
 */
private int waitForMuxing(Muxer muxer, long maxPosition, FileChannel fileChannel, int fileHandle, MuxedFile muxedFile)
        throws IOException, InterruptedException {
    long currentSize = 0;
    while (maxPosition >= (currentSize = fileChannel.size())) {
        State state = muxer.state();
        switch (state) {
            case RUNNING:
                logger.debug("Want to read @ {} (file is {}), so waiting for {}", maxPosition, currentSize, muxer);
                sleeper.sleep(MUX_WAIT_LOOP_MS);
                break;
            case SUCCESSFUL:
                logger.debug("Done waiting to read @ {}", maxPosition, muxer);
                return SUCCESS;
            case FAILED:
                return muxingFailed(fileHandle, muxedFile, muxer);
            default:
                logger.error("BUG: Unhandled state {} in muxer {}", state, muxer);
                return BUG;
        }
    }
    logger.debug("Done waiting to read @ {}", maxPosition, muxer);
    return SUCCESS;
}

@Test
public void shouldReleaseTaskStateDirectoryLock() throws Exception {
    final TaskId taskId = new TaskId(0, 0);
    final File taskDirectory = directory.directoryForTask(taskId);

    directory.lock(taskId, 1);
    directory.unlock(taskId);

    try (
        final FileChannel channel = FileChannel.open(
            new File(taskDirectory, StateDirectory.LOCK_FILE_NAME).toPath(),
            StandardOpenOption.CREATE,
            StandardOpenOption.WRITE)
    ) {
        channel.tryLock();
    }
}

/**
 * Delete Tag
 *
 * @param file to delete the tag from
 * @throws IOException if problem accessing the file
 *                     <p/>
 */
//TODO should clear all data and preferably recover lost space and go upto end of mp3s
public void delete(RandomAccessFile file) throws IOException {
    // this works by just erasing the "ID3" tag at the beginning
    // of the file
    byte[] buffer = new byte[FIELD_TAGID_LENGTH];
    //Read into Byte Buffer
    final FileChannel fc = file.getChannel();
    fc.position();
    ByteBuffer byteBuffer = ByteBuffer.allocate(TAG_HEADER_LENGTH);
    fc.read(byteBuffer, 0);
    byteBuffer.flip();
    if (seek(byteBuffer)) {
        file.seek(0L);
        file.write(buffer);
    }
}

/**
 * 读取文件到字节数组中
 *
 * @param file 文件
 * @return 字符数组
 */
public static byte[] readFile2BytesByMap(final File file) {
    if (!isFileExists(file)) return null;
    FileChannel fc = null;
    try {
        fc = new RandomAccessFile(file, "r").getChannel();
        int size = (int) fc.size();
        MappedByteBuffer mbb = fc.map(FileChannel.MapMode.READ_ONLY, 0, size).load();
        byte[] result = new byte[size];
        mbb.get(result, 0, size);
        return result;
    } catch (IOException e) {
        e.printStackTrace();
        return null;
    } finally {
        CloseUtils.closeIO(fc);
    }
}

/**
 * 将字节数组写入文件
 *
 * @param file    文件
 * @param bytes   字节数组
 * @param append  是否追加在文件末
 * @param isForce 是否写入文件
 * @return {@code true}: 写入成功<br>{@code false}: 写入失败
 */
public static boolean writeFileFromBytesByChannel(final File file, final byte[] bytes, final boolean append, final boolean isForce) {
    if (bytes == null) return false;
    FileChannel fc = null;
    try {
        fc = new FileOutputStream(file, append).getChannel();
        fc.position(fc.size());
        fc.write(ByteBuffer.wrap(bytes));
        if (isForce) fc.force(true);
        return true;
    } catch (IOException e) {
        e.printStackTrace();
        return false;
    } finally {
        CloseUtils.closeIO(fc);
    }
}

private static void testUnreadableChannel() throws Exception {
    File blah = File.createTempFile("blah2", null);
    blah.deleteOnExit();
    FileOutputStream fos = new FileOutputStream(blah);
    try {
        fos.write(new byte[128]);
        FileChannel fc = fos.getChannel();
        try {
            fc.read(ByteBuffer.allocate(256),1);
            throw new RuntimeException("Expected exception not thrown");
        } catch(NonReadableChannelException e) {
            // Correct result
        }
    } finally {
        fos.close();
        blah.delete();
    }
}

public boolean LockExclusive(File targetFile) {

            if (targetFile == null) {
                return false;
            }
            try {
                File lockFile = new File(targetFile.getParentFile().getAbsolutePath().concat("/lock"));
                if (!lockFile.exists()) {
                    lockFile.createNewFile();
                }
                RandomAccessFile randomAccessFile = new RandomAccessFile(lockFile.getAbsolutePath(), "rw");
                FileChannel channel = randomAccessFile.getChannel();
                java.nio.channels.FileLock lock = channel.lock();
                if (!lock.isValid()) {
                    return false;
                }
                RefCntInc(lockFile.getAbsolutePath(), lock, randomAccessFile, channel);
                return true;
            } catch (Exception e) {
                return false;
            }
        }

public FileChannel position(long newPosition) throws IOException {
    ensureOpen();
    if (newPosition < 0)
        throw new IllegalArgumentException();
    synchronized (positionLock) {
        long p = -1;
        int ti = -1;
        try {
            begin();
            ti = threads.add();
            if (!isOpen())
                return null;
            do {
                p  = position0(fd, newPosition);
            } while ((p == IOStatus.INTERRUPTED) && isOpen());
            return this;
        } finally {
            threads.remove(ti);
            end(p > -1);
            assert IOStatus.check(p);
        }
    }
}

/**
 * We can fit the metadata in under the meta item just by using some of the padding available in the {@code free}
 * atom under the {@code meta} atom need to take of the side of free header otherwise might end up with
 * solution where can fit in data, but can't fit in free atom header.
 *
 * @param fileReadChannel
 * @param fileWriteChannel
 * @param neroTagsHeader
 * @param sizeOfExistingMetaLevelFreeAtom
 * @param newIlstData
 * @param additionalSpaceRequiredForMetadata
 * @throws IOException
 * @throws CannotWriteException
 */
private void writeNewMetadataLargerButCanUseFreeAtom(FileChannel fileReadChannel, FileChannel fileWriteChannel, Mp4BoxHeader ilstHeader, Mp4BoxHeader neroTagsHeader, int sizeOfExistingMetaLevelFreeAtom, ByteBuffer newIlstData, int additionalSpaceRequiredForMetadata) throws IOException, CannotWriteException
{
    int newFreeSize = sizeOfExistingMetaLevelFreeAtom - (additionalSpaceRequiredForMetadata);
    logger.config("Writing:Option 5;Larger Size can use meta free atom need extra:" + newFreeSize + "bytes");

    writeDataUptoIncludingIlst(fileReadChannel, fileWriteChannel, ilstHeader, newIlstData);

    //Create an amended smaller freeBaos atom and write it to file
    Mp4FreeBox newFreeBox = new Mp4FreeBox(newFreeSize - Mp4BoxHeader.HEADER_LENGTH);
    fileWriteChannel.write(newFreeBox.getHeader().getHeaderData());
    fileWriteChannel.write(newFreeBox.getData());

    //Skip over the read channel old free atom
    fileReadChannel.position(fileReadChannel.position() + sizeOfExistingMetaLevelFreeAtom);
    writeDataAfterIlst(fileReadChannel, fileWriteChannel, neroTagsHeader);
}

public static void main(String[] args) throws Exception {
    try (RandomAccessFile aFile = new RandomAccessFile("data/data.txt", "rw")) {
        FileChannel inChannel = aFile.getChannel();

        ByteBuffer buf = ByteBuffer.allocate(1024);

        int bytesRead = inChannel.read(buf);
        while (bytesRead != -1) {

            System.out.println("Read " + bytesRead);
            buf.flip();

            while (buf.hasRemaining()) {
                System.out.print((char) buf.get());
            }
            System.out.print("\n");
            buf.clear();
            bytesRead = inChannel.read(buf);
        }
    }
}

private static void releaseFile(final FileChannel channel) {
  final File file = file2Channel.reversedGet(channel);
  if (file == null) return;
  synchronized (file) {
    Integer count = channel2ClientsCount.get(channel);
    if (count == null) return;//already removed
    if (count > 1) {
      channel2ClientsCount.put(channel, --count);
    } else {
      channel2ClientsCount.remove(channel);
      file2Channel.reversedRemove(channel);
      try {
        channel.close();
      } catch (IOException ex) {
        LogManager.log("can't close channel", ex);
      }
    }
  }
}

@Test
public void testRead()
        throws Exception {
    // Given
    FileHandleFiller filler = mock(FileHandleFiller.class);
    ArgumentCaptor<Integer> handleCaptor = ArgumentCaptor.forClass(Integer.class);
    doNothing().when(filler).setFileHandle(handleCaptor.capture());
    Path fooBar = mockPath("foo.bar");
    FileChannel fileChannel = mock(FileChannel.class);
    when(fileSystem.provider().newFileChannel(eq(fooBar), eq(set(StandardOpenOption.READ)))).thenReturn(fileChannel);
    fs.open("foo.bar", filler);
    Integer fileHandle = handleCaptor.getValue();
    ArgumentCaptor<ByteBuffer> bufferCaptor = ArgumentCaptor.forClass(ByteBuffer.class);
    when(fileChannel.read(bufferCaptor.capture(), eq(1234L))).thenReturn(10);
    // When
    int result = fs.read("foo.bar", (data) -> assertThat(data).hasSize(10), 10, 1234L, fileHandle);
    // Then
    assertThat(result).isEqualTo(10);
    verify(fileChannel).read(any(), eq(1234L));
    verifyNoMoreInteractions(fileChannel);
    assertThat(bufferCaptor.getValue().limit()).isEqualTo(10);
}

private FileChannel[] getFileChannels() throws Exception {
    FileChannel fc[] = new FileChannel[2];

    if (!PnfsUtil.isPnfs(this.source)) {
        dFileOut = new dCacheFileOutputStream(this.destination);
        fc[1] = dFileOut.getChannel();

        fileIn = new FileInputStream(this.source);
        fc[0] = fileIn.getChannel();
    } else {
        dFileIn = new dCacheFileInputStream(this.source);
        fc[0] = dFileIn.getChannel();

        fileOut = new FileOutputStream(this.destination);
        fc[1] = fileOut.getChannel();
    }

    return fc;
}

/**
 * Pretty print the file size
 */
public void printFileSize() throws IOException
{
    FileChannel fileChannel = new RandomAccessFile( dbFile, "r" ).getChannel();

    long l = fileChannel.size();

    fileChannel.close();

    String msg;

    if ( l < 1024 )
    {
        msg = l + " bytes";
    }
    else
    {
        msg = ( l / 1024 ) + " KB";
    }

    System.out.println( msg );

    fileChannel.close();
}

public static void main(String[] args) throws Throwable {
    try (FileChannel ch = FileChannel.open(BLK_PATH, READ);
         RandomAccessFile file = new RandomAccessFile(BLK_FNAME, "r")) {

        long size1 = ch.size();
        long size2 = file.length();
        if (size1 != size2) {
            throw new RuntimeException("size differs when retrieved" +
                    " in different ways: " + size1 + " != " + size2);
        }
        System.out.println("OK");

    } catch (NoSuchFileException nsfe) {
        System.err.println("File " + BLK_FNAME + " not found." +
                " Skipping test");
    } catch (AccessDeniedException ade) {
        System.err.println("Access to " + BLK_FNAME + " is denied." +
                " Run test as root.");
    }
}

/**
 * Delete Metadata tag
 *
 * @param tag
 * @param file
 * @throws CannotWriteException
 * @throws IOException
 */
@Override
protected void deleteTag(Tag tag, File file) throws CannotWriteException {
    try {
        FileChannel fc = new FileOutputStream(file.getAbsolutePath(), false).getChannel();
        DsdChunk dsd = DsdChunk.readChunk(Utils.readFileDataIntoBufferLE(fc, DsdChunk.DSD_HEADER_LENGTH));
        if (dsd != null) {
            if (dsd.getMetadataOffset() > 0) {
                fc.position(dsd.getMetadataOffset());
                ID3Chunk id3Chunk = ID3Chunk.readChunk(Utils.readFileDataIntoBufferLE(fc, (int) (fc.size() - fc.position())));
                if (id3Chunk != null) {
                    fc.truncate(dsd.getMetadataOffset());
                    //set correct value for fileLength and zero offset
                    dsd.setMetadataOffset(0);
                    dsd.setFileLength(fc.size());
                    fc.position(0);
                    fc.write(dsd.write());
                }
            } else {
                //Do Nothing;
            }
        }
    } catch (IOException ioe) {
        throw new CannotWriteException(file.getAbsolutePath() + ":" + ioe.getMessage());
    }
}

private void copyFile(File sourceFile, File destFile) throws IOException {
    if (!destFile.getParentFile().exists())
        destFile.getParentFile().mkdirs();

    if (!destFile.exists()) {
        destFile.createNewFile();
    }

    FileChannel source = null;
    FileChannel destination = null;

    try {
        source = new FileInputStream(sourceFile).getChannel();
        destination = new FileOutputStream(destFile).getChannel();
        destination.transferFrom(source, 0, source.size());
    } finally {
        if (source != null) {
            source.close();
        }
        if (destination != null) {
            destination.close();
        }
    }
}

@Test
public void testReleaseTwice()
        throws Exception {
    // Given
    FileHandleFiller filler = mock(FileHandleFiller.class);
    ArgumentCaptor<Integer> handleCaptor = ArgumentCaptor.forClass(Integer.class);
    doNothing().when(filler).setFileHandle(handleCaptor.capture());
    Path fooBar = mockPath(mirrorRoot, "foo.bar");
    when(fileSystem.provider().newFileChannel(eq(fooBar), eq(set(StandardOpenOption.READ)))).thenReturn(mock(FileChannel.class));
    fs.open("foo.bar", filler);
    fs.release("foo.bar", handleCaptor.getValue());
    // When
    int result = fs.release("foo.bar", handleCaptor.getValue());
    // Then
    assertThat(result).isEqualTo(-ErrorCodes.EBADF());
}

/**
 * Delete given {@link Tag} from file.
 *
 * @param tag  tag, must be instance of {@link AiffTag}
 * @param file
 * @throws java.io.IOException
 * @throws org.jaudiotagger.audio.exceptions.CannotWriteException
 */
public void delete(final Tag tag, File file) throws CannotWriteException {
    try {
        FileChannel fc = new FileOutputStream(file.getAbsolutePath(), false).getChannel();
        logger.severe(file.getAbsolutePath() + " Deleting tag from file");
        final AiffTag existingTag = getExistingMetadata(file);

        if (existingTag.isExistingId3Tag() && existingTag.getID3Tag().getStartLocationInFile() != null) {
            ChunkHeader chunkHeader = seekToStartOfMetadata(fc, existingTag, file.getAbsolutePath());
            if (isAtEndOfFileAllowingForPaddingByte(existingTag, fc)) {
                logger.severe(file.getAbsolutePath() + " Setting new length to:" + (existingTag.getStartLocationInFileOfId3Chunk()));
                fc.truncate(existingTag.getStartLocationInFileOfId3Chunk());
            } else {
                logger.severe(file.getAbsolutePath() + " Deleting tag chunk");
                deleteTagChunk(fc, existingTag, chunkHeader, file.getAbsolutePath());
            }
            rewriteRiffHeaderSize(fc);
        }
        logger.severe(file.getAbsolutePath() + " Deleted tag from file");
    } catch (IOException ioe) {
        throw new CannotWriteException(file.getAbsolutePath() + ":" + ioe.getMessage());
    }
}

public static void applyXOR(Path assemblyFile, long limit) throws URISyntaxException, IOException {
    Path inputFile = assemblyFile.resolveSibling(DLL_NAME + ".bak");
    URI xorUri = FileUtils.class.getResource("/" + XOR_NAME).toURI();

    Files.move(assemblyFile, inputFile, StandardCopyOption.REPLACE_EXISTING);

    try (FileSystem fileSystem = createFileSystem(xorUri); // Required to resolve xorUri
         FileChannel inputChannel = FileChannel.open(inputFile, READ);
         FileChannel xorChannel = FileChannel.open(Paths.get(xorUri), READ);
         FileChannel outputChannel = FileChannel.open(assemblyFile, WRITE, CREATE_NEW)) {

        xorFiles(inputChannel, xorChannel, outputChannel, limit);
    } catch (Throwable t) {
        Files.deleteIfExists(assemblyFile);
        Files.move(inputFile, assemblyFile);
        throw t;
    }
}

public static void fileCombination(File file, String targetFile) throws Exception {
    outFile = new File(targetFile); 
    outDest = new RandomAccessFile(outFile,"rw");     
    childReader = new BufferedReader(new FileReader(file));
    long temp = 0;
    while ((name = childReader.readLine()) != null) {
        cFile = new File(name);
        long size = cFile.length();
        fileReader = new FileInputStream(cFile);    
        read = fileReader.getChannel().map(FileChannel.MapMode.READ_ONLY, 0, size); 
        out = outDest.getChannel().map(FileChannel.MapMode.READ_WRITE, temp, size);     
        temp += size; 
        for (long j = 0; j < size; j++) {  
             byte b = read.get();    
             out.put(b);               
        }   
        fileReader.close();
        UnMap.unmap(read);
        UnMap.unmap(out);
    }           
    outDest.close();      
    childReader.close();
}

/**
 * Copies in to out using FileChannels.
 *
 * This tries to use channels for fast copying. If it doesn't work, fall back to
 * copyFileFallBack below.
 *
 * @param in the stream to copy from.
 * @param out the stream to copy to.
 * @throws IOException if both the normal and fallback methods raise exceptions.
 */
private static void copyFile(final InputStream in, final OutputStream out)
        throws IOException {
    DebugLogUtils.l("Copying files");
    if (!(in instanceof FileInputStream) || !(out instanceof FileOutputStream)) {
        DebugLogUtils.l("Not the right types");
        copyFileFallback(in, out);
    } else {
        try {
            final FileChannel sourceChannel = ((FileInputStream) in).getChannel();
            final FileChannel destinationChannel = ((FileOutputStream) out).getChannel();
            sourceChannel.transferTo(0, Integer.MAX_VALUE, destinationChannel);
        } catch (IOException e) {
            // Can't work with channels, or something went wrong. Copy by hand.
            DebugLogUtils.l("Won't work");
            copyFileFallback(in, out);
        }
    }
}

static void write(FileChannel fc, int b) throws IOException {
    ByteBuffer buf = ByteBuffer.allocate(1);
    buf.put((byte)b);
    buf.flip();
    if (rand.nextBoolean()) {
        ByteBuffer[] bufs = new ByteBuffer[1];
        bufs[0] = buf;
        fc.write(bufs);
    } else {
        fc.write(buf);
    }
}

private void saveCroppedImage(Uri croppedFileUri) {
    try {
        File saveFile = new File(previewFilePath);
        FileInputStream inStream = new FileInputStream(new File(croppedFileUri.getPath()));
        FileOutputStream outStream = new FileOutputStream(saveFile);
        FileChannel inChannel = inStream.getChannel();
        FileChannel outChannel = outStream.getChannel();
        inChannel.transferTo(0, inChannel.size(), outChannel);
        inStream.close();
        outStream.close();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

@Override
public int read(final ByteBuffer dst) throws IOException {
  return (int) doUninterruptibly(new FileOperation() {
    public long doOp(FileChannel channel) throws IOException {
      return channel.read(dst);
    }
  });
}

SectionEncoder(Marshaller<String> stringMarshaller) {
    super(stringMarshaller);
    this.layerEncoder = new LayerEncoder(stringMarshaller);
    FileChannel channel = stringMarshaller.getChannel();
    this.layerMarshaller = new Marshaller<Layer>(channel, layerEncoder);
    this.relationEncoder = new RelationEncoder(stringMarshaller);
    WriteCache<Relation> relationCache = MapWriteCache.hashMap();
    this.relationMarshaller = new Marshaller<Relation>(channel, relationEncoder, relationCache);
}

private FileChannelLinesSpliterator(FileChannel fc, Charset cs, int index, int fence, ByteBuffer buffer) {
    this.fc = fc;
    this.buffer = buffer;
    this.cs = cs;
    this.index = index;
    this.fence = fence;
}

@Override
public void onSubscribe(Flow.Subscription subscription) {
    this.subscription = subscription;
    try {
        out = FileChannel.open(file, options);
    } catch (IOException e) {
        result.completeExceptionally(e);
        subscription.cancel();
        return;
    }
    subscription.request(1);
}

/**
 * Attempt to read the same file using multiple channels concurrently
 */
public void testConcurrentFileReads() throws Exception
{
    // open the file for a read
    FileInputStream isA = new FileInputStream(file);
    FileInputStream isB = new FileInputStream(file);

    // get the channels
    FileChannel channelA = isA.getChannel();
    FileChannel channelB = isB.getChannel();

    // buffers for reading
    ByteBuffer bufferA = ByteBuffer.allocate(10);
    ByteBuffer bufferB = ByteBuffer.allocate(10);

    // read file into both buffers
    int countA = 0;
    int countB = 0;
    do
    {
        countA = channelA.read((ByteBuffer)bufferA.clear());
        countB = channelB.read((ByteBuffer)bufferB.clear());
        assertEquals("Should read same number of bytes", countA, countB);
    } while (countA > 6);

    // both buffers should be at the same marker 6
    assertEquals("BufferA marker incorrect", 6, bufferA.position());
    assertEquals("BufferB marker incorrect", 6, bufferB.position());
}

/**
 * Uploads an entire file, closing the descriptor when it is no longer needed.
 *
 * @param fd The file descriptor to upload
 * @throws IllegalArgumentException if {@code fd} is not a file.
 * @return A new UploadDataProvider for the given file descriptor
 */
public static UploadDataProvider create(final ParcelFileDescriptor fd) {
    return new FileUploadProvider(new FileChannelProvider() {
        @Override
        public FileChannel getChannel() throws IOException {
            if (fd.getStatSize() != -1) {
                return new ParcelFileDescriptor.AutoCloseInputStream(fd).getChannel();
            } else {
                fd.close();
                throw new IllegalArgumentException("Not a file: " + fd);
            }
        }
    });
}

/**
 * Transfers data from FileChannel using 
 * {@link FileChannel#transferTo(long, long, WritableByteChannel)}. 
 * 
 * Similar to readFully(), this waits till requested amount of 
 * data is transfered.
 * 
 * @param fileCh FileChannel to transfer data from.
 * @param position position within the channel where the transfer begins
 * @param count number of bytes to transfer.
 * 
 * @throws EOFException 
 *         If end of input file is reached before requested number of 
 *         bytes are transfered.
 *
 * @throws SocketTimeoutException 
 *         If this channel blocks transfer longer than timeout for 
 *         this stream.
 *          
 * @throws IOException Includes any exception thrown by 
 *         {@link FileChannel#transferTo(long, long, WritableByteChannel)}. 
 */
public void transferToFully(FileChannel fileCh, long position, int count) 
                            throws IOException {

  while (count > 0) {
    /* 
     * Ideally we should wait after transferTo returns 0. But because of
     * a bug in JRE on Linux (http://bugs.sun.com/view_bug.do?bug_id=5103988),
     * which throws an exception instead of returning 0, we wait for the
     * channel to be writable before writing to it. If you ever see 
     * IOException with message "Resource temporarily unavailable" 
     * thrown here, please let us know.
     * 
     * Once we move to JAVA SE 7, wait should be moved to correct place.
     */
    waitForWritable();
    int nTransfered = (int) fileCh.transferTo(position, count, getChannel());

    if (nTransfered == 0) {
      //check if end of file is reached.
      if (position >= fileCh.size()) {
        throw new EOFException("EOF Reached. file size is " + fileCh.size() + 
                               " and " + count + " more bytes left to be " +
                               "transfered.");
      }
      //otherwise assume the socket is full.
      //waitForWritable(); // see comment above.
    } else if (nTransfered < 0) {
      throw new IOException("Unexpected return of " + nTransfered + 
                            " from transferTo()");
    } else {
      position += nTransfered;
      count -= nTransfered;
    }
  }
}

@Override
public void pipe(Result result) throws PipeException {
    if (!baseDir.exists()) {
        throw new PipeException("Base directory not exists");
    }
    Page page = result.getPage();
    String savePath, saveType = "";
    if (page.getExtra() != null && page.getExtra().get("savePath") != null) {
        savePath = page.getExtra().get("savePath").toString();
    } else {
        String url = page.getTask().getUrl();
        String name = StringHelper.toEscapedFileName(url.substring(url.lastIndexOf('/') + 1));
        String dir = StringHelper.cutOffProtocolAndHost(url.substring(0, url.lastIndexOf('/') + 1));
        savePath = dir.concat(StringHelper.toEscapedFileName(name));
        saveType = page.getExtra().get("saveType") == null ? "" : page.getExtra().get("saveType").toString();
    }
    if (!saveType.equals("")) {
        savePath = savePath.endsWith(".".concat(saveType)) ? savePath : savePath.concat(".").concat(saveType);
    }
    savePath = savePath.startsWith("/") ? savePath.substring(1) : savePath;
    if (savePath.indexOf('/') != -1) {
        new File(basePath + savePath.substring(0, savePath.lastIndexOf('/') + 1)).mkdirs();
    }
    File file = new File(basePath + savePath);
    try {
        file.createNewFile();
        FileChannel channel = new FileOutputStream(file).getChannel();
        channel.write(ByteBuffer.wrap((byte[]) result.getData()));
        channel.close();
    } catch (IOException e) {
        throw new PipeException(e.getMessage(), e.getCause());
    }
}

@Override
public void transferFrom(FileChannel source, IoCallback callback) {
    if (inCall) {
        queue(source, callback);
        return;
    }
    performTransfer(source, callback);
}

public void index(Map<String, Object> context) throws Exception {
    long size;
    String content;
    String modified;
    File file = LoggerFactory.getFile();
    if (file != null && file.exists()) {
        FileInputStream fis = new FileInputStream(file);
        FileChannel channel = fis.getChannel();
        size = channel.size();
        ByteBuffer bb;
        if (size <= SHOW_LOG_LENGTH) {
            bb = ByteBuffer.allocate((int) size);
            channel.read(bb, 0);
        } else {
            int pos = (int) (size - SHOW_LOG_LENGTH);
            bb = ByteBuffer.allocate(SHOW_LOG_LENGTH);
            channel.read(bb, pos);
        }
        bb.flip();
        content = new String(bb.array()).replace("<", "&lt;").replace(">", "&gt;");
        modified = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date(file.lastModified()));
    } else {
        size = 0;
        content = "";
        modified = "Not exist";
    }
    Level level = LoggerFactory.getLevel();
    context.put("name", file == null ? "" : file.getAbsoluteFile());
    context.put("size", String.valueOf(size));
    context.put("level", level == null ? "" : level);
    context.put("modified", modified);
    context.put("content", content);
}