smsloadsvc.java

短信

package com.asiainfo.batchsend.sms.sms.db;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

import org.apache.log4j.Logger;

import com.asiainfo.batchsend.sms.boss.DBCon;
import com.asiainfo.batchsend.sms.sms.MessageInfo;
import com.asiainfo.databusi.util.DataBusiDbUtils;

/**
 * <p>Title: </p>
 * <p>Description: </p>
 * <p>Company: Asiainfo Technologies(China),Inc </p>
 * <p>Date: Nov 6, 2007 3:55:51 PM </p>
 * <p>Email to: [e-mail] </p>
 * @author jiangyl
 * @version 1.0
 */

public class SmsLoadSvc {
	private static final Logger log = Logger.getLogger(SmsLoadSvc.class);
	public static final int TODO = 0;
	public static final int DONE = 1;
	public static final int DOING = 3;
	public static final int FAIL = 2;
	private static String strDBType = DataBusiDbUtils.getDBType("back");
	
	private String taskId;
	private String channelId;
	private String sql = " select id, camp_id, obj_id, msisdn, send_msg, status, start_date,end_date from sms_push where status = ? and task_id = ? ";
	private String udpate = " update sms_push t set t.status = ? , send_count = send_count + 1 where t.id = ? ";

	
	public void setTaskId(String taskId) {
		this.taskId = taskId;
	}

	public SmsLoadSvc(String channelId){
		this.channelId = channelId;
		if (strDBType.equalsIgnoreCase("oracle")) {
			sql += " and rownum <  101";
		}else if(strDBType.equalsIgnoreCase("db2")) {
			sql += " fetch first 100 rows only ";
		}
	}
	
	
	//这里只根据任务提取数据，主要是因为根据调度任务传进来的taskId已经具备发送的条件

	
	public List loadData() {
		synchronized( channelId ){
			Connection conn = null;
			List list = new ArrayList();
			try {
				conn = DBCon.getShenYunDataSource().getConnection();
				conn.setAutoCommit(false);
				PreparedStatement ps = conn.prepareStatement(sql);
				PreparedStatement ups = conn.prepareStatement(udpate);
				ps.setInt(1, TODO);
				ps.setString(2, taskId);
				ResultSet rs = ps.executeQuery();
				while(rs.next()){
					MessageInfo sms = new MessageInfo();
					long id = rs.getLong("id");
					sms.setMessageid("" + id);
					sms.setPhonenumber( rs.getString("msisdn").trim());
					sms.setMessage( rs.getString("send_msg").trim());   
					list.add(sms);
					
					ups.setInt(1, DONE);
					ups.setLong(2, id);
					ups.addBatch();
				}
				ups.executeBatch();
				ps.close();
				ups.close();
				conn.commit();
				log.debug("深运平台从sms_push表在装载数据时获取" +  list.size() + "条记录,更新" + list.size() +"条记录");
			} catch (Exception e) {
				try{
					conn.rollback();
				}catch(Exception ee){
				}
				log.error("深运平台在装载数据时发生严重的错误",e);
			} finally {
				try {
					conn.close();
				} catch (SQLException e) {
				}
			}
			return list;
		}
	}
	
    //发送失败后需要如何处理数据
	public void failDone(List list, int sendLimit) {
		Connection conn = null;
		String update = "update sms_push t set t.status = (case when send_count >= " + sendLimit + " then 2 " +
		  			 " when send_count < " + sendLimit  + " then 0 end )  where t.id = ?  ";
		try {
			conn = DBCon.getShenYunDataSource().getConnection();
			conn.setAutoCommit(false);
			PreparedStatement ps = conn.prepareStatement(update);
			for (  int i = 0; i < list.size(); i++ ){
				ps.setLong(1, Long.parseLong(( (MessageInfo)list.get(i)).getMessageid()) );
				ps.addBatch();
			}
			ps.executeBatch();
			conn.commit();
		} catch (SQLException e) {
			try{
				conn.rollback();
			}catch(SQLException se){
			}
			log.error("深运平台在设置sms_push表发送失败状态的时候发生严重的错误",e);
		} finally {
			try {
				conn.close();
			} catch (SQLException e) {
			}
		}
	}
	
	public static boolean saveRecMessage(String src_addr,String message,String channeldId){
		boolean result = true;
		Connection con = null;
		String sql = "insert into sms_received (msisdn,received_msg,status,flag,received_time,camp_channel_id ) "
			+ " values(?,?,0,0,?,?)";
		if ( src_addr == null ){
			return false;
		}
		if(message == null)
			message = "";
		if ( src_addr.length() > 11 )
			src_addr = src_addr.substring(src_addr.length()-11);
		//如果没有找到
		if ( findRecMessage(src_addr,message))
			return true;
		try{ 
			log.info("msisdn,received_msg,camp_channel_id " + src_addr + "," + message + "," + channeldId );
			con = DBCon.getShenYunDataSource().getConnection();
			con.setAutoCommit(true);
			PreparedStatement ps = con.prepareStatement(sql);
			ps.setString(1, src_addr);
			ps.setString(2, message);
			ps.setTimestamp(3, new java.sql.Timestamp(System.currentTimeMillis()));
			ps.setString(4, channeldId);
			ps.executeUpdate();
			ps.close();
		    log.info("深运平台在sms_received表插入一条记录");
		}catch(Exception e){
			log.error("深运平台在设置sms_received表发送失败状态的时候发生严重的错误",e);
			result = false;
		} finally {
			try {
				con.close();
			} catch (SQLException e) {
			}
		}
		return result;
	}
	
	public static boolean findRecMessage(String src_addr,String message){
		boolean result = false;
		Connection con = null;
		String sql = "select msisdn from sms_received where msisdn = ? and received_msg = ? ";
		if (strDBType.equalsIgnoreCase("oracle")) {
			sql += " and to_char(received_time,'yyyy-MM-dd')= ? ";
		}else if(strDBType.equalsIgnoreCase("db2")) {
			sql += " and date(received_time)= ? ";
		}
		if ( src_addr == null ){
			return false;
		}
		if(message == null)
			message = "";
		if ( src_addr.length() > 11 )
			src_addr = src_addr.substring(src_addr.length()-11);
		try{
			con = DBCon.getShenYunDataSource().getConnection();
			con.setAutoCommit(true);
			PreparedStatement ps = con.prepareStatement(sql);
			ps.setString(1, src_addr);
			ps.setString(2, message);
			ps.setString(3, new java.sql.Date(System.currentTimeMillis())+"");
			ResultSet rs = ps.executeQuery();
			if ( rs.next()){
				result = true;
			}
			ps.close();
		    log.info("深运平台在sms_received表插入一条记录");
		}catch(Exception e){
			log.error("深运平台在设置sms_received表发送失败状态的时候发生严重的错误",e);
			result = false;
		} finally {
			try {
				con.close();
			} catch (SQLException e) {
			}
		}
		return result;
	}
}