class ChunkReader implements Closeable {

    public int readLine(Writable str, int maxBytesToConsume) throws IOException {
        ChunkWritable cw = (ChunkWritable) str;
        List<Node> list = new LinkedList<Node>();

        boolean newLine = false; // length of terminating newline
        long bytesConsumed = 0;

        do {
            int startPosn = bufferPosn; // starting from where we left off the
                                        // last time
            if (bufferPosn >= bufferLength) {
                startPosn = bufferPosn = 0;

                bufferLength = in.read(buffer);
                if (bufferLength <= 0) {
                    break; // EOF
                }
            }

            for (; bufferPosn < bufferLength; ++bufferPosn) { // search for
                                                              // newline
                if (buffer[bufferPosn] == LF) {
                    newLine = true;
                    ++bufferPosn; // at next invocation proceed from following
                                  // byte
                    break;
                }
            }

            int readLength = bufferPosn - startPosn;
            bytesConsumed += readLength;

            if (readLength > 0) {
                Node nd = new Node();
                nd.slice = new byte[readLength];
                nd.len = readLength;
                System.arraycopy(buffer, startPosn, nd.slice, 0, nd.len);
                list.add(nd);
            }
        } while (!newLine && bytesConsumed < maxBytesToConsume);

        if (list.size() > 0) {
            cw.box = new byte[(int) bytesConsumed];
            int pos = 0;
            for (int i = 0; i < list.size(); i++) {
                Node n = list.get(i);
                System.arraycopy(n.slice, 0, cw.box, pos, n.len);
                pos += n.len;
            }
        }

        return (int) bytesConsumed;
    }

    public  ChunkReader(InputStream in);

}