001
002 /*
003 * Copyright (C) 2011 Archie L. Cobbs. All rights reserved.
004 *
005 * $Id: InputStreamReader.java 162 2011-10-24 20:42:37Z archie.cobbs $
006 */
007
008 package org.dellroad.stuff.io;
009
010 import java.io.FilterInputStream;
011 import java.io.IOException;
012 import java.io.InputStream;
013
014 /**
015 * Reads zero or more {@link InputStream}s serialized inside an underlying {@link InputStream} by an {@link OutputStreamWriter}.
016 *
017 * <p>
018 * Instances of this class are thread safe, as are the {@link InputStream}s returned by {@link #read}.
019 * </p>
020 *
021 * @see OutputStreamWriter
022 */
023 public class InputStreamReader {
024
025 // Escape codes
026 static final int CONTROL_ESCAPE = 1;
027 static final int CONTROL_SEPARATOR = 2;
028
029 // Special return values
030 private static final int RESULT_EOF = -1; // read EOF on input
031 private static final int RESULT_SEPARATOR = -2; // read stream separator
032
033 private final RandomEscape randomEscape = new RandomEscape();
034 private final InputStream input;
035
036 private NestedInputStream current; // curently active nested input
037 private IOException exception; // exception thrown on input
038 private int escape; // current escape character
039 private boolean closed; // whether this instance is closed
040 private boolean eof; // whether this instance has read EOF
041
042 /**
043 * Constructor.
044 *
045 * @param input the underlying {@link InputStream} that will carry nested {@link InputStream}s within it
046 */
047 public InputStreamReader(InputStream input) {
048 this.input = input;
049 this.escape = this.randomEscape.next();
050 }
051
052 /**
053 * Read the next {@link InputStream}.
054 *
055 * <p>
056 * The returned {@link InputStream} will remain valid as long as it has not been closed, the underlying input
057 * has not been closed, and no subsequent invocation of this method has been made. As soon as any of those events occurs,
058 * all subsequent accesses to the previously returned {@link InputStream} will throw an exception.
059 * </p>
060 *
061 * <p>
062 * The returned {@link InputStream} may be closed prior to its EOF, in which case any remaining bytes will be skipped
063 * over during the next invocation of this method.
064 * </p>
065 *
066 * <p>
067 * Note: while any thread is blocked reading from the returned {@link InputStream}, this method will block as well.
068 * </p>
069 *
070 * @throws IOException if this instance is closed
071 * @throws IOException if the underlying {@link InputStream} has thrown an exception
072 * @throws return the next {@link InputStream}, or {@code null} if EOF has been reached on the underlying input
073 */
074 public synchronized InputStream read() throws IOException {
075
076 // Check state
077 if (this.closed)
078 throw new IOException("this instance is closed");
079 if (this.exception != null)
080 throw new IOException("exception on the underlying stream", this.exception);
081 if (this.eof)
082 return null;
083
084 // Close current stream (if any) and skip past abandoned bytes (if any)
085 if (this.current != null) {
086 this.current.close();
087 if (!this.current.isEOF()) {
088 while (true) {
089 if (this.readNext() < 0)
090 break;
091 }
092 }
093 }
094
095 // See what's next up
096 int firstValue = this.readNext();
097 if (firstValue == RESULT_EOF)
098 return null;
099
100 // Create new nested stream
101 this.current = new NestedInputStream(firstValue);
102 return this.current;
103 }
104
105 // Read next byte (unescaped) or special return value
106 private int readNext() throws IOException {
107
108 // Already read EOF?
109 if (this.eof)
110 return RESULT_EOF;
111
112 // Read next byte
113 int ch = this.input.read();
114 if (ch == -1) {
115 this.eof = true;
116 return RESULT_EOF;
117 }
118
119 // Check for escape byte
120 if (ch == this.escape) {
121
122 // Read escaped byte
123 if ((ch = this.input.read()) == -1) {
124 this.eof = true;
125 return RESULT_EOF;
126 }
127 ch ^= this.escape;
128
129 // Advance escape character for next time
130 int prevEscape = this.escape;
131 this.escape = this.randomEscape.next();
132
133 // Check control code
134 switch (ch) {
135 case CONTROL_ESCAPE:
136 return prevEscape;
137 case CONTROL_SEPARATOR:
138 return RESULT_SEPARATOR;
139 default:
140 break;
141 }
142 throw new IOException("rec'd unexpected escape code " + ch);
143 }
144
145 // Just a normal character
146 return ch;
147 }
148
149 /**
150 * Close this instance. Does nothing if already closed.
151 *
152 * <p>
153 * This closes the underlying input; however, if the {@link InputStream} most recently returned from {@link #read}
154 * is still open, the close of the underlying input will be postponed until it is no longer open.
155 * </p>
156 *
157 * @throws IOException if an there is an error closing the underlying {@link InputStream}
158 */
159 public synchronized void close() throws IOException {
160 if (this.closed)
161 return;
162 this.closed = true;
163 if (this.current != null)
164 this.current.checkInputClose();
165 }
166
167 private class NestedInputStream extends FilterInputStream {
168
169 private int firstValue;
170 private boolean firstRead = true;
171 private boolean closed;
172 private boolean eof;
173
174 public NestedInputStream(int firstValue) {
175 super(InputStreamReader.this.input);
176 this.firstValue = firstValue;
177 }
178
179 @Override
180 public int read() throws IOException {
181 synchronized (InputStreamReader.this) {
182
183 // Check state
184 if (this.closed)
185 throw new IOException("stream is closed");
186 if (InputStreamReader.this.exception != null)
187 throw new IOException("exception on the underlying stream", InputStreamReader.this.exception);
188 if (this.eof)
189 return -1;
190
191 // Read from the underlying stream
192 try {
193
194 // Read next unescaped byte or control code
195 int ch;
196 if (this.firstRead) {
197 ch = this.firstValue;
198 this.firstRead = false;
199 } else
200 ch = InputStreamReader.this.readNext();
201 switch (ch) {
202 case RESULT_EOF:
203 throw new IOException("underlying stream was truncated");
204 case RESULT_SEPARATOR:
205 this.eof = true;
206 return -1;
207 default:
208 break;
209 }
210
211 // Done
212 return ch;
213 } catch (IOException e) {
214 InputStreamReader.this.exception = e;
215 throw e;
216 }
217 }
218 }
219
220 /**
221 * Close this instance. Does nothing if already closed.
222 */
223 @Override
224 public void close() throws IOException {
225 synchronized (InputStreamReader.this) {
226 if (this.closed)
227 return;
228 this.closed = true;
229 this.checkInputClose();
230 }
231 }
232
233 @Override
234 public int read(byte[] buf, int off, int len) throws IOException {
235 if (len < 0 || off < 0 || off + len > buf.length)
236 throw new IndexOutOfBoundsException();
237 int count = 0;
238 while (count < len) {
239 int ch = this.read();
240 if (ch == -1) {
241 if (count == 0)
242 return -1;
243 break;
244 }
245 buf[off++] = (byte)ch;
246 count++;
247 }
248 return count;
249 }
250
251 @Override
252 public long skip(long num) throws IOException {
253 long count = 0;
254 while (count < num && this.read() != -1)
255 count++;
256 return count;
257 }
258
259 @Override
260 public boolean markSupported() {
261 return false;
262 }
263
264 @Override
265 public void mark(int readlimit) {
266 }
267
268 @Override
269 public void reset() throws IOException {
270 throw new IOException("mark/reset not supported");
271 }
272
273 public void checkInputClose() throws IOException {
274 synchronized (InputStreamReader.this) {
275 if (this.closed && InputStreamReader.this.closed)
276 this.in.close();
277 }
278 }
279
280 public boolean isEOF() {
281 return this.eof;
282 }
283 }
284 }
285