Given a Java InputStream, how can I determine the current offset in the stream?

Question

I'd like something like a generic, re-usable getPosition() method that will tell me the number of bytes read from the starting point of the stream. Ideally, I would prefer this to work with all InputStreams, so that I don't have to wrap each and every one of them as I get them from disparate sources.

Does such a beast exist? If not, can anyone recommend an existing implementation of a counting InputStream?

Answer 1

You'll need to follow the Decorator pattern established in java.io to implement this.

Let's give it a try here:

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

public final class PositionInputStream
  extends FilterInputStream
{

  private long pos = 0;

  private long mark = 0;

  public PositionInputStream(InputStream in)
  {
    super(in);
  }

  /**
   * <p>Get the stream position.</p>
   *
   * <p>Eventually, the position will roll over to a negative number.
   * Reading 1 Tb per second, this would occur after approximately three 
   * months. Applications should account for this possibility in their 
   * design.</p>
   *
   * @return the current stream position.
   */
  public synchronized long getPosition()
  {
    return pos;
  }

  @Override
  public synchronized int read()
    throws IOException
  {
    int b = super.read();
    if (b >= 0)
      pos += 1;
    return b;
  }

  @Override
  public synchronized int read(byte[] b, int off, int len)
    throws IOException
  {
    int n = super.read(b, off, len);
    if (n > 0)
      pos += n;
    return n;
  }

  @Override
  public synchronized long skip(long skip)
    throws IOException
  {
    long n = super.skip(skip);
    if (n > 0)
      pos += n;
    return n;
  }

  @Override
  public synchronized void mark(int readlimit)
  {
    super.mark(readlimit);
    mark = pos;
  }

  @Override
  public synchronized void reset()
    throws IOException
  {
    /* A call to reset can still succeed if mark is not supported, but the 
     * resulting stream position is undefined, so it's not allowed here. */
    if (!markSupported())
      throw new IOException("Mark not supported.");
    super.reset();
    pos = mark;
  }

}

The InputStreams are intended to be thread safe, so that accounts for the liberal use of synchronization. I played around with volatile and AtomicLong position variables, but synchronization is probably best because it allows one thread to operate on the stream and query its position without relinquishing the lock.

PositionInputStream is = …
synchronized (is) {
  is.read(buf);
  pos = is.getPosition();
}