socketio.java

一个实用工具类

				"write(byte[], int, int)",
				e.getMessage());

			close();
			throw e;
		}
	}
	/**
	 * flush socket output stream.  Not used anymore!
	 */
	public void flush() throws IOException {
	}
	/**
	 * get a summary information of this object.
	 * 
	 * @return String, the object summary information.
	 */
	public String toString() {
		if (f_socketDesc != null) {
			return f_socketDesc;
		}

		StringBuffer buffer = new StringBuffer("Socket: ");
		buffer.append("chunk read timeout=[");
		buffer.append(f_chunkReadTimeout);
		buffer.append("], buffer fill timeout=[");
		buffer.append(f_bufferReadWriteTimeout);
		buffer.append("]");

		Socket sock = getSocket();

		if (sock == null) {
			buffer.append(", NULL socket");

		} else {
			buffer.append(", Local Adr=[");
			buffer.append(sock.getLocalAddress());
			buffer.append(':');
			buffer.append(sock.getLocalPort());
			buffer.append("], Remote Adr=[");
			buffer.append(sock.getInetAddress());
			buffer.append(':');
			buffer.append(sock.getPort());
			buffer.append("]");
		}

		f_socketDesc = buffer.toString();

		return f_socketDesc;
	}
	/**
	 * Set this object socket.
	 * 
	 * @param Socket socket.
	 */
	public synchronized void setSocket(SocketChannel socket)
		throws IOException {

		f_socket = null;

		// valid that Buffer Fill timeout is at least equal
		// to one read tieout!
		if (f_chunkReadTimeout > f_bufferReadWriteTimeout) {
			String msg =
				"Buffer read timeout out must be greater or equal to chunk read timeout!";

			Log.logStr(this, Log.LOG_TYPE_ERROR, "setSocket()", msg);

			throw new IOException(msg);
		}

		f_socket = socket;
		Socket sock = getSocket();
		if (sock != null) {
			// This is mandatory for selectors to work...
			socket.configureBlocking(false);

			// the SO_LINGER is a bogus part of TCP/IP
			// it is NEVER EVER a good idea!
			sock.setSoLinger(false, 0);

			sock.setKeepAlive(f_keepAliveFlag);
			sock.setTcpNoDelay(f_tcpNoDelay);
		}
	}
	/**
	 * Set socket keep alive
	 * 
	 * @param flag boolean, flag
	 */
	public void setKeepAlive(boolean flag) {
		f_keepAliveFlag = flag;

		Socket socket = getSocket();

		if (socket != null) {
			try {
				socket.setKeepAlive(flag);
			} catch (SocketException e) {
				// error will be caught next read/write operation
			}
		}
	}
	/**
	 * Set socket tcp no delay
	 * 
	 * @param flag boolean, flag
	 */
	public void setTcpNoDelay(boolean flag) {
		f_tcpNoDelay = flag;

		Socket socket = getSocket();
		if (socket != null) {
			try {
				socket.setTcpNoDelay(flag);
			} catch (SocketException e) {
				// error will be caught next read/write operation
			}
		}
	}
	/**
	 * read buffer of bytes.
	 * Make one fast read without timeout mecanics.
	 * This is usefull to drain unwanted still bytes.
	 * If nothing to be read by 1/1000 of second it will
	 * be ok.
	 * 
	 * @param buffer byte[], buffer to fill
	 * @param offset int, start element in the array
	 * @param count int, number of bytes to read
	 * 
	 * @return int, number of read bytes.
	 */
	public int readFast(byte[] buf, int offset, int count) throws IOException {

		// work on copy of socket and input stream.
		// if socket are closed after our null test on it
		// we will catch it in the exception block
		// instead of having a NullPointer...
		SocketChannel socket = f_socket;

		ByteBuffer buffer = ByteBuffer.wrap(buf);
		buffer.position(offset);
		buffer.limit(count);

		Log.logStr(
			Log.LOG_LEVEL_LOW,
			this,
			Log.LOG_TYPE_INFO,
			"readFast(byte[], int, int)",
			"reading ...");

		int totalReadBytes = 0;
		/*
		 * There already some bytes to return to client.
		 * These bytes where kept because they were found after
		 * the pattern of the previous read.
		 * These bytes will be put in the buffer as a normal read
		 */
		if (f_nextRead != null) {
			int xcount;
			byte[] nr = null;
			if (f_nextRead.length >= count) {
				xcount = count;
				nr = new byte[f_nextRead.length - count];
				System.arraycopy(f_nextRead, 0, nr, 0, nr.length);
			} else {
				xcount = f_nextRead.length;
			}

			System.arraycopy(f_nextRead, 0, buffer.array(), offset, xcount);
			f_nextRead = nr;

			return xcount;
		}

		try {
			totalReadBytes = readBytes(buffer, offset, count, 0);
		} catch (InterruptedIOException e) {
			// OK - it was fast.

		} catch (IOException e) {
			Log.logStr(
				this,
				Log.LOG_TYPE_ERROR,
				"readFast(byte[], int, int, boolean, boolean)",
				e.getMessage());

			close();

			throw e;
		}

		if (Log.shouldLog(this.getClass().getName(), Log.LOG_LEVEL_LOW)) {
			Log.logStr(
				this,
				Log.LOG_TYPE_INFO,
				"readFast(byte[], int, int, boolean, boolean)",
				"read buffer: "
					+ (f_noLog ? "<data>" : new String(buffer.array())));
		}

		return totalReadBytes;
	}

	/**
	 * When called, this method will prevent any further
	 * logging of the data on the socket
	 */
	public void disableLogging() {
		Log.logStr(
			Log.LOG_LEVEL_MEDIUM,
			this,
			Log.LOG_TYPE_INFO,
			"disableLogging",
			toString());
		f_noLog = true;
	}

	/**
	 * When called, this method will enable logging
	 * of the data on the socket, the logging
	 * will correspond to the trace level of the class
	 */
	public void enableLogging() {
		Log.logStr(
			Log.LOG_LEVEL_MEDIUM,
			this,
			Log.LOG_TYPE_INFO,
			"enableLogging",
			toString());
		f_noLog = false;
	}

	/**
	 * This method will return the socket channel of this
	 * object or null if the socket is currently disconnected.
	 * This method should be used only by applications that
	 * manages their own "select" to minimize the number of
	 * threads e.g. the quote gateway...
	 * 
	 * @return The socket channel of this object or null
	 * if it is currently disconnected
	 */
	public SocketChannel getSocketChannel() {
		return f_socket;
	}

	/**
	 * This method is used to get the socket associated
	 * to the socket channel that is protected against the
	 * Channel being changed while the socket is retrieved
	 * 
	 * @return The socket associated to the socket channel
	 * or null if the channel is not opened
	 */
	protected Socket getSocket() {
		SocketChannel chan = f_socket;

		if (null != chan) {
			return chan.socket();
		}

		return null;
	}

	/**
	 * Simulate the regular java.net.Socket read behaviour of a blocking
	 * socket with the NIO implementation.
	 * 
	 * @param buffer The buffer into which to read
	 * @param offset The offset in the buffer to start to read to
	 * @param size The number of bytes to read
	 * @param timeout the timeout of the read operatino in
	 * millisconds 
	 * 
	 * @return The number of bytes read or -1, if the peer
	 * has properly closed the socket
	 * 
	 * @throws InterruptedIOExcpetion If there was a timeout in the
	 * read operation
	 * @throws IOException If there was a problem readi
	 */
	protected int readBytes(
		ByteBuffer buffer,
		int offset,
		int size,
		int timeout)
		throws IOException {
		SocketChannel socket = f_socket;

		if (null == socket) {
			throw new IOException("NULL InputStream!");
		}

		try {
			buffer.position(offset);
			buffer.limit(offset + size);

			Selector selector = getReadSelector();
			SelectionKey key = socket.register(selector, SelectionKey.OP_READ);
			int nReady = 0;

			Thread.interrupted();
			if (timeout > 0) {
				nReady = selector.select(timeout);
			} else {
				nReady = selector.selectNow();
			}
			if (nReady <= 0) {
				throw new InterruptedIOException("Simulated read timeout...");
			}

			Set s = selector.selectedKeys();
			if (null != s) {
				s.remove(key);
			}

			return (int) socket.read(buffer);
		} catch (CancelledKeyException e) {
			throw new IOException(e.getMessage());
		} catch (ClosedSelectorException e) {
			throw new IOException(e.getMessage());
		}
	}

	/**
	 * Simulate the regular java.net.Socket write behaviour of a blocking
	 * socket with the NIO implementation.
	 * 
	 * @param buffer The buffer to write to the socket
	 * @param offset The offset in the buffer to start to write
	 * @param size The number of bytes to read
	 * @param timeout the timeout of the read operatino in
	 * millisconds 
	 * 
	 * @return The number of bytes read or -1, if the peer
	 * has properly closed the socket
	 * 
	 * @throws InterruptedIOExcpetion If there was a timeout in the
	 * read operation
	 * @throws IOException If there was a problem readi
	 */
	protected int writeBytes(
		ByteBuffer buffer,
		int offset,
		int size,
		int timeout)
		throws IOException {
		SocketChannel socket = f_socket;

		if (null == socket) {
			throw new IOException("NULL OutputStream!");
		}

		int nbBytes;
		try {
			buffer.position(offset);
			buffer.limit(offset + size);
			Selector selector = getWriteSelector();
			SelectionKey key = socket.register(selector, SelectionKey.OP_WRITE);

			long start = System.currentTimeMillis();
			long elapsed = 0;
			nbBytes = 0;
			do {
				int nReady = 0;

				if (timeout > 0) {
					nReady = selector.select(timeout);
				} else {
					nReady = selector.selectNow();
				}
				if (nReady <= 0) {
					Log.logStr(
						this,
						Log.LOG_TYPE_WARN,
						"checkWriteTimeout()",
						"Write timed out on socket: " + this.toString());

					close();

					throw new WriteTimeoutException();
				}

				Set s = selector.selectedKeys();
				if (null != s) {
					s.remove(key);
				}

				nbBytes = (int) socket.write(buffer);

				elapsed = System.currentTimeMillis() - start;
			} while ((nbBytes < size) && (elapsed < timeout));
		} catch (CancelledKeyException e) {
			throw new IOException(e.getMessage());
		} catch (ClosedSelectorException e) {
			throw new IOException(e.getMessage());
		}

		return nbBytes;
	}

	protected Selector getReadSelector() throws IOException {
		Selector selector = f_readSelector;

		if (null == selector) {
			selector = Selector.open();
			f_readSelector = selector;
		}

		return selector;
	}

	protected Selector getWriteSelector() throws IOException {
		Selector selector = f_writeSelector;

		if (null == selector) {
			selector = Selector.open();
			f_writeSelector = selector;
		}

		return selector;
	}
}