View Javadoc
1   /*
2    * Licensed to the Apache Software Foundation (ASF) under one or more
3    * contributor license agreements.  See the NOTICE file distributed with
4    * this work for additional information regarding copyright ownership.
5    * The ASF licenses this file to You under the Apache License, Version 2.0
6    * (the "License"); you may not use this file except in compliance with
7    * the License.  You may obtain a copy of the License at
8    *
9    *      http://www.apache.org/licenses/LICENSE-2.0
10   *
11   * Unless required by applicable law or agreed to in writing, software
12   * distributed under the License is distributed on an "AS IS" BASIS,
13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14   * See the License for the specific language governing permissions and
15   * limitations under the License.
16   */
17  package org.apache.commons.vfs2.util;
18  
19  import java.io.BufferedInputStream;
20  import java.io.IOException;
21  import java.io.InputStream;
22  import java.util.concurrent.atomic.AtomicBoolean;
23  import java.util.concurrent.atomic.AtomicLong;
24  
25  /**
26   * An InputStream that provides buffering and end-of-stream monitoring.
27   */
28  public class MonitorInputStream extends BufferedInputStream {
29  
30      private static final int EOF_CHAR = -1;
31      private final AtomicBoolean finished = new AtomicBoolean(false);
32      private final AtomicLong atomicCount = new AtomicLong(0);
33  
34      /**
35       * Constructs a MonitorInputStream from the passed InputStream
36       *
37       * @param in The input stream to wrap.
38       */
39      public MonitorInputStream(final InputStream in) {
40          super(in);
41      }
42  
43      /**
44       * Constructs a MonitorInputStream from the passed InputStream and with the specified buffer size
45       *
46       * @param in The input stream to wrap.
47       * @param bufferSize The buffer size to use.
48       * @since 2.4
49       */
50      public MonitorInputStream(final InputStream in, final int bufferSize) {
51          super(in, bufferSize);
52      }
53  
54      /**
55       * Returns 0 if the stream is at EOF, else the underlying inputStream will be queried.
56       *
57       * @return The number of bytes that are available.
58       * @throws IOException if an error occurs.
59       * @since 2.0
60       */
61      @Override
62      public synchronized int available() throws IOException {
63          if (finished.get()) {
64              return 0;
65          }
66  
67          return super.available();
68      }
69  
70      /**
71       * Reads a character.
72       *
73       * @return The character that was read as an integer.
74       * @throws IOException if an error occurs.
75       */
76      @Override
77      public int read() throws IOException { // lgtm [java/non-sync-override]
78          if (finished.get()) {
79              return EOF_CHAR;
80          }
81  
82          final int ch = super.read();
83          if (ch != EOF_CHAR) {
84              atomicCount.incrementAndGet();
85              return ch;
86          }
87  
88          return EOF_CHAR;
89      }
90  
91      /**
92       * Reads bytes from this input stream.
93       *
94       * @param buffer A byte array in which to place the characters read.
95       * @param offset The offset at which to start reading.
96       * @param length The maximum number of bytes to read.
97       * @return The number of bytes read.
98       * @throws IOException if an error occurs.
99       */
100     @Override
101     public int read(final byte[] buffer, final int offset, final int length) throws IOException { // lgtm [java/non-sync-override]
102         if (finished.get()) {
103             return EOF_CHAR;
104         }
105 
106         final int nread = super.read(buffer, offset, length);
107         if (nread != EOF_CHAR) {
108             atomicCount.addAndGet(nread);
109             return nread;
110         }
111         return EOF_CHAR;
112     }
113 
114     /**
115      * Closes this input stream and releases any system resources associated with the stream.
116      *
117      * @throws IOException if an error occurs.
118      */
119     @Override
120     public void close() throws IOException {
121         final boolean closed = finished.getAndSet(true);
122         if (closed) {
123             return;
124         }
125 
126         // Close the stream
127         IOException exc = null;
128         try {
129             super.close();
130         } catch (final IOException ioe) {
131             exc = ioe;
132         }
133 
134         // Notify that the stream has been closed
135         try {
136             onClose();
137         } catch (final IOException ioe) {
138             exc = ioe;
139         }
140 
141         if (exc != null) {
142             throw exc;
143         }
144     }
145 
146     /**
147      * Called after the stream has been closed. This implementation does nothing.
148      *
149      * @throws IOException if an error occurs.
150      */
151     protected void onClose() throws IOException {
152         // noop
153     }
154 
155     /**
156      * Get the number of bytes read by this input stream.
157      *
158      * @return The number of bytes read by this input stream.
159      */
160     public long getCount() {
161         return atomicCount.get();
162     }
163 }