cmppasynchronizedio.java

短信网关发送接受平台。

import java.lang.*;
import java.net.*;
import java.io.*;

import CMPPThread;

import CMPPSynchronizedIO;

import CMPPInputStream;
import CMPPOutputStream;

public class CMPPAsynchronizedIO extends CMPPThread
{
	//错误记数
	public int m_nErrorCount = 0;
	//用户授权
	public boolean m_bAuthenticated = false;
	//记数器
	public int m_nSequenceID = 0;
	//关闭
	public boolean m_bTerminate = false;
	
	//超时设置
	static long m_lTimeout = 90; //million seconds
	
	//同步IO
	CMPPSynchronizedIO m_io = null;
	
	//双向异步数据环
	CMPPPacketCircle m_pcIn = null;//输入环
	CMPPPacketCircle m_pcOut = null;//输出环
	//输入输出流
	CMPPInputStream m_cmppis = null;//输入数据流
	CMPPOutputStream m_cmppos = null;//输出数据流
	
	public CMPPAsynchronizedIO(CMPPSynchronizedIO io)
		throws IOException, CMPPException
	{
		if(io == null)
		{
			throw new CMPPException("CMPPAsynchronizedIO.CMPPAsynchronizedIO : null synchronized io !");
		}
		m_io = io;
		
		m_pcIn = new CMPPPacketCircle();
		if(m_pcIn == null)
		{
			throw new CMPPException("CMPPAsynchronizedIO.CMPPAsynchronizedIO : cannot build packet circle in !");
		}
		
		m_pcOut = new CMPPPacketCircle();
		if(m_pcOut == null)
		{
			throw new CMPPException("CMPPAsynchronizedIO.CMPPAsynchronizedIO : cannot build packet circle out !");
		}
		
		m_cmppis = new CMPPInputStream(io,m_pcIn);
		if(m_cmppis == null)
		{
			throw new CMPPException("CMPPAsynchronizedIO.CMPPAsynchronizedIO : cannot build cmpp input stream !");
		}
		
		m_cmppos = new CMPPOutputStream(io,m_pcOut);
		if(m_cmppos == null)
		{
			throw new CMPPException("CMPPAsynchronizedIO.CMPPAsynchronizedIO : cannot build cmpp output stream !");
		}
	}

	public static void setTimeout(long lTimeout)
	{
		m_lTimeout = lTimeout;
	}
	
	public boolean write(CMPPPacket packet)
		throws InterruptedException
	{
		long lStartTime = System.currentTimeMillis();

		while(!m_pcOut.isWritable())
		{
			Thread.sleep(15);
			
			if((System.currentTimeMillis() - lStartTime) > m_lTimeout)
			//写超时
			{
				return false;
			}
		}
		
		return m_pcOut.write(packet);
	}
	
	public CMPPPacket read()
		throws InterruptedException
	{
		long lStartTime = System.currentTimeMillis();
		
		while(!m_pcIn.isReadable())
		{
			Thread.sleep(15);

			if((System.currentTimeMillis() - lStartTime) > m_lTimeout)
			//读超时
			{
				return null;
			}
		}

		return m_pcIn.read();
	}
	
	public CMPPPacket peer()
		throws InterruptedException
	{
		long lStartTime = System.currentTimeMillis();
		
		while(!m_pcIn.isReadable())
		{
			Thread.sleep(15);

			if((System.currentTimeMillis() - lStartTime) > m_lTimeout)
			//查超时
			{
				return null;
			}
		}
		
		return m_pcIn.peer();
	}
	
	public void run()
	{
		CMPPPacket packet = null;
		
		try
		{	
			m_nStatus = THREAD_RUNNING;
			while(isRunning())
			{
				//输入流异常
				if(!m_cmppis.isRunning())
				{
					throw new CMPPException("CMPPAsynchronizedIO.run : input stream exit unexpectedly !");
				}
				
				//输出流异常
				if(!m_cmppos.isRunning())
				{
					throw new CMPPException("CMPPAsynchronizedIO.run : output stream exit unexpectedly !");
				}

				//输入环被阻
				if(!m_pcIn.isWritable())
				{
					packet = m_pcIn.read();
					
					if(packet != null)
					{
						//Lost风险
						CMPPLog.log("CMPPAsynchronizedIO.run : packet lost in input stream !",
							LogRequest.LOG_EXCEPTION | CMPPLog.LOG_CMPP_ASYNCHRONIZED_IO);
					}
				}
				
				//输出环被阻
				if(!m_pcOut.isWritable())
				{
					packet = m_pcOut.read();
					
					if(packet != null)
					{
						//Leak风险
						CMPPLog.log("CMPPAsynchronizedIO.run : packet leaked in output stream !",
							LogRequest.LOG_EXCEPTION | CMPPLog.LOG_CMPP_ASYNCHRONIZED_IO);
					}
				}
				
				Thread.sleep(m_lTimeout);
			}
		}
		catch(Exception e)
		{
			CMPPLog.log(e.getMessage(),
				LogRequest.LOG_EXCEPTION | CMPPLog.LOG_CMPP_ASYNCHRONIZED_IO);
			CMPPLog.log("CMPPAsynchronizedIO.run : unexpected exit !",
				LogRequest.LOG_EXCEPTION | CMPPLog.LOG_CMPP_ASYNCHRONIZED_IO);
		}
		
		m_nStatus = THREAD_STOPPED;
		CMPPLog.log("CMPPAsynchronizedIO.run : thread stopped !",
			CMPPLog.LOG_CMPP_ASYNCHRONIZED_IO);
	}
	
	public void startup()
	{
		//启动输出数据环
		m_cmppos.startup();
		//启动输入数据环
		m_cmppis.startup();
		
		//启动守护线程
		super.startup();

		CMPPLog.log("CMPPAsynchronizedIO.startup : asynchronized io startup !",
			LogRequest.LOG_EXCEPTION | CMPPLog.LOG_CMPP_ASYNCHRONIZED_IO);
	}
	
	public void shutdown()
	{
		//停止守护线程
		super.shutdown();
		
		//停止输入环
		m_cmppis.shutdown();
		//停止输出环
		m_cmppos.shutdown();

		CMPPLog.log("CMPPAsynchronizedIO.shutdown : asynchronized io shutdown !",
			LogRequest.LOG_EXCEPTION | CMPPLog.LOG_CMPP_ASYNCHRONIZED_IO);
	}
}