channelmanager.java

thinking in java4 src

	public void registerThread(IChannelWorkerThread thr) throws IOException
	{
		synchronized (listenerThreads)
		{
			if (listenerThreadsAllowed == false)
				throw new IOException("Too late, this connection is closed.");
			listenerThreads.addElement(thr);
		}
	}

	public Channel openDirectTCPIPChannel(String host_to_connect, int port_to_connect, String originator_IP_address,
			int originator_port) throws IOException
	{
		Channel c = new Channel(this);

		synchronized (c)
		{
			c.localID = addChannel(c);
			// end of synchronized block forces writing out to main memory
		}

		PacketOpenDirectTCPIPChannel dtc = new PacketOpenDirectTCPIPChannel(c.localID, c.localWindow,
				c.localMaxPacketSize, host_to_connect, port_to_connect, originator_IP_address, originator_port);

		tm.sendMessage(dtc.getPayload());

		waitUntilChannelOpen(c);

		return c;
	}

	public Channel openSessionChannel() throws IOException
	{
		Channel c = new Channel(this);

		synchronized (c)
		{
			c.localID = addChannel(c);
			// end of synchronized block forces the writing out to main memory
		}

		if (log.isEnabled())
			log.log(50, "Sending SSH_MSG_CHANNEL_OPEN (Channel " + c.localID + ")");

		PacketOpenSessionChannel smo = new PacketOpenSessionChannel(c.localID, c.localWindow, c.localMaxPacketSize);
		tm.sendMessage(smo.getPayload());

		waitUntilChannelOpen(c);

		return c;
	}

	public void requestPTY(Channel c, String term, int term_width_characters, int term_height_characters,
			int term_width_pixels, int term_height_pixels, byte[] terminal_modes) throws IOException
	{
		PacketSessionPtyRequest spr;

		synchronized (c)
		{
			if (c.state != Channel.STATE_OPEN)
				throw new IOException("Cannot request PTY on this channel (" + c.getReasonClosed() + ")");

			spr = new PacketSessionPtyRequest(c.remoteID, true, term, term_width_characters, term_height_characters,
					term_width_pixels, term_height_pixels, terminal_modes);

			c.successCounter = c.failedCounter = 0;
		}

		synchronized (c.channelSendLock)
		{
			if (c.closeMessageSent)
				throw new IOException("Cannot request PTY on this channel (" + c.getReasonClosed() + ")");
			tm.sendMessage(spr.getPayload());
		}

		try
		{
			waitForChannelSuccessOrFailure(c);
		}
		catch (IOException e)
		{
			throw (IOException) new IOException("PTY request failed").initCause(e);
		}
	}

	public void requestX11(Channel c, boolean singleConnection, String x11AuthenticationProtocol,
			String x11AuthenticationCookie, int x11ScreenNumber) throws IOException
	{
		PacketSessionX11Request psr;

		synchronized (c)
		{
			if (c.state != Channel.STATE_OPEN)
				throw new IOException("Cannot request X11 on this channel (" + c.getReasonClosed() + ")");

			psr = new PacketSessionX11Request(c.remoteID, true, singleConnection, x11AuthenticationProtocol,
					x11AuthenticationCookie, x11ScreenNumber);

			c.successCounter = c.failedCounter = 0;
		}

		synchronized (c.channelSendLock)
		{
			if (c.closeMessageSent)
				throw new IOException("Cannot request X11 on this channel (" + c.getReasonClosed() + ")");
			tm.sendMessage(psr.getPayload());
		}

		if (log.isEnabled())
			log.log(50, "Requesting X11 forwarding (Channel " + c.localID + "/" + c.remoteID + ")");

		try
		{
			waitForChannelSuccessOrFailure(c);
		}
		catch (IOException e)
		{
			throw (IOException) new IOException("The X11 request failed.").initCause(e);
		}
	}

	public void requestSubSystem(Channel c, String subSystemName) throws IOException
	{
		PacketSessionSubsystemRequest ssr;

		synchronized (c)
		{
			if (c.state != Channel.STATE_OPEN)
				throw new IOException("Cannot request subsystem on this channel (" + c.getReasonClosed() + ")");

			ssr = new PacketSessionSubsystemRequest(c.remoteID, true, subSystemName);

			c.successCounter = c.failedCounter = 0;
		}

		synchronized (c.channelSendLock)
		{
			if (c.closeMessageSent)
				throw new IOException("Cannot request subsystem on this channel (" + c.getReasonClosed() + ")");
			tm.sendMessage(ssr.getPayload());
		}

		try
		{
			waitForChannelSuccessOrFailure(c);
		}
		catch (IOException e)
		{
			throw (IOException) new IOException("The subsystem request failed.").initCause(e);
		}
	}

	public void requestExecCommand(Channel c, String cmd) throws IOException
	{
		PacketSessionExecCommand sm;

		synchronized (c)
		{
			if (c.state != Channel.STATE_OPEN)
				throw new IOException("Cannot execute command on this channel (" + c.getReasonClosed() + ")");

			sm = new PacketSessionExecCommand(c.remoteID, true, cmd);

			c.successCounter = c.failedCounter = 0;
		}

		synchronized (c.channelSendLock)
		{
			if (c.closeMessageSent)
				throw new IOException("Cannot execute command on this channel (" + c.getReasonClosed() + ")");
			tm.sendMessage(sm.getPayload());
		}

		if (log.isEnabled())
			log.log(50, "Executing command (channel " + c.localID + ", '" + cmd + "')");

		try
		{
			waitForChannelSuccessOrFailure(c);
		}
		catch (IOException e)
		{
			throw (IOException) new IOException("The execute request failed.").initCause(e);
		}
	}

	public void requestShell(Channel c) throws IOException
	{
		PacketSessionStartShell sm;

		synchronized (c)
		{
			if (c.state != Channel.STATE_OPEN)
				throw new IOException("Cannot start shell on this channel (" + c.getReasonClosed() + ")");

			sm = new PacketSessionStartShell(c.remoteID, true);

			c.successCounter = c.failedCounter = 0;
		}

		synchronized (c.channelSendLock)
		{
			if (c.closeMessageSent)
				throw new IOException("Cannot start shell on this channel (" + c.getReasonClosed() + ")");
			tm.sendMessage(sm.getPayload());
		}

		try
		{
			waitForChannelSuccessOrFailure(c);
		}
		catch (IOException e)
		{
			throw (IOException) new IOException("The shell request failed.").initCause(e);
		}
	}

	public void msgChannelExtendedData(byte[] msg, int msglen) throws IOException
	{
		if (msglen <= 13)
			throw new IOException("SSH_MSG_CHANNEL_EXTENDED_DATA message has wrong size (" + msglen + ")");

		int id = ((msg[1] & 0xff) << 24) | ((msg[2] & 0xff) << 16) | ((msg[3] & 0xff) << 8) | (msg[4] & 0xff);
		int dataType = ((msg[5] & 0xff) << 24) | ((msg[6] & 0xff) << 16) | ((msg[7] & 0xff) << 8) | (msg[8] & 0xff);
		int len = ((msg[9] & 0xff) << 24) | ((msg[10] & 0xff) << 16) | ((msg[11] & 0xff) << 8) | (msg[12] & 0xff);

		Channel c = getChannel(id);

		if (c == null)
			throw new IOException("Unexpected SSH_MSG_CHANNEL_EXTENDED_DATA message for non-existent channel " + id);

		if (dataType != Packets.SSH_EXTENDED_DATA_STDERR)
			throw new IOException("SSH_MSG_CHANNEL_EXTENDED_DATA message has unknown type (" + dataType + ")");

		if (len != (msglen - 13))
			throw new IOException("SSH_MSG_CHANNEL_EXTENDED_DATA message has wrong len (calculated " + (msglen - 13)
					+ ", got " + len + ")");

		if (log.isEnabled())
			log.log(80, "Got SSH_MSG_CHANNEL_EXTENDED_DATA (channel " + id + ", " + len + ")");

		synchronized (c)
		{
			if (c.state == Channel.STATE_CLOSED)
				return; // ignore

			if (c.state != Channel.STATE_OPEN)
				throw new IOException("Got SSH_MSG_CHANNEL_EXTENDED_DATA, but channel is not in correct state ("
						+ c.state + ")");

			if (c.localWindow < len)
				throw new IOException("Remote sent too much data, does not fit into window.");

			c.localWindow -= len;

			System.arraycopy(msg, 13, c.stderrBuffer, c.stderrWritepos, len);
			c.stderrWritepos += len;

			c.notifyAll();
		}
	}

	/**
	 * Wait until for a condition.
	 * 
	 * @param c
	 *            Channel
	 * @param timeout
	 *            in ms, 0 means no timeout.
	 * @param condition_mask
	 *            minimum event mask
	 * @return all current events
	 * 
	 */
	public int waitForCondition(Channel c, long timeout, int condition_mask)
	{
		long end_time = 0;
		boolean end_time_set = false;

		synchronized (c)
		{
			while (true)
			{
				int current_cond = 0;

				int stdoutAvail = c.stdoutWritepos - c.stdoutReadpos;
				int stderrAvail = c.stderrWritepos - c.stderrReadpos;

				if (stdoutAvail > 0)
					current_cond = current_cond | ChannelCondition.STDOUT_DATA;

				if (stderrAvail > 0)
					current_cond = current_cond | ChannelCondition.STDERR_DATA;

				if (c.EOF)
					current_cond = current_cond | ChannelCondition.EOF;

				if (c.getExitStatus() != null)
					current_cond = current_cond | ChannelCondition.EXIT_STATUS;

				if (c.getExitSignal() != null)
					current_cond = current_cond | ChannelCondition.EXIT_SIGNAL;

				if (c.state == Channel.STATE_CLOSED)
					return current_cond | ChannelCondition.CLOSED | ChannelCondition.EOF;

				if ((current_cond & condition_mask) != 0)
					return current_cond;

				if (timeout > 0)
				{
					if (!end_time_set)
					{
						end_time = System.currentTimeMillis() + timeout;
						end_time_set = true;
					}
					else
					{
						timeout = end_time - System.currentTimeMillis();

						if (timeout <= 0)
							return current_cond | ChannelCondition.TIMEOUT;
					}
				}

				try
				{
					if (timeout > 0)
						c.wait(timeout);
					else
						c.wait();
				}
				catch (InterruptedException e)
				{
				}
			}
		}
	}

	public int getAvailable(Channel c, boolean extended) throws IOException
	{
		synchronized (c)
		{
			int avail;

			if (extended)
				avail = c.stderrWritepos - c.stderrReadpos;
			else
				avail = c.stdoutWritepos - c.stdoutReadpos;

			return ((avail > 0) ? avail : (c.EOF ? -1 : 0));
		}
	}

	public int getChannelData(Channel c, boolean extended, byte[] target, int off, int len) throws IOException
	{
		int copylen = 0;
		int increment = 0;
		int remoteID = 0;
		int localID = 0;

		synchronized (c)
		{
			int stdoutAvail = 0;
			int stderrAvail = 0;

			while (true)
			{
				/*
				 * Data available? We have to return remaining data even if the
				 * channel is already closed.
				 */

				stdoutAvail = c.stdoutWritepos - c.stdoutReadpos;
				stderrAvail = c.stderrWritepos - c.stderrReadpos;

				if ((!extended) && (stdoutAvail != 0))
					break;

				if ((extended) && (stderrAvail != 0))
					break;

				/* Do not wait if more data will never arrive (EOF or CLOSED) */

				if ((c.EOF) || (c.state != Channel.STATE_OPEN))
					return -1;

				try
				{
					c.wait();
				}
				catch (InterruptedException ignore)
				{
				}
			}

			/* OK, there is some data. Return it. */

			if (!extended)
			{
				copylen = (stdoutAvail > len) ? len : stdoutAvail;
				System.arraycopy(c.stdoutBuffer, c.stdoutReadpos, target, off, copylen);
				c.stdoutReadpos += copylen;

				if (c.stdoutReadpos != c.stdoutWritepos)

					System.arraycopy(c.stdoutBuffer, c.stdoutReadpos, c.stdoutBuffer, 0, c.stdoutWritepos
							- c.stdoutReadpos);

				c.stdoutWritepos -= c.stdoutReadpos;
				c.stdoutReadpos = 0;
			}
			else
			{
				copylen = (stderrAvail > len) ? len : stderrAvail;
				System.arraycopy(c.stderrBuffer, c.stderrReadpos, target, off, copylen);
				c.stderrReadpos += copylen;

				if (c.stderrReadpos != c.stderrWritepos)

					System.arraycopy(c.stderrBuffer, c.stderrReadpos, c.stderrBuffer, 0, c.stderrWritepos
							- c.stderrReadpos);

				c.stderrWritepos -= c.stderrReadpos;
				c.stderrReadpos = 0;
			}

			if (c.state != Channel.STATE_OPEN)
				return copylen;

			if (c.localWindow < ((Channel.CHANNEL_BUFFER_SIZE + 1) / 2))
			{
				int minFreeSpace = Math.min(Channel.CHANNEL_BUFFER_SIZE - c.stdoutWritepos, Channel.CHANNEL_BUFFER_SIZE
						- c.stderrWritepos);

				increment = minFreeSpace - c.localWindow;
				c.localWindow = minFreeSpace;
			}

			remoteID = c.remoteID; /* read while holding the lock */
			localID = c.localID; /* read while holding the lock */
		}

		/*
		 * If a consumer reads stdout and stdin in parallel, we may end up with
		 * sending two msgWindowAdjust messages. Luckily, it
		 * does not matter in which order they arrive at the server.
		 */

		if (increment > 0)
		{
			if (log.isEnabled())
				log.log(80, "Sending SSH_MSG_CHANNEL_WINDOW_ADJUST (channel " + localID + ", " + increment + ")");

			synchronized (c.channelSendLock)
			{
				byte[] msg = c.msgWindowAdjust;

				msg[0] = Packets.SSH_MSG_CHANNEL_WINDOW_ADJUST;
				msg[1] = (byte) (remoteID >> 24);
				msg[2] = (byte) (remoteID >> 16);
				msg[3] = (byte) (remoteID >> 8);
				msg[4] = (byte) (remoteID);
				msg[5] = (byte) (increment >> 24);
				msg[6] = (byte) (increment >> 16);
				msg[7] = (byte) (increment >> 8);
				msg[8] = (byte) (increment);

				if (c.closeMessageSent == false)
					tm.sendMessage(msg);
			}
		}

		return copylen;
	}

	public void msgChannelData(byte[] msg, int msglen) throws IOException
	{
		if (msglen <= 9)
			throw new IOException("SSH_MSG_CHANNEL_DATA message has wrong size (" + msglen + ")");

		int id = ((msg[1] & 0xff) << 24) | ((msg[2] & 0xff) << 16) | ((msg[3] & 0xff) << 8) | (msg[4] & 0xff);
		int len = ((msg[5] & 0xff) << 24) | ((msg[6] & 0xff) << 16) | ((msg[7] & 0xff) << 8) | (msg[8] & 0xff);

		Channel c = getChannel(id);

		if (c == null)
			throw new IOException("Unexpected SSH_MSG_CHANNEL_DATA message for non-existent channel " + id);

		if (len != (msglen - 9))
			throw new IOException("SSH_MSG_CHANNEL_DATA message has wrong len (calculated " + (msglen - 9) + ", got "
					+ len + ")");

		if (log.isEnabled())
			log.log(80, "Got SSH_MSG_CHANNEL_DATA (channel " + id + ", " + len + ")");

		synchronized (c)
		{
			if (c.state == Channel.STATE_CLOSED)
				return; // ignore

			if (c.state != Channel.STATE_OPEN)
				throw new IOException("Got SSH_MSG_CHANNEL_DATA, but channel is not in correct state (" + c.state + ")");

			if (c.localWindow < len)
				throw new IOException("Remote sent too much data, does not fit into window.");

			c.localWindow -= len;

			System.arraycopy(msg, 9, c.stdoutBuffer, c.stdoutWritepos, len);
			c.stdoutWritepos += len;

			c.notifyAll();
		}
	}

	public void msgChannelWindowAdjust(byte[] msg, int msglen) throws IOException
	{
		if (msglen != 9)
			throw new IOException("SSH_MSG_CHANNEL_WINDOW_ADJUST message has wrong size (" + msglen + ")");