TelnetInputStream.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.commons.net.telnet;

import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InterruptedIOException;

final class TelnetInputStream extends BufferedInputStream implements Runnable {
    /** End of file has been reached */
    private static final int EOF = -1;

    /** Read would block */
    private static final int WOULD_BLOCK = -2;

    // TODO should these be private enums?
    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,
            STATE_IAC_SB = 9;

    private boolean hasReachedEOF; // @GuardedBy("queue")
    private volatile boolean isClosed;
    private boolean readIsWaiting;
    private int receiveState, queueHead, queueTail, bytesAvailable;
    private final int[] queue;
    private final TelnetClient client;
    private final Thread thread;
    private IOException ioException;

    /* TERMINAL-TYPE option (start) */
    private final int suboption[];
    private int suboptionCount;
    /* TERMINAL-TYPE option (end) */

    private volatile boolean threaded;

    TelnetInputStream(final InputStream input, final TelnetClient client) {
        this(input, client, true);
    }

    TelnetInputStream(final InputStream input, final TelnetClient client, final boolean readerThread) {
        super(input);
        this.client = client;
        this.receiveState = STATE_DATA;
        this.isClosed = true;
        this.hasReachedEOF = false;
        // Make it 2049, because when full, one slot will go unused, and we
        // want a 2048 byte buffer just to have a round number (base 2 that is)
        this.queue = new int[2049];
        this.queueHead = 0;
        this.queueTail = 0;
        this.suboption = new int[client.maxSubnegotiationLength];
        this.bytesAvailable = 0;
        this.ioException = null;
        this.readIsWaiting = false;
        this.threaded = false;
        if (readerThread) {
            this.thread = new Thread(this);
        } else {
            this.thread = null;
        }
    }

    @Override
    public int available() throws IOException {
        // Critical section because run() may change bytesAvailable
        synchronized (queue) {
            if (threaded) { // Must not call super.available when running threaded: NET-466
                return bytesAvailable;
            }
            return bytesAvailable + super.available();
        }
    }

    // Cannot be synchronized. Will cause deadlock if run() is blocked
    // in read because BufferedInputStream read() is synchronized.
    @Override
    public void close() throws IOException {
        // Completely disregard the fact thread may still be running.
        // We can't afford to block on this close by waiting for
        // thread to terminate because few if any JVM's will actually
        // interrupt a system read() from the interrupt() method.
        super.close();

        synchronized (queue) {
            hasReachedEOF = true;
            isClosed = true;

            if (thread != null && thread.isAlive()) {
                thread.interrupt();
            }

            queue.notifyAll();
        }

    }

    /** Returns false. Mark is not supported. */
    @Override
    public boolean markSupported() {
        return false;
    }

    // synchronized(client) critical sections are to protect against
    // TelnetOutputStream writing through the telnet client at same time
    // as a processDo/Will/etc. command invoked from TelnetInputStream
    // tries to write. Returns true if buffer was previously empty.
    private boolean processChar(final int ch) throws InterruptedException {
        // Critical section because we're altering bytesAvailable,
        // queueTail, and the contents of _queue.
        final boolean bufferWasEmpty;
        synchronized (queue) {
            bufferWasEmpty = bytesAvailable == 0;
            while (bytesAvailable >= queue.length - 1) {
                // The queue is full. We need to wait before adding any more data to it. Hopefully the stream owner
                // will consume some data soon!
                if (!threaded) {
                    // We've been asked to add another character to the queue, but it is already full and there's
                    // no other thread to drain it. This should not have happened!
                    throw new IllegalStateException("Queue is full! Cannot process another character.");
                }
                queue.notify();
                try {
                    queue.wait();
                } catch (final InterruptedException e) {
                    throw e;
                }
            }

            // Need to do this in case we're not full, but block on a read
            if (readIsWaiting && threaded) {
                queue.notify();
            }

            queue[queueTail] = ch;
            ++bytesAvailable;

            if (++queueTail >= queue.length) {
                queueTail = 0;
            }
        }
        return bufferWasEmpty;
    }

    @Override
    public int read() throws IOException {
        // Critical section because we're altering bytesAvailable,
        // queueHead, and the contents of _queue in addition to
        // testing value of hasReachedEOF.
        synchronized (queue) {

            while (true) {
                if (ioException != null) {
                    final IOException e;
                    e = ioException;
                    ioException = null;
                    throw e;
                }

                if (bytesAvailable == 0) {
                    // Return EOF if at end of file
                    if (hasReachedEOF) {
                        return EOF;
                    }

                    // Otherwise, we have to wait for queue to get something
                    if (threaded) {
                        queue.notify();
                        try {
                            readIsWaiting = true;
                            queue.wait();
                            readIsWaiting = false;
                        } catch (final InterruptedException e) {
                            throw new InterruptedIOException("Fatal thread interruption during read.");
                        }
                    } else {
                        // alreadyread = false;
                        readIsWaiting = true;
                        int ch;
                        boolean mayBlock = true; // block on the first read only

                        do {
                            try {
                                if ((ch = read(mayBlock)) < 0) { // must be EOF
                                    if (ch != WOULD_BLOCK) {
                                        return ch;
                                    }
                                }
                            } catch (final InterruptedIOException e) {
                                synchronized (queue) {
                                    ioException = e;
                                    queue.notifyAll();
                                    try {
                                        queue.wait(100);
                                    } catch (final InterruptedException interrupted) {
                                        // Ignored
                                    }
                                }
                                return EOF;
                            }

                            try {
                                if (ch != WOULD_BLOCK) {
                                    processChar(ch);
                                }
                            } catch (final InterruptedException e) {
                                if (isClosed) {
                                    return EOF;
                                }
                            }

                            // Reads should not block on subsequent iterations. Potentially, this could happen if the
                            // remaining buffered socket data consists entirely of Telnet command sequence and no "user" data.
                            mayBlock = false;

                        }
                        // Continue reading as long as there is data available and the queue is not full.
                        while (super.available() > 0 && bytesAvailable < queue.length - 1);

                        readIsWaiting = false;
                    }
                    continue;
                }
                final int ch;

                ch = queue[queueHead];

                if (++queueHead >= queue.length) {
                    queueHead = 0;
                }

                --bytesAvailable;

                // Need to explicitly notify() so available() works properly
                if (bytesAvailable == 0 && threaded) {
                    queue.notify();
                }

                return ch;
            }
        }
    }

    // synchronized(client) critical sections are to protect against
    // TelnetOutputStream writing through the telnet client at same time
    // as a processDo/Will/etc. command invoked from TelnetInputStream
    // tries to write.
    /**
     * Gets the next byte of data. IAC commands are processed internally and do not return data.
     *
     * @param mayBlock true if method is allowed to block
     * @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
     */
    private int read(final boolean mayBlock) throws IOException {
        int ch;

        while (true) {

            // If there is no more data AND we were told not to block,
            // just return WOULD_BLOCK (-2). (More efficient than exception.)
            if (!mayBlock && super.available() == 0) {
                return WOULD_BLOCK;
            }

            // Otherwise, exit only when we reach end of stream.
            if ((ch = super.read()) < 0) {
                return EOF;
            }

            ch &= 0xff;

            /* Code Section added for supporting AYT (start) */
            synchronized (client) {
                client.processAYTResponse();
            }
            /* Code Section added for supporting AYT (end) */

            /* Code Section added for supporting spystreams (start) */
            client.spyRead(ch);
            /* Code Section added for supporting spystreams (end) */

            switch (receiveState) {

            case STATE_CR:
                if (ch == '\0') {
                    // Strip null
                    continue;
                }
                // How do we handle newline after cr?
                // else if (ch == '\n' && _requestedDont(TelnetOption.ECHO) &&

                // Handle as normal data by falling through to _STATE_DATA case

                //$FALL-THROUGH$
            case STATE_DATA:
                if (ch == TelnetCommand.IAC) {
                    receiveState = STATE_IAC;
                    continue;
                }

                if (ch == '\r') {
                    synchronized (client) {
                        if (client.requestedDont(TelnetOption.BINARY)) {
                            receiveState = STATE_CR;
                        } else {
                            receiveState = STATE_DATA;
                        }
                    }
                } else {
                    receiveState = STATE_DATA;
                }
                break;

            case STATE_IAC:
                switch (ch) {
                case TelnetCommand.WILL:
                    receiveState = STATE_WILL;
                    continue;
                case TelnetCommand.WONT:
                    receiveState = STATE_WONT;
                    continue;
                case TelnetCommand.DO:
                    receiveState = STATE_DO;
                    continue;
                case TelnetCommand.DONT:
                    receiveState = STATE_DONT;
                    continue;
                /* TERMINAL-TYPE option (start) */
                case TelnetCommand.SB:
                    suboptionCount = 0;
                    receiveState = STATE_SB;
                    continue;
                /* TERMINAL-TYPE option (end) */
                case TelnetCommand.IAC:
                    receiveState = STATE_DATA;
                    break; // exit to enclosing switch to return IAC from read
                case TelnetCommand.SE: // unexpected byte! ignore it (don't send it as a command)
                    receiveState = STATE_DATA;
                    continue;
                default:
                    receiveState = STATE_DATA;
                    client.processCommand(ch); // Notify the user
                    continue; // move on the next char
                }
                break; // exit and return from read
            case STATE_WILL:
                synchronized (client) {
                    client.processWill(ch);
                    client.flushOutputStream();
                }
                receiveState = STATE_DATA;
                continue;
            case STATE_WONT:
                synchronized (client) {
                    client.processWont(ch);
                    client.flushOutputStream();
                }
                receiveState = STATE_DATA;
                continue;
            case STATE_DO:
                synchronized (client) {
                    client.processDo(ch);
                    client.flushOutputStream();
                }
                receiveState = STATE_DATA;
                continue;
            case STATE_DONT:
                synchronized (client) {
                    client.processDont(ch);
                    client.flushOutputStream();
                }
                receiveState = STATE_DATA;
                continue;
            /* TERMINAL-TYPE option (start) */
            case STATE_SB:
                switch (ch) {
                case TelnetCommand.IAC:
                    receiveState = STATE_IAC_SB;
                    continue;
                default:
                    // store suboption char
                    if (suboptionCount < suboption.length) {
                        suboption[suboptionCount++] = ch;
                    }
                    break;
                }
                receiveState = STATE_SB;
                continue;
            case STATE_IAC_SB: // IAC received during SB phase
                switch (ch) {
                case TelnetCommand.SE:
                    synchronized (client) {
                        client.processSuboption(suboption, suboptionCount);
                        client.flushOutputStream();
                    }
                    receiveState = STATE_DATA;
                    continue;
                case TelnetCommand.IAC: // De-dup the duplicated IAC
                    if (suboptionCount < suboption.length) {
                        suboption[suboptionCount++] = ch;
                    }
                    break;
                default: // unexpected byte! ignore it
                    break;
                }
                receiveState = STATE_SB;
                continue;
            /* TERMINAL-TYPE option (end) */
            }

            break;
        }

        return ch;
    }

    /**
     * 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.
     *
     * @param buffer The byte array in which to store the data.
     * @return The number of bytes read. Returns -1 if the end of the message has been reached.
     * @throws IOException If an error occurs in reading the underlying stream.
     */
    @Override
    public int read(final byte buffer[]) throws IOException {
        return read(buffer, 0, buffer.length);
    }

    /**
     * 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.
     * The characters are stored in the array starting from the given offset and up to the length specified.
     *
     * @param buffer The byte array in which to store the data.
     * @param offset The offset into the array at which to start storing data.
     * @param length The number of bytes to read.
     * @return The number of bytes read. Returns -1 if the end of the stream has been reached.
     * @throws IOException If an error occurs while reading the underlying stream.
     */
    @Override
    public int read(final byte buffer[], int offset, int length) throws IOException {
        int ch;
        final int off;

        if (length < 1) {
            return 0;
        }

        // Critical section because run() may change bytesAvailable
        synchronized (queue) {
            if (length > bytesAvailable) {
                length = bytesAvailable;
            }
        }

        if ((ch = read()) == EOF) {
            return EOF;
        }

        off = offset;

        do {
            buffer[offset++] = (byte) ch;
        } while (--length > 0 && (ch = read()) != EOF);

        // client._spyRead(buffer, off, offset - off);
        return offset - off;
    }

    @Override
    public void run() {
        int ch;

        try {
            _outerLoop: while (!isClosed) {
                try {
                    if ((ch = read(true)) < 0) {
                        break;
                    }
                } catch (final InterruptedIOException e) {
                    synchronized (queue) {
                        ioException = e;
                        queue.notifyAll();
                        try {
                            queue.wait(100);
                        } catch (final InterruptedException interrupted) {
                            if (isClosed) {
                                break _outerLoop;
                            }
                        }
                        continue;
                    }
                } catch (final RuntimeException re) {
                    // We treat any runtime exceptions as though the
                    // stream has been closed. We close the
                    // underlying stream just to be sure.
                    super.close();
                    // Breaking the loop has the effect of setting
                    // the state to closed at the end of the method.
                    break _outerLoop;
                }

                // Process new character
                boolean notify = false;
                try {
                    notify = processChar(ch);
                } catch (final InterruptedException e) {
                    if (isClosed) {
                        break _outerLoop;
                    }
                }

                // Notify input listener if buffer was previously empty
                if (notify) {
                    client.notifyInputListener();
                }
            }
        } catch (final IOException ioe) {
            synchronized (queue) {
                ioException = ioe;
            }
            client.notifyInputListener();
        }

        synchronized (queue) {
            isClosed = true; // Possibly redundant
            hasReachedEOF = true;
            queue.notify();
        }

        threaded = false;
    }

    void start() {
        if (thread == null) {
            return;
        }

        int priority;
        isClosed = false;
        // TODO remove this
        // Need to set a higher priority in case JVM does not use pre-emptive
        // threads. This should prevent scheduler induced deadlock (rather than
        // deadlock caused by a bug in this code).
        priority = Thread.currentThread().getPriority() + 1;
        if (priority > Thread.MAX_PRIORITY) {
            priority = Thread.MAX_PRIORITY;
        }
        thread.setPriority(priority);
        thread.setDaemon(true);
        thread.start();
        threaded = true; // tell _processChar that we are running threaded
    }
}