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
18 package org.apache.commons.net.telnet;
19
20 import java.io.BufferedInputStream;
21 import java.io.IOException;
22 import java.io.InputStream;
23 import java.io.InterruptedIOException;
24
25 final class TelnetInputStream extends BufferedInputStream implements Runnable {
26 /** End of file has been reached */
27 private static final int EOF = -1;
28
29 /** Read would block */
30 private static final int WOULD_BLOCK = -2;
31
32 // TODO should these be private enums?
33 static final int STATE_DATA = 0, STATE_IAC = 1, STATE_WILL = 2, STATE_WONT = 3, STATE_DO = 4, STATE_DONT = 5, STATE_SB = 6, STATE_SE = 7, STATE_CR = 8,
34 STATE_IAC_SB = 9;
35
36 private boolean hasReachedEOF; // @GuardedBy("queue")
37 private volatile boolean isClosed;
38 private boolean readIsWaiting;
39 private int receiveState, queueHead, queueTail, bytesAvailable;
40 private final int[] queue;
41 private final TelnetClient client;
42 private final Thread thread;
43 private IOException ioException;
44
45 /* TERMINAL-TYPE option (start) */
46 private final int suboption[];
47 private int suboptionCount;
48 /* TERMINAL-TYPE option (end) */
49
50 private volatile boolean threaded;
51
52 TelnetInputStream(final InputStream input, final TelnetClient client) {
53 this(input, client, true);
54 }
55
56 TelnetInputStream(final InputStream input, final TelnetClient client, final boolean readerThread) {
57 super(input);
58 this.client = client;
59 this.receiveState = STATE_DATA;
60 this.isClosed = true;
61 this.hasReachedEOF = false;
62 // Make it 2049, because when full, one slot will go unused, and we
63 // want a 2048 byte buffer just to have a round number (base 2 that is)
64 this.queue = new int[2049];
65 this.queueHead = 0;
66 this.queueTail = 0;
67 this.suboption = new int[client.maxSubnegotiationLength];
68 this.bytesAvailable = 0;
69 this.ioException = null;
70 this.readIsWaiting = false;
71 this.threaded = false;
72 if (readerThread) {
73 this.thread = new Thread(this);
74 } else {
75 this.thread = null;
76 }
77 }
78
79 @Override
80 public int available() throws IOException {
81 // Critical section because run() may change bytesAvailable
82 synchronized (queue) {
83 if (threaded) { // Must not call super.available when running threaded: NET-466
84 return bytesAvailable;
85 }
86 return bytesAvailable + super.available();
87 }
88 }
89
90 // Cannot be synchronized. Will cause deadlock if run() is blocked
91 // in read because BufferedInputStream read() is synchronized.
92 @Override
93 public void close() throws IOException {
94 // Completely disregard the fact thread may still be running.
95 // We can't afford to block on this close by waiting for
96 // thread to terminate because few if any JVM's will actually
97 // interrupt a system read() from the interrupt() method.
98 super.close();
99
100 synchronized (queue) {
101 hasReachedEOF = true;
102 isClosed = true;
103
104 if (thread != null && thread.isAlive()) {
105 thread.interrupt();
106 }
107
108 queue.notifyAll();
109 }
110
111 }
112
113 /** Returns false. Mark is not supported. */
114 @Override
115 public boolean markSupported() {
116 return false;
117 }
118
119 // synchronized(client) critical sections are to protect against
120 // TelnetOutputStream writing through the telnet client at same time
121 // as a processDo/Will/etc. command invoked from TelnetInputStream
122 // tries to write. Returns true if buffer was previously empty.
123 private boolean processChar(final int ch) throws InterruptedException {
124 // Critical section because we're altering bytesAvailable,
125 // queueTail, and the contents of _queue.
126 final boolean bufferWasEmpty;
127 synchronized (queue) {
128 bufferWasEmpty = bytesAvailable == 0;
129 while (bytesAvailable >= queue.length - 1) {
130 // The queue is full. We need to wait before adding any more data to it. Hopefully the stream owner
131 // will consume some data soon!
132 if (!threaded) {
133 // We've been asked to add another character to the queue, but it is already full and there's
134 // no other thread to drain it. This should not have happened!
135 throw new IllegalStateException("Queue is full! Cannot process another character.");
136 }
137 queue.notify();
138 try {
139 queue.wait();
140 } catch (final InterruptedException e) {
141 throw e;
142 }
143 }
144
145 // Need to do this in case we're not full, but block on a read
146 if (readIsWaiting && threaded) {
147 queue.notify();
148 }
149
150 queue[queueTail] = ch;
151 ++bytesAvailable;
152
153 if (++queueTail >= queue.length) {
154 queueTail = 0;
155 }
156 }
157 return bufferWasEmpty;
158 }
159
160 @Override
161 public int read() throws IOException {
162 // Critical section because we're altering bytesAvailable,
163 // queueHead, and the contents of _queue in addition to
164 // testing value of hasReachedEOF.
165 synchronized (queue) {
166
167 while (true) {
168 if (ioException != null) {
169 final IOException e;
170 e = ioException;
171 ioException = null;
172 throw e;
173 }
174
175 if (bytesAvailable == 0) {
176 // Return EOF if at end of file
177 if (hasReachedEOF) {
178 return EOF;
179 }
180
181 // Otherwise, we have to wait for queue to get something
182 if (threaded) {
183 queue.notify();
184 try {
185 readIsWaiting = true;
186 queue.wait();
187 readIsWaiting = false;
188 } catch (final InterruptedException e) {
189 throw new InterruptedIOException("Fatal thread interruption during read.");
190 }
191 } else {
192 // alreadyread = false;
193 readIsWaiting = true;
194 int ch;
195 boolean mayBlock = true; // block on the first read only
196
197 do {
198 try {
199 if ((ch = read(mayBlock)) < 0) { // must be EOF
200 if (ch != WOULD_BLOCK) {
201 return ch;
202 }
203 }
204 } catch (final InterruptedIOException e) {
205 synchronized (queue) {
206 ioException = e;
207 queue.notifyAll();
208 try {
209 queue.wait(100);
210 } catch (final InterruptedException interrupted) {
211 // Ignored
212 }
213 }
214 return EOF;
215 }
216
217 try {
218 if (ch != WOULD_BLOCK) {
219 processChar(ch);
220 }
221 } catch (final InterruptedException e) {
222 if (isClosed) {
223 return EOF;
224 }
225 }
226
227 // Reads should not block on subsequent iterations. Potentially, this could happen if the
228 // remaining buffered socket data consists entirely of Telnet command sequence and no "user" data.
229 mayBlock = false;
230
231 }
232 // Continue reading as long as there is data available and the queue is not full.
233 while (super.available() > 0 && bytesAvailable < queue.length - 1);
234
235 readIsWaiting = false;
236 }
237 continue;
238 }
239 final int ch;
240
241 ch = queue[queueHead];
242
243 if (++queueHead >= queue.length) {
244 queueHead = 0;
245 }
246
247 --bytesAvailable;
248
249 // Need to explicitly notify() so available() works properly
250 if (bytesAvailable == 0 && threaded) {
251 queue.notify();
252 }
253
254 return ch;
255 }
256 }
257 }
258
259 // synchronized(client) critical sections are to protect against
260 // TelnetOutputStream writing through the telnet client at same time
261 // as a processDo/Will/etc. command invoked from TelnetInputStream
262 // tries to write.
263 /**
264 * Gets the next byte of data. IAC commands are processed internally and do not return data.
265 *
266 * @param mayBlock true if method is allowed to block
267 * @return the next byte of data, or -1 (EOF) if end of stread reached, or -2 (WOULD_BLOCK) if mayBlock is false and there is no data available
268 */
269 private int read(final boolean mayBlock) throws IOException {
270 int ch;
271
272 while (true) {
273
274 // If there is no more data AND we were told not to block,
275 // just return WOULD_BLOCK (-2). (More efficient than exception.)
276 if (!mayBlock && super.available() == 0) {
277 return WOULD_BLOCK;
278 }
279
280 // Otherwise, exit only when we reach end of stream.
281 if ((ch = super.read()) < 0) {
282 return EOF;
283 }
284
285 ch &= 0xff;
286
287 /* Code Section added for supporting AYT (start) */
288 synchronized (client) {
289 client.processAYTResponse();
290 }
291 /* Code Section added for supporting AYT (end) */
292
293 /* Code Section added for supporting spystreams (start) */
294 client.spyRead(ch);
295 /* Code Section added for supporting spystreams (end) */
296
297 switch (receiveState) {
298
299 case STATE_CR:
300 if (ch == '\0') {
301 // Strip null
302 continue;
303 }
304 // How do we handle newline after cr?
305 // else if (ch == '\n' && _requestedDont(TelnetOption.ECHO) &&
306
307 // Handle as normal data by falling through to _STATE_DATA case
308
309 //$FALL-THROUGH$
310 case STATE_DATA:
311 if (ch == TelnetCommand.IAC) {
312 receiveState = STATE_IAC;
313 continue;
314 }
315
316 if (ch == '\r') {
317 synchronized (client) {
318 if (client.requestedDont(TelnetOption.BINARY)) {
319 receiveState = STATE_CR;
320 } else {
321 receiveState = STATE_DATA;
322 }
323 }
324 } else {
325 receiveState = STATE_DATA;
326 }
327 break;
328
329 case STATE_IAC:
330 switch (ch) {
331 case TelnetCommand.WILL:
332 receiveState = STATE_WILL;
333 continue;
334 case TelnetCommand.WONT:
335 receiveState = STATE_WONT;
336 continue;
337 case TelnetCommand.DO:
338 receiveState = STATE_DO;
339 continue;
340 case TelnetCommand.DONT:
341 receiveState = STATE_DONT;
342 continue;
343 /* TERMINAL-TYPE option (start) */
344 case TelnetCommand.SB:
345 suboptionCount = 0;
346 receiveState = STATE_SB;
347 continue;
348 /* TERMINAL-TYPE option (end) */
349 case TelnetCommand.IAC:
350 receiveState = STATE_DATA;
351 break; // exit to enclosing switch to return IAC from read
352 case TelnetCommand.SE: // unexpected byte! ignore it (don't send it as a command)
353 receiveState = STATE_DATA;
354 continue;
355 default:
356 receiveState = STATE_DATA;
357 client.processCommand(ch); // Notify the user
358 continue; // move on the next char
359 }
360 break; // exit and return from read
361 case STATE_WILL:
362 synchronized (client) {
363 client.processWill(ch);
364 client.flushOutputStream();
365 }
366 receiveState = STATE_DATA;
367 continue;
368 case STATE_WONT:
369 synchronized (client) {
370 client.processWont(ch);
371 client.flushOutputStream();
372 }
373 receiveState = STATE_DATA;
374 continue;
375 case STATE_DO:
376 synchronized (client) {
377 client.processDo(ch);
378 client.flushOutputStream();
379 }
380 receiveState = STATE_DATA;
381 continue;
382 case STATE_DONT:
383 synchronized (client) {
384 client.processDont(ch);
385 client.flushOutputStream();
386 }
387 receiveState = STATE_DATA;
388 continue;
389 /* TERMINAL-TYPE option (start) */
390 case STATE_SB:
391 switch (ch) {
392 case TelnetCommand.IAC:
393 receiveState = STATE_IAC_SB;
394 continue;
395 default:
396 // store suboption char
397 if (suboptionCount < suboption.length) {
398 suboption[suboptionCount++] = ch;
399 }
400 break;
401 }
402 receiveState = STATE_SB;
403 continue;
404 case STATE_IAC_SB: // IAC received during SB phase
405 switch (ch) {
406 case TelnetCommand.SE:
407 synchronized (client) {
408 client.processSuboption(suboption, suboptionCount);
409 client.flushOutputStream();
410 }
411 receiveState = STATE_DATA;
412 continue;
413 case TelnetCommand.IAC: // De-dup the duplicated IAC
414 if (suboptionCount < suboption.length) {
415 suboption[suboptionCount++] = ch;
416 }
417 break;
418 default: // unexpected byte! ignore it
419 break;
420 }
421 receiveState = STATE_SB;
422 continue;
423 /* TERMINAL-TYPE option (end) */
424 }
425
426 break;
427 }
428
429 return ch;
430 }
431
432 /**
433 * Reads the next number of bytes from the stream into an array and returns the number of bytes read. Returns -1 if the end of the stream has been reached.
434 *
435 * @param buffer The byte array in which to store the data.
436 * @return The number of bytes read. Returns -1 if the end of the message has been reached.
437 * @throws IOException If an error occurs in reading the underlying stream.
438 */
439 @Override
440 public int read(final byte buffer[]) throws IOException {
441 return read(buffer, 0, buffer.length);
442 }
443
444 /**
445 * Reads the next number of bytes from the stream into an array and returns the number of bytes read. Returns -1 if the end of the message has been reached.
446 * The characters are stored in the array starting from the given offset and up to the length specified.
447 *
448 * @param buffer The byte array in which to store the data.
449 * @param offset The offset into the array at which to start storing data.
450 * @param length The number of bytes to read.
451 * @return The number of bytes read. Returns -1 if the end of the stream has been reached.
452 * @throws IOException If an error occurs while reading the underlying stream.
453 */
454 @Override
455 public int read(final byte buffer[], int offset, int length) throws IOException {
456 int ch;
457 final int off;
458
459 if (length < 1) {
460 return 0;
461 }
462
463 // Critical section because run() may change bytesAvailable
464 synchronized (queue) {
465 if (length > bytesAvailable) {
466 length = bytesAvailable;
467 }
468 }
469
470 if ((ch = read()) == EOF) {
471 return EOF;
472 }
473
474 off = offset;
475
476 do {
477 buffer[offset++] = (byte) ch;
478 } while (--length > 0 && (ch = read()) != EOF);
479
480 // client._spyRead(buffer, off, offset - off);
481 return offset - off;
482 }
483
484 @Override
485 public void run() {
486 int ch;
487
488 try {
489 _outerLoop: while (!isClosed) {
490 try {
491 if ((ch = read(true)) < 0) {
492 break;
493 }
494 } catch (final InterruptedIOException e) {
495 synchronized (queue) {
496 ioException = e;
497 queue.notifyAll();
498 try {
499 queue.wait(100);
500 } catch (final InterruptedException interrupted) {
501 if (isClosed) {
502 break _outerLoop;
503 }
504 }
505 continue;
506 }
507 } catch (final RuntimeException re) {
508 // We treat any runtime exceptions as though the
509 // stream has been closed. We close the
510 // underlying stream just to be sure.
511 super.close();
512 // Breaking the loop has the effect of setting
513 // the state to closed at the end of the method.
514 break _outerLoop;
515 }
516
517 // Process new character
518 boolean notify = false;
519 try {
520 notify = processChar(ch);
521 } catch (final InterruptedException e) {
522 if (isClosed) {
523 break _outerLoop;
524 }
525 }
526
527 // Notify input listener if buffer was previously empty
528 if (notify) {
529 client.notifyInputListener();
530 }
531 }
532 } catch (final IOException ioe) {
533 synchronized (queue) {
534 ioException = ioe;
535 }
536 client.notifyInputListener();
537 }
538
539 synchronized (queue) {
540 isClosed = true; // Possibly redundant
541 hasReachedEOF = true;
542 queue.notify();
543 }
544
545 threaded = false;
546 }
547
548 void start() {
549 if (thread == null) {
550 return;
551 }
552
553 int priority;
554 isClosed = false;
555 // TODO remove this
556 // Need to set a higher priority in case JVM does not use pre-emptive
557 // threads. This should prevent scheduler induced deadlock (rather than
558 // deadlock caused by a bug in this code).
559 priority = Thread.currentThread().getPriority() + 1;
560 if (priority > Thread.MAX_PRIORITY) {
561 priority = Thread.MAX_PRIORITY;
562 }
563 thread.setPriority(priority);
564 thread.setDaemon(true);
565 thread.start();
566 threaded = true; // tell _processChar that we are running threaded
567 }
568 }