socketconnection.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;


import java.io.IOException;
import java.nio.channels.SocketChannel;
import java.util.Enumeration;
import java.util.Hashtable;

import org.butor.log.Log;
import org.butor.socket.tcp.lowlevel.ReadTimeoutException;
import org.butor.socket.tcp.lowlevel.SocketIO;
import org.butor.socket.tcp.lowlevel.WriteTimeoutException;

/**
 * Tcp socket manager object. The goal of this object is
 * to simplify and standardize socket handling.
 * 
 * This object handle socket operations such
 * creation, logging, read/write, timeout checking, ...
 * 
 * Client using this object can register within this class
 * to be informed about this class activities and events such as
 * connection/disconnection events and read/write operations ...
 * 
 * @author Aiman Sawan
 */

public class SocketConnection {

	protected SocketIO f_socketIO;
	protected Hashtable f_socketEventListeners;

	protected int f_readWriteTimeout = SocketIO.DEFAULT_BUFFER_READ_WRITE_TIMEOUT;
	protected boolean f_tcpNoDelay = true;
	protected boolean f_keepAlive = true;
	protected int f_chunkReadTimeout = SocketIO.DEFAULT_CHUNK_READ_TIMEOUT;
	protected int f_chunkSize = SocketIO.DEFAULT_CHUNK_SIZE;
	protected boolean f_noLog = false;
	
	/**
	 * Constructor.
	 * 
	 * @param host String, host name to connect
	 * @param port int, port number to connect
	 */
	public SocketConnection() {
		super();

		Log.logStr(
			Log.LOG_LEVEL_MEDIUM,
			this,
			Log.LOG_TYPE_INFO,
			"Constructor()",
			"");

		f_socketEventListeners = new Hashtable();
	}
	/**
	 * Set this object socket.
	 * 
	 * @param Socket socket.
	 */
	public boolean setSocket(SocketChannel socket) {

		Log.logStr(Log.LOG_LEVEL_MEDIUM, this, Log.LOG_TYPE_INFO, "setSocket()", "");

		try {
			if (f_socketIO == null) {
				f_socketIO = new SocketIO(socket);
			} else {
				f_socketIO.setSocket(socket);
			}

			// apply client config
			if (f_noLog) {
				f_socketIO.disableLogging();
			} else {
				f_socketIO.enableLogging();
			}
			f_socketIO.setBufferReadWriteTimeout(f_readWriteTimeout);
			f_socketIO.setTcpNoDelay(f_tcpNoDelay);
			f_socketIO.setKeepAlive(f_keepAlive);
			f_socketIO.setChunkReadTimeout(f_chunkReadTimeout);
			f_socketIO.setChunkSize(f_chunkSize);
	
		} catch (IOException e) {
			Log.logStr(this, Log.LOG_TYPE_ERROR, "setSocket()", "");

			return false;
		}

		return true;
	}
	/**
	 * Register a socket events org.butor.web.listener.
	 * The org.butor.web.listener will be notified when registered events occure.
	 * Registered events may be overriden at any moment.
	 * 
	 * @see SocketEvent.
	 * 
	 * @param org.butor.web.listener ISocketEventListener, a org.butor.web.listener object.
	 * @param eventsBitSet int, events bitset. Example if a org.butor.web.listener
	 * 		need to be notified for connect and disconnect and readTimeout events
	 * 		it must register like this: 
	 * 			registerSocketEventListener(org.butor.web.listener, 
	 * 				SocketEvent.EVENT_CONNECTED |
	 *				SocketEvent.EVENT_DISCONNECTED |
	 *				SocketEvent.EVENT_READ_TIMEOUT)
	 */
	public synchronized void registerSocketEventListener(
		ISocketEventListener listener,
		int eventsBitSet) {
		f_socketEventListeners.put(listener, new Integer(eventsBitSet));
	}
	/**
	 * Unregister a socket events org.butor.web.listener.
	 * 
	 * @param org.butor.web.listener ISocketEventListener, a org.butor.web.listener object.
	 */
	public synchronized void unregisterSocketEventListener(ISocketEventListener listener) {
		if (f_socketEventListeners.containsKey(listener)) {
			f_socketEventListeners.remove(listener);
		}
	}
	/**
	 * Close connection
	 */
	public void disconnect() {

		Log.logStr(
			Log.LOG_LEVEL_LOW,
			this,
			Log.LOG_TYPE_INFO,
			"disconnect()",
			"disconnecting " +this.toString() +" ...");

		if (f_socketIO != null) {
			f_socketIO.close();
			fireSocketEvent(SocketEvent.EVENT_DISCONNECTED);
		}	
	}
	/**
	 * Fire events to listeners
	 * 
	 * @param id int, events id
	 */
	protected void fireSocketEvent(int id) {
		this.fireSocketEvent(id, null, 0, 0);
	}
	/**
	 * Fire events to listeners
	 * 
	 * @param id int, events id
	 * @param buffer byte[], data related to events
	 */
	protected void fireSocketEvent(int id, byte[] buffer) {
		this.fireSocketEvent(id, buffer, 0, (buffer == null ? 0 : buffer.length));
	}
	/**
	 * Fire events to listeners
	 * 
	 * @param id int, events id
	 * @param data byte[], data related to events
	 */
	protected void fireSocketEvent(int id, byte[] buffer, int offset, int count) {
		SocketEvent event = null;
		Integer eventsBitSet = null;

		Enumeration listeners = f_socketEventListeners.keys();
		while (listeners.hasMoreElements()) {

			if (event == null) {
				event = new SocketEvent(this);
				event.setId(id);
				if (buffer != null && offset > -1 && count > -1 && count <= buffer.length) {
					byte[] eventBytes = new byte[count];
					System.arraycopy(buffer, offset, eventBytes, 0, count);
					event.setBytes(eventBytes);
				}
			}

			ISocketEventListener listener = (ISocketEventListener) listeners.nextElement();
			// check if org.butor.web.listener are registered for this event
			// bitset flag and test.
			eventsBitSet = (Integer) f_socketEventListeners.get(listener);
			if ((eventsBitSet.intValue() & id) == id) {
				listener.handleSocketEvent(event);
			}
		}
	}
	/**
	 * Check connection status.
	 * 
	 * @return boolean, true if it is sane, false otherwise.
	 */
	public boolean isSane() {
		return (f_socketIO != null) && f_socketIO.isSane();
	}
	/**
	 * Set chunk read size.
	 * 
	 * @param size int, one chunk read size.
	 */
	public void setChunkSize(int size) {
		f_chunkSize = size;
		if (f_socketIO != null) {
			f_socketIO.setChunkSize(size);
		}
	}
	/**
	 * Get chunk read size.
	 * 
	 * @return int, chunk read size
	 */
	public int getChunkSize() {
		if (f_socketIO != null) {
			f_socketIO.getChunkSize();
		}
		return f_chunkSize;
	}
	/**
	 * read buffer of bytes.
	 * 
	 * @param buffer byte[], buffer to fill
	 * 
	 * @return int, number of read bytes.
	 */
	public int read(byte[] buffer) throws IOException {
		return this.read(buffer, 0, buffer.length);
	}
	/**
	 * 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.
	 * 
	 * @return byte[], read buffer ending with expected bytes.
	 */
	public int readUntilPattern(byte[] buffer, byte[] pattern, int patternSearchOffset)
		throws IOException, ReadTimeoutException {
		Log.logStr(
			Log.LOG_LEVEL_MEDIUM,
			this,
			Log.LOG_TYPE_INFO,
			"readUntilPattern(byte[], byte[], int)",
			"reading ...");

		if (f_socketIO == null) {
			throw new IOException("NULL Socket!");
		}

		try {
			int bytesCount = f_socketIO.readUntilPattern(buffer, pattern, patternSearchOffset);
			fireSocketEvent(
				(bytesCount > 0 ? SocketEvent.EVENT_READ_DATA : SocketEvent.EVENT_READ_FAILED),
				buffer, 0, bytesCount);
			return bytesCount;

		} catch (ReadTimeoutException e) {
			fireSocketEvent(SocketEvent.EVENT_READ_TIMEOUT);
			throw e;

		} catch (IOException e) {
			fireSocketEvent(SocketEvent.EVENT_READ_FAILED);
			throw e;
		}
	}
	/**
	 * Get pattern offset.
	 * If last readUntilPattern() contain the pattern
	 * expected then this method will return its starting position.
	 * 
	 * @return int.
	 */
	public int getPatternEndOffset() {
		return (f_socketIO != null ? f_socketIO.getPatternEndOffset() : -1);
	}
	/**
	 * read buffer of bytes.
	 * 
	 * @param buffer byte[], buffer to read into
	 * @param offset int, reading start element in the buffer
	 * @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 {

		if (f_socketIO == null) {
			throw new IOException("NULL Socket!");
		}

		try {
			int bytesCount = f_socketIO.read(buffer, offset, count);
			fireSocketEvent(
				(bytesCount > 0 ? SocketEvent.EVENT_READ_DATA : SocketEvent.EVENT_READ_FAILED),
				buffer, offset, count);
			return bytesCount;

		} catch (ReadTimeoutException e) {
			fireSocketEvent(SocketEvent.EVENT_READ_TIMEOUT, buffer, offset, count);
			throw e;

		} catch (IOException e) {
			fireSocketEvent(SocketEvent.EVENT_READ_FAILED, buffer, offset, count);
			throw e;
		}
	}
	/**
	 * write a byte array onto the socket.
	 * 
	 * @param buffer byte[], buffer to write
	 * 
	 * @return boolean, write success status.
	 */
	public boolean write(byte[] buffer) throws IOException {

		if (buffer == null) {
			Log.logStr(
				this,
				Log.LOG_TYPE_ERROR,
				"write(byte[])",
				"Got NULL buffer to write");
			throw new IOException("Got NULL buffer to write");
		}

		return this.write(buffer, 0, buffer.length);		
	}
	/**
	 * write an array of bytes onto the socket.
	 * 
	 * @param buffer byte[], buffer to write
	 * @param offset int, starting byte in the array
	 * @param count int, number of bytes to write
	 * 
	 * @return boolean, write success status.
	 */
	public boolean write(byte[] buffer, int offset, int count)
		throws IOException {

		if (buffer == null) {
			Log.logStr(
				this,
				Log.LOG_TYPE_ERROR,
				"write(byte[], int, int)",
				"Got NULL buffer to write");
			throw new IOException("Got NULL buffer to write");
		}

		if (f_socketIO != null) {
			try {
				f_socketIO.write(buffer, offset, count);
				fireSocketEvent(SocketEvent.EVENT_WRITE_DATA, buffer, offset, count);
				return true;
				
			} catch (WriteTimeoutException e) {
				Log.logStr(this, Log.LOG_TYPE_ERROR, "write(byte[], int, int)", "WriteTimeoutException: " +e.getMessage());
				fireSocketEvent(SocketEvent.EVENT_WRITE_TIMEOUT, buffer, offset, count);
				throw e;
				
			} catch (IOException e) {
				Log.logStr(this, Log.LOG_TYPE_ERROR, "write(byte[], int, int)", "IOException: " +e.getMessage());
				fireSocketEvent(SocketEvent.EVENT_WRITE_FAILED, buffer, offset, count);
				throw e;
			}

		} else {
			fireSocketEvent(SocketEvent.EVENT_WRITE_FAILED, buffer, offset, count);
			return false;
		}
	}
	/**
	 * Set sockey keep alive
	 * 
	 * @param flag boolean, flag
	 */
	public void setKeepAlive(boolean flag) {
		f_keepAlive = flag;
		if (f_socketIO != null) {
			f_socketIO.setKeepAlive(flag);
		}
	}
	/**
	 * Set socket tcp no delay
	 * 
	 * @param flag boolean, flag
	 */
	public void setTcpNoDelay(boolean flag) {
		f_tcpNoDelay = flag;
		if (f_socketIO != null) {
			f_socketIO.setTcpNoDelay(flag);
		}
	}
	/**
	 * Set read chunk of bytes timeout.
	 * 
	 * @param milliseconds int, read timeout in milliseconds
	 */
	public void setChunkReadTimeout(int milliseconds) {
		f_chunkReadTimeout = milliseconds;
		if (f_socketIO != null) {
			f_socketIO.setChunkReadTimeout(milliseconds);
		}
	}
	/**
	 * Get read chunk of bytes timeout.
	 * 
	 * @return int, read timeout in milliseconds
	 */
	public int getChunkReadTimeout() {
		if (f_socketIO != null) {
			return f_socketIO.getChunkReadTimeout();
		}

		return f_chunkReadTimeout;
	}
	/**
	 * Set read and write operations timeout.
	 * 
	 * @param milliseconds int, read timeout in milliseconds
	 */
	public void setReadWriteTimeout(int milliseconds) {
		f_readWriteTimeout = milliseconds;
		if (f_socketIO != null) {
			f_socketIO.setBufferReadWriteTimeout(milliseconds);
		}
	}
	/**
	 * Get read and write operations timeout.
	 * 
	 * @return int, read timeout in milliseconds
	 */
	public int getReadWriteTimeout() {
		if (f_socketIO != null) {
			return f_socketIO.getBufferReadWriteTimeout();
		}

		return f_readWriteTimeout;
	}

	/**
	 * Flush socket output stream.
	 */
	public void flush() throws IOException {
		if (f_socketIO != null) {
			f_socketIO.flush();
		}
	}
	/**
	 * 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
	 * 
	 * @return int, number of read bytes.
	 */
	public int readFast(byte[] buffer) {
		if (buffer == null) {
			Log.logStr(this, Log.LOG_TYPE_ERROR, "readFast(byte[])", 
				"Got NULL buffer to read in.");
			return 0;
		}
		
		if (!isSane()) {
			fireSocketEvent(SocketEvent.EVENT_READ_FAILED, null);
			return 0;
		}

		try {
			int bytesCount = f_socketIO.readFast(buffer, 0, buffer.length);			
			fireSocketEvent(SocketEvent.EVENT_READ_DATA, buffer, 0, bytesCount);
			return bytesCount;
			
		} catch (ReadTimeoutException e) {
			// It is ok because there in nothing to read fast.
			
		} catch (IOException e) {
			Log.logStr(this, Log.LOG_TYPE_ERROR, "readFast(byte[])", e.getMessage());
			fireSocketEvent(SocketEvent.EVENT_READ_FAILED, null);
		}
		
		return 0;
	}
	
	/**
	 * When called, this method will prevent any further
	 * logging of the data on the socket
	 */
	public void disableLogging() {
		f_noLog = true;
		
		SocketIO io = f_socketIO;
		if (null != io) {
			io.disableLogging();
		}
	}
	
	/**
	 * 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() {
		f_noLog = false;
		
		SocketIO io = f_socketIO;
		if (null != io) {
			io.enableLogging();
		}
	}
	
	/**
	 * Returns the current socket channel of this socket
	 * if there is one or null if there is none
	 * 
	 * @return The socket channel of this socket
	 */
	public SocketChannel currentChannel() {
		SocketIO socket = f_socketIO;
		
		if (null != socket) {
			return socket.getSocketChannel();
		}
		
		return null;
	}
	
	/**
	 * 
	 */
	public String toString() {
		if (f_socketIO != null) {		
			return f_socketIO.toString();
		}
		
		return super.toString();
	}
}