新增航班动态信息处理

src/main/java/com/gzzn/omms/msgexchangeapi/domain/primary/entity/msg/schd/dnld/DnldMsg.java

@@ -3,7 +3,17 @@ package com.gzzn.omms.msgexchangeapi.domain.primary.entity.msg.schd.dnld;
 import com.gzzn.omms.msgexchangeapi.domain.primary.entity.msg.Msg;
 
 public class DnldMsg extends Msg {
-
+    public static final String TYPE = "SCHD";
+    public static final String SUBTYPE = "DNLD";
+	
 	private DnldMsgBody schd;
+
+	public DnldMsgBody getSchd() {
+		return schd;
+	}
+
+	public void setSchd(DnldMsgBody schd) {
+		this.schd = schd;
+	}
 	
 }

src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/dao/FlightSchdInfoDao.java

@@ -0,0 +1,9 @@
+package com.gzzn.omms.msgexchangeapi.domain.secondary.dao;
+
+import org.springframework.data.repository.CrudRepository;
+
+import com.gzzn.omms.msgexchangeapi.domain.secondary.entity.FlightSchdInfo;
+
+public interface FlightSchdInfoDao extends CrudRepository<FlightSchdInfo, String>{
+
+}

src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/entity/FlightSchdInfo.java

@@ -0,0 +1,34 @@
+package com.gzzn.omms.msgexchangeapi.domain.secondary.entity;
+
+import javax.persistence.Entity;
+import javax.persistence.Id;
+import javax.persistence.Table;
+
+/**
+ * 航班计划信息
+ * @author 
+ *
+ */
+@Entity
+@Table(name="flightschdinfo")
+public class FlightSchdInfo {
+    
+	@Id
+   	private String flightId;//航班id,用于索引
+	
+	private String context;//内容
+
+	
+	public String getFlightId() {
+		return flightId;
+	}
+	public void setFlightId(String flightId) {
+		this.flightId = flightId;
+	}
+	public String getContext() {
+		return context;
+	}
+	public void setContext(String context) {
+		this.context = context;
+	}
+}

src/main/java/com/gzzn/omms/msgexchangeapi/service/ExchangeServiceImpl.java

@@ -22,9 +22,10 @@ public class ExchangeServiceImpl implements IExchangeService {
 			Map map;
 			ObjectMapper xmlMapper = new XmlMapper();
 			map = xmlMapper.readValue(xml, Map.class);
+			map.remove("noNamespaceSchemaLocation");
 			
 			ObjectMapper jsonMapper = new ObjectMapper();
-			String json = jsonMapper.writeValueAsString(map);
+			String json = jsonMapper.writer().withDefaultPrettyPrinter().writeValueAsString(map);
 			return json;
 		} catch (JsonParseException e) {
 			throw new ExchangeServiceException(e.getMessage(),e);

src/main/java/com/gzzn/omms/msgexchangeapi/task/ExchangeTask.java

@@ -3,6 +3,7 @@ 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;
@@ -11,12 +12,17 @@ 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.service.ICminmsgService;
 import com.gzzn.omms.msgexchangeapi.service.IExchangeService;
 import com.gzzn.omms.msgexchangeapi.service.IKafkaService;
 import com.gzzn.omms.msgexchangeapi.utils.DateTimeUtil;
+import com.gzzn.omms.msgexchangeapi.utils.JsonUtil;
 
 
 /**
@@ -39,8 +45,11 @@ public class ExchangeTask {
 	 
 	 @Autowired
 	 private ContextDao contextDao;
+	 
+	 @Autowired
+	 private FlightSchdInfoDao flightSchdInfoDao;
 	  
-	@Scheduled(cron="0 0/1 * * * ?")
+	//@Scheduled(cron="0 0/1 * * * ?")
 	public void corn()
 	{
 		logger.info("定时任务启动....");
@@ -48,29 +57,79 @@ public class ExchangeTask {
 		Context isWaitSchd = contextDao.findOne(Context.KEY_ISWAITSCHD);
 		Context msgProgress = contextDao.findOne(Context.KEY_MSGPROGRESS);
 		
-		
+		Long beginId = null;
 		while (isWaitSchd.getValue().equalsIgnoreCase("true")) {
-			
 			//查找回复的日计划消息
 			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().findFirst();
+		    Optional<Cminmsg>  opCminmsg = lsCminmsgs
-		    
+		    		                       .stream()
-		    //如果找到，添加到内存数据库动态航班信息表
+		    		                       .filter(x->{
-		    
+		    		                    	   String msgBlob = x.getCminmsgsClobMsg();
-		    //更新状态
+		    		                    	   Msg msg = exchangeService.xmlstrToObject(msgBlob, Msg.class);
-			isWaitSchd.setValue("false");
+		    		                    	   String type = msg.getMeta().getType();
-			contextDao.save(isWaitSchd);
+		    		                    	   String subType = msg.getMeta().getStyp();
+		    		                    	   return type.equals("SCHD") && subType.equals("RESP");
+		    		                       })
+		    		                       .findFirst();
+		    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);
+		    	
+		    	//
+		    	beginId = opCminmsg.get().getCminmsgsId();
+		    	
+			    //更新状态
+				isWaitSchd.setValue("false");
+				contextDao.save(isWaitSchd);
+		    }
+		    else {
+				//等待xx秒继续循环过程
+		    	try {
+					Thread.sleep(3000);
+				} catch (InterruptedException e) {
+					logger.warn("Thread.sleep:暂停线程异常");
+				}
+			}
+		} // 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();
+			
 		}
 		
+		//新增或变更 航班动态信息  发送 到kafka的SCHD topic
+		for (Cminmsg cminmsg : lsCminmsgs) {
+			//更新动态航班消息
+		}
 		
+		//发送动态消息到kafka
+		for (Cminmsg cminmsg : lsCminmsgs) {
+			//更新动态航班消息
+		}
 		
-		
-		
-		//
 		logger.info("同步完成");
 	}
 }

src/main/java/com/gzzn/omms/msgexchangeapi/utils/JsonUtil.java

@@ -9,14 +9,18 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.ObjectWriter;
 
 public class JsonUtil {
-  public static String getString(Object object) throws JsonProcessingException
+  public static String getString(Object object) 
   {
-	  ObjectMapper mapper = new ObjectMapper();
+	try {
-      ObjectWriter ow = mapper.writer().withDefaultPrettyPrinter();
+		ObjectMapper mapper = new ObjectMapper();
-      java.lang.String jsonStr = ow.writeValueAsString(object);
+	    ObjectWriter ow = mapper.writer().withDefaultPrettyPrinter();
-      
+	    java.lang.String jsonStr;
-      return jsonStr;
+		jsonStr = ow.writeValueAsString(object);
-  }
+	    return jsonStr;
+	} catch (JsonProcessingException e) {
+		throw new RuntimeException(e);
+	}
+  } //end function
   
   
   public static <T>  T getObject(String str,Class<T> valueType) throws JsonParseException, JsonMappingException, IOException

src/test/java/com/gzzn/omms/msgexchangeapi/tools/ToolTest.java

@@ -3,7 +3,10 @@ package com.gzzn.omms.msgexchangeapi.tools;
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -35,56 +38,93 @@ public class ToolTest {
 	@Autowired
 	private IExchangeService exchangeService;
 	
+	/**
+	 * 每种消息类型导出一个xml文件即可
+	 */
 	@Test
-	public void test() {
+	public void exportOneTypeToJsonFiles() {
+		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);
+			mapCminmsg.putIfAbsent(msg.getMeta().getType() + "_" + msg.getMeta().getStyp(), x);
+		}
+		
+		List<Cminmsg> ls = new ArrayList(mapCminmsg.values());
+		toJsonFiles(ls);
 	}
 	
-	@Test
+	/**
+	 * 导出CMINMSGS数据中的消息到单独的Xml文件
+	 * @throws JsonParseException
+	 * @throws JsonMappingException
+	 * @throws IOException
+	 */
 	public void exportToXmlFiles() throws JsonParseException, JsonMappingException, IOException
 	{
 		List<Cminmsg> lsCmin = (List<Cminmsg>) cminmsgDao.findAll();
 		 
-		for(Cminmsg x : lsCmin)
+		toJsonFiles(lsCmin);
-		{
-			String json = exchangeService.xmlToJson(x.getCminmsgsClobMsg());
-			Msg msg =JsonUtil.getObject(json,Msg.class);
-			String Dttm =  msg.getMeta().getDttm();
-			
-			String dir = "xml/" + Dttm.substring(0, 10);
-			
-			StringBuilder path = new StringBuilder();
-			path.append(dir);
-			path.append("/");
-			path.append(msg.getMeta().getDttm().substring(10, Dttm.length()));//time
-			path.append("-"+x.getCminmsgsId());
-			path.append("-"+msg.getMeta().getType());//Type
-			path.append("-"+msg.getMeta().getStyp());//Subtype
-			path.append(".xml");
-			
-			//父目录是否存在
-			File fileDir = new File(dir);
-			if(fileDir.exists() == false)
-			{
-				fileDir.mkdirs();
-			}
-			
-			//
-			File f = new File(path.toString());
-			try (FileOutputStream fop = new FileOutputStream(f)) {
-				byte[] contentInBytes = x.getCminmsgsClobMsg().getBytes("UTF-8");
-
-				fop.write(contentInBytes);
-				fop.flush();
-				fop.close();
-			}
-			catch (ExchangeServiceException e) {
-				  logger.error(e.getMessage());
-		    }
-			 catch (IOException e) {
-				logger.error(e.getMessage());
-			} 
-		}
 			
 	} //end function
+	
+	/**
+	 * 生成xml文件
+	 * @param lsCmin
+	 */
+	private void toJsonFiles(List<Cminmsg> lsCmin)
+	{
+		try {
+			for(Cminmsg x : lsCmin)
+			{
+				String json = exchangeService.xmlToJson(x.getCminmsgsClobMsg());
+				Msg msg;
+			
+				msg = JsonUtil.getObject(json,Msg.class);
+				String Dttm =  msg.getMeta().getDttm();
+				
+				String dir = "json";
+				
+				StringBuilder path = new StringBuilder();
+				path.append(dir);
+				path.append("/");
+				path.append(msg.getMeta().getType());//Type
+				path.append("-"+msg.getMeta().getStyp());//Subtype
+				path.append(".json");
+				
+				//父目录是否存在
+				File fileDir = new File(dir);
+				if(fileDir.exists() == false)
+				{
+					fileDir.mkdirs();
+				}
+				
+				//
+				File f = new File(path.toString());
+				try (FileOutputStream fop = new FileOutputStream(f)) {
+					String jsonString = exchangeService.xmlToJson(x.getCminmsgsClobMsg());
+					jsonString = jsonString.replace("\"noNamespaceSchemaLocation\":\"unisysaodbsis.xsd\",", "");
+					byte[] contentInBytes = jsonString.getBytes("UTF-8");
+	
+					fop.write(contentInBytes);
+					fop.flush();
+					fop.close();
+				}
+				catch (ExchangeServiceException e) {
+					  logger.error(e.getMessage());
+			    }
+				 catch (IOException e) {
+					logger.error(e.getMessage());
+				} 
+			}
+		} catch (JsonParseException e1) {
+			throw new RuntimeException(e1.getMessage());
+		} catch (JsonMappingException e1) {
+			throw new RuntimeException(e1);
+		} catch (IOException e1) {
+			throw new RuntimeException(e1);
+		}
+	}
 }