// File created: 2010-08-09 14:34:08

package fi.tkk.ics.hadoop.bam;

import java.io.InputStream;
import java.io.IOException;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.Seekable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import net.sf.samtools.SAMFileReader;
import net.sf.samtools.util.BlockCompressedInputStream;
import net.sf.samtools.util.SeekableStream;

import fi.tkk.ics.hadoop.bam.customsamtools.BAMRecordCodec;
import fi.tkk.ics.hadoop.bam.customsamtools.SAMRecord;

/** The key is the bitwise OR of the reference sequence ID in the upper 32 bits
 * and the 0-based leftmost coordinate in the lower.
 */
public class BAMRecordReader
	extends RecordReader<LongWritable,SAMRecordWritable>
{
	private final LongWritable key = new LongWritable();
	private final SAMRecordWritable record = new SAMRecordWritable();

	private BlockCompressedInputStream bci;
	private BAMRecordCodec codec;
	private float fileSizeFloat;

	@Override public void initialize(InputSplit spl, TaskAttemptContext ctx)
		throws IOException
	{
		final FileVirtualSplit split = (FileVirtualSplit)spl;

		final Path file = split.getPath();
		final FileSystem fs = FileSystem.get(ctx.getConfiguration());

		final long fileSize = fs.getFileStatus(file).getLen();
		fileSizeFloat = fileSize;

		final FSDataInputStream in = fs.open(file);
		codec = new BAMRecordCodec(new SAMFileReader(in).getFileHeader());

		in.seek(0);
		bci =
			new BlockCompressedInputStream(
				new WrapSeekable<FSDataInputStream>(in, fileSize, file));

		bci.seek(split.getStartVirtualOffset());
		codec.setInputStream(bci);
	}
	@Override public void close() throws IOException { bci.close(); }

	/** Inexact; generally never reaches 1. */
	@Override public float getProgress() {
		return (float)(bci.getFilePointer() >>> 16) / fileSizeFloat;
	}
	@Override public LongWritable      getCurrentKey  () { return key; }
	@Override public SAMRecordWritable getCurrentValue() { return record; }

	@Override public boolean nextKeyValue() {
		SAMRecord r = codec.decode();
		if (r == null)
			return false;
		key.set((long)r.getReferenceIndex() << 32
		            | r.getAlignmentStart() - 1);
		record.set(r);
		return true;
	}
}

// Hadoop and the SAM tools each have their own "seekable stream" abstraction.
// This class wraps Hadoop's so that we can give such a one to
// BlockCompressedInputStream and retain seekability.
class WrapSeekable<S extends InputStream & Seekable> extends SeekableStream {
	private final S    stm;
	private final long len;
	private final Path path;

	public WrapSeekable(final S s, long length, Path p) {
		stm  = s;
		len  = length;
		path = p;
	}

	@Override public String getSource() { return path.toString(); }
	@Override public long   length   () { return len; }

	@Override public void    close() throws IOException { stm.close(); }
	@Override public boolean eof  () throws IOException {
		return stm.getPos() == length();
	}
	@Override public void seek(long pos) throws IOException {
		stm.seek(pos);
	}
	@Override public int read() throws IOException {
		return stm.read();
	}
	@Override public int read(byte[] buf, int offset, int len)
		throws IOException
	{
		return stm.read(buf, offset, len);
	}
}

@@ -1157,27 +1157,34 @@
 ate 
-f
 lo
-at
+ng
  fileS
-izeFloat
+tart, virtualEnd
 ;%0A%0A%09
@@ -1443,94 +1443,8 @@
 );%0A%0A
-%09%09final long fileSize = fs.getFileStatus(file).getLen();%0A%09%09fileSizeFloat = fileSize;%0A%0A
 %09%09fi
@@ -1656,320 +1656,715 @@
 am%3E(
-in, fileSize, file));%0A%0A%09%09bci.seek(split.getStartVirtualOffset());%0A%09%09codec.setInputStream(bci);%0A%09%7D%0A%09@Override public void close() throws IOException %7B bci.close(); %7D%0A%0A%09/** Inexact; generally never reaches 1. */%0A%09@Override public float getProgress() %7B%0A%09%09return (float)(bci.getFilePointer() %3E%3E%3E 16) / fileSizeFloat;
+%0A%09%09%09%09%09in, fs.getFileStatus(file).getLen(), file));%0A%0A%09%09final long virtualStart = split.getStartVirtualOffset();%0A%0A%09%09fileStart  = virtualStart %3E%3E%3E 16;%0A%09%09virtualEnd = split.getEndVirtualOffset();%0A%0A%09%09bci.seek(virtualStart);%0A%09%09codec.setInputStream(bci);%0A%09%7D%0A%09@Override public void close() throws IOException %7B bci.close(); %7D%0A%0A%09@Override public float getProgress() %7B%0A%09%09final long virtPos = bci.getFilePointer();%0A%09%09final long filePos = virtPos %3E%3E%3E 16;%0A%09%09if (virtPos %3E= virtualEnd)%0A%09%09%09return 1;%0A%09%09else %7B%0A%09%09%09final long fileEnd = virtualEnd %3E%3E%3E 16;%0A%09%09%09// Add 1 to the denominator to make sure it doesn't reach 1 here when%0A%09%09%09// filePos == fileEnd.%0A%09%09%09return (float)(filePos - fileStart) / (fileEnd - fileStart + 1);%0A%09%09%7D
 %0A%09%7D%0A
@@ -2552,16 +2552,82 @@
 e() %7B%0A%09%09
+if (bci.getFilePointer() %3E= virtualEnd)%0A%09%09%09return false;%0A%0A%09%09final 
 SAMRecor

stop reading at split end, not file end.