ssh2streamchannel.java

来自「一个非常好的ssh客户端实现」· Java 代码 · 共 324 行

/******************************************************************************
 *
 * Copyright (c) 1999-2003 AppGate Network Security AB. All Rights Reserved.
 * 
 * This file contains Original Code and/or Modifications of Original Code as
 * defined in and that are subject to the MindTerm Public Source License,
 * Version 2.0, (the 'License'). You may not use this file except in compliance
 * with the License.
 * 
 * You should have received a copy of the MindTerm Public Source License
 * along with this software; see the file LICENSE.  If not, write to
 * AppGate Network Security AB, Otterhallegatan 2, SE-41118 Goteborg, SWEDEN
 *
 *****************************************************************************/

package com.mindbright.ssh2;

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

import com.mindbright.util.Queue;

/**
 * Class implementing streams-based channels. That is channels which
 * locally are connected to a pair of Input/Output streams. It is also
 * possible to apply filters to the channels.
 */
public class SSH2StreamChannel extends SSH2Channel {
    protected InputStream  in;
    protected OutputStream out;

    protected Thread transmitter;
    protected Thread receiver;
    protected Queue  rxQueue;
    protected long   txCounter;
    protected long   rxCounter;

    /*
     * NOTE, if enabled can cause dead-lock when doing re-keyexchange
     * if we initiate it, hence SSH2Transport.incompatibleCantReKey is
     * set accordingly
     */
    private boolean rxChanIsQueued;

    /**
     * Create a new stream channel of the given type. The channel is
     * associated with an ssh connection. Channel types are
     * defined in <code>SSH2Connection</code> and starts with
     * <code>CH_TYPE</code>.
     *
     * @param channelType Type of channel to create.
     * @param connection The ssh connection to associate the channel with.
     * @param creator The object the channel is created from.
     * @param in The input stream from which data to be sent over the
     * channel is read.
     * @param out The output stream onto which data received from the
     * channel is written.
     */
    protected SSH2StreamChannel(int channelType, SSH2Connection connection,
				Object creator,
				InputStream in, OutputStream out) {
	super(channelType, connection, creator);

	rxChanIsQueued =
	    "true".equals(connection.getPreferences().
			  getPreference(SSH2Preferences.QUEUED_RX_CHAN));

	this.in  = in;
	this.out = out;
	createStreams();
    }

    /**
     * Apply the given filter to this channel.
     *
     * @param filter Filter to apply.
     */
    public void applyFilter(SSH2StreamFilter filter) {
	if(filter != null) {
	    in  = filter.getInputFilter(in);
	    out = filter.getOutputFilter(out);
	}
    }

    private void channelTransmitLoop() {
	connection.getLog().debug("SSH2StreamChannel",
				  "starting ch. #" + channelId +
				  " (" + getType() + ") transmitter");
	Thread.yield();
	try {
	    SSH2TransportPDU pdu;
	    int              maxSz = 0;
	    int              rcvSz = 0;
            boolean          interrupted = false;
	    while(!eofSent && !closeSent) {
		pdu =
		    SSH2TransportPDU.createOutgoingPacket(SSH2.MSG_CHANNEL_DATA,
							  txMaxPktSz + 256);
		pdu.writeInt(peerChanId);
		maxSz = checkTxWindowSize(rcvSz);
                do {                        
                    try {
                        rcvSz = in.read(pdu.data, pdu.wPos + 4, maxSz);
                        interrupted = false;
                    } catch (InterruptedIOException e) {
                        interrupted = true;
                    }
                } while (interrupted);
		if(rcvSz == -1) {
		    sendEOF();
		} else if (!eofSent && !closeSent) {
		    pdu.writeInt(rcvSz);
		    pdu.wPos  += rcvSz;
		    txCounter += rcvSz;
		    transmit(pdu);
		}
	    }
	} catch (IOException e) {
	    if(!eofSent) {
		connection.getLog().error("SSH2StreamChannel",
					  "channelTransmitLoop",
					  e.toString());
	    }
	} finally {
	    try { in.close(); } catch (IOException e) { /* don't care */ };
	    sendClose();
	}
	connection.getLog().debug("SSH2StreamChannel",
				  "exiting ch. #" +
				  channelId + " (" + getType() +
				  ") transmitter, " + txCounter +
				  " bytes tx");
    }

    private void channelReceiveLoop() {
	connection.getLog().debug("SSH2StreamChannel",
				  "starting ch. #" + channelId +
				  " (" + getType() + ") receiver");
	Thread.yield();
	try {
	    SSH2TransportPDU pdu;
	    while((pdu = (SSH2TransportPDU)rxQueue.getFirst()) != null) {
		rxWrite(pdu);
	    }
	} catch (IOException e) {
	    connection.getLog().error("SSH2StreamChannel",
				      "channelReceiveLoop",
				      e.toString());
	} finally {
	    rxClosing();
	}
	connection.getLog().debug("SSH2StreamChannel",
				  "exiting ch. #" +
				  channelId + " (" + getType() +
				  ") receiver, " + rxCounter +
				  " bytes rx");
    }

    private final void rxWrite(SSH2TransportPDU pdu) throws IOException {
	int len = pdu.readInt();
	int off = pdu.getRPos();
	rxCounter += len;
	out.write(pdu.data, off, len);
	pdu.release();
	checkRxWindowSize(len);
    }

    private final void rxClosing() {
	// Signal to transmitter that this is an orderly shutdown
	//
	eofSent = true;
	try { out.close(); } catch (IOException e) { /* don't care */ }
	try { in.close(); } catch (IOException e) { /* don't care */ }
	outputClosed();

	// there is a slight chance that the transmitter is waiting for
	// window adjust in which case we must interrupt it here so it
	// doesn't hang
	//
	if(txCurrWinSz == 0) {
	    txCurrWinSz = -1;
	    transmitter.interrupt();
	}
    }

    private final synchronized int checkTxWindowSize(int lastSz) {
	txCurrWinSz -= lastSz;
	while(txCurrWinSz == 0) {
	    // Our window is full, wait for ACK from peer
	    try {
		this.wait();
	    } catch (InterruptedException e) {
		if(!eofSent) {
		    connection.getLog().error("SSH2StreamChannel",
					      "checkTxWindowSize",
					      "window adjust wait interrupted");
		}
	    }
	}
	// Try sending remaining window size or max packet size before ACK
	//
	int dataSz = (txCurrWinSz < txMaxPktSz ? txCurrWinSz : txMaxPktSz);
	return dataSz;
    }

    protected final void checkRxWindowSize(int len) {
	rxCurrWinSz -= len;
	if(rxCurrWinSz < 0) {
	    connection.fatalDisconnect(SSH2.DISCONNECT_PROTOCOL_ERROR,
				       "Peer overflowed window");
	} else if(rxCurrWinSz <= (rxInitWinSz >>> 1)) {
	    // ACK on >= 50% of window received
	    SSH2TransportPDU pdu =
		SSH2TransportPDU.createOutgoingPacket(SSH2.MSG_CHANNEL_WINDOW_ADJUST);
	    pdu.writeInt(peerChanId);
	    pdu.writeInt(rxInitWinSz - rxCurrWinSz);
	    transmit(pdu);
	    rxCurrWinSz = rxInitWinSz; 
	}
    }

    protected void data(SSH2TransportPDU pdu) {
	if(rxChanIsQueued) {
	    rxQueue.putLast(pdu);
	} else {
	    try {
		rxWrite(pdu);
	    } catch (IOException e) {
		connection.getLog().error("SSH2StreamChannel",
					  "data",
					  e.toString());
		rxClosing();
	    }
	}
    }

    protected void openConfirmationImpl(SSH2TransportPDU pdu) {
	startStreams();
    }

    protected boolean openFailureImpl(int reasonCode, String reasonText,
				   String langTag) {
	// Just return false since we don't want to keep the channel,
	// handle in derived class if needed
	return false;
    }

    protected synchronized void windowAdjustImpl(int inc) {
	txCurrWinSz += inc;
	this.notify();
    }

    protected void eofImpl() {
	if(rxChanIsQueued) {
	    rxQueue.setBlocking(false);
	} else {
	    rxClosing();
	}
    }

    protected void closeImpl() {
	eofImpl();
    }

    /**
     * Called when no more data can be written to the channel.
     */
    protected void outputClosed() {
	// Do nothing, handle in derived class if needed
    }

    protected void handleRequestImpl(String type, boolean wantReply,
				     SSH2TransportPDU pdu) {
	// Do nothing, handle in derived class if needed
    }

    /**
     * Create the transmitter and receiver threads.
     */
    protected void createStreams() {
	if(rxChanIsQueued) {
	    receiver = new Thread(new Runnable() {
		    public void run() {
			channelReceiveLoop();
		    }
		}, "SSH2StreamRX_" + getType() + "_" + channelId);
	    receiver.setDaemon(false);
	    rxQueue =
		new Queue(connection.getPreferences().
			  getIntPreference(SSH2Preferences.QUEUE_DEPTH),
			  connection.getPreferences().
			  getIntPreference(SSH2Preferences.QUEUE_HIWATER));
	}
	transmitter = new Thread(new Runnable() {
		public void run() {
		    channelTransmitLoop();
		}
	    }, "SSH2StreamTX_" + getType() + "_" + channelId);
	transmitter.setDaemon(false);
    }

    /**
     * Starts the transmitter and receiver threads.
     */
    protected void startStreams() {
	transmitter.start();
	if(rxChanIsQueued) {
	    receiver.start();
	}
    }

    public void waitUntilClosed() {
	super.waitUntilClosed();
	if(rxChanIsQueued) {
	    try { receiver.join(); } catch (InterruptedException e) { }
	}
    }

}