java/src/main/com/google/appengine/api/files/BufferedFileReadChannelImpl.java

   1 package com.google.appengine.api.files;
   2
   3 import java.io.IOException;
   4 import java.nio.ByteBuffer;
   5
   6 /**
   7  * Wraps a FileReadChannel to provide buffering of reads.
   8  */
   9 public class BufferedFileReadChannelImpl implements FileReadChannel {
  10
  11   private final FileReadChannel readChannel;
  12   private final int bufferSize;
  13   private final ByteBuffer buffer;
  14   private boolean encounteredEof;
  15   private final Object lock = new Object();
  16
  17   public BufferedFileReadChannelImpl(FileReadChannel readChannel, int bufferSize) {
  18     this.readChannel = readChannel;
  19     this.bufferSize = bufferSize;
  20     this.buffer = ByteBuffer.allocate(bufferSize);
  21     makeEmpty();
  22     this.encounteredEof = false;
  23   }
  24
  25   private void makeEmpty() {
  26     this.buffer.position(buffer.limit());
  27   }
  28
  29   @Override
  30   public int read(ByteBuffer dst) throws IOException {
  31     synchronized (lock) {
  32       if (encounteredEof && buffer.remaining() == 0) {
  33         return -1;
  34       }
  35       if (dst.remaining() == 0) {
  36         if (buffer.remaining() == 0) {
  37           refillBuffer();
  38           if (encounteredEof) {
  39             return -1;
  40           } else {
  41             return 0;
  42           }
  43         } else {
  44           return 0;
  45         }
  46       }
  47       int bytesRead;
  48       if (bufferHasMoreThanNeeded(dst)) {
  49         bytesRead = readIntoShortBuffer(dst);
  50       } else {
  51         bytesRead = buffer.remaining();
  52         dst.put(buffer);
  53         assert buffer.remaining() == 0;
  54         while (dst.remaining() > bufferSize && !encounteredEof) {
  55           int bytesFromChannel = readChannel.read(dst);
  56           if (bytesFromChannel == -1) {
  57             encounteredEof = true;
  58           } else {
  59             bytesRead += bytesFromChannel;
  60           }
  61         }
  62         if (dst.remaining() > 0 && dst.remaining() <= bufferSize && !encounteredEof) {
  63           refillBuffer();
  64           bytesRead += readIntoShortBuffer(dst);
  65         }
  66       }
  67       return (bytesRead == 0 && encounteredEof) ? -1 : bytesRead;
  68     }
  69   }
  70
  71   private boolean bufferHasMoreThanNeeded(ByteBuffer dst) {
  72     return dst.remaining() <= buffer.remaining();
  73   }
  74
  75   private int readIntoShortBuffer(ByteBuffer dst) {
  76     int oldLimit = buffer.limit();
  77     int newLimit = Math.min(oldLimit, buffer.position() + dst.remaining());
  78     int bytesRead = Math.min(dst.remaining(), buffer.remaining());
  79     buffer.limit(newLimit);
  80     try {
  81       dst.put(buffer);
  82     } finally {
  83       buffer.limit(oldLimit);
  84     }
  85     return bytesRead;
  86   }
  87
  88   private void refillBuffer() throws IOException {
  89     assert buffer.remaining() == 0;
  90     buffer.clear();
  91     int read = readChannel.read(buffer);
  92     buffer.flip();
  93     if (read == -1) {
  94       encounteredEof = true;
  95     }
  96   }
  97
  98   /**
  99    * {@inheritDoc}
 100    */
 101   @Override
 102   public boolean isOpen() {
 103     return readChannel.isOpen();
 104   }
 105
 106   /**
 107    * {@inheritDoc}
 108    */
 109   @Override
 110   public void close() throws IOException {
 111     readChannel.close();
 112   }
 113
 114   @Override
 115   public long position() throws IOException {
 116     synchronized (lock) {
 117       long position = readChannel.position();
 118       return position - buffer.remaining();
 119     }
 120   }
 121
 122   @Override
 123   public FileReadChannel position(long newPosition) throws IOException {
 124     synchronized (lock) {
 125       long startingPosition = position();
 126       int relativePos = (int) (newPosition - startingPosition);
 127       if (relativePos < buffer.remaining() && relativePos >= 0) {
 128         buffer.position(buffer.position() + relativePos);
 129       } else {
 130         buffer.clear();
 131         makeEmpty();
 132         readChannel.position(newPosition);
 133       }
 134       return this;
 135     }
 136   }
 137
 138 }