xynetsocket.cs

一个打包好的.NET环境下的TCP插口组件

using System;
using System.Threading;
using System.Collections;
using System.Net;
using System.Net.Sockets;
using XYThreadPoolLib;

namespace XYNetSocketLib
{
	public delegate void ExceptionHandlerDelegate(Exception oBug);
	public delegate void ConnectionFilterDelegate(String sRemoteAddress, int nRemotePort, Socket sock);
	public delegate void BinaryInputHandlerDelegate(String sRemoteAddress, int nRemotePort, Byte[] pData);
	public delegate void StringInputHandlerDelegate(String sRemoteAddress, int nRemotePort, String sData);

	public class XYNetServer
	{
		private delegate void AcceptClientsDelegate();
		private delegate void DetectInputDelegate();
		private delegate void ProcessInputDelegate(Socket sock, IPEndPoint ipe);
		
		private const int m_nServerPause = 25;
		private const int m_nListenBacklog = 32;
		private const int m_nArrayCapacity = 512;
		private int m_nReadTimeout = 30;
		private int m_nMaxDataSize = 4*1024*1024;
		private int m_nMinThreadCount = 7;
		private int m_nMaxThreadCount = 12;
		private String m_sAddress = "";
		private int m_nPort = 0;
		private Socket m_socketServer = null;
		private XYThreadPool m_threadPool = new XYThreadPool();
		private Hashtable m_htSockets = new Hashtable(m_nArrayCapacity);
		private ArrayList m_listSockets = new ArrayList(m_nArrayCapacity);
		private ConnectionFilterDelegate m_delegateConnectionFilter = null;
		private ExceptionHandlerDelegate m_delegateExceptionHandler = null;
		private BinaryInputHandlerDelegate m_delegateBinaryInputHandler = null;
		private StringInputHandlerDelegate m_delegateStringInputHandler = null;
		private Exception m_exception = null;
		
		public XYNetServer(String sAddress, int nPort, int nMinThreadCount, int nMaxThreadCount)
		{
			if(sAddress!=null) m_sAddress = sAddress;
			if(nPort>0) m_nPort = nPort;
			if(nMinThreadCount>0) m_nMinThreadCount = nMinThreadCount+2;
			if(nMinThreadCount>0&&nMaxThreadCount>nMinThreadCount) m_nMaxThreadCount = nMaxThreadCount+2;
			else m_nMaxThreadCount = 2*(m_nMinThreadCount-2)+2;
		}

		~XYNetServer()
		{
			StopServer();
		}

		public Exception GetLastException()
		{
			Monitor.Enter(this);
			Exception exp = m_exception;
			Monitor.Exit(this);
			return exp;
		}

		public void StopServer()
		{
			try
			{
				Monitor.Enter(this);
				m_threadPool.StopThreadPool();
				if(m_socketServer!=null)
				{
					Socket sock = null;
					for(int i=0;i<m_listSockets.Count;i++)
					{
						sock = (Socket)(m_listSockets[i]);
						try
						{
							sock.Shutdown(SocketShutdown.Both);
							sock.Close();
						}
						catch(Exception) {}
					}
					try
					{
						m_socketServer.Shutdown(SocketShutdown.Both);
						m_socketServer.Close();
					}
					catch(Exception) {}
				}
			}
			catch(Exception) {}
			finally 
			{
				try
				{
					m_socketServer = null;
					m_htSockets.Clear();
					m_listSockets.Clear();
				}
				catch(Exception) {}
				Monitor.Exit(this); 
			}
		}

		public bool StartServer()
		{
			try
			{
				Monitor.Enter(this);
				XYNetCommon.SetSocketPermission();
				StopServer();
				m_threadPool.SetThreadErrorHandler(new ThreadErrorHandlerDelegate(ThreadErrorHandler));
				m_threadPool.StartThreadPool(m_nMinThreadCount, m_nMaxThreadCount);
				m_socketServer = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
				m_socketServer.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, 1);
				IPEndPoint myEnd = (m_sAddress=="")?(new IPEndPoint(Dns.GetHostByName(Dns.GetHostName()).AddressList[0].Address, m_nPort)):(new IPEndPoint(IPAddress.Parse(m_sAddress), m_nPort));
				m_socketServer.Bind(myEnd);
				m_socketServer.Listen(m_nListenBacklog);
				m_threadPool.InsertWorkItem("Accept Clients", new AcceptClientsDelegate(AcceptClients), null, false);
				m_threadPool.InsertWorkItem("Detect Input", new DetectInputDelegate(DetectInput), null, false);
				return true;
			}
			catch(Exception oBug) 
			{ 
				m_exception = oBug; 
				return false;
			}
			finally { Monitor.Exit(this); }
		}

		public void SetReadTimeout(int nReadTimeout)
		{
			Monitor.Enter(this);
			if(nReadTimeout>=5&&nReadTimeout<=1200) m_nReadTimeout = nReadTimeout;
			Monitor.Exit(this);
		}

		public void SetMaxDataSize(int nMaxDataSize)
		{
			Monitor.Enter(this);
			if(nMaxDataSize>=1024) m_nMaxDataSize = nMaxDataSize;
			Monitor.Exit(this);
		}

		public void SetConnectionFilter(ConnectionFilterDelegate pMethod)
		{
			Monitor.Enter(this);
			if(m_delegateConnectionFilter==null) m_delegateConnectionFilter = pMethod;
			Monitor.Exit(this);
		}

		public void SetExceptionHandler(ExceptionHandlerDelegate pMethod)
		{
			Monitor.Enter(this);
			if(m_delegateExceptionHandler==null) m_delegateExceptionHandler = pMethod;
			Monitor.Exit(this);
		}

		public void SetBinaryInputHandler(BinaryInputHandlerDelegate pMethod)
		{
			Monitor.Enter(this);
			if(m_delegateBinaryInputHandler==null) m_delegateBinaryInputHandler = pMethod;
			Monitor.Exit(this);
		}

		public void SetStringInputHandler(StringInputHandlerDelegate pMethod)
		{
			Monitor.Enter(this);
			if(m_delegateStringInputHandler==null) m_delegateStringInputHandler = pMethod;
			Monitor.Exit(this);
		}

		private void ThreadErrorHandler(ThreadPoolWorkItem oWorkItem, Exception oBug)
		{
			try
			{
				Monitor.Enter(this);
				if(m_delegateExceptionHandler!=null)
				{
					m_threadPool.InsertWorkItem("Handle Exception", m_delegateExceptionHandler, new Object[1]{oBug}, false);
				}
				else m_exception = oBug;
			}
			catch(Exception) { }
			finally { Monitor.Exit(this); }
		}

		private void AcceptClients()
		{
			while(true)
			{
				bool bHasNewClient = false;
				Socket sock = null;
				try
				{
					if(m_socketServer.Poll(m_nServerPause, SelectMode.SelectRead))
					{
						bHasNewClient = true;
						sock = m_socketServer.Accept();
						IPEndPoint ipe = (IPEndPoint)(sock.RemoteEndPoint);
						if(m_delegateConnectionFilter!=null)
						{
							m_delegateConnectionFilter.DynamicInvoke(new Object[3]{ipe.Address.ToString(), ipe.Port, sock});
						}
						if(sock.Connected)
						{
							String sKey = ipe.Address.ToString() + ":" + ipe.Port.ToString();
							Monitor.Enter(this);
							m_htSockets.Add(sKey, sock);
							m_listSockets.Add(sock);
							Monitor.Exit(this);
						}
					}
				}
				catch(Exception oBug)
				{
					if(sock!=null)
					{
						try
						{
							sock.Shutdown(SocketShutdown.Both);
							sock.Close();
						}
						catch(Exception) {}
					}
					if(m_delegateExceptionHandler!=null)
					{
						m_threadPool.InsertWorkItem("Handle Exception", m_delegateExceptionHandler, new Object[1]{oBug}, false);
					}
					else 
					{
						Monitor.Enter(this);
						m_exception = oBug;
						Monitor.Exit(this);
					}
				}
				if(bHasNewClient) Thread.Sleep(m_nServerPause);
				else Thread.Sleep(10*m_nServerPause);
			}
		}

		private void DetectInput()
		{
			while(true)
			{
				bool bNoData = true;
				Socket sock = null;
				IPEndPoint ipe = null;
				int i = -1;
				try
				{
					for(i=m_listSockets.Count-1;i>=0;i--)
					{
						sock = (Socket)(m_listSockets[i]);
						ipe = (IPEndPoint)(sock.RemoteEndPoint);
						if(!sock.Connected) throw new Exception("Connection to client closed");
						if(sock.Available>0)
						{
							Monitor.Enter(this);
							m_listSockets.RemoveAt(i);
							Monitor.Exit(this);
							bNoData = false;
							m_threadPool.InsertWorkItem("Process Input", new ProcessInputDelegate(ProcessInput), new Object[2] {sock, ipe}, false);
						}
					}
				}
				catch(Exception oBug)
				{
					if(i>=0&&sock!=null)
					{
						Monitor.Enter(this);
						m_listSockets.RemoveAt(i);
						m_htSockets.Remove(ipe.Address.ToString() + ":" + ipe.Port.ToString());
						Monitor.Exit(this);
						try
						{
							sock.Shutdown(SocketShutdown.Both);
							sock.Close();
						}
						catch(Exception) {}
					}
					if(m_delegateExceptionHandler!=null)
					{
						m_threadPool.InsertWorkItem("Handle Exception", m_delegateExceptionHandler, new Object[1]{oBug}, false);
					}
					else 
					{
						Monitor.Enter(this);
						m_exception = oBug;
						Monitor.Exit(this);
					}
				}
				if(bNoData) Thread.Sleep(10*m_nServerPause);
				else Thread.Sleep(m_nServerPause);
			}
		}

		private void ProcessInput(Socket sock, IPEndPoint ipe)
		{
			try
			{
				Byte[] pHeader = new Byte[4];
				int nPos = 0;
				long nStart = DateTime.Now.Ticks;
				while(nPos<4)
				{
					if(sock.Available>0)
					{
						nPos += sock.Receive(pHeader, nPos, Math.Min(sock.Available,(4-nPos)), SocketFlags.None);
						if((pHeader[0]%16)>1) 
						{
							sock.Shutdown(SocketShutdown.Both);
							sock.Close();
							Monitor.Enter(this);
							m_htSockets.Remove(ipe.Address.ToString() + ":" + ipe.Port.ToString());
							Monitor.Exit(this);
							return;
						}
					}
					else Thread.Sleep(m_nServerPause);
					if(nPos<4&&((DateTime.Now.Ticks-nStart)/10000)>m_nReadTimeout*1000) throw new Exception("Timeout while receiving incoming data");
				}
				int nSize = pHeader[1] + pHeader[2]*256 + pHeader[3]*65536+(pHeader[0]/16)*16777216;
				if(nSize>m_nMaxDataSize) throw new Exception("Data size too large");
				Byte[] pData = new Byte[nSize];
				nPos = 0;
				nStart = DateTime.Now.Ticks;
				while(nPos<nSize)
				{
					if(sock.Available>0)
					{
						nPos += sock.Receive(pData, nPos, Math.Min(sock.Available, (nSize-nPos)), SocketFlags.None);
					}
					else Thread.Sleep(m_nServerPause);
					if(nPos<nSize&&((DateTime.Now.Ticks-nStart)/10000)>m_nReadTimeout*1000) throw new Exception("Timeout while receiving incoming data");
				}
				Monitor.Enter(this);
				m_listSockets.Add(sock);
				Monitor.Exit(this);
				if((pHeader[0]%16)==1)
				{
					if(m_delegateBinaryInputHandler!=null) 
					{
						m_threadPool.InsertWorkItem("Handle Binary Input", new BinaryInputHandlerDelegate(m_delegateBinaryInputHandler), new Object[3] {ipe.Address.ToString(), ipe.Port, pData}, false);
					}
					else throw new Exception("No binary input handler");
				}
				else
				{
					if(m_delegateStringInputHandler!=null)