dbreader.java

短信发送

/**
 * Created at Nov 30, 2008
 */
package com.jdev.app.db;

import java.io.InputStream;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.util.Properties;

import com.jdev.net.event.Notifier;
import com.jdev.app.db.queue.SmsQueue;
import com.jdev.app.db.queue.SmsQueueFactory;
import com.jdev.db.ConnectPool;
import com.jdev.net.connector.Connection;
import com.jdev.net.connector.ConnectionFactory;
import com.jdev.net.event.EventAdapter;
import com.jdev.net.queue.Request;
import com.jdev.net.queue.Response;
import com.jdev.util.Debug;


/**
 * <p>Title: DbReader</p>
 * <p>Description: </p>
 * @author Lawrence
 * @version 1.0
 */
public class DbReader extends EventAdapter implements Runnable {

	private SmsQueueHelper smsDataHelper;
	private FidsQueueHelper fidsDataHelper;
	private EmailQueueHelper emailDataHelper;

	public final static int DEF = 0;
	public final static int SMS = 1;
	public final static int EMAIL = 2;
	public final static int FIDS = 3;
	private Connection server;

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.jdev.net.event.EventAdapter#onRead(com.jdev.net.queue.Request)
	 */
	@Override
	public void onRead(Request request) throws Exception {
		// TODO Auto-generated method stub
//		super.onRead(request);

		String recv = new String(request.getDataInput());
		int len = recv.toUpperCase().indexOf("FIDS");
		int len2 = recv.toUpperCase().indexOf("SMS");
		int len3 = recv.toUpperCase().indexOf("EMAIL");
		if (len >= 0) {
			request.setDataType(FIDS);
		} else if (len2 >= 0) {
			request.setDataType(SMS);
		} else if (len3 >= 0) {
			request.setDataType(EMAIL);
		}
		// if(recv.equalsIgnoreCase("fids")) {
		// request.setDataType(FIDS);
		// } else if (recv.equalsIgnoreCase("sms")) {
		// request.setDataType(SMS);
		// } else if (recv.equalsIgnoreCase("email")) {
		// request.setDataType(EMAIL);
		// }

	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.jdev.net.event.EventAdapter#onWrite(com.jdev.net.queue.Request,
	 *      com.jdev.net.queue.Response)
	 */
	@Override
	public void onWrite(Request request, Response response) throws Exception {
		if(request == null)return;
		int type = request.getDataType();
		byte[] bout = null;
		if (type == SMS) {
			bout = smsDataHelper.readSend().getBytes();
		} else if (type == EMAIL) {
			bout = emailDataHelper.readSend().getBytes();
		} else if (type == FIDS) {
			bout = fidsDataHelper.readSend().getBytes();
		}

		if (bout != null)
			response.setDataInput(bout);
		else
			response.setDataInput("OK".getBytes());
//		super.onWrite(request, response);
	}

	private java.sql.Connection conn;
	private ConnectPool connectionManager;
	private String sms;
	private String fids;
	private String email;
	private ConnectionFactory connFactory = ConnectionFactory
			.getInstance(ConnectionFactory.TCPSERVER);

	/**
	 * 
	 */
	public DbReader() {
		// 获取事件触发器
		notifier = Notifier.getNotifier();
		connectionManager = ConnectPool.getInstance();
		smsDataHelper = new SmsQueueHelper(ConnectionFactory.TCPSERVER);
		fidsDataHelper = new FidsQueueHelper(ConnectionFactory.TCPSERVER);
		emailDataHelper = new EmailQueueHelper(ConnectionFactory.TCPSERVER);
		server = connFactory.getConnection();
	}

	private final static String module = DbReader.class.getName();
	protected Notifier notifier;

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.lang.Runnable#run()
	 */
	public void run() {

		try {
			this.openDb("oracle.jdbc.driver.OracleDriver",
					"jdbc:oracle:thin:@192.168.168.121:1521:GWTWOC", "gwtexpo",
					"gwtexpo");
			
		} catch (Exception e) {
			notifier.fireOnError("-->Error occured in DbReader openDb: "
					+ e.getMessage());
		}
		// 监听
		while (!Thread.interrupted()) {
			try {
				Thread.sleep(1000);
				if (conn == null || conn.isClosed()) {
					Thread.sleep(10000);
					try {
						this
								.openDb(
										"oracle.jdbc.driver.OracleDriver",
										"jdbc:oracle:thin:@192.168.168.121:1521:GWTWOC",
										"gwtexpo", "gwtexpo");

					} catch (Exception e) {
						notifier.fireOnError("-->Error occured in DbReader openDb: "
								+ e.getMessage());
					}

					continue;
				}
				// 3. 获取表达式
				java.sql.Statement stmt = conn.createStatement();
				ResultSet rs = null;
				int i = 0;
				if (sms.length() > 0) {
					StringBuilder strUpdate = new StringBuilder(
							"update TB_SMS_Receiver set RECEIVESTATE='1' where CLIENTINFOID in (");
					// 读取sms
					rs = stmt.executeQuery(sms);
					while (rs.next()) {
						i++;
						StringBuilder str = new StringBuilder(rs.getString(4)
								+ "," + rs.getString(7) + "," + rs.getString(9));
						strUpdate.append(rs.getString(7) + ",");

//						smsDataHelper.putSend(str.toString().getBytes());
						smsDataHelper.writeSend(str.toString());
					}
					if(i>0) {
						int len = strUpdate.length();
						strUpdate.delete(len - 1, len);
						strUpdate.append(")");
						Debug.logVerbose(strUpdate.toString());
						len = stmt.executeUpdate(strUpdate.toString());
					}
				}

				if (fids.length() > 0) {
					// 读取fids
					StringBuilder strUpdate = new StringBuilder(
							"update TB_SMS_Receiver set RECEIVESTATE='1' where CLIENTINFOID in (");
					rs = stmt.executeQuery(fids);
					while (rs.next()) {
						i++;
						StringBuilder str = new StringBuilder(rs.getString(4)
								+ "," + rs.getString(7) + "," + rs.getString(9));
						strUpdate.append(rs.getString(7) + ",");

//						fidsDataHelper.putSend(str.toString().getBytes());
						fidsDataHelper.writeSend(str.toString());
					}
					if(i>0) {
						int len = strUpdate.length();
						strUpdate.delete(len - 1, len);
						strUpdate.append(")");
						Debug.logVerbose(strUpdate.toString());
						len = stmt.executeUpdate(strUpdate.toString());
					}
				}

				if (email.length() > 0) {
					// 读取fids
					StringBuilder strUpdate = new StringBuilder(
							"update TB_SMS_Receiver set RECEIVESTATE='1' where CLIENTINFOID in (");
					rs = stmt.executeQuery(email);
					while (rs.next()) {
						i++;
						StringBuilder str = new StringBuilder(rs.getString(4)
								+ "," + rs.getString(7) + "," + rs.getString(9));
						strUpdate.append(rs.getString(7) + ",");

						emailDataHelper.writeSend(str.toString());
					}
					if(i>0) {
						int len = strUpdate.length();
						strUpdate.delete(len - 1, len);
						strUpdate.append(")");
						Debug.logVerbose(strUpdate.toString());
						len = stmt.executeUpdate(strUpdate.toString());
					}
				}
				
				// 6. 释放资源
				rs.close();
				stmt.close();

			} catch (Exception e) {
				notifier.fireOnError("-->Error occured in DbReader: "
						+ e.getMessage());
				continue;

			}
		}
	}

	public void openDb(String DriverName, String connString, String userName,
			String password) throws Exception {
		// // 1. 注册驱动
		// try {
		// Class.forName(DriverName);
		// } catch (ClassNotFoundException e) {
		// e.printStackTrace();
		// }// Mysql 的驱动
		//		
		// try {
		// // 2. 获取数据库的连接
		// conn = DriverManager.getConnection(
		// connString,userName,password);
		//			
		// } catch (Exception e) {
		// Debug.logError(e, module);
		// throw new Exception(e);
		// }

		conn = connectionManager.getConnection("test");
		InputStream is = getClass().getClassLoader().getResourceAsStream(
				"db.properties");
		Properties dbProps = new Properties();
		try {
			dbProps.load(is);
		} catch (Exception e) {
			Debug.logError("不能读取属性文件. " + "请确保db.properties在CLASSPATH指定的路径中");
		}
		sms = dbProps
				.getProperty(
						"sms",
						"SELECT * FROM MV_SMS_Receiver where receivestate = '0' and id < (select min(id) from mv_sms_receiver where receivestate = '0')+10");
		fids = dbProps.getProperty("fids", "");
		email = dbProps.getProperty("email", "SELECT * FROM MV_SMS_Receiver where receivestate = '0' and id < (select min(id) from mv_sms_receiver where receivestate = '0')+10");
	}

	public void close() {
		try {

		} catch (Exception e) {
			Debug.logError(e, module);
		}
	}

	public static void main(String[] args) throws Exception {
		try {
//			ReadMailBody r = new ReadMailBody();
//			Thread rs = new Thread(r);
//			rs.start();

			DbReader d = new DbReader();
			Notifier notifier = Notifier.getNotifier();
			notifier.addListener(d);
			Thread s = new Thread(d);
			s.start();

		} catch (Exception e) {
			throw new Exception(e);
		}
	}

}