windy/msgexchange-api
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

提交整体消息转换分发处理框架

Browse Source
This commit is contained in:
zhouxiunai
2018-12-03 10:44:14 +08:00
parent 48d37b2bf7
commit 021de12235
9 changed files with 267 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/dao/FlightSchdInfoDao.java → src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/dao/FlightInfoDao.java
+2 -2
View File
@@ -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<FlightSchdInfo, String>{
public interface FlightInfoDao extends CrudRepository<FlightInfo, String>{
}
src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/entity/FlightSchdInfo.java → src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/entity/FlightInfo.java
+1 -1
View File
@@ -11,7 +11,7 @@ import javax.persistence.Table;
*/
@Entity
@Table(name="flightschdinfo")
public class FlightSchdInfo {
public class FlightInfo {
@Id
private String flightId;//航班id,用于索引
src/main/java/com/gzzn/omms/msgexchangeapi/msghandler/MsgHandlerDispatcher.java
+27
View File
@@ -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<FlightInfo> dispatch(String xmlMsg)
{
Msg msg = exchangeService.xmlstrToObject(xmlMsg, Msg.class);
String type = msg.getMeta().getType();
String subType = msg.getMeta().getStyp();
return Collections.emptyList();
}
}
src/main/java/com/gzzn/omms/msgexchangeapi/service/CminmsgServiceImpl.java
+22 -1
View File
@@ -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<Cminmsg> getRespSchdCminmsg(Date date) {
List<Cminmsg> lsCminmsgs = getNewMsgsAfterDate(date);
Optional<Cminmsg> 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
}
src/main/java/com/gzzn/omms/msgexchangeapi/service/FlightInfoServiceImpl.java
+61
View File
@@ -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<FlightInfo> 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<FlightInfo> flightInfos = (List<FlightInfo>) flightInfoDao.save(flightSchdInfo);
return flightInfos != null;
} //end function
@Override
public List<FlightInfo> updateByCminmsgs(List<Cminmsg> lsCminmsgs) {
MsgHandlerDispatcher msgHandlerDispatcher = new MsgHandlerDispatcher();
for (Cminmsg cminmsg : lsCminmsgs) {
//更新动态航班消息
String strClobMsg = cminmsg.getCminmsgsClobMsg();
msgHandlerDispatcher.dispatch(strClobMsg);
}
return null;
}//end function
}
src/main/java/com/gzzn/omms/msgexchangeapi/service/ICminmsgService.java
+9
View File
@@ -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<Cminmsg> getRespSchdCminmsg(Date date);
/**
* 获取指定时间后的新消息
* @return
src/main/java/com/gzzn/omms/msgexchangeapi/service/IFlightInfoService.java
+20
View File
@@ -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<FlightInfo> updateByCminmsgs(List<Cminmsg> lsCminmsgs);
}
src/main/java/com/gzzn/omms/msgexchangeapi/task/ExchangeTask.java
+23 -44
View File
@@ -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<Cminmsg> lsCminmsgs = cminmsgService.getNewMsgsAfterDate(date);
Optional<Cminmsg> 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<Cminmsg> opCminmsg = cminmsgService.getRespSchdCminmsg(date);
if(opCminmsg.isPresent())
{
//如果找到，添加到内存数据库动态航班信息表
String clobMsg = opCminmsg.get().getCminmsgsClobMsg();
DnldMsg dnldMsg = exchangeService.xmlstrToObject(clobMsg, DnldMsg.class);
List<FlightSchdInfo> 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<Cminmsg> 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<Cminmsg> lsCminmsgs = cminmsgService.getNewMsgsAfterId(beginId);
List<FlightInfo> lsUpdatedFlightInfo = flightInfoService.updateByCminmsgs(lsCminmsgs);
//新增或变更 航班动态信息 发送 到kafka的SCHD topic
for (Cminmsg cminmsg : lsCminmsgs) {
for (FlightInfo flightInfo : lsUpdatedFlightInfo) {
//更新动态航班消息
}
@@ -130,6 +106,9 @@ public class ExchangeTask {
//更新动态航班消息
}
//更新消息处理状态
logger.info("同步完成");
}
}
src/test/java/com/gzzn/omms/msgexchangeapi/tools/ToolTest.java
+102 -3
View File
@@ -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<Cminmsg> lsCmin = (List<Cminmsg>) cminmsgDao.findAll();
Map<String, Cminmsg> 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<Cminmsg> lsCmin = (List<Cminmsg>) cminmsgDao.findAll();
toJsonFiles(lsCmin);
//提取某个航班的信息
List<Cminmsg> 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<Cminmsg> lsCmin)
@@ -126,5 +145,85 @@ public class ToolTest {
} catch (IOException e1) {
throw new RuntimeException(e1);
}
} //end function
private void flopMsgToXmlFiles(List<Cminmsg> 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();
}
}