socketio.java

一个实用工具类

/*
 * Copyright (C) butor.com. All rights reserved.
 *
 * This software is published under the terms of the GNU Library General
 * Public License (GNU LGPL), a copy of which has been included with this
 * distribution in the LICENSE.txt file. 
 */


package org.butor.socket.tcp.lowlevel;


import java.io.IOException;
import java.io.InterruptedIOException;
import java.net.Socket;
import java.net.SocketException;
import java.nio.ByteBuffer;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.ClosedSelectorException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Set;

import org.butor.log.Log;

/**
 * This object handle basic socket read/write operations.
 * Timeout on read/write are implemented.
 * 
 * Read timeout composed from two timeouts:
 * 1. timeout to read one chunk of bytes.
 * 2. timeout to fill the entire buffer.
 * 
 * Write timeout are implemented as follow:
 * start write time are initialized before the write
 * begin. A thread "SocketIOWriteTimeoutMonitor" check
 * at regular interval if the operation timedout. If there a timeout
 * then the socket will be closed.
 * 
 * @see SocketIOManager
 * @see SocketIOWriteTimeoutMonitor
 * 
 * @author Aiman Sawan
 */
public class SocketIO {
	protected static final String CLASS_NAME = SocketIO.class.getName();

	public static final int DEFAULT_CHUNK_SIZE = 256;
	public static final int DEFAULT_CHUNK_READ_TIMEOUT = 3000;
	public static final int DEFAULT_BUFFER_READ_WRITE_TIMEOUT = 5000;

	protected SocketChannel f_socket;
	protected int f_chunkReadTimeout;
	protected int f_bufferReadWriteTimeout;
	protected int f_chunkSize;
	protected boolean f_keepAliveFlag;
	protected boolean f_tcpNoDelay;

	protected int f_patternEndOffset = -1;
	protected int f_patternNextByteToCheck = 0;

	protected byte[] f_nextRead = null;

	protected String f_socketDesc = null;

	protected boolean f_noLog = false;

	protected Selector f_readSelector = null;
	protected Selector f_writeSelector = null;

	/**
	 * Constructor.
	 * 
	 * @param socket Socket, the socket to warp for reading and writing.
	 */
	public SocketIO(SocketChannel socket) throws IOException {

		// default settings			
		f_chunkSize = DEFAULT_CHUNK_SIZE;
		f_chunkReadTimeout = DEFAULT_CHUNK_READ_TIMEOUT;
		f_bufferReadWriteTimeout = DEFAULT_BUFFER_READ_WRITE_TIMEOUT;

		f_keepAliveFlag = true;

		setSocket(socket);
	}
	/**
	 * Get connection sanity status after last operation.
	 * 
	 * @return boolean, true if connection is sane, false otherwise.
	 */
	public boolean isSane() {
		return (f_socket != null);
	}
	/**
	 * Set read chunk of bytes timeout.
	 * 
	 * @param milliseconds int, read timeout in milliseconds
	 */
	public void setChunkReadTimeout(int milliseconds) {
		f_chunkReadTimeout = milliseconds;
		// force socket description to be rebuilt
		f_socketDesc = null;
	}
	/**
	 * Get read chunk of bytes timeout.
	 * 
	 * @return int, read timeout in milliseconds
	 */
	public int getChunkReadTimeout() {
		return f_chunkReadTimeout;
	}
	/**
	 * Set one read chunk size.
	 * 
	 * @param size int, read chunk size
	 */
	public void setChunkSize(int size) {
		f_chunkSize = size;
	}
	/**
	 * Get one read chunk size.
	 * 
	 * @return int, read chunk size
	 */
	public int getChunkSize() {
		return f_chunkSize;
	}
	/**
	 * Set buffer read / write timeout.
	 * This is the timeout to read and fill a buffer.
	 * 
	 * @param milliseconds int, read timeout in milliseconds
	 */
	public void setBufferReadWriteTimeout(int milliseconds) {
		f_bufferReadWriteTimeout = milliseconds;
		// force socket description to be rebuilt
		f_socketDesc = null;
	}
	/**
	 * Get buffer read / write timeout.
	 * 
	 * @return int, read timeout in milliseconds
	 */
	public int getBufferReadWriteTimeout() {
		return f_bufferReadWriteTimeout;
	}
	/**
	 * Close the socket.
	 */
	public synchronized void close() {

		Log.logStr(
			Log.LOG_LEVEL_LOW,
			this,
			Log.LOG_TYPE_INFO,
			"close()",
			"closing socket ...");

		SocketChannel socket = f_socket;
		if (socket == null) {
			return;
		}

		synchronized (socket) {
			try {
				socket.socket().close();
			} catch (IOException e) {
				// Don't care...
			}

			try {
				socket.close();
			} catch (IOException e) {
				// Don't care...
			}

			f_socket = null;

			if (null != f_readSelector) {
				try {
					f_readSelector.wakeup();
					f_readSelector.close();
				} catch (IOException e) {
					// Don't care...
				} finally {
					f_readSelector = null;
				}
			}

			if (null != f_writeSelector) {
				try {
					f_writeSelector.wakeup();
					f_writeSelector.close();
				} catch (IOException e) {
					// Don't care...
				} finally {
					f_writeSelector = null;
				}
			}
		}
	}
	/**
	 * Get pattern end offset.
	 * If last readUntilPattern() contain the pattern
	 * expected then this method will return its end position.
	 * 
	 * @return int.
	 */
	public int getPatternEndOffset() {
		return f_patternEndOffset;
	}
	/**
	 * Read until expected bytes are encountered or timeout occure.
	 * will read until the sequence of bytes are encountered.
	 * For example read one line will stop reading and 
	 * return a line when the char '\n' will be read.
	 * 
	 * @param pattern byte[], Expected sequence of bytes.
	 * @param patternSearchOffset int, start seraching of the pattern at this 
	 * offset FROM THE END of buffer. Go to the end of the buffer then 
	 * go back this param value and then search the pattern. This is usefull
	 * when client is SURE that the pattern is in the end. This is fast because
	 * no need to scan all the buffer chars looking for the pattern.
	 * 
	 * @return byte[], read buffer ending with expected bytes.
	 */
	public int readUntilPattern(
		byte[] buffer,
		byte[] pattern,
		int patternSearchOffsetFromTheEnd)
		throws IOException {
		Log.logStr(
			Log.LOG_LEVEL_MEDIUM,
			this,
			Log.LOG_TYPE_INFO,
			"readUntilPattern(byte[], byte[], int)",
			"reading ...");

		return this.readBytes(
			buffer,
			0,
			buffer.length,
			pattern,
			patternSearchOffsetFromTheEnd);
	}

	/**
	 * read buffer of bytes.
	 * 
	 * @param buffer byte[], buffer to fill
	 * 
	 * @return int, number of read bytes.
	 */
	public int read(byte[] buffer) throws IOException {

		Log.logStr(
			Log.LOG_LEVEL_MEDIUM,
			this,
			Log.LOG_TYPE_INFO,
			"read(byte[])",
			"reading ...");

		// get bytes from buffered lines.
		return readBytes(buffer, 0, buffer.length, null, 0);
	}
	/**
	 * read buffer of bytes.
	 * 
	 * @param buffer byte[], buffer to read into
	 * @param offset int, start element in the array
	 * @param count int, number of bytes to read
	 * 
	 * @return int, number of read bytes.
	 */
	public int read(byte[] buffer, int offset, int count) throws IOException {
		return this.readBytes(buffer, offset, count, null, 0);
	}
	/**
	 * read buffer of bytes.
	 * There is two timeouts. one to read a chunk of bytes and the other
	 * to fill the buffer. The first one is used to return after
	 * an elapse of time if there is nothing to read. The other is checked
	 * while filling the buffer. 
	 * The read my be unterrupted if a pattern of bytes is found. You can get
	 * the position of the end of the pattern by calling getPatternEndOffset().
	 * A pattern may be searched on two reads (two successives calls) This is occure when
	 * the pattern start at the end of the buffer and continue at the beginig of the buffer
	 * of the next read.
	 * 
	 * @param buffer byte[], buffer to fill
	 * @param offset int, start element in the array
	 * @param count int, number of bytes to read
	 * @param pattern byte[], Expected sequence of bytes to stop reading.
	 * 		this may be null if there is no need to pattern search.
	 * 		
	 * @param patternSearchOffset int, start seraching of the pattern at this 
	 * offset FROM THE END of buffer. Go to the end of the buffer then 
	 * go back this param value and then search the pattern. This is usefull
	 * when client is SURE that the pattern is in the end. This is fast because
	 * no need to scan all the buffer chars looking for the pattern.
	 * 
	 * @return int, number of read bytes.
	 */
	protected int readBytes(
		byte[] buf,
		int offset,
		int count,
		byte[] pattern,
		int patternSearchOffsetFromTheEnd)
		throws IOException {

		if (Log
			.shouldLog(this.getClass().getName(), Log.LOG_LEVEL_MEDIUM)) {
			Log.logStr(
				this,
				Log.LOG_TYPE_INFO,
				"readBytes(byte[], int, int, byte[], int)",
				"reading: offset=["
					+ offset
					+ "] count=["
					+ count
					+ "] buffer size=["
					+ buf.length
					+ "] pattern=["
					+ (pattern != null ? new String(pattern) : "NULL")
					+ "]");
		}

		if (pattern != null
			&& pattern.length > patternSearchOffsetFromTheEnd) {
			/**
			 * Impossible to find the pattern in the last 'patternSearchOffsetFromTheEnd'
			 * chars because number of chars to search in is less than the pattern length.
			 */
			Log.logStr(
				this,
				Log.LOG_TYPE_WARN,
				"readBytes(byte[], int, int, byte[], int)",
				"patternSearchOffsetFromTheEnd is less than the pattern length! Correct please.");
			throw new IOException("patternSearchOffsetFromTheEnd is less than the pattern length! Correct please.");
		}

		if (offset < 0 || offset > buf.length) {
			/**
			 * Trying to read outside buffer limits!
			 */
			Log.logStr(
				this,
				Log.LOG_TYPE_WARN,
				"readBytes(byte[], int, int, byte[], int)",
				"Trying to read outside buffer limits! offset is greater than buffer length!");
			throw new IOException("Trying to read outside buffer! offset is greater than buffer length!");
		}

		if (buf.length - offset < count) {
			/**
			 * The count exceed the available space in buffer!
			 */
			Log.logStr(
				this,
				Log.LOG_TYPE_WARN,
				"readBytes(byte[], int, int, byte[], int)",
				"The count exceed the available space in buffer!");
			throw new IOException("The count exceed the available space in buffer!");
		}

		ByteBuffer buffer = ByteBuffer.wrap(buf);

		// reset last found pattern offset.
		f_patternEndOffset = -1;

		if (buffer == null) {
			throw new IOException("Got NULL buffer to read into.");
		}

		boolean isReadChunckTimedout = false;

		// 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;

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

		Log.logStr(
			Log.LOG_LEVEL_LOW,
			this,
			Log.LOG_TYPE_INFO,
			"readBytes(byte[], int, int, byte[], 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
		 * and the next processing will check for the presence of pattern
		 * (if apply). If again bytes will be found after the pattern, theses
		 * bytes will be returned back into the f_nextRead buffer (at the beginig)
		 * for further reads.
		 */
		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, buf, offset, xcount);
			f_nextRead = nr;
			totalReadBytes = xcount;
		}

		long startTime = System.currentTimeMillis();