tcptransport.java

jxme的一些相关程序,主要是手机上程序开发以及手机和计算机通信的一些程序资料,程序编译需要Ant支持

			// The multicast socket needs first to be allocated.
			try {
				multicastSocket = new MulticastSocket(propagatePort);
				multicastSocket.setInterface( usingInterface );
				multicastSocket.joinGroup(propagateInetAddress);
			} catch (Exception e) {
				if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug(e);
			}
		}
		try {
			multicastSocket.send(packet);
			//XXX: Shouldn't we close the socket?!
		} catch (Exception e) {
			if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug(e);
		}
	}


	/**
	 *  Return the protocol name we support with this instance. For the TCP
	 *  transport this is the only protocol supported so we return the same
	 *  result as getSupportedProtocolName()
	 *
	 *  @return String containing the name of the protocol we are supporting.
	 **/
	public String getProtocolName() {
		return protocolName;
	}

	/**
	 * Return the endpoint address by which this peer is accessible via this
	 * transport.
	 */
	public EndpointAddress getPublicAddress() {
		return publicAddress;
	}

	public boolean isConnectionOriented() {
		return true;
	}

	/**
	 * Returns true if the endpoint protocol can be used by the EndpointRouter
	 *
	 * @return boolean true if the protocol can be used by the EndpointRouter
	 */
	public boolean allowRouting() {
		return true;
	}


	/**
	 * Ping a remote host.
	 *
	 * This implementation tries to open a connection, and after tests the result. Note
	 * if there is already an open connection, no new connection is actually created.
	 **/
	public  boolean ping(EndpointAddress addr) {

		try {
			if (LOG.isEnabledFor(Priority.DEBUG))
				LOG.debug("ping to " + addr.toString());
		} catch (Exception e) {
			if (LOG.isEnabledFor(Priority.DEBUG))
				LOG.debug("ping cannot print address because: ", e);
		}

		// XXX: the following is a work around in order to
		//      strip out peers that advertise a local IP address
		//      as their EndpointService (127.0.0.1)
		// [e-mail]

		try {
			String tmpAddr = addr.getProtocolAddress();
			if (tmpAddr.indexOf("127.0.0.1") >= 0) {
				if (LOG.isEnabledFor(Priority.DEBUG))
					LOG.debug("   pinging localhost - discard");
				return false;
			}
		} catch (Exception e) {
			if (LOG.isEnabledFor(Priority.DEBUG))
				LOG.debug("  failed ", e);
			return false;
		}

		TcpConnection conn = connManager.getTcpConnection(addr);
		if (conn == null) {
			if (LOG.isEnabledFor(Priority.DEBUG))
				LOG.debug("  no connection return false ");
			return false;
		}

		if (LOG.isEnabledFor(Priority.DEBUG))
			LOG.debug("  return " + conn.isConnected());
		return conn.isConnected();
	}

	// If there is already a TcpConnection for the given host, notify it
	// that the connection might be up again (if it was disconnected)
	private void checkConnection(EndpointAddress addr) {

		TcpConnection conn = connManager.getTcpConnection(addr);
		if (conn == null) {
			// No pending connection. Nothing to do.
			return;
		}

		conn.notifyActivity();
	}

	public class IncomingUnicastThreads implements Runnable {

		protected Vector sockets = null;
		protected TcpTransport tp = null;
		private volatile int waitingThreads = 0;
		private   int nbOfThreads = 0;
		private   int nbOfKeepAlive = 0;

		public IncomingUnicastThreads(TcpTransport tp, int initialNumber) {

			this.tp = tp;
			sockets = new Vector();

			for (int i = 0; i < initialNumber; ++i) {
				++nbOfThreads;
				Thread t = new Thread(this, "TCP Incoming Unicast:" + nbOfThreads);
				t.start();
			}
		}

		public void run() {

			try {
				Socket socket = null;
				while (true) {
					socket = waitForSocket();
					if (socket == null) {
						// That means that this thread must die
						decThreads();
						return;
					}
					boolean keep = doesKeepAlive();
					tp.runReceive(socket, keep);
					if (keep) {
						stopKeepAlive();
					}
				}
			} catch ( Throwable all ) {
				if (LOG.isEnabledFor(Priority.FATAL)) LOG.fatal( "Uncaught Throwable in thread :" + Thread.currentThread().getName(), all );
			}
		}

		synchronized private void decThreads() {
			--nbOfThreads;
		}

		synchronized protected Socket waitForSocket() {

			Socket socket = null;

			while (true) {

				if (sockets.size() > 0) {
					try {
						socket = (Socket) sockets.elementAt(0);
					} catch (Exception e) {
						// Strange error. XXX: to handle better
						if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("waitForSocket error: " + e);
					}
					if (socket != null) {
						try {
							sockets.removeElementAt(0);
						} catch (Exception e) {
							if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("waitForSocket error: " + e);
						}
						return socket;
					}
					// We should not get here.
					if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("waitForSocket cannot get incoming socket");
				}

				// No more socket to consume... Just wait.
				// Check if this thread must die or wait for a new socket
				if (mustDie()) {
					if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("waitForSocket: killing thread");
					return null;
				}
				try {
					++waitingThreads;
					wait();
					--waitingThreads;
				} catch (Exception e) {
				}
			}
		}

		synchronized private boolean mustDie() {

			if ((nbOfThreads > MaxNbOfUnicastThreads) ||
			        (waitingThreads > 1)) {
				return true;
			}
			return false;
		}

		synchronized private boolean doesKeepAlive() {

			if (nbOfKeepAlive >= MaxNbOfUnicastKeepAliveThreads) {
				return false;
			}
			++nbOfKeepAlive;
			return true;
		}

		synchronized private void stopKeepAlive() {
			--nbOfKeepAlive;
		}


		synchronized protected void newSocket(Socket socket) {

			if (sockets.size() > MaxNbOfPendingSockets) {
				// Too many request. Just drop it.
				try {
					socket.close();
				} catch (Exception e) {
					if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("newSocket socket.close failed with " + e);
				}
				if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("newSocket: too many incoming request, dropped incoming message");
				return;
			}

			sockets.addElement(socket);
			// Check if we need to allocate a new Thread
			if (waitingThreads == 0) {
				++nbOfThreads;
				// Allocate a new thread.
				if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("newSocket: create a new Thread");
				Thread t = new Thread(this, "TCP Incoming Unicast:" + nbOfThreads );
				t.start();
			}
			try {
				notify();
			} catch (Exception e) {
			}
		}
	}



	void processReceivingSocket(Socket socket) {

		unicastThreads.newSocket(socket);
	}


	public class IncomingMulticastThreads implements Runnable {

		protected Vector buffers = null;
		protected TcpTransport tp = null;
		private volatile int waitingThreads = 0;
		private   int nbOfThreads = 0;

		public IncomingMulticastThreads(TcpTransport tp, int initialNumber) {

			this.tp = tp;
			buffers = new Vector();

			for (int i = 0; i < initialNumber; ++i) {
				++nbOfThreads;
				Thread t = new Thread(this, "TCP Incoming Multicast:" + nbOfThreads);
				t.start();
			}
		}

		public void run() {

			try {
				byte[] buffer = null;
				while (true) {
					buffer = waitForBuffer();
					if (buffer == null) {
						// That means that this thread must die
						decThreads();
						return;
					}
					tp.processMulticast(buffer);
				}
			} catch ( Throwable all ) {
				if (LOG.isEnabledFor(Priority.FATAL)) LOG.fatal( "Uncaught Throwable in thread :" + Thread.currentThread().getName(), all );
			}
		}

		synchronized private void decThreads() {
			--nbOfThreads;
		}

		synchronized protected byte[] waitForBuffer() {

			byte[] buffer = null;

			while (true) {

				if (buffers.size() > 0) {
					try {
						buffer = (byte[]) buffers.elementAt(0);
					} catch (Exception e) {
						// Strange error. XXX: to handle better
						if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("waitForBuffer error: " + e);
					}
					if (buffer != null) {
						try {
							buffers.removeElementAt(0);
						} catch (Exception e) {
							if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("waitForBuffer error: " + e);
						}
						return buffer;
					}
					// We should not get here.
					if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("waitForBuffer cannot get incoming buffer");
				}

				// No more socket to consume... Just wait.
				// Check if this thread must die or wait for a new socket
				if (mustDie()) {
					if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("waitForBuffer: killing thread " + Thread.currentThread().getName() );
					return null;
				}
				try {
					++waitingThreads;
					wait();
					--waitingThreads;
				} catch (Exception e) {
				}
			}
		}

		synchronized private boolean mustDie() {

			if ((nbOfThreads > MaxNbOfMulticastThreads) ||
			        (waitingThreads > 0)) {
				return true;
			}
			return false;
		}

		synchronized protected void newBuffer(byte[] buffer) {

			buffers.addElement(buffer);
			// Check if we need to allocate a new Thread
			if (waitingThreads == 0) {
				++nbOfThreads;
				// Allocate a new thread.
				if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("newBuffer: create a new Multicast receive Thread");
				Thread t = new Thread(this, "TCP Incoming Multicast:" + nbOfThreads);
				t.start();
			}
			try {
				notify();
			} catch (Exception e) {
			}
		}
	}

	void processReceivingBuffer(byte[] buffer) {

		multicastThreads.newBuffer(buffer);
	}

}