From 021de12235638f8fd0fec7c4c815fce8785ebda8 Mon Sep 17 00:00:00 2001 From: zhouxiunai <154707516@qq.com> Date: Mon, 3 Dec 2018 10:44:14 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8F=90=E4=BA=A4=E6=95=B4=E4=BD=93=E6=B6=88?= =?UTF-8?q?=E6=81=AF=E8=BD=AC=E6=8D=A2=E5=88=86=E5=8F=91=E5=A4=84=E7=90=86?= =?UTF-8?q?=E6=A1=86=E6=9E=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...ghtSchdInfoDao.java => FlightInfoDao.java} | 4 +- .../{FlightSchdInfo.java => FlightInfo.java} | 2 +- .../msghandler/MsgHandlerDispatcher.java | 27 +++++ .../service/CminmsgServiceImpl.java | 23 +++- .../service/FlightInfoServiceImpl.java | 61 ++++++++++ .../service/ICminmsgService.java | 9 ++ .../service/IFlightInfoService.java | 20 ++++ .../msgexchangeapi/task/ExchangeTask.java | 67 ++++------- .../omms/msgexchangeapi/tools/ToolTest.java | 105 +++++++++++++++++- 9 files changed, 267 insertions(+), 51 deletions(-) rename src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/dao/{FlightSchdInfoDao.java => FlightInfoDao.java} (64%) rename src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/entity/{FlightSchdInfo.java => FlightInfo.java} (90%) create mode 100644 src/main/java/com/gzzn/omms/msgexchangeapi/msghandler/MsgHandlerDispatcher.java create mode 100644 src/main/java/com/gzzn/omms/msgexchangeapi/service/FlightInfoServiceImpl.java create mode 100644 src/main/java/com/gzzn/omms/msgexchangeapi/service/IFlightInfoService.java diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/dao/FlightSchdInfoDao.java b/src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/dao/FlightInfoDao.java similarity index 64% rename from src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/dao/FlightSchdInfoDao.java rename to src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/dao/FlightInfoDao.java index e81c5c3c..656a97db 100644 --- a/src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/dao/FlightSchdInfoDao.java +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/dao/FlightInfoDao.java @@ -2,8 +2,8 @@ package com.gzzn.omms.msgexchangeapi.domain.secondary.dao; import org.springframework.data.repository.CrudRepository; -import com.gzzn.omms.msgexchangeapi.domain.secondary.entity.FlightSchdInfo; +import com.gzzn.omms.msgexchangeapi.domain.secondary.entity.FlightInfo; -public interface FlightSchdInfoDao extends CrudRepository{ +public interface FlightInfoDao extends CrudRepository{ } diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/entity/FlightSchdInfo.java b/src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/entity/FlightInfo.java similarity index 90% rename from src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/entity/FlightSchdInfo.java rename to src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/entity/FlightInfo.java index c7376790..fa8a90da 100644 --- a/src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/entity/FlightSchdInfo.java +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/entity/FlightInfo.java @@ -11,7 +11,7 @@ import javax.persistence.Table; */ @Entity @Table(name="flightschdinfo") -public class FlightSchdInfo { +public class FlightInfo { @Id private String flightId;//航班id,用于索引 diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/msghandler/MsgHandlerDispatcher.java b/src/main/java/com/gzzn/omms/msgexchangeapi/msghandler/MsgHandlerDispatcher.java new file mode 100644 index 00000000..ebe4840d --- /dev/null +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/msghandler/MsgHandlerDispatcher.java @@ -0,0 +1,27 @@ +package com.gzzn.omms.msgexchangeapi.msghandler; + +import java.util.Collections; +import java.util.List; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Service; + +import com.gzzn.omms.msgexchangeapi.domain.primary.entity.msg.Msg; +import com.gzzn.omms.msgexchangeapi.domain.secondary.entity.FlightInfo; +import com.gzzn.omms.msgexchangeapi.service.IExchangeService; + +@Service +public class MsgHandlerDispatcher { + @Autowired + IExchangeService exchangeService; + + public List dispatch(String xmlMsg) + { + + Msg msg = exchangeService.xmlstrToObject(xmlMsg, Msg.class); + String type = msg.getMeta().getType(); + String subType = msg.getMeta().getStyp(); + + return Collections.emptyList(); + } +} diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/service/CminmsgServiceImpl.java b/src/main/java/com/gzzn/omms/msgexchangeapi/service/CminmsgServiceImpl.java index 911ed69f..2bab9c7c 100644 --- a/src/main/java/com/gzzn/omms/msgexchangeapi/service/CminmsgServiceImpl.java +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/service/CminmsgServiceImpl.java @@ -3,19 +3,23 @@ package com.gzzn.omms.msgexchangeapi.service; import java.util.Collections; import java.util.Date; import java.util.List; +import java.util.Optional; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import com.gzzn.omms.msgexchangeapi.domain.primary.dao.CminmsgDao; import com.gzzn.omms.msgexchangeapi.domain.primary.entity.Cminmsg; +import com.gzzn.omms.msgexchangeapi.domain.primary.entity.msg.Msg; @Service public class CminmsgServiceImpl implements ICminmsgService { @Autowired private CminmsgDao cminmsgDao; - + @Autowired + IExchangeService exchangeService; + @Override public Cminmsg sendXmlMsg(String xmlMsg) { Cminmsg cminmsg = new Cminmsg(); @@ -49,4 +53,21 @@ public class CminmsgServiceImpl implements ICminmsgService { return lsCminmsgs; } + + + @Override + public Optional getRespSchdCminmsg(Date date) { + List lsCminmsgs = getNewMsgsAfterDate(date); + Optional opCminmsg = lsCminmsgs + .stream() + .filter(x->{ + String msgBlob = x.getCminmsgsClobMsg(); + Msg msg = exchangeService.xmlstrToObject(msgBlob, Msg.class); + String type = msg.getMeta().getType(); + String subType = msg.getMeta().getStyp(); + return type.equals("SCHD") && subType.equals("RESP"); + }) + .findFirst(); + return opCminmsg; + }//end function } diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/service/FlightInfoServiceImpl.java b/src/main/java/com/gzzn/omms/msgexchangeapi/service/FlightInfoServiceImpl.java new file mode 100644 index 00000000..39695cb5 --- /dev/null +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/service/FlightInfoServiceImpl.java @@ -0,0 +1,61 @@ +package com.gzzn.omms.msgexchangeapi.service; + +import java.util.List; +import java.util.stream.Collectors; + +import org.springframework.beans.factory.annotation.Autowired; + +import com.gzzn.omms.msgexchangeapi.domain.primary.entity.Cminmsg; +import com.gzzn.omms.msgexchangeapi.domain.primary.entity.msg.Msg; +import com.gzzn.omms.msgexchangeapi.domain.primary.entity.msg.schd.dnld.DnldMsg; +import com.gzzn.omms.msgexchangeapi.domain.secondary.dao.FlightInfoDao; +import com.gzzn.omms.msgexchangeapi.domain.secondary.entity.FlightInfo; +import com.gzzn.omms.msgexchangeapi.msghandler.MsgHandlerDispatcher; +import com.gzzn.omms.msgexchangeapi.utils.JsonUtil; + +public class FlightInfoServiceImpl implements IFlightInfoService { + @Autowired + private FlightInfoDao flightInfoDao; + + @Autowired + IExchangeService exchangeService; + + @Autowired + MsgHandlerDispatcher msgHandlerDispatcher; + + @Override + public Boolean updateByDaySchd(Cminmsg cminmsg) { + String clobMsg = cminmsg.getCminmsgsClobMsg(); + DnldMsg dnldMsg = exchangeService.xmlstrToObject(clobMsg, DnldMsg.class); + + List flightSchdInfo = dnldMsg + .getSchd() + .getFltr() + .stream() + .map(x->{ + FlightInfo node = new FlightInfo(); + node.setFlightId(x.getFlid()); + node.setContext(JsonUtil.getString(x)); + return node; + }).collect(Collectors.toList()); + + List flightInfos = (List) flightInfoDao.save(flightSchdInfo); + return flightInfos != null; + } //end function + + + @Override + public List updateByCminmsgs(List lsCminmsgs) { + MsgHandlerDispatcher msgHandlerDispatcher = new MsgHandlerDispatcher(); + + for (Cminmsg cminmsg : lsCminmsgs) { + //更新动态航班消息 + String strClobMsg = cminmsg.getCminmsgsClobMsg(); + msgHandlerDispatcher.dispatch(strClobMsg); + + + } + + return null; + }//end function +} diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/service/ICminmsgService.java b/src/main/java/com/gzzn/omms/msgexchangeapi/service/ICminmsgService.java index 8a9a5aac..bf71b70c 100644 --- a/src/main/java/com/gzzn/omms/msgexchangeapi/service/ICminmsgService.java +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/service/ICminmsgService.java @@ -2,6 +2,7 @@ package com.gzzn.omms.msgexchangeapi.service; import java.util.Date; import java.util.List; +import java.util.Optional; import com.gzzn.omms.msgexchangeapi.domain.primary.entity.Cminmsg; @@ -14,6 +15,14 @@ public interface ICminmsgService { */ public Cminmsg sendXmlMsg(String xmlMsg); + /** + * 获取回复的航班计划信息 + * @param date + * @return + */ + public Optional getRespSchdCminmsg(Date date); + + /** * 获取指定时间后的新消息 * @return diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/service/IFlightInfoService.java b/src/main/java/com/gzzn/omms/msgexchangeapi/service/IFlightInfoService.java new file mode 100644 index 00000000..e448eed9 --- /dev/null +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/service/IFlightInfoService.java @@ -0,0 +1,20 @@ +package com.gzzn.omms.msgexchangeapi.service; + +import java.util.List; + +import com.gzzn.omms.msgexchangeapi.domain.primary.entity.Cminmsg; +import com.gzzn.omms.msgexchangeapi.domain.secondary.entity.FlightInfo; + +public interface IFlightInfoService { + /** + * 同步日计划更新航班信息 + */ + public Boolean updateByDaySchd(Cminmsg cminmsg); + + /** + * 根据消息更新航班信息 + * @param lsCminmsgs + * @return 返回被更新了的动态航班信息 + */ + public List updateByCminmsgs(List lsCminmsgs); +} diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/task/ExchangeTask.java b/src/main/java/com/gzzn/omms/msgexchangeapi/task/ExchangeTask.java index 6240efb2..497e3ee6 100644 --- a/src/main/java/com/gzzn/omms/msgexchangeapi/task/ExchangeTask.java +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/task/ExchangeTask.java @@ -3,26 +3,21 @@ package com.gzzn.omms.msgexchangeapi.task; import java.util.Date; import java.util.List; import java.util.Optional; -import java.util.stream.Collectors; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import com.gzzn.omms.msgexchangeapi.domain.primary.entity.Cminmsg; -import com.gzzn.omms.msgexchangeapi.domain.primary.entity.msg.Msg; -import com.gzzn.omms.msgexchangeapi.domain.primary.entity.msg.schd.dnld.DnldMsg; import com.gzzn.omms.msgexchangeapi.domain.secondary.dao.ContextDao; -import com.gzzn.omms.msgexchangeapi.domain.secondary.dao.FlightSchdInfoDao; import com.gzzn.omms.msgexchangeapi.domain.secondary.entity.Context; -import com.gzzn.omms.msgexchangeapi.domain.secondary.entity.FlightSchdInfo; +import com.gzzn.omms.msgexchangeapi.domain.secondary.entity.FlightInfo; import com.gzzn.omms.msgexchangeapi.service.ICminmsgService; import com.gzzn.omms.msgexchangeapi.service.IExchangeService; +import com.gzzn.omms.msgexchangeapi.service.IFlightInfoService; import com.gzzn.omms.msgexchangeapi.service.IKafkaService; import com.gzzn.omms.msgexchangeapi.utils.DateTimeUtil; -import com.gzzn.omms.msgexchangeapi.utils.JsonUtil; /** @@ -47,7 +42,7 @@ public class ExchangeTask { private ContextDao contextDao; @Autowired - private FlightSchdInfoDao flightSchdInfoDao; + private IFlightInfoService flightInfoService; //@Scheduled(cron="0 0/1 * * * ?") public void corn() @@ -58,39 +53,19 @@ public class ExchangeTask { Context msgProgress = contextDao.findOne(Context.KEY_MSGPROGRESS); Long beginId = null; - while (isWaitSchd.getValue().equalsIgnoreCase("true")) { + Integer tryTimes = 0; + while (isWaitSchd.getValue().equalsIgnoreCase("true") && tryTimes < 10) { + tryTimes++; + //查找回复的日计划消息 String strDate = msgProgress.getValue(); Date date = DateTimeUtil.toDate(strDate, "yyyy-MM-dd hh:mm:ss"); - List lsCminmsgs = cminmsgService.getNewMsgsAfterDate(date); - Optional opCminmsg = lsCminmsgs - .stream() - .filter(x->{ - String msgBlob = x.getCminmsgsClobMsg(); - Msg msg = exchangeService.xmlstrToObject(msgBlob, Msg.class); - String type = msg.getMeta().getType(); - String subType = msg.getMeta().getStyp(); - return type.equals("SCHD") && subType.equals("RESP"); - }) - .findFirst(); + Optional opCminmsg = cminmsgService.getRespSchdCminmsg(date); if(opCminmsg.isPresent()) { //如果找到，添加到内存数据库动态航班信息表 - String clobMsg = opCminmsg.get().getCminmsgsClobMsg(); - DnldMsg dnldMsg = exchangeService.xmlstrToObject(clobMsg, DnldMsg.class); - List flightSchdInfo = dnldMsg - .getSchd() - .getFltr() - .stream() - .map(x->{ - FlightSchdInfo node = new FlightSchdInfo(); - node.setFlightId(x.getFlid()); - node.setContext(JsonUtil.getString(x)); - return node; - }).collect(Collectors.toList()); - - flightSchdInfoDao.save(flightSchdInfo); + flightInfoService.updateByDaySchd(opCminmsg.get()); // beginId = opCminmsg.get().getCminmsgsId(); @@ -98,6 +73,8 @@ public class ExchangeTask { //更新状态 isWaitSchd.setValue("false"); contextDao.save(isWaitSchd); + + break; //跳出循环 } else { //等待xx秒继续循环过程 @@ -109,19 +86,18 @@ public class ExchangeTask { } } // end while 获取航班日记录 - //获取航班动态消息 - List lsCminmsgs = cminmsgService.getNewMsgsAfterId(beginId); - for (Cminmsg cminmsg : lsCminmsgs) { - //更新动态航班消息 - String strClobMsg = cminmsg.getCminmsgsClobMsg(); - Msg msg = exchangeService.xmlstrToObject(strClobMsg, Msg.class); - String type = msg.getMeta().getType(); - String subType = msg.getMeta().getStyp(); - + if(beginId == null) + { + //获取返回的航班计划回复失败,发送报警信息退出整个程序 + return ; } + //获取航班动态消息 + List lsCminmsgs = cminmsgService.getNewMsgsAfterId(beginId); + List lsUpdatedFlightInfo = flightInfoService.updateByCminmsgs(lsCminmsgs); + //新增或变更 航班动态信息 发送 到kafka的SCHD topic - for (Cminmsg cminmsg : lsCminmsgs) { + for (FlightInfo flightInfo : lsUpdatedFlightInfo) { //更新动态航班消息 } @@ -130,6 +106,9 @@ public class ExchangeTask { //更新动态航班消息 } + + //更新消息处理状态 + logger.info("同步完成"); } } diff --git a/src/test/java/com/gzzn/omms/msgexchangeapi/tools/ToolTest.java b/src/test/java/com/gzzn/omms/msgexchangeapi/tools/ToolTest.java index 7bce973c..b6764b8a 100644 --- a/src/test/java/com/gzzn/omms/msgexchangeapi/tools/ToolTest.java +++ b/src/test/java/com/gzzn/omms/msgexchangeapi/tools/ToolTest.java @@ -4,9 +4,15 @@ import java.io.File; import java.io.FileOutputStream; import java.io.IOException; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; +import java.util.Iterator; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.stream.Collector; +import java.util.stream.Collectors; import org.junit.Test; import org.junit.runner.RunWith; @@ -15,6 +21,7 @@ import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.util.StringUtils; import com.fasterxml.jackson.core.JsonParseException; import com.fasterxml.jackson.databind.JsonMappingException; @@ -46,6 +53,7 @@ public class ToolTest { List lsCmin = (List) cminmsgDao.findAll(); Map mapCminmsg = new HashMap(); + //每种类型提取一个 for(Cminmsg x : lsCmin) { Msg msg = exchangeService.xmlstrToObject(x.getCminmsgsClobMsg(),Msg.class); @@ -62,16 +70,27 @@ public class ToolTest { * @throws JsonMappingException * @throws IOException */ + @Test public void exportToXmlFiles() throws JsonParseException, JsonMappingException, IOException { List lsCmin = (List) cminmsgDao.findAll(); - - toJsonFiles(lsCmin); + + //提取某个航班的信息 + List lsFlopTypeCmin = lsCmin.parallelStream() + .filter(node->{ + Msg msg = exchangeService.xmlstrToObject(node.getCminmsgsClobMsg(), Msg.class); + return msg.getMeta().getType().equals("FLOP"); + }) + .collect(Collectors.toList()); + + + + flopMsgToXmlFiles(lsFlopTypeCmin); } //end function /** - * 生成xml文件 + * 生成json文件 * @param lsCmin */ private void toJsonFiles(List lsCmin) @@ -126,5 +145,85 @@ public class ToolTest { } catch (IOException e1) { throw new RuntimeException(e1); } + } //end function + + + private void flopMsgToXmlFiles(List lsCmin) + { + FileOutputStream fop = null; + try { + for(Cminmsg x : lsCmin) + { + Msg msg = exchangeService.xmlstrToObject(x.getCminmsgsClobMsg(), Msg.class); + Map msgMap = exchangeService.xmlstrToObject(x.getCminmsgsClobMsg(), Map.class); + + + Map mapFlop = (Map)msgMap.get("FLOP"); + String flid = String.valueOf(mapFlop.get("FLID")); + if(null == flid || flid.equalsIgnoreCase("null") || flid.equals("")) { + logger.warn("没有flid的消息:"+x.getCminmsgsClobMsg()); + continue; + } + + + String type = "xml"; + String path = getPathGroupByFlid(msg,type,flid); + + //父目录是否存在 + File fileDir = new File(path); + if(fileDir.getParentFile().exists() == false) + { + fileDir.getParentFile().mkdirs(); + } + + // + byte[] contentInBytes = x.getCminmsgsClobMsg().getBytes("UTF-8"); + + File f = new File(path.toString()); + fop = new FileOutputStream(f); + fop.write(contentInBytes); + fop.flush(); + fop.close(); + }//end for + } + catch (Exception e) { + logger.error(""+JsonUtil.getString(e)); + } + } //end function + + + private String getPath(Msg msg,String type) + { + String dir = type; + + StringBuilder path = new StringBuilder(); + path.append(dir); + path.append("/"); + path.append(msg.getMeta().getType());//Type + path.append("-"+msg.getMeta().getStyp());//Subtype + path.append("."+dir); + + return path.toString(); } + + + private String getPathGroupByFlid(Msg msg,String type,String flid) + { + String dir = type; + + StringBuilder path = new StringBuilder(); + path.append(dir); + path.append("/"); + path.append(flid); + path.append("/"); + path.append(msg.getMeta().getDttm());//dttm + path.append("-"+msg.getMeta().getType());//Type + path.append("-"+msg.getMeta().getStyp());//Subtype + path.append("."+dir); + + return path.toString(); + } + + + }