fix recv msg

liuxiaolong

2019-11-04 5a8bbcb100f37b8e17d6fda0d851d96dbace46e4

fix recv msg

 src/main/java/com/cloud/count/listener/BeanContext.java

@@ -1,4 +1,29 @@
package com.cloud.count.listener;

public class BeanContext {
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

@Component
public class BeanContext implements ApplicationContextAware {

    private static ApplicationContext applicationContext;

    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        BeanContext.applicationContext = applicationContext;
    }

    public static ApplicationContext getApplicationContext(){
        return applicationContext;
    }

    @SuppressWarnings("unchecked")
    public static <T> T getBean(String name) throws BeansException {
        return (T)applicationContext.getBean(name);
    }

    public static <T> T getBean(Class<T> clz) throws BeansException {
        return applicationContext.getBean(clz);
    }
}

 src/main/java/com/cloud/count/listener/CountListener.java

@@ -10,6 +10,8 @@
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Service;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
@@ -21,15 +23,9 @@
 */

public class CountListener implements ApplicationListener<ContextRefreshedEvent> {
    private Logger logger = LoggerFactory.getLogger(this.getClass());
    @Autowired
    private CountDao dao;

    static final String CRLF = System.getProperty("line.separator");
    static ServerSocket server;

    static Integer port = 2010;
    static boolean isServerReboot = false;

    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {

@@ -46,157 +42,17 @@
                catch (IOException e){
                    e.printStackTrace();
                }
                String encoding = "UTF-8";
                BufferedReader iReader = null;
                BufferedWriter iWriter = null;
                StringBuilder sb = null;
                String lineContent = null;

                while (true){
                    Socket client = null;
                    try{
                        client = server.accept();

                        iReader = new BufferedReader(new InputStreamReader(client.getInputStream(), encoding));
                        iWriter = new BufferedWriter(new OutputStreamWriter(client.getOutputStream(), encoding));

                        InputStream is=client.getInputStream();
                        byte[] b=new byte[1024];
                        is.read(b);
                        String str = new String(b);

                        sendAck(iWriter);
                        try{
                            String xmlStr = XmlUtil.toXmlStr(str);

                            // 如果存在TotalEnters。则证明返回的是实时数据的xml
                            People people = null;
                            if(xmlStr.indexOf("TotalEnters")>-1){
                                people = XmlUtil.realTimeXmlToPeople(xmlStr);
                                int inCreEnter = CountServiceImpl.addEnter(people.getEnters());//进入人数增量
                                int inCreExit = CountServiceImpl.addExit(people.getExits());//出去人数增量
                                System.out.println("TotalEnters:"+people.getEnters());
                                CountServiceImpl.totalEnters += inCreEnter;
                                CountServiceImpl.totalExits += inCreExit;


                                people.setEnters(CountServiceImpl.totalEnters);//将累加后的数量存到数据库中
                                people.setExits(CountServiceImpl.totalExits);
                                boolean resultB = dao.savePeople(people);

                                //总进入数量或者总出去数量，不一定哪一个先到2字节int的上限65535
//                                if( CountServiceImpl.totalEnters < CountServiceImpl.baseEnters){
//                                    //表示统计设备重新启动了,统计参数归零
//                                    CountServiceImpl.initCountArgs(people);
//                                }
                               op: if(!isServerReboot){
                                    isServerReboot = true;
                                    CountState state = dao.getCountState();//恢复统计状态
                                    Date nowDate = new Date();
                                    Date BaseDate = new Date(state.getBaseTime()*1000L);
                                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                                    if(!sdf.format(nowDate).equals(sdf.format(BaseDate))){//历史数据不等于当天
                                        CountServiceImpl.baseEnters  = people.getEnters();
                                        CountServiceImpl.baseExits = people.getExits();
                                        CountServiceImpl.realtimeBaseEnters = people.getEnters();
                                        CountServiceImpl.realtimebBaseExits =  people.getExits();
                                        CountServiceImpl.errorCount = 0;
                                        break op;
                                    }
                                    //还原状态
                                    CountServiceImpl.baseEnters  = state.getBaseEnters();
                                    CountServiceImpl.baseExits = state.getBaseExits();
                                    CountServiceImpl.totalEnters = state.getTotalEnters();
                                    CountServiceImpl.totalExits = state.getTotalExits();
                                    CountServiceImpl.realtimeBaseEnters = state.getRealtimeBaseEnters();
                                    CountServiceImpl.realtimebBaseExits = state.getRealtimeBaseExits();
                                    CountServiceImpl.errorCount = state.getErrorCount();
                                }

                                //总进入人数小于总出去的人数，表示统计设备在放学时间（出去的人数大于进入的人数）重启了
                                if(CountServiceImpl.totalEnters < CountServiceImpl.totalExits){
                                    CountServiceImpl.errorCount = CountServiceImpl.totalExits - CountServiceImpl.totalEnters;
                                }else
                                    CountServiceImpl.errorCount = 0;

                                CountState currentState = new CountState();
                                currentState.setBaseEnters(CountServiceImpl.baseEnters);
                                currentState.setBaseExits(CountServiceImpl.baseExits);
                                currentState.setTotalEnters(CountServiceImpl.totalEnters);
                                currentState.setTotalExits(CountServiceImpl.totalExits);
                                currentState.setRealtimeBaseEnters(CountServiceImpl.realtimeBaseEnters);
                                currentState.setRealtimeBaseExits(CountServiceImpl.realtimebBaseExits);
                                currentState.setErrorCount(CountServiceImpl.errorCount);
                                currentState.setBaseTime(people.getTimestamp());
                                dao.updateCountState(currentState);//更新当前统计状态
                            } else {
                                System.out.println("noTotalEners");
                            }
                            // 如果存在ReportData。则证明返回的是统计数据的xml
                            if(xmlStr.indexOf("ReportData")>-1){
                                people = XmlUtil.xmlToPeople(xmlStr);
                                dao.savePeople(people);
                            }
                        }catch (Exception e){
                            System.out.println(e.getMessage());
                        }

                        new Thread(new CountThread(client)).start();
                    }catch (IOException ex){

                    }finally {
                        try {
                            if(iWriter !=null)
                                iWriter.close();
                            if(iReader !=null)
                                iReader.close();
//                            if(client !=null)
//                                client.close();

                        }catch (IOException e){

                        }
                    }
                }
            }
        }).start();
    }

    public static void sendAck(BufferedWriter iWriter) {
        String statusLine = "HTTP/1.0 200 OK"+CRLF;
        String contentTypeLine = "Content-type: text/html"+CRLF ;
        String connectionClose = "Connection: Keep-Alive"+CRLF ;
        String contentLengthLine = "Content-Length: 0"+CRLF +CRLF;;

        try {
            iWriter.write(statusLine);
            iWriter.write(contentTypeLine);
            iWriter.write(connectionClose);
            iWriter.write(contentLengthLine);
            //iWriter.write(content);
            iWriter.flush();
            iWriter.close();
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }

    // Send nack
    public static void sendNack(BufferedWriter iWriter) {
        String statusLine = "HTTP/1.1 400 Bad Request" + CRLF;
        String contentTypeLine = "Content-type: text/html" + CRLF;
        String connectionClose = "Connection: close" + CRLF;
        String content = "<ErrorList><Error>error</Error></ErrorList>";
        String contentLengthLine = "Content-Length: "
                + (new Integer(content.length())).toString() + CRLF + CRLF;

        try {
            iWriter.write(statusLine);
            iWriter.write(contentTypeLine);
            iWriter.write(connectionClose);
            iWriter.write(contentLengthLine);
            iWriter.write(content);
            iWriter.flush();
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }
}

 src/main/java/com/cloud/count/listener/CountThread.java

@@ -1,4 +1,172 @@
package com.cloud.count.listener;

public class CountThread {
import com.cloud.count.dao.CountDao;
import com.cloud.count.model.CountState;
import com.cloud.count.model.People;
import com.cloud.count.service.serviceImpl.CountServiceImpl;
import com.cloud.count.utils.XmlUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.*;
import java.net.Socket;
import java.text.SimpleDateFormat;
import java.util.Date;

public class CountThread implements Runnable {
    private Logger logger = LoggerFactory.getLogger(this.getClass());
    static boolean isServerReboot = false;

    private CountDao dao;

    static final String CRLF = System.getProperty("line.separator");

    private Socket client;

    public CountThread(Socket socket) {
        this.dao = BeanContext.getBean("countDao");
        this.client = socket;
    }

    String encoding = "UTF-8";
    BufferedReader iReader = null;
    BufferedWriter iWriter = null;

    @Override
    public void run() {
        try{
            iReader = new BufferedReader(new InputStreamReader(client.getInputStream(), encoding));
            iWriter = new BufferedWriter(new OutputStreamWriter(client.getOutputStream(), encoding));

            while (true) {
                String str = iReader.readLine();
                sendAck(iWriter);

                String xmlStr = XmlUtil.toXmlStr(str);

                // 如果存在TotalEnters。则证明返回的是实时数据的xml
                People people = null;
                if(xmlStr.indexOf("TotalEnters")>-1){
                    people = XmlUtil.realTimeXmlToPeople(xmlStr);
                    int inCreEnter = CountServiceImpl.addEnter(people.getEnters());//进入人数增量
                    int inCreExit = CountServiceImpl.addExit(people.getExits());//出去人数增量
                    CountServiceImpl.totalEnters += inCreEnter;
                    CountServiceImpl.totalExits += inCreExit;

                    people.setEnters(CountServiceImpl.totalEnters);//将累加后的数量存到数据库中
                    people.setExits(CountServiceImpl.totalExits);
                    boolean resultB = dao.savePeople(people);
                    System.out.println("totalEnters:"+CountServiceImpl.totalEnters);

                    //总进入数量或者总出去数量，不一定哪一个先到2字节int的上限65535
//                                if( CountServiceImpl.totalEnters < CountServiceImpl.baseEnters){
//                                    //表示统计设备重新启动了,统计参数归零
//                                    CountServiceImpl.initCountArgs(people);
//                                }
                    op: if(!isServerReboot){
                        isServerReboot = true;
                        CountState state = dao.getCountState();//恢复统计状态
                        Date nowDate = new Date();
                        Date BaseDate = new Date(state.getBaseTime()*1000L);
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                        if(!sdf.format(nowDate).equals(sdf.format(BaseDate))){//历史数据不等于当天
                            CountServiceImpl.baseEnters  = people.getEnters();
                            CountServiceImpl.baseExits = people.getExits();
                            CountServiceImpl.realtimeBaseEnters = people.getEnters();
                            CountServiceImpl.realtimebBaseExits =  people.getExits();
                            CountServiceImpl.errorCount = 0;
                            break op;
                        }
                        //还原状态
                        CountServiceImpl.baseEnters  = state.getBaseEnters();
                        CountServiceImpl.baseExits = state.getBaseExits();
                        CountServiceImpl.totalEnters = state.getTotalEnters();
                        CountServiceImpl.totalExits = state.getTotalExits();
                        CountServiceImpl.realtimeBaseEnters = state.getRealtimeBaseEnters();
                        CountServiceImpl.realtimebBaseExits = state.getRealtimeBaseExits();
                        CountServiceImpl.errorCount = state.getErrorCount();
                    }

                    //总进入人数小于总出去的人数，表示统计设备在放学时间（出去的人数大于进入的人数）重启了
                    if(CountServiceImpl.totalEnters < CountServiceImpl.totalExits){
                        CountServiceImpl.errorCount = CountServiceImpl.totalExits - CountServiceImpl.totalEnters;
                    }else
                        CountServiceImpl.errorCount = 0;

                    CountState currentState = new CountState();
                    currentState.setBaseEnters(CountServiceImpl.baseEnters);
                    currentState.setBaseExits(CountServiceImpl.baseExits);
                    currentState.setTotalEnters(CountServiceImpl.totalEnters);
                    currentState.setTotalExits(CountServiceImpl.totalExits);
                    currentState.setRealtimeBaseEnters(CountServiceImpl.realtimeBaseEnters);
                    currentState.setRealtimeBaseExits(CountServiceImpl.realtimebBaseExits);
                    currentState.setErrorCount(CountServiceImpl.errorCount);
                    currentState.setBaseTime(people.getTimestamp());
                    dao.updateCountState(currentState);//更新当前统计状态
                } else {
                    System.out.println("noTotalEners");
                }
                // 如果存在ReportData。则证明返回的是统计数据的xml
                if(xmlStr.indexOf("ReportData")>-1){
                    people = XmlUtil.xmlToPeople(xmlStr);
                    dao.savePeople(people);
                }
            }

        }catch (IOException ex){

        }catch (Exception e){
            System.out.println("exception:"+e.getMessage());
        }
        finally {
            try {
                if(iWriter !=null)
                    iWriter.close();
                if(iReader !=null)
                    iReader.close();
                if(client !=null)
                    client.close();

            }catch (IOException e){
                System.out.println("finally exception:"+e.getMessage());
            }
        }
    }

    public static void sendAck(BufferedWriter iWriter) {
        String statusLine = "HTTP/1.0 200 OK"+CRLF;
        String contentTypeLine = "Content-type: text/html"+CRLF ;
        String connectionClose = "Connection: Keep-Alive"+CRLF ;
        String contentLengthLine = "Content-Length: 0"+CRLF +CRLF;;

        try {
            iWriter.write(statusLine);
            iWriter.write(contentTypeLine);
            iWriter.write(connectionClose);
            iWriter.write(contentLengthLine);
            iWriter.flush();
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }

    // Send nack
    public static void sendNack(BufferedWriter iWriter) {
        String statusLine = "HTTP/1.1 400 Bad Request" + CRLF;
        String contentTypeLine = "Content-type: text/html" + CRLF;
        String connectionClose = "Connection: close" + CRLF;
        String content = "<ErrorList><Error>error</Error></ErrorList>";
        String contentLengthLine = "Content-Length: "
                + (new Integer(content.length())).toString() + CRLF + CRLF;

        try {
            iWriter.write(statusLine);
            iWriter.write(contentTypeLine);
            iWriter.write(connectionClose);
            iWriter.write(contentLengthLine);
            iWriter.write(content);
            iWriter.flush();
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }
}