+ * This class is based on AutoCloseInputStream in HttpClient 3.1,
+ * but has notable differences. It does not allow mark/reset, distinguishes
+ * different kinds of event, and does not always close the underlying stream
+ * on EOF. That decision is left to the {@link EofSensorWatcher watcher}.
+ *
+ * @see EofSensorWatcher
+ *
+ * @since 4.0
+ */
+// don't use FilterInputStream as the base class, we'd have to
+// override markSupported(), mark(), and reset() to disable them
+@NotThreadSafe
+public class EofSensorInputStream extends InputStream implements ConnectionReleaseTrigger {
+
+ /**
+ * The wrapped input stream, while accessible.
+ * The value changes to null when the wrapped stream
+ * becomes inaccessible.
+ */
+ protected InputStream wrappedStream;
+
+ /**
+ * Indicates whether this stream itself is closed.
+ * If it isn't, but {@link #wrappedStream wrappedStream}
+ * is null, we're running in EOF mode.
+ * All read operations will indicate EOF without accessing
+ * the underlying stream. After closing this stream, read
+ * operations will trigger an {@link IOException IOException}.
+ *
+ * @see #isReadAllowed isReadAllowed
+ */
+ private boolean selfClosed;
+
+ /** The watcher to be notified, if any. */
+ private final EofSensorWatcher eofWatcher;
+
+ /**
+ * Creates a new EOF sensor.
+ * If no watcher is passed, the underlying stream will simply be
+ * closed when EOF is detected or {@link #close close} is called.
+ * Otherwise, the watcher decides whether the underlying stream
+ * should be closed before detaching from it.
+ *
+ * @param in the wrapped stream
+ * @param watcher the watcher for events, or null for
+ * auto-close behavior without notification
+ */
+ public EofSensorInputStream(final InputStream in,
+ final EofSensorWatcher watcher) {
+ if (in == null) {
+ throw new IllegalArgumentException
+ ("Wrapped stream may not be null.");
+ }
+
+ wrappedStream = in;
+ selfClosed = false;
+ eofWatcher = watcher;
+ }
+
+ /**
+ * Checks whether the underlying stream can be read from.
+ *
+ * @return true if the underlying stream is accessible,
+ * false if this stream is in EOF mode and
+ * detached from the underlying stream
+ *
+ * @throws IOException if this stream is already closed
+ */
+ protected boolean isReadAllowed() throws IOException {
+ if (selfClosed) {
+ throw new IOException("Attempted read on closed stream.");
+ }
+ return (wrappedStream != null);
+ }
+
+ @Override
+ public int read() throws IOException {
+ int l = -1;
+
+ if (isReadAllowed()) {
+ try {
+ l = wrappedStream.read();
+ checkEOF(l);
+ } catch (IOException ex) {
+ checkAbort();
+ throw ex;
+ }
+ }
+
+ return l;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ int l = -1;
+
+ if (isReadAllowed()) {
+ try {
+ l = wrappedStream.read(b, off, len);
+ checkEOF(l);
+ } catch (IOException ex) {
+ checkAbort();
+ throw ex;
+ }
+ }
+
+ return l;
+ }
+
+ @Override
+ public int read(byte[] b) throws IOException {
+ int l = -1;
+
+ if (isReadAllowed()) {
+ try {
+ l = wrappedStream.read(b);
+ checkEOF(l);
+ } catch (IOException ex) {
+ checkAbort();
+ throw ex;
+ }
+ }
+ return l;
+ }
+
+ @Override
+ public int available() throws IOException {
+ int a = 0; // not -1
+
+ if (isReadAllowed()) {
+ try {
+ a = wrappedStream.available();
+ // no checkEOF() here, available() can't trigger EOF
+ } catch (IOException ex) {
+ checkAbort();
+ throw ex;
+ }
+ }
+
+ return a;
+ }
+
+ @Override
+ public void close() throws IOException {
+ // tolerate multiple calls to close()
+ selfClosed = true;
+ checkClose();
+ }
+
+ /**
+ * Detects EOF and notifies the watcher.
+ * This method should only be called while the underlying stream is
+ * still accessible. Use {@link #isReadAllowed isReadAllowed} to
+ * check that condition.
+ *
+ * If EOF is detected, the watcher will be notified and this stream
+ * is detached from the underlying stream. This prevents multiple
+ * notifications from this stream.
+ *
+ * @param eof the result of the calling read operation.
+ * A negative value indicates that EOF is reached.
+ *
+ * @throws IOException
+ * in case of an IO problem on closing the underlying stream
+ */
+ protected void checkEOF(int eof) throws IOException {
+
+ if ((wrappedStream != null) && (eof < 0)) {
+ try {
+ boolean scws = true; // should close wrapped stream?
+ if (eofWatcher != null)
+ scws = eofWatcher.eofDetected(wrappedStream);
+ if (scws)
+ wrappedStream.close();
+ } finally {
+ wrappedStream = null;
+ }
+ }
+ }
+
+ /**
+ * Detects stream close and notifies the watcher.
+ * There's not much to detect since this is called by {@link #close close}.
+ * The watcher will only be notified if this stream is closed
+ * for the first time and before EOF has been detected.
+ * This stream will be detached from the underlying stream to prevent
+ * multiple notifications to the watcher.
+ *
+ * @throws IOException
+ * in case of an IO problem on closing the underlying stream
+ */
+ protected void checkClose() throws IOException {
+
+ if (wrappedStream != null) {
+ try {
+ boolean scws = true; // should close wrapped stream?
+ if (eofWatcher != null)
+ scws = eofWatcher.streamClosed(wrappedStream);
+ if (scws)
+ wrappedStream.close();
+ } finally {
+ wrappedStream = null;
+ }
+ }
+ }
+
+ /**
+ * Detects stream abort and notifies the watcher.
+ * There's not much to detect since this is called by
+ * {@link #abortConnection abortConnection}.
+ * The watcher will only be notified if this stream is aborted
+ * for the first time and before EOF has been detected or the
+ * stream has been {@link #close closed} gracefully.
+ * This stream will be detached from the underlying stream to prevent
+ * multiple notifications to the watcher.
+ *
+ * @throws IOException
+ * in case of an IO problem on closing the underlying stream
+ */
+ protected void checkAbort() throws IOException {
+
+ if (wrappedStream != null) {
+ try {
+ boolean scws = true; // should close wrapped stream?
+ if (eofWatcher != null)
+ scws = eofWatcher.streamAbort(wrappedStream);
+ if (scws)
+ wrappedStream.close();
+ } finally {
+ wrappedStream = null;
+ }
+ }
+ }
+
+ /**
+ * Same as {@link #close close()}.
+ */
+ public void releaseConnection() throws IOException {
+ close();
+ }
+
+ /**
+ * Aborts this stream.
+ * This is a special version of {@link #close close()} which prevents
+ * re-use of the underlying connection, if any. Calling this method
+ * indicates that there should be no attempt to read until the end of
+ * the stream.
+ */
+ public void abortConnection() throws IOException {
+ // tolerate multiple calls
+ selfClosed = true;
+ checkAbort();
+ }
+
+}
+