Upload the seed code to gerrit for the first time.
Change-Id: I2e3bd269d9a0daca63faef838356b5a113bdae9b
Issue-ID: VFC-59
Signed-off-by: ying.yunlong <ying.yunlong@zte.com.cn>
--- /dev/null
+<?xml version='1.0' encoding='GBK'?>
+<!--
+
+ Copyright 2017 BOCO Corporation.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+
+-->
+<collectSources>
+ <ems name="1234">
+ <collect type ="cm" crontab="0 0/1 * * * ?">
+ <ip>192.168.6.48</ip>
+ <port>21</port>
+ <user>gcp</user>
+ <password>gcp</password>
+ <remotepath>/opt/Gcp/data/</remotepath>
+ <match>*</match>
+ <passive>true</passive>
+ <ftptype>ftp</ftptype>
+ <granularity>15</granularity>
+ </collect>
+ <collect type ="pm" crontab="">
+ <ip>192.168.1.16</ip>
+ <port>21</port>
+ <user>gcp</user>
+ <password>gcp</password>
+ <remotepath>/var/vsftp/gcp/ftproot/GD/WX/HW/JS_OMC2/</remotepath>
+ <match>*</match>
+ <passive>true</passive>
+ <ftptype>ftp</ftptype>
+ <granularity>15</granularity>
+ </collect>
+ <collect type ="alarm" iscollect = "false">
+ <ip>127.0.0.1</ip>
+ <port>9997</port>
+ <user>yiyang</user>
+ <password>123456</password>
+ <readtimeout>6</readtimeout>;
+ </collect>
+ </ems>
+</collectSources>
\ No newline at end of file
--- /dev/null
+
+# Copyright 2017 BOCO Corporation.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+msbAddress: 10.74.205.123:80
+defaultName: EmsDriver-Stranger
+
+
--- /dev/null
+
+# Copyright 2017 BOCO Corporation.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+ftp_ip = 192.168.6.48
+ftp_port = 21
+ftp_user = gcp
+ftp_password = gcp
+ftp_remote_path = /opt/Gcp/
+ftp_passive = true
+ftp_type = ftp
+
+#
+msbAddress: 10.74.205.123:80
+dataNotifyUrl:/dataNotify
+alarmUrl:/alarm
\ No newline at end of file
--- /dev/null
+# Copyright 2017 BOCO Corporation.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+log4j.rootCategory=debug,R,E,stdout
+
+#### \u7b2c\u4e00\u4e2a appender\u5c06log\u5199\u5230\u5c4f\u5e55
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=[%p] [%d] [%c] - [%m]%n
+log4j.appender.stdout.Threshold=INFO
+
+#### \u7b2c\u4e8c\u4e2a appender \u5c06\u5168\u90e8log\u5199\u5230\u6587\u4ef6
+log4j.appender.R=org.apache.log4j.RollingFileAppender
+log4j.appender.R.File=../log/monitor.log
+log4j.appender.R.MaxFileSize=20MB
+log4j.appender.R.MaxBackupIndex=50
+log4j.appender.R.layout=org.apache.log4j.PatternLayout
+log4j.appender.R.layout.ConversionPattern=[%p] [%d] [%t] [%c] - [%m]%n
+log4j.appender.R.Threshold=INFO
+
+#### \u7b2c\u4e09\u4e2a appender \u5c06\u9519\u8beflog\u5199\u5230\u6587\u4ef6
+log4j.appender.E=org.apache.log4j.RollingFileAppender
+log4j.appender.E.File=../log/error.log
+log4j.appender.E.MaxFileSize=20MB
+log4j.appender.E.MaxBackupIndex=10
+log4j.appender.E.layout=org.apache.log4j.PatternLayout
+log4j.appender.E.layout.ConversionPattern=[%p] [%d] [%t] [%c] - [%m]%n
+log4j.appender.E.Threshold=ERROR
--- /dev/null
+<?xml version="1.0" encoding="GBK"?>
+<!--
+
+ Copyright 2017 BOCO Corporation.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+
+-->
+<beans xmlns="http://www.springframework.org/schema/beans"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd"
+ default-lazy-init="true">
+
+ <bean id="configurationManager" class="org.openo.nfvo.emsdriver.configmgr.ConfigurationManager">
+ <property name="run" value="true" />
+
+ </bean>
+ <bean id="collectMsgReceiverThread" class="org.openo.nfvo.emsdriver.collector.CollectMsgReceiverThread">
+ <property name="run" value="true" />
+ <property name="thread_max_num" value="100" />
+ </bean>
+
+ <bean id="configurationImp" class="org.openo.nfvo.emsdriver.configmgr.ConfigurationImp">
+ </bean>
+ <bean id="alarmManager" class="org.openo.nfvo.emsdriver.collector.alarm.AlarmManager">
+ <property name="run" value="true" />
+ <property name="configurationInterface" ref="configurationImp" />
+ </bean>
+ <bean id="collectManager" class="org.openo.nfvo.emsdriver.taskscheduler.CollectManager">
+ <property name="run" value="true" />
+ <property name="configurationInterface" ref="configurationImp" />
+ </bean>
+ <bean id="northMessageMgr" class="org.openo.nfvo.emsdriver.northbound.client.NorthMessageMgr">
+ <property name="run" value="true" />
+ <property name="configurationInterface" ref="configurationImp" />
+ </bean>
+
+
+</beans>
--- /dev/null
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Copyright (c) 2017, BOCO.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <parent>
+ <groupId>org.openo.nfvo</groupId>
+ <artifactId>nfvo-root</artifactId>
+ <version>1.1.0-SNAPSHOT</version>
+ <relativePath>../../../../..</relativePath>
+ </parent>
+
+ <modelVersion>4.0.0</modelVersion>
+ <groupId>org.openo.nfvo</groupId>
+ <artifactId>ems-driver</artifactId>
+ <version>1.1.0-SNAPSHOT</version>
+ <packaging>pom</packaging>
+ <name>nfvo/drivers/ems/sems/boco/ems-driver</name>
+ <description>nfvo ems-driver</description>
+ <url>http://maven.apache.org</url>
+ <properties>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ </properties>
+
+ <dependencies>
+
+ <dependency>
+ <groupId>commons-lang</groupId>
+ <artifactId>commons-lang</artifactId>
+ <version>2.6</version>
+ </dependency>
+ <dependency>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ <version>1.2.17</version>
+ </dependency>
+ <dependency>
+ <groupId>com.alibaba</groupId>
+ <artifactId>fastjson</artifactId>
+ <version>1.2.24</version>
+ </dependency>
+ <dependency>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ <version>2.5</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.ant</groupId>
+ <artifactId>ant</artifactId>
+ <version>1.10.1</version>
+ </dependency>
+ <dependency>
+ <groupId>jdom</groupId>
+ <artifactId>jdom</artifactId>
+ <version>1.1</version>
+ </dependency>
+ <dependency>
+ <groupId>commons-net</groupId>
+ <artifactId>commons-net</artifactId>
+ <version>3.6</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-context</artifactId>
+ <version>4.3.6.RELEASE</version>
+ </dependency>
+ <dependency>
+ <groupId>org.quartz-scheduler</groupId>
+ <artifactId>quartz</artifactId>
+ <version>1.7.3</version>
+ </dependency>
+ <dependency>
+ <groupId>io.dropwizard</groupId>
+ <artifactId>dropwizard-core</artifactId>
+ <version>1.0.6</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ <version>4.5.2</version>
+ </dependency>
+ </dependencies>
+</project>
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.openo.nfvo.emsdriver.commons.constant.Constant;
+import org.openo.nfvo.emsdriver.commons.utils.DriverThread;
+import org.openo.nfvo.emsdriver.northbound.service.EmsDriverApplication;
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.support.FileSystemXmlApplicationContext;
+
+
+
+public class EMSDriverApp {
+
+ private static Log log = LogFactory.getLog(DriverThread.class);
+ public static ApplicationContext context = null;
+ static {
+ try {
+ /** spring bean applicationContext **/
+ context = new FileSystemXmlApplicationContext("file:" + Constant.SYS_CFG+ "spring.xml");
+
+ } catch (BeansException e) {
+ log.error("spring.xml is fail ", e);
+ System.exit(1);
+ } catch (Exception e) {
+ log.error("spring.xml is fail", e);
+ System.exit(1);
+ }
+ }
+
+
+ /**
+ * @param args
+ */
+ public static void main(String[] args) {
+ String[] allThreadName = context.getBeanNamesForType(DriverThread.class);
+
+ log.info("worker num :" + allThreadName.length);
+
+ for (String threadName : allThreadName) {
+ DriverThread thread = (DriverThread) context.getBean(threadName);
+ if (thread == null) {
+ log.error(threadName + "Thread start error,system exit");
+ System.exit(1);
+ }
+ thread.setName(threadName);
+ thread.start();
+ }
+
+ try {
+ new EmsDriverApplication().run(args);
+ } catch (Exception e) {
+ log.error("EmsDriverApplication.run is fail", e);
+ System.exit(1);
+ }
+ log.info("the workerThreads start sucess" );
+ }
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.collector;
+
+import org.openo.nfvo.emsdriver.commons.constant.Constant;
+import org.openo.nfvo.emsdriver.commons.model.CollectMsg;
+import org.openo.nfvo.emsdriver.commons.utils.DriverThread;
+import org.openo.nfvo.emsdriver.messagemgr.MessageChannel;
+import org.openo.nfvo.emsdriver.messagemgr.MessageChannelFactory;
+
+public class CollectMsgReceiverThread extends DriverThread{
+
+ private long timeStamp = System.currentTimeMillis();
+
+ private MessageChannel collectChannel;
+
+ private TaskThreadService taskService;
+
+ private int thread_max_num = 100;
+
+
+
+ @Override
+ public void dispose() {
+ collectChannel = MessageChannelFactory.getMessageChannel(Constant.COLLECT_CHANNEL_KEY);
+
+ taskService = TaskThreadService.getInstance(thread_max_num);
+ taskService.start();
+
+ while (isRun()) {
+
+ try {
+ if(System.currentTimeMillis() - timeStamp > Constant.ONEMINUTE){
+ timeStamp = System.currentTimeMillis();
+
+ log.debug("COLLECT_CHANNEL Msg size :"+collectChannel.size());
+ }
+
+ Object obj = collectChannel.poll();
+ if(obj == null){
+ continue;
+ }
+ if(obj != null && obj instanceof CollectMsg){
+ CollectMsg collectMsg = (CollectMsg)obj;
+ taskService.add(collectMsg);
+ log.debug("receive a CollectMsg id = "+collectMsg.getId());
+ }else{
+ log.error("receive Objcet not CollectMsg "+obj);
+ }
+
+ } catch (Exception e) {
+ log.error("dispatch alarm exception",e);
+
+ }
+ }
+
+ }
+
+
+
+ /**
+ * @return the thread_max_num
+ */
+ public int getThread_max_num() {
+ return thread_max_num;
+ }
+
+
+
+ /**
+ * @param thread_max_num the thread_max_num to set
+ */
+ public void setThread_max_num(int thread_max_num) {
+ this.thread_max_num = thread_max_num;
+ }
+
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.collector;
+
+import java.io.BufferedOutputStream;
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.OutputStreamWriter;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Properties;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import javax.xml.stream.XMLInputFactory;
+import javax.xml.stream.XMLStreamConstants;
+import javax.xml.stream.XMLStreamReader;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.openo.nfvo.emsdriver.commons.constant.Constant;
+import org.openo.nfvo.emsdriver.commons.ftp.AFtpRemoteFile;
+import org.openo.nfvo.emsdriver.commons.ftp.FTPInterface;
+import org.openo.nfvo.emsdriver.commons.ftp.FTPSrv;
+import org.openo.nfvo.emsdriver.commons.ftp.SFTPSrv;
+import org.openo.nfvo.emsdriver.commons.model.CollectMsg;
+import org.openo.nfvo.emsdriver.commons.model.CollectVo;
+import org.openo.nfvo.emsdriver.commons.utils.StringUtil;
+import org.openo.nfvo.emsdriver.commons.utils.UnZip;
+import org.openo.nfvo.emsdriver.commons.utils.Zip;
+import org.openo.nfvo.emsdriver.configmgr.ConfigurationImp;
+import org.openo.nfvo.emsdriver.configmgr.ConfigurationInterface;
+import org.openo.nfvo.emsdriver.messagemgr.MessageChannel;
+import org.openo.nfvo.emsdriver.messagemgr.MessageChannelFactory;
+
+
+public class TaskThread implements Runnable{
+
+ public Log log = LogFactory.getLog(TaskThread.class);
+
+ private MessageChannel collectResultChannel;
+
+ private CollectMsg data;
+
+ private ConfigurationInterface configurationInterface = new ConfigurationImp();
+
+ private String localPath = Constant.SYS_DATA_TEMP;
+ private String resultPath = Constant.SYS_DATA_RESULT;
+
+ private SimpleDateFormat dateFormat = new SimpleDateFormat("yyyyMMddHHmmss");
+
+ private SimpleDateFormat dateFormat2 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+
+ public TaskThread(CollectMsg data) {
+ this.data = data;
+ }
+
+ @Override
+ public void run(){
+
+ collectResultChannel = MessageChannelFactory.getMessageChannel(Constant.COLLECT_RESULT_CHANNEL_KEY);
+
+ try {
+ collectMsgHandle(data);
+ } catch (Exception e) {
+ log.error("",e);
+ }
+ }
+
+ private void collectMsgHandle(CollectMsg collectMsg) {
+ String emsName = collectMsg.getEmsName();
+ String type = collectMsg.getType();
+ CollectVo collectVo = configurationInterface.getCollectVoByEmsNameAndType(emsName, type);
+
+ //ftp download
+ List<String> downloadfiles = this.ftpDownload(collectVo);
+ //paser ftp update message send
+ for(String fileName :downloadfiles){
+ this.parseFtpAndSendMessage(fileName,collectVo);
+ }
+ }
+
+ private void parseFtpAndSendMessage(String fileName, CollectVo collectVo) {
+ //
+ List<File> filelist = decompressed(fileName);
+
+ for (File tempfile : filelist) {
+
+ String unfileName = tempfile.getName();
+
+ Pattern pa = Pattern.compile(".*-(.*)-\\w{2}-");
+ Matcher ma = pa.matcher(unfileName);
+ if (!ma.find())
+ continue;
+ String nename = ma.group(1);
+ boolean parseResult = false;
+ if("CM".equalsIgnoreCase(collectVo.getType())){
+ parseResult = processCMXml(tempfile, nename,"CM");
+ }else{
+ parseResult = processPMCsv(tempfile, nename,"PM");
+ }
+
+ if (parseResult){
+ log.info("parser "+tempfile+" sucess");
+ }else {
+ log.info("parser "+tempfile+" fail");
+ }
+
+ }
+ }
+
+ private boolean processPMCsv(File tempfile, String nename,String type) {
+
+ String csvpath = localPath+nename+"/"+type+"/";
+ File csvpathfile = new File(csvpath);
+ if(!csvpathfile.exists()){
+ csvpathfile.mkdirs();
+ }
+ String csvFileName = nename +dateFormat.format(new Date())+ System.nanoTime();
+ String csvpathAndFileName = csvpath+csvFileName;
+ BufferedOutputStream bos = null;
+ FileOutputStream fos = null;
+ try {
+ fos = new FileOutputStream(csvpathAndFileName,false);
+ bos = new BufferedOutputStream(fos, 10240);
+ } catch (FileNotFoundException e1) {
+ log.error("FileNotFoundException "+StringUtil.getStackTrace(e1));
+ }
+
+ FileInputStream brs = null;
+ InputStreamReader isr = null;
+ BufferedReader br = null;
+
+ List<String> columnNames = new ArrayList<String>();
+ List<String> commonValues = new ArrayList<String>();
+ try {
+
+ brs = new FileInputStream(tempfile);
+ isr = new InputStreamReader(brs, Constant.ENCODING_UTF8);
+ br = new BufferedReader(isr);
+ //common field
+ String commonField = br.readLine();
+ String[] fields = commonField.split("|",-1);
+ for(String com : fields){
+ String[] comNameAndValue = com.split("=",2);
+ columnNames.add(comNameAndValue[0].trim());
+ commonValues.add(comNameAndValue[1]);
+ }
+ //column names
+ String columnName = br.readLine();
+ String[] names = columnName.split("|",-1);
+ for(String name : names){
+ columnNames.add(name);
+ }
+
+ String xmlPathAndFileName = this.setColumnNames(nename, columnNames,type);
+
+ String valueLine = "";
+ List<String> valuelist = new ArrayList<String>();
+ int countNum = 0 ;
+ while (br.readLine() != null) {
+
+ if (valueLine.trim().equals("")) {
+ continue;
+ }
+ countNum ++;
+ String [] values = valueLine.split("|",-1);
+
+ valuelist.addAll(commonValues);
+ for(String value : values){
+ valuelist.add(value);
+ }
+ this.appendLine(valuelist, bos);
+
+ valuelist.clear();
+ }
+
+ if(bos != null){
+ bos.close();
+ bos = null;
+ }
+ if(fos != null){
+ fos.close();
+ fos = null;
+ }
+
+ String[] fileKeys = this.createZipFile(csvpathAndFileName,xmlPathAndFileName,nename);
+ //ftp store
+ Properties ftpPro = configurationInterface.getProperties();
+ String ip = ftpPro.getProperty("ftp_ip");
+ String port = ftpPro.getProperty("ftp_port");
+ String ftp_user = ftpPro.getProperty("ftp_user");
+ String ftp_password = ftpPro.getProperty("ftp_password");
+
+ String ftp_passive = ftpPro.getProperty("ftp_passive");
+ String ftp_type = ftpPro.getProperty("ftp_type");
+ String remoteFile = ftpPro.getProperty("ftp_remote_path");
+ this.ftpStore(fileKeys,ip,port,ftp_user,ftp_password,ftp_passive,ftp_type,remoteFile);
+ //create Message
+ String message = this.createMessage(fileKeys[1], ftp_user, ftp_password, ip, port, countNum,nename);
+
+ //set message
+ this.setMessage(message);
+ } catch (IOException e) {
+ log.error("processPMCsv is fail ",e);
+ return false;
+ }finally{
+ try{
+ if (br != null)
+ br.close();
+ if (isr != null)
+ isr.close();
+ if (brs != null)
+ brs.close();
+ if(bos != null){
+ bos.close();
+ }
+
+ if(fos != null){
+ fos.close();
+ }
+ } catch (Exception e){
+ log.error(e);
+ }
+ }
+ return true;
+
+ }
+
+ private boolean processCMXml(File tempfile, String nename, String type) {
+
+ String csvpath = localPath+nename+"/"+type+"/";
+ File csvpathfile = new File(csvpath);
+ if(!csvpathfile.exists()){
+ csvpathfile.mkdirs();
+ }
+ String csvFileName = nename +dateFormat.format(new Date())+ System.nanoTime();
+ String csvpathAndFileName = csvpath+csvFileName+".csv";
+ BufferedOutputStream bos = null;
+ FileOutputStream fos = null;
+ try {
+ fos = new FileOutputStream(csvpathAndFileName,false);
+ bos = new BufferedOutputStream(fos, 10240);
+ } catch (FileNotFoundException e1) {
+ log.error("FileNotFoundException "+StringUtil.getStackTrace(e1));
+ }
+
+ boolean FieldNameFlag = false;
+ boolean FieldValueFlag = false;
+ //line num
+ int countNum = 0;
+ String xmlPathAndFileName = null;
+ String localName = null;
+ String endLocalName = null;
+ String rmUID = null;
+ int index = -1;
+ ArrayList<String> names = new ArrayList<String>();// colname
+ LinkedHashMap<String, String> nameAndValue = new LinkedHashMap<String, String>();
+
+
+ FileInputStream fis = null;
+ InputStreamReader isr = null;
+ XMLStreamReader reader = null;
+ try{
+ fis = new FileInputStream(tempfile);
+ isr = new InputStreamReader(fis, Constant.ENCODING_UTF8);
+ XMLInputFactory fac = XMLInputFactory.newInstance();
+ reader = fac.createXMLStreamReader(isr);
+ int event = -1;
+ boolean setcolum = true;
+ while (reader.hasNext()){
+ try{
+ event = reader.next();
+ switch (event){
+ case XMLStreamConstants.START_ELEMENT:
+ localName = reader.getLocalName();
+ if ("FieldName".equalsIgnoreCase(localName)){
+ FieldNameFlag = true;
+ }
+ if (FieldNameFlag){
+ if ("N".equalsIgnoreCase(localName)){
+ String colName = reader.getElementText().trim();
+ names.add(colName);
+ }
+ }
+ if ("FieldValue".equalsIgnoreCase(localName)){
+ FieldValueFlag = true;
+
+ }
+ if (FieldValueFlag){
+ if(setcolum){
+ xmlPathAndFileName = this.setColumnNames(nename, names,type);
+ setcolum = false;
+ }
+
+ if ("Object".equalsIgnoreCase(localName)){
+ int ac = reader.getAttributeCount();
+ for (int i = 0; i < ac; i++){
+ if ("rmUID".equalsIgnoreCase(reader.getAttributeLocalName(i))){
+ rmUID = reader.getAttributeValue(i).trim();
+ }
+ }
+ nameAndValue.put("rmUID", rmUID);
+ }
+ if ("V".equalsIgnoreCase(localName)) {
+ index = Integer.parseInt(reader
+ .getAttributeValue(0)) - 1;
+ String currentName = names.get(index);
+ String v = reader.getElementText().trim();
+ nameAndValue.put(currentName, v);
+ }
+ }
+ break;
+ case XMLStreamConstants.CHARACTERS:
+ break;
+ case XMLStreamConstants.END_ELEMENT:
+ endLocalName = reader.getLocalName();
+
+ if ("FieldName".equalsIgnoreCase(endLocalName)){
+ FieldNameFlag = false;
+ }
+ if ("FieldValue".equalsIgnoreCase(endLocalName)){
+ FieldValueFlag = false;
+ }
+ if ("Object".equalsIgnoreCase(endLocalName)){
+ countNum ++;
+ this.appendLine(nameAndValue,bos);
+ nameAndValue.clear();
+ }
+ break;
+ }
+ } catch (Exception e)
+ {
+ log.error(""+StringUtil.getStackTrace(e));
+ event = reader.next();
+ }
+ }
+
+
+ if(bos != null){
+ bos.close();
+ bos = null;
+ }
+ if(fos != null){
+ fos.close();
+ fos = null;
+ }
+
+ String[] fileKeys = this.createZipFile(csvpathAndFileName,xmlPathAndFileName,nename);
+ //ftp store
+ Properties ftpPro = configurationInterface.getProperties();
+ String ip = ftpPro.getProperty("ftp_ip");
+ String port = ftpPro.getProperty("ftp_port");
+ String ftp_user = ftpPro.getProperty("ftp_user");
+ String ftp_password = ftpPro.getProperty("ftp_password");
+
+ String ftp_passive = ftpPro.getProperty("ftp_passive");
+ String ftp_type = ftpPro.getProperty("ftp_type");
+ String remoteFile = ftpPro.getProperty("ftp_remote_path");
+ this.ftpStore(fileKeys,ip,port,ftp_user,ftp_password,ftp_passive,ftp_type,remoteFile);
+ //create Message
+ String message = this.createMessage(fileKeys[1], ftp_user, ftp_password, ip, port, countNum,nename);
+
+ //set message
+ this.setMessage(message);
+ } catch (Exception e){
+ log.error(""+StringUtil.getStackTrace(e));
+ return false;
+ } finally{
+ try{
+ if (reader != null){
+ reader.close();
+ }
+ if (isr != null){
+ isr.close();
+ }
+ if (fis != null){
+ fis.close();
+ }
+ if(bos != null){
+ bos.close();
+ }
+
+ if(fos != null){
+ fos.close();
+ }
+ } catch (Exception e){
+ log.error(e);
+ }
+ }
+ return true;
+ }
+
+ private void setMessage(String message) {
+
+ try {
+ collectResultChannel.put(message);
+ } catch (Exception e) {
+ log.error("collectResultChannel.put(message) is error "+StringUtil.getStackTrace(e));
+ }
+ }
+
+ private String createMessage(String zipName,String user,String pwd,String ip, String port,int countNum, String nename) {
+
+ StringBuffer strBuffer = new StringBuffer();
+ strBuffer
+ .append("<?xml version=\"1.0\" encoding=\"UTF-8\"?>"
+ + "<FILE_DATA_READY_UL xmlns:xsi=\" http://www.w3.org/2001/XMLSchema-instance\">"
+ + "<Header SessionID=\"");
+ strBuffer.append("");
+ strBuffer.append("\" LicenceID=\"");
+ strBuffer.append("");
+ strBuffer.append("\" SystemID=\"");
+ strBuffer.append("");
+ strBuffer.append("\" Time=\"");
+ strBuffer.append( dateFormat2.format(new Date()));
+ strBuffer.append("\" PolicyID=\"");
+ strBuffer.append("");
+ strBuffer.append("\"/><Body>");
+ strBuffer.append("<DataCatalog>");
+ strBuffer.append("");
+ strBuffer.append("</DataCatalog><GroupID>");
+ strBuffer.append(nename);
+ strBuffer.append("</GroupID><DataSourceName>");
+ strBuffer.append("");
+ strBuffer.append("</DataSourceName><InstanceID>");
+ strBuffer.append("");
+ strBuffer.append("</InstanceID><FileFormat>");
+ strBuffer.append("csv");
+ strBuffer.append("</FileFormat><CharSet>");
+ strBuffer.append("gbk");
+ strBuffer.append("</CharSet><FieldSeparator>");
+ strBuffer.append("|");
+ strBuffer.append("</FieldSeparator><IsCompressed>");
+ strBuffer.append("true");
+ strBuffer.append("</IsCompressed><StartTime>");
+ strBuffer.append(dateFormat2.format(new Date()));
+ strBuffer.append("</StartTime><EndTime>");
+ strBuffer.append("");
+ strBuffer.append("</EndTime><FileList>");
+ strBuffer.append(zipName);
+ strBuffer.append("</FileList><ConnectionString>");
+ strBuffer.append("ftp://" + user + ":" + pwd + "@" + ip + ":" + port);
+ strBuffer.append("</ConnectionString>");
+ strBuffer.append("<DataCount>");
+ strBuffer.append(countNum);
+ strBuffer.append("</DataCount>");
+
+ strBuffer.append("<FileSize>").append("").append("</FileSize>");
+ strBuffer.append("<DataGranularity>").append("").append("</DataGranularity>");
+
+
+ strBuffer.append("</Body></FILE_DATA_READY_UL>");
+ return strBuffer.toString();
+
+ }
+
+ private void ftpStore(String[] fileKeys, String ip, String port, String ftp_user, String ftp_password,
+ String ftp_passive, String ftp_type, String remoteFile) {
+ String zipFilePath = fileKeys[0];
+
+
+ FTPInterface ftpClient;
+ if("ftp".equalsIgnoreCase(ftp_type)){
+ ftpClient = new FTPSrv();
+ }else{
+ ftpClient = new SFTPSrv();
+ }
+
+ //login
+ try {
+ ftpClient.login(ip, Integer.parseInt(port), ftp_user, ftp_password, "GBK", Boolean.parseBoolean(ftp_passive), 5*60*1000);
+ } catch (Exception e) {
+ log.error("login fail,ip=["+ip+"] port=["+port+"] user=["+ftp_user+"]pwd=["+ftp_password+"]"+StringUtil.getStackTrace(e));
+ return;
+ }
+ ftpClient.store(zipFilePath, remoteFile);
+ log.debug("store ["+zipFilePath+"]to["+remoteFile+"]");
+
+ FileUtils.deleteQuietly(new File(zipFilePath));
+
+
+ }
+
+ private String[] createZipFile(String csvpathAndFileName,String xmlPathAndFileName,String nename) {
+
+ String zipPath = resultPath+nename +dateFormat.format(new Date())+"_"+System.nanoTime();
+
+ File destDir = new File(zipPath);
+ destDir.mkdirs();
+
+ try {
+ FileUtils.copyFileToDirectory(new File(csvpathAndFileName), destDir);
+ FileUtils.copyFileToDirectory(new File(xmlPathAndFileName), destDir);
+ } catch (IOException e) {
+
+ }
+
+ String destFilePath = zipPath + ".zip";
+ try {
+ Zip zip = new Zip(destDir.getAbsolutePath(), destFilePath);
+ zip.setCompressLevel(9);
+ zip.compress();
+
+ FileUtils.deleteDirectory(destDir);
+ } catch (IOException e) {
+ log.error("zip.compress() is fail "+StringUtil.getStackTrace(e));
+ }
+ return new String[] { destFilePath, zipPath + ".zip"};
+ }
+
+
+ private String setColumnNames(String nename, List<String> names,String type) {
+ //write xml
+ String xmlpath = localPath+nename +"/"+type+"/";
+ File xmlpathfile = new File(xmlpath);
+ if(!xmlpathfile.exists()){
+ xmlpathfile.mkdirs();
+ }
+ String xmlFileName = nename +dateFormat.format(new Date())+ System.nanoTime();
+ String fieldLine = "";
+ for (int i = 0; i < names.size(); i++) {
+ String field = "\t<Field>\r\n" + "\t\t<FieldNo>" + i
+ + "</FieldNo>\r\n" + "\t\t<FieldName>"
+ + names.get(i) + "</FieldName>\r\n"
+ + "\t\t<FieldType>" + names.get(i)
+ + "</FieldType>\r\n"
+ + "\t\t<FieldNameOther>" + names.get(i)
+ + "</FieldNameOther>\r\n" +
+ "\t</Field>\r\n";
+ fieldLine = fieldLine + field;
+ }
+
+ String str = "<?xml version=\"1.0\" encoding=\"gbk\"?>\r\n"
+ + "<xml>\r\n" + "<FILE_STRUCTURE>\r\n" + fieldLine
+ + "</FILE_STRUCTURE>\r\n" + "</xml>\r\n";
+ String xmlPathAndFileName = xmlpath+xmlFileName+".xml";
+ try {
+ this.writeDetail(xmlPathAndFileName,str);
+ } catch (Exception e) {
+ log.error("writeDetail is fail ,xmlFileName="+xmlFileName +StringUtil.getStackTrace(e));
+ }
+
+ return xmlPathAndFileName;
+ }
+
+ private void writeDetail(String detailFileName,String str) throws Exception {
+ OutputStreamWriter writer = null;
+ OutputStream readOut = null;
+ try {
+ readOut = new FileOutputStream(new File(detailFileName), false);
+ writer = new OutputStreamWriter(readOut);
+ writer.write(str);
+ writer.flush();
+ } finally {
+
+ if(null != writer){
+ writer.close();
+ }
+ if(readOut != null){
+ readOut.close();
+ }
+
+ }
+
+ }
+
+
+ private void appendLine(LinkedHashMap<String, String> nameAndValue,BufferedOutputStream bos) {
+ StringBuilder lineDatas = new StringBuilder();
+
+ for (String key : nameAndValue.keySet()) {
+ lineDatas.append(nameAndValue.get(key)).append("|");
+ }
+ try {
+ bos.write(lineDatas.toString().getBytes());
+ bos.write("\n".getBytes());
+ } catch (IOException e) {
+ log.error("appendLine error "+StringUtil.getStackTrace(e));
+ }
+ }
+
+ private void appendLine(List<String> values,BufferedOutputStream bos) {
+ StringBuilder lineDatas = new StringBuilder();
+
+ for (String value : values) {
+ lineDatas.append(value).append("|");
+ }
+ try {
+ bos.write(lineDatas.toString().getBytes());
+ bos.write("\n".getBytes());
+ } catch (IOException e) {
+ log.error("appendLine error "+StringUtil.getStackTrace(e));
+ }
+ }
+
+ public List<File> decompressed(String fileName){
+ List<File> filelist = new ArrayList<File>();
+
+ if (fileName.indexOf(".gz") > 1)
+ {
+// decompressFile = deGz(file);
+ } else if (fileName.indexOf(".zip") > 1)
+ {
+ try {
+ File[] files = deZip(new File(fileName));
+ for(File temp :files){
+ filelist.add(temp);
+ }
+ } catch (Exception e) {
+ log.error("decompressed is fail "+StringUtil.getStackTrace(e));
+ }
+ }
+ else {
+ filelist.add(new File(fileName));
+ }
+
+ return filelist;
+ }
+
+ public File[] deZip(File file) throws Exception{
+
+ String regx = "(.*).zip";
+ Pattern p = Pattern.compile(regx);
+ Matcher m = p.matcher(file.getName());
+ if (m.find())
+ {
+ String orgFile = localPath + m.group(1) + "/";
+ UnZip unzip = new UnZip(file.getAbsolutePath(), orgFile);
+ unzip.deCompress();
+ file = new File(orgFile);
+ }
+ File[] files = file.listFiles();
+
+ return files;
+
+ }
+
+ private List<String> ftpDownload(CollectVo collectVo) {
+
+ List<String> fileList = new ArrayList<String>();
+ //IP
+ String ip = collectVo.getIP();
+ //port
+ String port = collectVo.getPort();
+ //user
+ String user = collectVo.getUser();
+ //password
+ String password = collectVo.getPassword();
+ //isPassiveMode
+ String passivemode = collectVo.getPassive();
+
+ String ftpType = collectVo.getFtptype();
+ FTPInterface ftpClient = null;
+ if("ftp".equalsIgnoreCase(ftpType)){
+ ftpClient = new FTPSrv();
+ }else{
+ ftpClient = new SFTPSrv();
+ }
+
+ //login
+ try {
+ ftpClient.login(ip, Integer.parseInt(port), user, password, "GBK", Boolean.parseBoolean(passivemode), 5*60*1000);
+ } catch (Exception e) {
+ log.error("login fail,ip=["+ip+"] port=["+port+"] user=["+user+"]password=["+password+"]"+StringUtil.getStackTrace(e));
+ return fileList;
+ }
+
+ //download
+ String dir = collectVo.getRemotepath();
+ boolean cdsucess = ftpClient.chdir(dir);
+ if(cdsucess){
+ AFtpRemoteFile[] remoteFiles = (AFtpRemoteFile[]) ftpClient.list();
+
+ for(AFtpRemoteFile ftpRemoteFile: remoteFiles){
+ if(!new File(localPath).exists()){
+ try {
+ new File(localPath).mkdir();
+ } catch (Exception e) {
+ log.error("create localPath is fail localPath="+localPath+" "+StringUtil.getStackTrace(e));
+ }
+ }
+
+ if(!new File(localPath).exists()){
+ new File(localPath).mkdirs();
+ }
+
+ String localFileName = localPath + ftpRemoteFile.getFileName();
+ File loaclFile = new File(localFileName);
+ if (loaclFile.exists()) {
+ loaclFile.delete();
+ }
+
+ boolean flag = ftpClient.downloadFile(ftpRemoteFile.getAbsFileName(), localFileName);
+
+ if(flag){
+ fileList.add(localFileName);
+ }else{
+ log.error("download file fail fileName="+ftpRemoteFile.getAbsFileName());
+ }
+ }
+
+ }else{
+ log.error("chdir is faill dir =["+dir+"]");
+ }
+
+ return fileList;
+ }
+
+
+}
\ No newline at end of file
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.collector;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.openo.nfvo.emsdriver.commons.constant.Constant;
+import org.openo.nfvo.emsdriver.commons.model.CollectMsg;
+import org.openo.nfvo.emsdriver.commons.utils.StringUtil;
+
+
+
+public class TaskThreadService extends Thread {
+
+ public Log log = LogFactory.getLog(TaskThreadService.class);
+ private final ExecutorService pool;
+
+ private BlockingQueue<CollectMsg> queue = new LinkedBlockingQueue<CollectMsg>();
+ private boolean startFlag = true;
+
+ public static TaskThreadService getInstance(int poolSize) {
+ return new TaskThreadService(poolSize);
+ }
+ private TaskThreadService(int poolSize) {
+ pool = Executors.newFixedThreadPool(poolSize);
+ }
+
+ private long timeStamp = System.currentTimeMillis();
+ public void run() { // run the service
+ try {
+ while(startFlag) {
+
+ try {
+ if(System.currentTimeMillis() - timeStamp > Constant.ONEMINUTE){
+ timeStamp = System.currentTimeMillis();
+ log.debug("task queue size " + queue.size());
+ }
+
+ CollectMsg data = receive();
+ if(data == null){
+ continue;
+ }
+
+ pool.execute(new TaskThread(data));
+
+
+ } catch (Exception e) {
+ log.error("collect task process fail!"+StringUtil.getStackTrace(e));
+ }
+
+ }
+
+ } catch (Exception ex) {
+ log.error("task ThreadService error "+StringUtil.getStackTrace(ex));
+ pool.shutdown();
+ }
+ log.error("Task ThreadService exit");
+ }
+
+
+
+ public CollectMsg receive() {
+ try {
+ return queue.poll(100, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ log.error("queue.poll is error"+StringUtil.getStackTrace(e));
+ }
+ return null;
+ }
+
+
+
+ public void add(CollectMsg data){
+ try {
+ queue.put(data);
+ } catch (InterruptedException e) {
+ log.error("queue.put is error"+StringUtil.getStackTrace(e));
+ }
+ }
+
+
+ public int size(){
+ return queue.size();
+ }
+
+ public void stopTask(){
+ startFlag = false;
+ }
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.collector.alarm;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.openo.nfvo.emsdriver.commons.constant.Constant;
+import org.openo.nfvo.emsdriver.commons.model.CollectVo;
+import org.openo.nfvo.emsdriver.commons.model.EMSInfo;
+import org.openo.nfvo.emsdriver.commons.utils.DriverThread;
+import org.openo.nfvo.emsdriver.configmgr.ConfigurationInterface;
+
+public class AlarmManager extends DriverThread{
+
+ private ConfigurationInterface configurationInterface;
+
+ @Override
+ public void dispose() {
+ log.debug("AlarmManager is start");
+ //get alarm config
+ List<EMSInfo> emsInfos = configurationInterface.getAllEMSInfo();
+ while(isRun() && emsInfos.size() == 0){
+ emsInfos = configurationInterface.getAllEMSInfo();
+ if(emsInfos.size() == 0){
+ try {
+ Thread.sleep(1000);
+ log.debug("config is not load");
+ } catch (InterruptedException e) {
+ }
+ }
+ }
+ List<CollectVo> collectVos = new ArrayList<CollectVo>();
+ for(EMSInfo emsInfo : emsInfos){
+ //alarm
+ CollectVo CollectVo = emsInfo.getCollectVoByType(Constant.COLLECT_TYPE_ALARM);
+ if(CollectVo != null){
+ CollectVo.setEmsName(emsInfo.getName());
+ collectVos.add(CollectVo);
+ }
+ }
+
+ for(CollectVo collectVo : collectVos){
+ AlarmTaskThread alarm = new AlarmTaskThread(collectVo);
+ alarm.setName(collectVo.getIP()+collectVo.getPort());
+ alarm.start();
+ }
+
+ }
+
+ /**
+ * @return the configurationInterface
+ */
+ public ConfigurationInterface getConfigurationInterface() {
+ return configurationInterface;
+ }
+
+ /**
+ * @param configurationInterface the configurationInterface to set
+ */
+ public void setConfigurationInterface(
+ ConfigurationInterface configurationInterface) {
+ this.configurationInterface = configurationInterface;
+ }
+
+
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.collector.alarm;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.IOException;
+import java.net.Socket;
+import java.net.SocketException;
+import java.net.UnknownHostException;
+import java.util.Set;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.openo.nfvo.emsdriver.commons.constant.Constant;
+import org.openo.nfvo.emsdriver.commons.model.CollectVo;
+import org.openo.nfvo.emsdriver.commons.utils.StringUtil;
+import org.openo.nfvo.emsdriver.messagemgr.MessageChannel;
+import org.openo.nfvo.emsdriver.messagemgr.MessageChannelFactory;
+
+import com.alibaba.fastjson.JSONObject;
+
+
+public class AlarmTaskThread extends Thread{
+ public Log log = LogFactory.getLog(AlarmTaskThread.class);
+
+ private HeartBeat heartBeat = null;
+
+ private boolean isStop = false;
+ private CollectVo collectVo = null;
+ private int read_timeout = Constant.READ_TIMEOUT_MILLISECOND;
+ private int reqId;
+
+ private Socket socket = null;
+ private BufferedInputStream is = null;
+ private BufferedOutputStream dos = null;
+
+ private MessageChannel alarmChannel;
+
+ public AlarmTaskThread(CollectVo collectVo) {
+
+ this.collectVo = collectVo;
+ }
+
+ public void run() {
+ alarmChannel = MessageChannelFactory.getMessageChannel(Constant.ALARM_CHANNEL_KEY);
+ try {
+ this.init();
+ while(!this.isStop){
+ String body;
+ try {
+ body = this.receive();
+ String alarm120 = this.build120Alarm(body);
+
+ this.send120Alarm(alarm120);
+ } catch (Exception e) {
+ reinit();
+ }
+ }
+ } catch (Exception e) {
+ log.error(StringUtil.getStackTrace(e));
+ }
+ }
+
+ private void send120Alarm(String alarm120) {
+
+ try {
+ alarmChannel.put(alarm120);
+ } catch (InterruptedException e) {
+ log.error(StringUtil.getStackTrace(e));
+ }
+ }
+
+ private String build120Alarm(String body) {
+ StringBuilder content = new StringBuilder(
+ "<?xml version='1.0' encoding='iso-8859-1'?>\n")
+ .append("<WholeMsg MsgMark='120' Priority='2' FieldNum='5'><FM_ALARM_MSG>\n");
+
+
+ JSONObject reagobj = JSONObject.parseObject(body);
+
+ Set<String> keys = reagobj.keySet();
+
+ for (String key : keys) {
+
+ String value = (String)reagobj.get(key);
+ content.append("<").append(key).append(">");
+ content.append(value);
+ content.append("</").append(key).append(">\n");
+ }
+ content.append("</FM_ALARM_MSG></WholeMsg>");
+
+ return content.toString();
+
+ }
+
+ public String receive() throws Exception {
+
+ Msg msg =null;
+ String retString = null;
+
+ while (retString == null && !this.isStop) {
+
+ msg = MessageUtil.readOneMsg(is);
+
+ if("ackLoginAlarm".equalsIgnoreCase(msg.getMsgType().name)){
+ log.debug("receive login ack");
+ boolean suc = this.ackLoginAlarm(msg);
+ if(suc){
+
+ if(reqId == Integer.MAX_VALUE){
+ reqId = 0;
+ }
+ reqId ++;
+ Msg msgheart = MessageUtil.putHeartBeatMsg(reqId);
+ heartBeat = new HeartBeat(socket,msgheart);
+ heartBeat.setName("CMCC_JT_HeartBeat");
+ // start heartBeat
+ heartBeat.start();
+ }
+ retString = null;
+ }
+
+ if("ackHeartBeat".equalsIgnoreCase(msg.getMsgType().name)){
+ log.debug("received heartBeat��"+msg.getBody());
+ retString = null;
+ }
+
+
+
+ if("realTimeAlarm".equalsIgnoreCase(msg.getMsgType().name)){
+ log.debug("received alarm message");
+ retString = msg.getBody();
+ }
+ }
+ return retString;
+ }
+
+ public void init() throws Exception {
+ isStop = false;
+ //host
+ String host = collectVo.getIP();
+ //port
+ String port = collectVo.getPort();
+ //user
+ String user = collectVo.getUser();
+ //password
+ String password = collectVo.getPassword();
+
+ String read_timeout = collectVo.getRead_timeout();
+ if ((read_timeout != null) && (read_timeout.trim().length() > 0)) {
+ try {
+ this.read_timeout = Integer.parseInt(read_timeout);
+ } catch (NumberFormatException e) {
+ log.error(StringUtil.getStackTrace(e));
+ }
+ }
+ log.debug("socket connect host=" + host + ", port=" + port);
+ try {
+ int portInt = Integer.parseInt(port);
+ socket = new Socket(host, portInt);
+
+ } catch (UnknownHostException e) {
+ throw new Exception("remote host [" + host + "]connect fail" + StringUtil.getStackTrace(e));
+ } catch (IOException e) {
+ throw new Exception("create socket IOException " + StringUtil.getStackTrace(e));
+ }
+ try {
+ socket.setSoTimeout(this.read_timeout);
+ socket.setTcpNoDelay(true);
+ socket.setKeepAlive(true);
+ } catch (SocketException e) {
+ throw new Exception(" SocketException " + StringUtil.getStackTrace(e));
+ }
+ try {
+ dos = new BufferedOutputStream(socket.getOutputStream());
+
+ Msg msg = MessageUtil.putLoginMsg(user,password);
+
+ try {
+ log.debug("send login message "+msg.toString(false));
+ MessageUtil.writeMsg(msg,dos);
+
+ } catch (Exception e) {
+ log.error("send login message is fail "+StringUtil.getStackTrace(e));
+ }
+
+ is = new BufferedInputStream(socket.getInputStream());
+
+ } catch (SocketException e) {
+ throw new Exception(StringUtil.getStackTrace(e));
+ }
+ }
+
+ private boolean ackLoginAlarm(Msg msg) throws Exception {
+
+ boolean is_success = false;
+ try {
+ String loginres = msg.getBody();
+ //ackLoginAlarm; result=fail(succ); resDesc=username-error
+ String [] loginbody = loginres.split(";");
+ if(loginbody.length > 1){
+ for(String str :loginbody){
+ if(str.contains("=")){
+ String [] paras1 = str.split("=",-1);
+ if("result".equalsIgnoreCase(paras1[0].trim())){
+ if("succ".equalsIgnoreCase(paras1[1].trim())){
+ is_success = true;
+ }else{
+ is_success = false;
+ }
+ }
+ }
+ }
+ }else {
+ log.error("login ack body Incorrect formatbody=" + loginres);
+ }
+
+
+ } catch (Exception e) {
+ log.error("pocess login ack fail"+StringUtil.getStackTrace(e));
+ }
+ if (is_success) {
+ log.info("login sucess receive login ack " + msg.getBody());
+ } else {
+ log.error("login fail receive login ack " + msg.getBody());
+ this.close();
+ this.isStop = true;
+ throw new Exception("login fail quit");
+ }
+ return is_success;
+ }
+
+ public void close() {
+
+ if(heartBeat != null){
+ heartBeat.setStop(true);
+ }
+
+ if (is != null) {
+ try {
+ is.close();
+ } catch (IOException e) {
+ } finally {
+ is = null;
+ }
+ }
+
+ if (dos != null) {
+ try {
+ dos.close();
+ } catch (IOException e) {
+ } finally {
+ dos = null;
+ }
+ }
+
+ if (socket != null) {
+ try {
+ socket.close();
+ } catch (IOException e) {
+ } finally {
+ socket = null;
+ }
+
+ }
+ }
+
+ public void reinit() {
+ int time = 0;
+ close();
+ while(!this.isStop) {
+ close();
+ time++;
+ try {
+ Thread.sleep(1000 * 10);
+ init();
+ return;
+ } catch (Exception e) {
+ log.error("Number ["+time+"]reconnect ["+collectVo.getIP()+"]fail" );
+ }
+ }
+ }
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.collector.alarm;
+
+import java.io.BufferedOutputStream;
+import java.net.Socket;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.openo.nfvo.emsdriver.commons.constant.Constant;
+
+public class HeartBeat extends Thread{
+ public Log log = LogFactory.getLog(HeartBeat.class);
+ private BufferedOutputStream out = null;
+ private Socket socket=null;
+ private Msg heartStr ;
+ private boolean stop = false;
+ public boolean isStop(){
+ return this.stop;
+ }
+ public void setStop(boolean stop){
+ this.stop = stop;
+ }
+
+ public HeartBeat( Socket socket,Msg heatMessage){
+ this.socket=socket;
+ this.heartStr = heatMessage;
+ }
+
+ public void run(){
+ log.debug("HeartBeat start heartStr:"+heartStr.toString(false));
+ this.stop=false;
+ try {
+ while(!this.isStop()){
+ out = new BufferedOutputStream(socket.getOutputStream());
+ MessageUtil.writeMsg(heartStr,out);
+ log.debug("send HeartBeat heartStr:"+heartStr.toString(false));
+ Thread.sleep(Constant.ONEMINUTE);
+ }
+ } catch (Exception e) {
+ log.error("send HeartBeat fail ",e);
+ }
+ log.debug("HeartBeat thread stop");
+ }
+
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.collector.alarm;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+
+
+
+public class MessageUtil
+{
+ public static String MSG_BODY_ENCODE_CHARSET="UTF-8";
+ public static int MSG_BUF_SIZE=8096 ;
+
+ public static Msg putLoginMsg(String user,String passwd)
+ {
+ String body = String.format(Msg.reqLoginAlarm, user,passwd,"msg");
+ Msg msg = new Msg(body,MsgType.reqLoginAlarm);
+
+
+ return msg;
+
+ }
+ public static Msg putLoginFtp(String user,String passwd)
+ {
+ String body = String.format(Msg.reqLoginAlarm, user,passwd,"ftp");
+ Msg msg = new Msg(body,MsgType.reqLoginAlarm);
+
+
+ return msg;
+
+ }
+
+ public static Msg putSyncMsg(int reqId,int alarmSeq)
+ {
+ String body = String.format(Msg.syncAlarmMessageMsg, reqId,alarmSeq);
+ Msg msg = new Msg(body,MsgType.reqSyncAlarmMsg);
+
+
+ return msg;
+
+ }
+
+ public static Msg putHeartBeatMsg(int reqId)
+ {
+ String body = String.format(Msg.reqHeartBeat, reqId);
+ Msg msg = new Msg(body,MsgType.reqHeartBeat);
+ return msg;
+
+ }
+
+ public static Msg reqSyncAlarmFile(int reqId, String startTime,String endTime) {
+ String body = String.format(Msg.syncActiveAlarmFileMsg, reqId,startTime,endTime);
+ Msg msg = new Msg(body,MsgType.reqSyncAlarmFile);
+ return msg;
+ }
+
+ public static Msg reqSyncAlarmFileByAlarmSeq(int reqId, int alarmSeq) {
+ String body = String.format(Msg.syncAlarmMessageByalarmSeq, reqId,alarmSeq);
+ Msg msg = new Msg(body,MsgType.reqSyncAlarmFile);
+ return msg;
+ }
+
+ public static Msg reqSyncAlarmFileByTime(int reqId, String startTime,String endTime) {
+ String body = String.format(Msg.syncAlarmFileMsg, reqId,startTime,endTime);
+ Msg msg = new Msg(body,MsgType.reqSyncAlarmFile);
+ return msg;
+ }
+
+ public static Msg closeConnAlarmMsg()
+ {
+ String body = String.format(Msg.disconnectMsg);
+ Msg msg = new Msg(body,MsgType.closeConnAlarm);
+ return msg;
+ }
+
+ public static Msg readOneMsg(BufferedInputStream is) throws Exception
+ {
+ byte[] inputB = new byte[9];
+
+ ByteArrayInputStream bais = null;
+ DataInputStream ois = null;
+
+ Msg msg = new Msg();
+ try {
+ DataInputStream dis = new DataInputStream(is);
+ dis.readFully(inputB);
+ bais = new ByteArrayInputStream(inputB);
+ ois = new DataInputStream(bais);
+ short StartSign = ois.readShort();
+ if (StartSign != Msg.StartSign) {
+ throw new Exception("start sign is [" + Msg.StartSign
+ + "],not is [" + StartSign + "]");
+ }
+ int msgType = ois.readByte();
+ msg.setMsgType(MsgType.getMsgTypeValue(msgType));
+ int timeStamp = ois.readInt();
+ msg.setTimeStamp(timeStamp);
+ int bodylength = ois.readShort();
+ msg.setLenOfBody(bodylength);
+ byte b[] = new byte[bodylength];
+ dis.readFully(b);
+ msg.newBodyfromBytes(b);
+ } catch (Exception e) {
+ throw new Exception(e);
+ }finally{
+ if(bais != null){
+ bais.close();
+ }
+ if(ois != null){
+ ois.close();
+ }
+ }
+
+ return msg;
+ }
+
+ public static void writeMsg(Msg msg,BufferedOutputStream dout) throws Exception{
+
+ ByteArrayOutputStream byteOutStream = null;
+ DataOutputStream oos = null;
+ try {
+ byteOutStream = new ByteArrayOutputStream(9);
+ oos = new DataOutputStream(byteOutStream);
+ oos.writeShort(Msg.StartSign);
+ oos.writeByte(msg.getMsgType().value);
+ oos.writeInt(Msg.creatMsgTimeStamp());
+ oos.writeShort(msg.getBodyLenNow());
+
+ dout.write(byteOutStream.toByteArray());
+
+ dout.write(msg.getBodyBytes());
+ dout.flush();
+ } catch (Exception e) {
+ throw new Exception(e);
+ }finally{
+ if(oos != null){
+ oos.close();
+ }
+ if(byteOutStream != null){
+ byteOutStream.close();
+ }
+ }
+
+ }
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.collector.alarm;
+
+import java.io.UnsupportedEncodingException;
+
+import org.openo.nfvo.emsdriver.commons.constant.Constant;
+
+
+public class Msg {
+
+ public static short StartSign = (short)0xffff;
+
+ public final static String reqLoginAlarm = "reqLoginAlarm;user=%s;key=%s;type=%s";
+
+ public final static String reqHeartBeat = "reqHeartBeat;reqId=%s";
+ public final static String disconnectMsg = "closeConnAlarm";
+
+ public final static String syncAlarmMessageMsg = "reqSyncAlarmMsg;reqID=%s;alarmSeq=%s";
+ public final static String syncAlarmMessageByalarmSeq = "reqSyncAlarmFile;reqID=%s;alarmSeq=%s;syncSource=1";
+ public final static String syncActiveAlarmFileMsg = "reqSyncAlarmFile;reqID=%s;startTime=%s;endTime=%s;syncSource=0";
+ public final static String syncAlarmFileMsg = "reqSyncAlarmFile;reqID=%s;startTime=%s;endTime=%s;syncSource=1";
+
+
+
+ private MsgType msgType;
+ private int timeStamp = 0;
+ private int lenOfBody = 0;
+ private String body = null;
+ public Msg(){}
+ public Msg(String body,MsgType msgType ){
+ this.body = body;
+ this.setMsgType(msgType);
+ }
+
+ public void newBodyfromBytes(byte b[]) throws UnsupportedEncodingException{
+ this.body = new String(b,Constant.ENCODING_UTF8);
+ }
+ public static int creatMsgTimeStamp(){
+ return (int)System.currentTimeMillis()/1000;
+ }
+
+ public int getBodyLenNow(){
+ return getBody().getBytes().length;
+ }
+
+
+ public void setTimeStamp(int timeStamp) {
+ this.timeStamp = timeStamp;
+ }
+
+ public int getTimeStamp() {
+ return timeStamp;
+ }
+
+ public void setLenOfBody(int lenOfBody) {
+ this.lenOfBody = lenOfBody;
+ }
+
+ public int getLenOfBody() {
+ return lenOfBody;
+ }
+
+ public byte[] getBodyBytes() throws UnsupportedEncodingException {
+ return getBody().getBytes(Constant.ENCODING_UTF8);
+ }
+
+ public void setBody(String body) {
+ this.body = body;
+ }
+
+ public String getBody() {
+ return body;
+ }
+
+ public void setMsgType(MsgType msgType) {
+ this.msgType = msgType;
+ }
+
+ public MsgType getMsgType() {
+ return msgType;
+ }
+
+ public String toString(boolean isRead){
+ StringBuilder sb = new StringBuilder();
+ sb.append("StartSign[").append(StartSign).append("]msgType[").append(msgType.value).append("]timeStamp[");
+ if(isRead){
+ sb.append(timeStamp).append("]lenOfBody[").append(lenOfBody);
+ }else{
+ sb.append(creatMsgTimeStamp()).append("]lenOfBody[").append(getBodyLenNow());
+ }
+ sb.append("]body[").append(body).append("]");
+ return sb.toString();
+ }
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.collector.alarm;
+
+public enum MsgType {
+
+ reqLoginAlarm("reqLoginAlarm",1,"all"),
+ ackLoginAlarm("ackLoginAlarm",2,"all"),
+ reqSyncAlarmMsg("reqSyncAlarmMsg",3,"msg"),
+ ackSyncAlarmMsg("ackSyncAlarmMsg",4,"msg"),
+ reqSyncAlarmFile("reqSyncAlarmFile",5,"file"),
+ ackSyncAlarmFile("ackSyncAlarmFile",6,"file"),
+ ackSyncAlarmFileResult("ackSyncAlarmFileResult",7,"file"),
+ reqHeartBeat("reqHeartBeat",8,"all"),
+ ackHeartBeat("ackHeartBeat",9,"all"),
+ closeConnAlarm("closeConnAlarm",10,"all"),
+ realTimeAlarm("realTimeAlarm",0,"all"),
+ undefined("undefined",-1,"all");
+
+ public int value = -1;
+ public String name = null;
+ public String type = null;
+
+ MsgType(String name,int value,String type){this.name = name;this.value = value;this.type = type;}
+
+ public static MsgType getMsgTypeValue(int msgTypeValue){
+
+ for(MsgType msgType : MsgType.values()){
+ if(msgType.value == msgTypeValue){
+ return msgType;
+ }
+ }
+ return undefined;
+ }
+
+ public static MsgType getMsgTypeName(String msgTypeName){
+
+ for(MsgType msgType : MsgType.values()){
+ if(msgType.name.toLowerCase().equals(msgTypeName.toLowerCase())){
+ return msgType;
+ }
+ }
+ return undefined;
+ }
+
+ public String toString(){
+ return this.name;
+ }
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.commons.constant;
+
+import java.io.File;
+
+public class Constant {
+
+ public static String SYS_HOME = System.getenv("EMSDRIVER_HOME")==null?"E:/workspace/ems-driver/":System.getenv("EMSDRIVER_HOME");
+
+ public static String SYS_CFG = SYS_HOME + File.separator + "cfg" + File.separator;
+ public static String SYS_DATA = SYS_HOME + "data" + File.separator;
+ public static String SYS_DATA_TEMP = SYS_DATA + File.separator + "temp" + File.separator;
+ public static String SYS_DATA_RESULT = SYS_DATA + File.separator + "RESULT" + File.separator;
+ public static String COLLECT_TYPE_CM = "cm";
+ public static String COLLECT_TYPE_PM = "pm";
+ public static String COLLECT_TYPE_ALARM = "alarm";
+
+ public static String ENCODING_UTF8 = "UTF-8";
+ public static String ENCODING_GBK = "GBK";
+
+ public static final String COLLECT_CHANNEL_KEY = "COLLECT_CHANNEL_KEY";
+ public static final String COLLECT_RESULT_CHANNEL_KEY = "COLLECT_RESULT_CHANNEL_KEY";
+ public static final String ALARM_CHANNEL_KEY = "ALARM_CHANNEL_KEY";
+
+
+ public static final String MSBAPIROOTDOMAIN = "/openoapi/microservices/v1/services";
+
+ //alarm
+ public static final int READ_TIMEOUT_MILLISECOND = 180000;
+ public static final long ONEMINUTE = 60000;
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.commons.ftp;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Date;
+
+import org.apache.commons.net.ftp.FTPClient;
+import org.apache.commons.net.ftp.FTPFile;
+
+public class AFtpRemoteFile implements RemoteFile{
+ protected FTPClient ftpClient = null;
+ protected FTPFile ftpFile = null;
+ protected String currDir = null;
+
+ public AFtpRemoteFile(FTPFile rfile, FTPClient ftpClient, String currDir)
+ throws IOException {
+ this.ftpClient = ftpClient;
+ this.ftpFile = rfile;
+ this.currDir = currDir;
+ }
+
+ public long getSize() {
+ return ftpFile.getSize();
+ }
+
+ public String getFileName() {
+ return ftpFile.getName();
+ }
+
+ public String getAbsFileName() {
+ return currDir.concat(getFileName());
+ }
+
+ public boolean isDirectory() {
+ return ftpFile.isDirectory();
+ }
+ public boolean isFile() {
+ return ftpFile.isFile();
+ }
+
+ public String getOwner() {
+ return ftpFile.getUser();
+ }
+
+ public Date getModifyDate() {
+ return ftpFile.getTimestamp().getTime();
+ }
+ public boolean renameTo(String newName) throws IOException {
+ return ftpClient.rename(
+ currDir.concat(getFileName()), newName);
+ }
+ public boolean remove() throws IOException {
+ return ftpClient.deleteFile(
+ currDir.concat(getFileName()));
+ }
+
+ public InputStream getInputStream() throws IOException {
+ return ftpClient.retrieveFileStream(this.getAbsFileName());
+ }
+
+ public void release() {
+ ftpClient = null;
+ ftpFile = null;
+ currDir = null;
+ }
+}
\ No newline at end of file
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.commons.ftp;
+
+import org.apache.commons.net.ftp.Configurable;
+import org.apache.commons.net.ftp.FTPClientConfig;
+import org.apache.commons.net.ftp.FTPFileEntryParser;
+import org.apache.commons.net.ftp.parser.CompositeFileEntryParser;
+import org.apache.commons.net.ftp.parser.DefaultFTPFileEntryParserFactory;
+import org.apache.commons.net.ftp.parser.MVSFTPEntryParser;
+import org.apache.commons.net.ftp.parser.NTFTPEntryParser;
+import org.apache.commons.net.ftp.parser.OS2FTPEntryParser;
+import org.apache.commons.net.ftp.parser.OS400FTPEntryParser;
+import org.apache.commons.net.ftp.parser.ParserInitializationException;
+import org.apache.commons.net.ftp.parser.UnixFTPEntryParser;
+import org.apache.commons.net.ftp.parser.VMSVersioningFTPEntryParser;
+ /**
+ * This is the default implementation of the
+ * FTPFileEntryParserFactory interface. This is the
+ * implementation that will be used by
+ * org.apache.commons.net.ftp.FTPClient.listFiles()
+ * if no other implementation has been specified.
+ *
+ * @see org.apache.commons.net.ftp.FTPClient#listFiles
+ * @see org.apache.commons.net.ftp.FTPClient#setParserFactory
+ */
+public class ExtendsDefaultFTPFileEntryParserFactory extends
+ DefaultFTPFileEntryParserFactory {
+
+ private FTPClientConfig config = null;
+
+ /**
+ * This default implementation of the FTPFileEntryParserFactory
+ * interface works according to the following logic:
+ * First it attempts to interpret the supplied key as a fully
+ * qualified classname of a class implementing the
+ * FTPFileEntryParser interface. If that succeeds, a parser
+ * object of this class is instantiated and is returned;
+ * otherwise it attempts to interpret the key as an identirier
+ * commonly used by the FTP SYST command to identify systems.
+ * <p/>
+ * If <code>key</code> is not recognized as a fully qualified
+ * classname known to the system, this method will then attempt
+ * to see whether it <b>contains</b> a string identifying one of
+ * the known parsers. This comparison is <b>case-insensitive</b>.
+ * The intent here is where possible, to select as keys strings
+ * which are returned by the SYST command on the systems which
+ * the corresponding parser successfully parses. This enables
+ * this factory to be used in the auto-detection system.
+ * <p/>
+ *
+ * @param key should be a fully qualified classname corresponding to
+ * a class implementing the FTPFileEntryParser interface<br/>
+ * OR<br/>
+ * a string containing (case-insensitively) one of the
+ * following keywords:
+ * <ul>
+ * <li>{@link FTPClientConfig#SYST_UNIX UNIX}</li>
+ * <li>{@link FTPClientConfig#SYST_NT WINDOWS}</li>
+ * <li>{@link FTPClientConfig#SYST_OS2 OS/2}</li>
+ * <li>{@link FTPClientConfig#SYST_OS400 OS/400}</li>
+ * <li>{@link FTPClientConfig#SYST_VMS VMS}</li>
+ * <li>{@link FTPClientConfig#SYST_MVS MVS}</li>
+ * </ul>
+ * @return the FTPFileEntryParser corresponding to the supplied key.
+ * @throws ParserInitializationException thrown if for any reason the factory cannot resolve
+ * the supplied key into an FTPFileEntryParser.
+ * @see FTPFileEntryParser
+ */
+ public FTPFileEntryParser createFileEntryParser(String key)
+ {
+ @SuppressWarnings("rawtypes")
+ Class parserClass = null;
+ FTPFileEntryParser parser = null;
+ try
+ {
+ parserClass = Class.forName(key);
+ parser = (FTPFileEntryParser) parserClass.newInstance();
+ }
+ catch (ClassNotFoundException e)
+ {
+ String ukey = null;
+ if (null != key)
+ {
+ ukey = key.toUpperCase();
+ }
+ if (ukey.indexOf(FTPClientConfig.SYST_UNIX) >= 0)
+ {
+ parser = createUnixFTPEntryParser();
+ }
+ else if (ukey.indexOf(FTPClientConfig.SYST_VMS) >= 0)
+ {
+ parser = createVMSVersioningFTPEntryParser();
+ }
+ else if (ukey.indexOf(FTPClientConfig.SYST_NT) >= 0 || ukey.indexOf("DOPRA") >= 0 || ukey.indexOf("MSDOS") >= 0)
+ {
+ parser = createNTFTPEntryParser();
+ }
+ else if (ukey.indexOf(FTPClientConfig.SYST_OS2) >= 0)
+ {
+ parser = createOS2FTPEntryParser();
+ }
+ else if (ukey.indexOf(FTPClientConfig.SYST_OS400) >= 0)
+ {
+ parser = createOS400FTPEntryParser();
+ }
+ else if (ukey.indexOf(FTPClientConfig.SYST_MVS) >= 0)
+ {
+ parser = createMVSEntryParser();
+ }
+ else
+ {
+ throw new ParserInitializationException("Unknown parser type: " + key);
+ }
+ }
+ catch (ClassCastException e)
+ {
+ throw new ParserInitializationException(parserClass.getName()
+ + " does not implement the interface "
+ + "org.apache.commons.net.ftp.FTPFileEntryParser.", e);
+ }
+ catch (Throwable e)
+ {
+ throw new ParserInitializationException("Error initializing parser", e);
+ }
+
+ if (parser instanceof Configurable) {
+ ((Configurable)parser).configure(this.config);
+ }
+ return parser;
+ }
+
+ /**
+ * <p>Implementation extracts a key from the supplied
+ * {@link FTPClientConfig FTPClientConfig}
+ * parameter and creates an object implementing the
+ * interface FTPFileEntryParser and uses the supplied configuration
+ * to configure it.
+ * </p><p>
+ * Note that this method will generally not be called in scenarios
+ * that call for autodetection of parser type but rather, for situations
+ * where the user knows that the server uses a non-default configuration
+ * and knows what that configuration is.
+ * </p>
+ * @param config A {@link FTPClientConfig FTPClientConfig}
+ * used to configure the parser created
+ *
+ * @return the @link FTPFileEntryParser FTPFileEntryParser} so created.
+ * @exception ParserInitializationException
+ * Thrown on any exception in instantiation
+ * @since 1.4
+ */
+ public FTPFileEntryParser createFileEntryParser(FTPClientConfig config)
+ throws ParserInitializationException
+ {
+ this.config = config;
+ String key = config.getServerSystemKey();
+ return createFileEntryParser(key);
+ }
+
+
+ public FTPFileEntryParser createUnixFTPEntryParser()
+ {
+ return (FTPFileEntryParser) new UnixFTPEntryParser();
+ }
+
+ public FTPFileEntryParser createVMSVersioningFTPEntryParser()
+ {
+ return (FTPFileEntryParser) new VMSVersioningFTPEntryParser();
+ }
+
+ public FTPFileEntryParser createNTFTPEntryParser()
+ {
+ if (config != null && FTPClientConfig.SYST_NT.equals(
+ config.getServerSystemKey()))
+ {
+ return new NTFTPEntryParser();
+ } else {
+ return new CompositeFileEntryParser(new FTPFileEntryParser[]
+ {
+ new NTFTPEntryParser(),
+ new UnixFTPEntryParser()
+ });
+ }
+ }
+
+ public FTPFileEntryParser createOS2FTPEntryParser()
+ {
+ return (FTPFileEntryParser) new OS2FTPEntryParser();
+ }
+
+ public FTPFileEntryParser createOS400FTPEntryParser()
+ {
+ if (config != null &&
+ FTPClientConfig.SYST_OS400.equals(config.getServerSystemKey()))
+ {
+ return new OS400FTPEntryParser();
+ } else {
+ return new CompositeFileEntryParser(new FTPFileEntryParser[]
+ {
+ new OS400FTPEntryParser(),
+ new UnixFTPEntryParser()
+ });
+ }
+ }
+
+ public FTPFileEntryParser createMVSEntryParser()
+ {
+ return new MVSFTPEntryParser();
+ }
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.commons.ftp;
+
+public interface FTPInterface {
+
+ /**
+ * login ftp
+ * @throws Exception
+ */
+ public void login(String host, int port, String user, String pwd, String encode,boolean isPassiveMode, int timeout) throws Exception;
+
+ /**
+ * close ftp
+ */
+ public void logout();
+
+ /**
+ * download file
+ * @return
+ */
+ public boolean downloadFile(String remoteFile,String localFile);
+
+ public boolean chdir(String dir);
+
+ public RemoteFile[] list();
+
+ public RemoteFile[] list(String dir);
+
+ public boolean store(String localFile,String remoteFile);
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.commons.ftp;
+
+import java.io.BufferedOutputStream;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.TimeZone;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.commons.net.ftp.FTP;
+import org.apache.commons.net.ftp.FTPClient;
+import org.apache.commons.net.ftp.FTPClientConfig;
+import org.apache.commons.net.ftp.FTPFile;
+import org.apache.commons.net.ftp.FTPReply;
+import org.openo.nfvo.emsdriver.commons.utils.StringUtil;
+
+
+public class FTPSrv implements FTPInterface{
+ private Log log = LogFactory.getLog(FTPSrv.class);
+ private FTPClient ftpClient = null;
+
+
+ /**
+ * login FTP
+ * @param host
+ * @param port
+ * @param user
+ * @param pwd
+ * @param encode
+ * @param timeout
+ * @throws Exception
+ */
+ public void login(String host, int port, String user, String pwd, String encode, boolean isPassiveMode,int timeout) throws Exception {
+ ftpClient = new FTPClient();
+
+ FTPClientConfig ftpClientConfig = new FTPClientConfig();
+ ftpClientConfig.setServerTimeZoneId(TimeZone.getDefault().getID());
+ this.ftpClient.setControlEncoding("GBK");
+ this.ftpClient.configure(ftpClientConfig);
+ ftpClient.setParserFactory(new ExtendsDefaultFTPFileEntryParserFactory());
+
+ if(encode!=null && encode.length()>0){
+ ftpClient.setControlEncoding(encode);
+ }
+
+ ftpClient.connect(host, port);
+ int reply = this.ftpClient.getReplyCode();
+ if (!FTPReply.isPositiveCompletion(reply)) {
+ this.ftpClient.disconnect();
+ return ;
+ }
+
+ if(!ftpClient.login(user, pwd)){
+ throw new Exception("login["+host+"],port["+port+"] fail, please check user and password");
+ }
+ if(isPassiveMode){
+ ftpClient.enterLocalPassiveMode();
+ }else{
+ ftpClient.enterLocalActiveMode();
+ }
+ ftpClient.setFileType(FTP.BINARY_FILE_TYPE);
+ this.ftpClient.setBufferSize(1024 * 2);
+ this.ftpClient.setDataTimeout(3*60 * 1000);
+ try{
+ this.ftpClient.setSoTimeout(timeout);
+ }catch(Exception e){
+ e.printStackTrace();
+ }
+ }
+
+
+ /**
+ * logout
+ */
+ public void logout(){
+ if(ftpClient != null){
+ try {
+ ftpClient.logout();
+ ftpClient.disconnect();
+ }catch(Exception e){
+ }
+ ftpClient = null;
+ }
+ }
+
+
+ public boolean chdir(String dir) {
+ boolean sucess = false;
+ try {
+ if(ftpClient.changeWorkingDirectory(dir)){
+ sucess = true;
+ }else{
+ sucess = false;
+ }
+ } catch (IOException e) {
+ log.error("chdir dir ="+dir+" is error"+StringUtil.getStackTrace(e));
+ sucess = false;
+ }
+
+ return sucess;
+ }
+
+
+ public boolean downloadFile(String remoteFile, String localFile) {
+ boolean sucess = false;
+ BufferedOutputStream toLfileOutput = null;
+ try {
+ toLfileOutput = new BufferedOutputStream(new FileOutputStream(localFile));
+ ftpClient.setFileType(FTPClient.BINARY_FILE_TYPE);
+ if (ftpClient.retrieveFile(remoteFile, toLfileOutput)){
+ sucess = true;
+ }else{
+ sucess = false;
+ }
+ } catch (Exception ioe) {
+ sucess = false;
+ log.error("downloadFile remoteFile ="+remoteFile +" is fail ",ioe);
+ } finally {
+ if (toLfileOutput != null)
+ try {
+ toLfileOutput.close();
+ } catch (IOException e) {
+ }
+ }
+
+ return sucess;
+ }
+
+
+ public RemoteFile[] list() {
+ AFtpRemoteFile[] ftpRemoteFiles = null;
+ String currdir = null;
+ try {
+ currdir = ftpClient.printWorkingDirectory();
+ if (currdir.endsWith("/") == false) {
+ currdir = currdir + "/";
+ }
+ FTPFile[] rfileList = null;
+ rfileList = ftpClient.listFiles(currdir);
+ ftpRemoteFiles = new AFtpRemoteFile[rfileList.length];
+ for (int i=0; i<rfileList.length; i++){
+ ftpRemoteFiles[i] = new AFtpRemoteFile(rfileList[i], ftpClient, currdir);
+ }
+ } catch (IOException e) {
+ log.error("Ftp list currdir = "+currdir+" is fail "+StringUtil.getStackTrace(e));
+ }
+ return ftpRemoteFiles;
+ }
+
+
+ public RemoteFile[] list(String dir) {
+ return null;
+ }
+
+
+ public boolean store(String localFile, String remoteFile) {
+
+ boolean sucess = false;
+ FileInputStream lfileInput = null;
+ try {
+ lfileInput = new FileInputStream(localFile);
+ ftpClient.setFileType(FTPClient.BINARY_FILE_TYPE);
+
+ if (ftpClient.storeFile(remoteFile, lfileInput)){
+ sucess = true;
+ }else{
+ sucess = false;
+ }
+ } catch (Exception ioe) {
+ sucess = false;
+ log.error("store localFile = "+localFile+" is fail "+StringUtil.getStackTrace(ioe));
+ } finally {
+ if (lfileInput != null)
+ try {
+ lfileInput.close();
+ } catch (IOException e) {
+ }
+ }
+ return sucess;
+ }
+
+}
+
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.commons.ftp;
+
+public interface RemoteFile {
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.commons.ftp;
+
+public class SFTPSrv implements FTPInterface{
+
+ public boolean chdir(String dir) {
+ // TODO Auto-generated method stub
+ return false;
+ }
+
+ public boolean downloadFile(String remoteFile, String localFile) {
+ // TODO Auto-generated method stub
+ return false;
+ }
+
+ public RemoteFile[] list() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ public RemoteFile[] list(String dir) {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ public void login(String host, int port, String user, String pwd,
+ String encode, boolean isPassiveMode, int timeout) throws Exception {
+ // TODO Auto-generated method stub
+
+ }
+
+ public void logout() {
+ // TODO Auto-generated method stub
+
+ }
+
+ public boolean store(String localFile, String remoteFile) {
+ // TODO Auto-generated method stub
+ return false;
+ }
+
+
+}
+
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.commons.model;
+
+public class CollectMsg {
+
+ private long id;
+
+ private String emsName;
+
+ private String type;
+
+ /**
+ * @return the id
+ */
+ public long getId() {
+ return id;
+ }
+
+ /**
+ * @param id the id to set
+ */
+ public void setId(long id) {
+ this.id = id;
+ }
+
+ /**
+ * @return the emsName
+ */
+ public String getEmsName() {
+ return emsName;
+ }
+
+ /**
+ * @param emsName the emsName to set
+ */
+ public void setEmsName(String emsName) {
+ this.emsName = emsName;
+ }
+
+ /**
+ * @return the type
+ */
+ public String getType() {
+ return type;
+ }
+
+ /**
+ * @param type the type to set
+ */
+ public void setType(String type) {
+ this.type = type;
+ }
+
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.commons.model;
+
+/**
+ * @author boco
+ *
+ */
+public class CollectVo {
+
+ private String emsName;
+
+ private String type;
+
+ private String crontab;
+
+ private String IP;
+
+ private String port;
+
+ private String user;
+
+ private String password;
+
+ private String remotepath;
+
+ private String match;
+
+ private String passive;
+
+ private String ftptype;
+
+ private String granularity;
+
+ private boolean iscollect = false;
+
+ private String read_timeout;
+
+
+
+ /**
+ * @return the iscollect
+ */
+ public boolean isIscollect() {
+ return iscollect;
+ }
+
+ /**
+ * @param iscollect the iscollect to set
+ */
+ public void setIscollect(boolean iscollect) {
+ this.iscollect = iscollect;
+ }
+
+ /**
+ * @return the type
+ */
+ public String getType() {
+ return type;
+ }
+
+ /**
+ * @param type the type to set
+ */
+ public void setType(String type) {
+ this.type = type;
+ }
+
+ /**
+ * @return the crontab
+ */
+ public String getCrontab() {
+ return crontab;
+ }
+
+ /**
+ * @param crontab the crontab to set
+ */
+ public void setCrontab(String crontab) {
+ this.crontab = crontab;
+ }
+
+ /**
+ * @return the iP
+ */
+ public String getIP() {
+ return IP;
+ }
+
+ /**
+ * @param ip the iP to set
+ */
+ public void setIP(String ip) {
+ IP = ip;
+ }
+
+ /**
+ * @return the port
+ */
+ public String getPort() {
+ return port;
+ }
+
+ /**
+ * @param port the port to set
+ */
+ public void setPort(String port) {
+ this.port = port;
+ }
+
+ /**
+ * @return the user
+ */
+ public String getUser() {
+ return user;
+ }
+
+ /**
+ * @param user the user to set
+ */
+ public void setUser(String user) {
+ this.user = user;
+ }
+
+ /**
+ * @return the password
+ */
+ public String getPassword() {
+ return password;
+ }
+
+ /**
+ * @param password the password to set
+ */
+ public void setPassword(String password) {
+ this.password = password;
+ }
+
+ /**
+ * @return the remotepath
+ */
+ public String getRemotepath() {
+ return remotepath;
+ }
+
+ /**
+ * @param remotepath the remotepath to set
+ */
+ public void setRemotepath(String remotepath) {
+ this.remotepath = remotepath;
+ }
+
+ /**
+ * @return the match
+ */
+ public String getMatch() {
+ return match;
+ }
+
+ /**
+ * @param match the match to set
+ */
+ public void setMatch(String match) {
+ this.match = match;
+ }
+
+ /**
+ * @return the passive
+ */
+ public String getPassive() {
+ return passive;
+ }
+
+ /**
+ * @param passive the passive to set
+ */
+ public void setPassive(String passive) {
+ this.passive = passive;
+ }
+
+ /**
+ * @return the ftptype
+ */
+ public String getFtptype() {
+ return ftptype;
+ }
+
+ /**
+ * @param ftptype the ftptype to set
+ */
+ public void setFtptype(String ftptype) {
+ this.ftptype = ftptype;
+ }
+
+ /**
+ * @return the granularity
+ */
+ public String getGranularity() {
+ return granularity;
+ }
+
+ /**
+ * @param granularity the granularity to set
+ */
+ public void setGranularity(String granularity) {
+ this.granularity = granularity;
+ }
+
+
+ public String toString(){
+ StringBuffer sb = new StringBuffer();
+ sb.append("CollectVo:").append("\n");
+ if("alarm".equalsIgnoreCase(type)){
+ sb.append("type = ").append(type).append("\n");
+ sb.append("ip = ").append(IP).append("\n");
+ sb.append("port = ").append(port).append("\n");
+ sb.append("user = ").append(user).append("\n");
+ sb.append("password = ").append(password).append("\n");
+ sb.append("iscollect = ").append(iscollect).append("\n");
+ }else{
+ sb.append("type = ").append(type).append("\n");
+ sb.append("ip = ").append(IP).append("\n");
+ sb.append("port = ").append(port).append("\n");
+ sb.append("user = ").append(user).append("\n");
+
+ sb.append("password = ").append(password).append("\n");
+ sb.append("remotepath = ").append(remotepath).append("\n");
+ sb.append("match = ").append(match).append("\n");
+ sb.append("passive = ").append(passive).append("\n");
+ sb.append("ftptype = ").append(ftptype).append("\n");
+ sb.append("granularity = ").append(type).append("\n");
+ }
+
+
+ return sb.toString();
+
+ }
+
+ /**
+ * @return the emsName
+ */
+ public String getEmsName() {
+ return emsName;
+ }
+
+ /**
+ * @param emsName the emsName to set
+ */
+ public void setEmsName(String emsName) {
+ this.emsName = emsName;
+ }
+
+ /**
+ * @return the read_timeout
+ */
+ public String getRead_timeout() {
+ return read_timeout;
+ }
+
+ /**
+ * @param read_timeout the read_timeout to set
+ */
+ public void setRead_timeout(String read_timeout) {
+ this.read_timeout = read_timeout;
+ }
+
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.commons.model;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class EMSInfo {
+
+ private String name;
+
+ private Map<String,CollectVo> collectMap = new HashMap<String,CollectVo>();
+
+ /**
+ * @return the name
+ */
+ public String getName() {
+ return name;
+ }
+
+ /**
+ * @param name the name to set
+ */
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public CollectVo getCollectVoByType(String type){
+ CollectVo collectVo = null;
+ collectVo = this.collectMap.get(type);
+ return collectVo;
+ }
+
+ public void putCollectMap(String type,CollectVo collectVo) {
+
+ this.collectMap.put(type, collectVo);
+ }
+
+ public String toString(){
+ StringBuffer sb = new StringBuffer();
+ sb.append("name = ").append(name).append("\n");
+ sb.append("CollectMap = ").append(collectMap);
+
+ return sb.toString();
+ }
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.commons.utils;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+public abstract class DriverThread implements Runnable{
+ protected Log log = LogFactory.getLog(this.getClass());
+ private String name = null;
+ private Thread t = null;
+ private boolean run = false;
+ private boolean end = false;
+
+ public synchronized void start() {
+ t = new Thread(this);
+ t.start();
+ }
+ public void setName(String name) {
+ this.name = name;
+ if (t != null)
+ t.setName(name);
+ }
+
+ public String getName() {
+ if (t != null)
+ return t.getName();
+ return name;
+ }
+
+ public abstract void dispose();
+
+ final public void run() {
+ t = Thread.currentThread();
+ if (name != null)
+ t.setName(name);
+
+ try {
+ dispose();
+ } catch (Throwable e) {
+ e.printStackTrace();
+ }
+ this.setEnd(true);
+
+ }
+
+ public boolean stop(){
+
+ this.setRun(false);
+ while(!isEnd()){
+ try {
+ Thread.sleep(1);
+ } catch (InterruptedException e) {
+ log.error("InterruptedException :"+StringUtil.getStackTrace(e));
+ }
+ }
+ return end;
+ }
+
+ public void interrupt() {
+ if (t != null)
+ t.interrupt();
+ }
+
+ public void setRun(boolean run) {
+ this.run = run;
+ }
+
+ public boolean isRun() {
+ return run;
+ }
+ public void setEnd(boolean end) {
+ this.end = end;
+ }
+ public boolean isEnd() {
+ return end;
+ }
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.commons.utils;
+
+import java.io.File;
+import java.io.PrintWriter;
+import java.io.StringWriter;
+
+public class StringUtil {
+
+ public static String getStackTrace(Throwable t){
+
+ StringWriter sw = null;
+ PrintWriter pw = null;
+ try {
+ sw = new StringWriter();
+ pw = new PrintWriter(sw);
+ t.printStackTrace(pw);
+ pw.flush();
+ sw.flush();
+ return sw.getBuffer().toString();
+ } catch (Exception e) {
+
+ }finally{
+ try {
+ if(pw != null) pw.close();
+ if(sw != null) sw.close();
+ } catch (Exception e2) {
+
+ }
+ }
+ return null;
+ }
+
+ public static String addSlash(String dirName) {
+ if (dirName.endsWith(File.separator))
+ return dirName;
+ return dirName + File.separator;
+ }
+
+ public static boolean isBank(String str){
+
+ if(str == null || str.trim().length() == 0){
+
+ return true;
+ }
+ return false;
+ }
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.commons.utils;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Enumeration;
+
+import org.apache.tools.zip.ZipEntry;
+import org.apache.tools.zip.ZipFile;
+
+
+public class UnZip {
+ protected String deCompressPath = null;
+ protected String zipFilePath = null;
+
+ /**
+ *
+ */
+ public UnZip(String zipFilePath, String toPath) throws IOException {
+ File zipFile = new File(
+ new File(zipFilePath).getAbsolutePath());
+ if (!zipFile.isFile())
+ throw new IOException("not found file '"+zipFilePath+"'");
+
+ this.deCompressPath = toPath;
+ this.zipFilePath = zipFile.getAbsolutePath();
+
+ if (deCompressPath == null)
+ deCompressPath = zipFile.getParent() + File.separator;
+
+ else if (deCompressPath.charAt(deCompressPath.length()-1) != '/')
+ deCompressPath = deCompressPath + File.separator;
+ }
+
+ /**
+ *
+ */
+ public void deCompress() throws IOException {
+ ZipFile zipFile = new ZipFile(zipFilePath);
+ try{
+ Enumeration<ZipEntry> e = zipFile.getEntries();
+ for (ZipEntry entry; e.hasMoreElements(); ) {
+ if (!(entry=e.nextElement()).isDirectory()) {
+ String toPath = new StringBuffer(
+ deCompressPath).append(entry.getName()).toString();
+ toPath = toPath.replace("\\", File.separator);
+ deCompressFile(zipFile.getInputStream(entry), toPath);
+ }
+ }
+ }catch(IOException e){
+ throw e;
+ }finally{
+ zipFile.close();
+ }
+ }
+
+ /**
+ *
+ */
+ protected void deCompressFile(InputStream input, String toPath)
+ throws IOException {
+ byte byteBuf[] = new byte[2048];
+ String path = new File(toPath).getParent();
+ if(!new File(path).exists()){
+ new File(path).mkdirs();
+ }
+ FileOutputStream output = new FileOutputStream(toPath, false);
+ try{
+ for (int count=0; (count=input.read(byteBuf,0,byteBuf.length))!=-1;)
+ output.write(byteBuf, 0, count);
+ }catch(IOException e){
+ throw e;
+ }finally{
+ output.close();
+ input.close();
+ }
+ }
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.commons.utils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.Reader;
+
+import javax.xml.stream.XMLStreamException;
+
+import org.jdom.Document;
+import org.jdom.JDOMException;
+import org.jdom.input.SAXBuilder;
+
+
+public class XmlUtil {
+
+ public static Document getDocument(InputStream is) throws XMLStreamException, JDOMException, IOException{
+
+ SAXBuilder builder = new SAXBuilder();
+ Document doc = builder.build(is);
+ return doc;
+ }
+ public static Document getDocument(Reader reader) throws XMLStreamException, JDOMException, IOException{
+ SAXBuilder builder = new SAXBuilder();
+ Document doc = builder.build(reader);
+ return doc;
+ }
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.commons.utils;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipOutputStream;
+
+
+public class Zip {
+ protected int compressDirectoryCount = 0;
+ protected int compressFileCount = 0;
+
+ protected int relativeAddrIdx = 0;
+ protected int compressLevel = 6;
+ protected String zipFilePath = null;
+ protected String compressPath = null;
+
+ protected ZipOutputStream zipOutput = null;
+
+ /**
+ *
+ */
+ public Zip(String compressPath, String zipFilePath) throws IOException{
+ File compressFile = new File(compressPath);
+ if (!compressFile.exists())
+ throw new IOException("the file or directory '"+compressPath+"' not found!");
+
+ this.zipFilePath = zipFilePath;
+ this.compressPath = compressFile.getAbsolutePath();
+
+ if (this.zipFilePath == null) {
+ StringBuffer zipFilePathBuf = new StringBuffer(this.compressPath);
+ int bufLen = zipFilePathBuf.length();
+ if (zipFilePathBuf.charAt(bufLen-1) == '/')
+ zipFilePathBuf.deleteCharAt(bufLen-1);
+ this.zipFilePath = zipFilePathBuf.append(".zip").toString();
+ }
+ relativeAddrIdx = this.compressPath.lastIndexOf(File.separator)+1;
+ }
+
+ /**
+ *
+ */
+ public void compress() throws IOException {
+ File theFile = new File(zipFilePath);
+
+ if (!theFile.exists()) {
+ String parentPath = theFile.getParent();
+ if (parentPath != null)
+ new File(parentPath).mkdirs();
+ theFile.createNewFile();
+ }
+ zipOutput = new ZipOutputStream(new FileOutputStream(zipFilePath));
+ zipOutput.setMethod(ZipOutputStream.DEFLATED);
+ zipOutput.setLevel(compressLevel);
+ compressDirectory(new File(compressPath));
+ zipOutput.close();
+ }
+
+ protected void compressDirectory(File directoryPath) throws IOException {
+ if (directoryPath.isFile()) {
+ compressFile(directoryPath.getAbsolutePath());
+ }else{
+ File listFiles[] = directoryPath.listFiles();
+ for (int i=0; i<listFiles.length; i++)
+ if (listFiles[i].isFile()) {
+ compressFile(listFiles[i].getAbsolutePath());
+ }else {
+ compressDirectoryCount ++;
+ compressDirectory(listFiles[i]);
+ }
+ }
+
+
+ }
+ protected void compressFile(String absolutePath) throws IOException {
+ compressFileCount ++;
+ byte byteBuf[] = new byte[2048];
+ zipOutput.putNextEntry(new ZipEntry(absolutePath.substring(relativeAddrIdx)));
+
+ FileInputStream input= new FileInputStream(absolutePath);
+ for (int count=0; (count=input.read(byteBuf,0,byteBuf.length))!=-1;)
+ zipOutput.write(byteBuf, 0, count);
+ input.close();
+ zipOutput.closeEntry();
+ }
+
+ public void setCompressLevel(int level) {
+ compressLevel = level;
+ }
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.configmgr;
+
+import java.util.List;
+import java.util.Properties;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.openo.nfvo.emsdriver.commons.model.CollectVo;
+import org.openo.nfvo.emsdriver.commons.model.EMSInfo;
+
+
+public class ConfigurationImp implements ConfigurationInterface{
+
+ private Log log = LogFactory.getLog(ConfigurationImp.class);
+
+ @Override
+ public List<EMSInfo> getAllEMSInfo() {
+ List<EMSInfo> emsInfos = ConfigurationManager.getAllEMSInfos();
+ return emsInfos;
+ }
+
+ @Override
+ public CollectVo getCollectVoByEmsNameAndType(String emsName, String type) {
+ CollectVo collectVo = null;
+
+ EMSInfo emsInfo = ConfigurationManager.getEMSInfoByName(emsName);
+ if(emsInfo != null){
+ collectVo = emsInfo.getCollectVoByType(type);
+ }else{
+ log.error("ConfigurationManager.getEMSInfoByName return null");
+ }
+ return collectVo;
+ }
+
+ @Override
+ public Properties getProperties() {
+ Properties p = ConfigurationManager.getProperties();
+ return p;
+ }
+
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.configmgr;
+
+import java.util.List;
+import java.util.Properties;
+
+import org.openo.nfvo.emsdriver.commons.model.CollectVo;
+import org.openo.nfvo.emsdriver.commons.model.EMSInfo;
+
+
+public interface ConfigurationInterface {
+
+ public List<EMSInfo> getAllEMSInfo();
+ public CollectVo getCollectVoByEmsNameAndType(String emsName,String type);
+
+ public Properties getProperties();
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.configmgr;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.jdom.Document;
+import org.jdom.Element;
+import org.openo.nfvo.emsdriver.commons.constant.Constant;
+import org.openo.nfvo.emsdriver.commons.model.CollectVo;
+import org.openo.nfvo.emsdriver.commons.model.EMSInfo;
+import org.openo.nfvo.emsdriver.commons.utils.DriverThread;
+import org.openo.nfvo.emsdriver.commons.utils.StringUtil;
+import org.openo.nfvo.emsdriver.commons.utils.XmlUtil;
+
+
+public class ConfigurationManager extends DriverThread{
+
+ /**
+ * ESM Cache
+ */
+ private static Map<String, EMSInfo> emsInfoCache = new ConcurrentHashMap<String, EMSInfo>();
+
+ private static Properties properties = null;
+
+ private final static String ftpconfig = Constant.SYS_CFG + "ftpconfig.properties";
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void dispose() {
+ String path = Constant.SYS_CFG + "EMSInfo.xml";
+ File cfg = new File(path);
+ this.log.debug("start loading " + path);
+ if(!cfg.exists() || !cfg.isFile()){
+ this.log.debug("not exists " + path);
+ return;
+ }
+
+ InputStream is = null;
+ Map<String, EMSInfo> tmpcache = new HashMap<String, EMSInfo>();
+
+ try {
+ is = new FileInputStream(cfg);
+ Document doc = XmlUtil.getDocument(is);
+
+ Element root = doc.getRootElement();
+
+ List<Element> children = root.getChildren();
+
+ for(Iterator<Element> it = children.iterator();it.hasNext();){
+ EMSInfo emsInfo = new EMSInfo();
+ Element child = it.next();
+ String name = child.getAttributeValue("name");
+ if(StringUtil.isBank(name)){
+ continue;
+ }
+ emsInfo.setName(name);
+
+ tmpcache.put(name, emsInfo);
+
+ List<Element> collectList = child.getChildren();
+ for(Element collect : collectList){
+
+ CollectVo collectVo = new CollectVo();
+
+ String type = collect.getAttributeValue("type");
+ if("alarm".equalsIgnoreCase(type)){
+ boolean iscollect = Boolean.parseBoolean(collect.getAttributeValue("iscollect"));
+ if(iscollect){
+ collectVo.setIscollect(iscollect);
+ }else{
+ continue;
+ }
+ collectVo.setType(type);
+ collectVo.setIP(collect.getChildText("ip"));
+ collectVo.setPort(collect.getChildText("port"));
+ collectVo.setUser(collect.getChildText("user"));
+ collectVo.setPassword(collect.getChildText("password"));
+ collectVo.setRead_timeout(collect.getChildText("readtimeout"));
+ }else{
+ String crontab = collect.getAttributeValue("crontab");
+ if(!StringUtil.isBank(type) && !StringUtil.isBank(crontab)){
+ collectVo.setType(type);
+ collectVo.setCrontab(crontab);
+ }else{
+ continue;
+ }
+ collectVo.setIP(collect.getChildText("ip"));
+ collectVo.setPort(collect.getChildText("port"));
+ collectVo.setUser(collect.getChildText("user"));
+ collectVo.setPassword(collect.getChildText("password"));
+ collectVo.setRemotepath(collect.getChildText("remotepath"));
+ collectVo.setMatch(collect.getChildText("match"));
+ collectVo.setPassive(collect.getChildText("passive"));
+ collectVo.setFtptype(collect.getChildText("ftptype"));
+ collectVo.setGranularity(collect.getChildText("granularity"));
+ }
+
+ emsInfo.putCollectMap(type, collectVo);
+ }
+ tmpcache.put(name, emsInfo);
+ }
+ emsInfoCache.putAll(tmpcache);
+
+ } catch (Exception e) {
+ log.error("load EMSInfo.xml is error "+StringUtil.getStackTrace(e));
+ }finally{
+ tmpcache.clear();
+ try {
+ if(is != null){
+ is.close();
+ is = null;
+ }
+ } catch (Exception e2) {
+ }
+ cfg = null;
+ }
+
+
+ //this.log.debug("start loading " + cacheFilePath);
+ File file = new File(ftpconfig);
+ if(!file.exists() || !file.isFile()){
+ this.log.error("cacheFilePath " + ftpconfig+"not exist or is not File");
+ return;
+ }
+ InputStream in = null;
+ try{
+ properties = new Properties();
+ in = new FileInputStream(file);
+ properties.load(in);
+
+ }catch(Exception e) {
+ log.error("read ["+file.getAbsolutePath()+"]Exception :",e);
+ }finally {
+ if(in != null) {
+ try {
+ in.close();
+ } catch (Exception e) {
+ }
+ }
+ }
+ }
+
+ public static synchronized List<EMSInfo> getAllEMSInfos(){
+ List<EMSInfo> list = new ArrayList<EMSInfo>();
+ for(EMSInfo emsinfo :emsInfoCache.values()){
+ list.add(emsinfo);
+ }
+ return list;
+ }
+
+ public static synchronized EMSInfo getEMSInfoByName(String emsName){
+ EMSInfo emsInfo= emsInfoCache.get(emsName);
+ return emsInfo;
+ }
+
+ public static synchronized Properties getProperties() {
+ return properties;
+ }
+
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.messagemgr;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+
+public class MessageChannel {
+
+ private BlockingQueue<Object> queue = null;
+
+ public MessageChannel(int size){
+ if(size>0){
+ queue = new LinkedBlockingQueue<Object>(size);
+ }else{
+ queue = new LinkedBlockingQueue<Object>();
+ }
+ }
+
+ public MessageChannel(){
+ queue = new LinkedBlockingQueue<Object>();
+ }
+ public void put(Object msg) throws InterruptedException{
+ while(!queue.offer(msg)){
+ queue.poll();
+ }
+ }
+
+ public Object get() throws InterruptedException{
+ return queue.take();
+ }
+
+ public Object poll() throws InterruptedException{
+ return queue.poll(100, TimeUnit.MILLISECONDS);
+ }
+
+ public int size(){
+ return queue.size();
+ }
+
+ public void clear(){
+ queue.clear();
+ }
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.messagemgr;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class MessageChannelFactory {
+
+ public static Map<String, MessageChannel> map = new HashMap<String, MessageChannel>();
+
+ public synchronized static MessageChannel getMessageChannel(String key,Integer size){
+ if(map.get(key) != null){
+ return map.get(key);
+ }
+ MessageChannel mc = null;
+ if(size != null && size > 0){
+ mc = new MessageChannel(size);
+ }else{
+ mc = new MessageChannel();
+ }
+
+ map.put(key, mc);
+ return mc;
+ }
+
+ public synchronized static MessageChannel getMessageChannel(String key){
+ if(map.get(key) != null){
+ return map.get(key);
+ }
+ MessageChannel mc = new MessageChannel();
+
+ map.put(key, mc);
+ return mc;
+ }
+
+ public synchronized static boolean destroyMessageChannel(String key){
+ if(map.get(key) != null){
+ map.remove(key);
+ return true;
+ }
+ return false;
+ }
+
+ public synchronized static void clean(){
+ map.clear();
+ }
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.northbound.client;
+
+import java.security.cert.CertificateException;
+import java.security.cert.X509Certificate;
+
+import javax.net.ssl.SSLContext;
+
+import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
+import org.apache.http.conn.ssl.TrustStrategy;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.ssl.SSLContextBuilder;
+
+/**
+ * HttpClient
+ */
+public class HttpClientFactory{
+
+
+ public static CloseableHttpClient getSSLClientFactory() throws Exception {
+
+ SSLContext sslContext = new SSLContextBuilder().loadTrustMaterial(null, new TrustStrategy() {
+ //信任所有
+ public boolean isTrusted(X509Certificate[] chain,
+ String authType) throws CertificateException {
+ return true;
+ }
+ }).build();
+ SSLConnectionSocketFactory sslsf = new SSLConnectionSocketFactory(sslContext);
+ CloseableHttpClient httpclient = HttpClients.custom().setSSLSocketFactory(sslsf).build();
+
+ return httpclient;
+ }
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.northbound.client;
+
+import java.io.IOException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.http.HttpEntity;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpDelete;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.util.EntityUtils;
+
+/*
+ * HttpClient post request
+ */
+public class HttpClientUtil {
+
+ private static Log log = LogFactory.getLog(HttpClientUtil.class);
+
+ public static String doPost(String url,String json,String charset){
+ CloseableHttpClient httpClient = null;
+ HttpPost httpPost = null;
+ String result = null;
+ try{
+ httpClient = HttpClientFactory.getSSLClientFactory();
+ httpPost = new HttpPost(url);
+ if (null != json) {
+ StringEntity s = new StringEntity(json);
+ s.setContentEncoding("UTF-8");
+ s.setContentType("application/json"); // set contentType
+ httpPost.setEntity(s);
+ }
+ CloseableHttpResponse response = httpClient.execute(httpPost);
+ try {
+ if(response != null){
+ HttpEntity resEntity = response.getEntity();
+ if(resEntity != null){
+ result = EntityUtils.toString(resEntity,charset);
+ }
+ }
+ } catch (Exception e) {
+ log.error("httpClient.execute(httpPost) is fail",e);
+ }finally{
+ if(response != null){
+ response.close();
+ }
+ }
+ }catch(Exception e){
+ log.error("doPost is fail ",e);
+ }finally{
+ if(httpClient != null){
+ try {
+ httpClient.close();
+ } catch (IOException e) {
+ }
+ }
+
+ }
+ return result;
+ }
+
+ public static String doDelete(String url ,String charset){
+ CloseableHttpClient httpClient = null;
+ HttpDelete httpDelete = null;
+ String result = null;
+ try{
+ httpClient = HttpClientFactory.getSSLClientFactory();
+ httpDelete = new HttpDelete(url);
+
+ CloseableHttpResponse response = httpClient.execute(httpDelete);
+
+ try {
+ if(response != null){
+ HttpEntity resEntity = response.getEntity();
+ if(resEntity != null){
+ result = EntityUtils.toString(resEntity,charset);
+ }
+ }
+ } catch (Exception e) {
+ log.error("",e);
+ }finally{
+ if(response != null){
+ response.close();
+ }
+ }
+ }catch(Exception e){
+ log.error("doDelete is fail ",e);
+ }finally{
+ if(httpClient != null){
+ try {
+ httpClient.close();
+ } catch (IOException e) {
+ }
+ }
+ }
+ return result;
+ }
+
+ public static String doGet(String url, String charset){
+ CloseableHttpClient httpClient = null;
+ HttpGet httpGet = null;
+ String result = null;
+ try{
+ httpClient = HttpClientFactory.getSSLClientFactory();
+ httpGet = new HttpGet(url);
+
+ CloseableHttpResponse response = httpClient.execute(httpGet);
+
+ try {
+ if(response != null){
+ HttpEntity resEntity = response.getEntity();
+ if(resEntity != null){
+ result = EntityUtils.toString(resEntity,charset);
+ }
+ }
+ } catch (Exception e) {
+ log.error("",e);
+ }finally{
+ if(response != null){
+ response.close();
+ }
+ }
+ }catch(Exception e){
+ log.error("doGet is fail ",e);
+ }finally{
+ if(httpClient != null){
+ try {
+ httpClient.close();
+ } catch (IOException e) {
+ }
+ }
+ }
+ return result;
+ }
+
+}
\ No newline at end of file
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.northbound.client;
+
+import java.util.Properties;
+
+import org.openo.nfvo.emsdriver.commons.constant.Constant;
+import org.openo.nfvo.emsdriver.commons.utils.DriverThread;
+import org.openo.nfvo.emsdriver.configmgr.ConfigurationInterface;
+import org.openo.nfvo.emsdriver.messagemgr.MessageChannel;
+import org.openo.nfvo.emsdriver.messagemgr.MessageChannelFactory;
+
+public class NorthMessageMgr extends DriverThread{
+
+ private MessageChannel alarmChannel = MessageChannelFactory.getMessageChannel(Constant.ALARM_CHANNEL_KEY);
+ private MessageChannel collectResultChannel = MessageChannelFactory.getMessageChannel(Constant.COLLECT_RESULT_CHANNEL_KEY);
+ private ConfigurationInterface configurationInterface ;
+
+ private boolean threadStop = false;
+
+ @Override
+ public void dispose() {
+ //
+ new AlarmMessageRecv().start();
+
+ new ResultMessageRecv().start();
+ }
+
+
+ class AlarmMessageRecv extends Thread{
+ long timeStamp = System.currentTimeMillis();
+
+ public void run() {
+ while(!threadStop){
+
+ try {
+ if(System.currentTimeMillis() - timeStamp > Constant.ONEMINUTE){
+ timeStamp = System.currentTimeMillis();
+
+ log.debug("ALARM_CHANNEL Msg size :"+alarmChannel.size());
+ }
+
+ Object obj = alarmChannel.poll();
+ if(obj == null){
+ continue;
+ }
+ if(obj instanceof String){
+ //http
+ Properties properties = configurationInterface.getProperties();
+ String msbAddress = properties.getProperty("msbAddress");
+ String url = properties.getProperty("alarm");
+ String postUrl = "http://"+msbAddress+url;
+
+ HttpClientUtil.doPost(postUrl, (String)obj, Constant.ENCODING_UTF8);
+ }
+
+ } catch (Exception e) {
+ log.error("AlarmMessageRecv exception",e);
+ }
+ }
+ }
+ }
+
+ class ResultMessageRecv extends Thread{
+ long timeStamp = System.currentTimeMillis();
+
+ public void run() {
+ while(!threadStop){
+
+ try {
+ if(System.currentTimeMillis() - timeStamp > Constant.ONEMINUTE){
+ timeStamp = System.currentTimeMillis();
+
+ log.debug("COLLECT_RESULT_CHANNEL Msg size :"+collectResultChannel.size());
+ }
+
+ Object obj = collectResultChannel.poll();
+ if(obj == null){
+ continue;
+ }
+ if(obj instanceof String){
+ //http
+ Properties properties = configurationInterface.getProperties();
+ String msbAddress = properties.getProperty("msbAddress");
+ String url = properties.getProperty("dataNotifyUrl");
+ String postUrl = "http://"+msbAddress+url;
+ HttpClientUtil.doPost(postUrl, (String)obj, Constant.ENCODING_UTF8);
+ }
+
+ } catch (Exception e) {
+ log.error("AlarmMessageRecv exception",e);
+ }
+ }
+ }
+ }
+
+ /**
+ * @return the configurationInterface
+ */
+ public ConfigurationInterface getConfigurationInterface() {
+ return configurationInterface;
+ }
+
+ /**
+ * @param configurationInterface the configurationInterface to set
+ */
+ public void setConfigurationInterface(
+ ConfigurationInterface configurationInterface) {
+ this.configurationInterface = configurationInterface;
+ }
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.northbound.service;
+
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.core.MediaType;
+
+import com.codahale.metrics.annotation.Timed;
+
+@Path("/ems-driver")
+@Produces(MediaType.APPLICATION_JSON)
+public class CommandResource {
+
+
+ @GET
+ @Timed
+ public String executeCommand(@QueryParam("command") String command) {
+
+ System.out.println("receiver command = "+command);
+ return command;
+ }
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.northbound.service;
+
+import io.dropwizard.Application;
+import io.dropwizard.jetty.HttpConnectorFactory;
+import io.dropwizard.server.DefaultServerFactory;
+import io.dropwizard.setup.Bootstrap;
+import io.dropwizard.setup.Environment;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.openo.nfvo.emsdriver.serviceregister.MsbConfiguration;
+import org.openo.nfvo.emsdriver.serviceregister.MsbRestServiceProxy;
+import org.openo.nfvo.emsdriver.serviceregister.model.MsbRegisterVo;
+import org.openo.nfvo.emsdriver.serviceregister.model.ServiceNodeVo;
+
+public class EmsDriverApplication extends Application<EmsDriverConfiguration> {
+
+ protected static Log log = LogFactory.getLog(EmsDriverApplication.class);
+
+ public static void main(String[] args) throws Exception {
+ new EmsDriverApplication().run(args);
+ }
+
+ @Override
+ public String getName() {
+ return "ems-driver";
+ }
+
+ @Override
+ public void initialize(Bootstrap<EmsDriverConfiguration> bootstrap) {
+ // nothing to do yet
+ }
+
+ @Override
+ public void run(EmsDriverConfiguration configuration,Environment environment) {
+ // register CommandResource
+ environment.jersey().register(new CommandResource());
+
+
+ MsbConfiguration.setMsbAddress(configuration.getMsbAddress());
+ //MSB register
+ this.msbRegisteEmsDriverService(configuration);
+ }
+
+ private void msbRegisteEmsDriverService(EmsDriverConfiguration configuration) {
+ DefaultServerFactory defaultServerFactory = (DefaultServerFactory)configuration.getServerFactory();
+ HttpConnectorFactory connector = (HttpConnectorFactory)defaultServerFactory.getAdminConnectors().get(0);
+ MsbRegisterVo registerVo = new MsbRegisterVo();
+ ServiceNodeVo serviceNode = new ServiceNodeVo();
+ String ip = "";
+ try {
+ ip = InetAddress.getLocalHost().getHostAddress();
+ } catch (UnknownHostException e) {
+ log.error("Unable to get host ip: " + e.getMessage());
+ }
+ if(ip.equals("")){
+ ip = connector.getBindHost();
+ }
+ serviceNode.setIp(ip);
+ serviceNode.setPort(String.valueOf(connector.getPort()));
+ serviceNode.setTtl(0);
+
+ List<ServiceNodeVo> nodeList = new ArrayList<ServiceNodeVo>();
+ nodeList.add(serviceNode);
+ registerVo.setServiceName("emsdriver");
+ registerVo.setUrl("/openoapi/emsdriver/v1");
+ registerVo.setNodes(nodeList);
+
+ MsbRestServiceProxy.registerService(registerVo);
+ log.info("register monitor-umc service to msb finished.");
+
+ }
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.northbound.service;
+
+import io.dropwizard.Configuration;
+
+import org.hibernate.validator.constraints.NotEmpty;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class EmsDriverConfiguration extends Configuration{
+
+ @NotEmpty
+ private String msbAddress;
+
+ @NotEmpty
+ private String defaultName = "EmsDriver-Stranger";
+
+ @JsonProperty
+ public String getMsbAddress() {
+ return msbAddress;
+ }
+
+ @JsonProperty
+ public void setMsbAddress(String msbAddress) {
+ this.msbAddress = msbAddress;
+ }
+
+ @JsonProperty
+ public String getDefaultName() {
+ return defaultName;
+ }
+
+ @JsonProperty
+ public void setDefaultName(String name) {
+ this.defaultName = name;
+ }
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.serviceregister;
+
+public class MsbConfiguration {
+ protected static String msbAddress;
+
+ public static String getMsbAddress() {
+ return msbAddress;
+ }
+
+ public static void setMsbAddress(String msbAddress) {
+ MsbConfiguration.msbAddress ="http://"+ msbAddress;
+ }
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.serviceregister;
+
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.openo.nfvo.emsdriver.commons.constant.Constant;
+import org.openo.nfvo.emsdriver.northbound.client.HttpClientUtil;
+import org.openo.nfvo.emsdriver.serviceregister.model.MsbRegisterVo;
+import org.openo.nfvo.emsdriver.serviceregister.model.ServiceNodeVo;
+
+import com.alibaba.fastjson.JSON;
+
+public class MsbRestServiceProxy {
+
+ public static String registerService(MsbRegisterVo registerInfo){
+ String url = MsbConfiguration.getMsbAddress()+Constant.MSBAPIROOTDOMAIN;
+ String registerObj = JSON.toJSONString(registerInfo);
+
+ String registerResponse = HttpClientUtil.doPost(url, registerObj, Constant.ENCODING_UTF8);
+ return registerResponse;
+ }
+
+ public static void unRegiserService(String serviceName,String version,String ip,String port){
+ String url = MsbConfiguration.getMsbAddress()+Constant.MSBAPIROOTDOMAIN+"/"+serviceName+"/version/"+version+"/nodes/"+ip+"/"+port;
+ HttpClientUtil.doDelete(url, Constant.ENCODING_UTF8);
+ }
+
+ public static List<String> queryService(String serviceName,String version){
+ List<String> ipList = new ArrayList<String>();
+ String url = MsbConfiguration.getMsbAddress()+Constant.MSBAPIROOTDOMAIN+"/"+serviceName+"/version/"+version;
+ String response = HttpClientUtil.doGet(url, Constant.ENCODING_UTF8);
+ if(!response.equals("")){
+ MsbRegisterVo msbRegisterVo = JSON.parseObject(response,MsbRegisterVo.class);
+ List<ServiceNodeVo> nodeList = msbRegisterVo.getNodes();
+
+ for(ServiceNodeVo node :nodeList){
+ String ip = node.getIp();
+ ipList.add(ip);
+ }
+ }
+ return ipList;
+
+ }
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.serviceregister.model;
+
+import java.util.List;
+
+
+public class MsbRegisterVo {
+ private String serviceName = "emsdriver";
+ private String version = "v1";
+ private String url = "/openoapi/emsdriver/v1";
+ private String protocol = "REST";
+ private String visualRange = "1";
+ private List<ServiceNodeVo> nodes;
+
+
+ /**
+ * @return the serviceName
+ */
+ public String getServiceName() {
+ return serviceName;
+ }
+ /**
+ * @param serviceName the serviceName to set
+ */
+ public void setServiceName(String serviceName) {
+ this.serviceName = serviceName;
+ }
+ /**
+ * @return the version
+ */
+ public String getVersion() {
+ return version;
+ }
+ /**
+ * @param version the version to set
+ */
+ public void setVersion(String version) {
+ this.version = version;
+ }
+ /**
+ * @return the url
+ */
+ public String getUrl() {
+ return url;
+ }
+ /**
+ * @param url the url to set
+ */
+ public void setUrl(String url) {
+ this.url = url;
+ }
+ /**
+ * @return the protocol
+ */
+ public String getProtocol() {
+ return protocol;
+ }
+ /**
+ * @param protocol the protocol to set
+ */
+ public void setProtocol(String protocol) {
+ this.protocol = protocol;
+ }
+ /**
+ * @return the visualRange
+ */
+ public String getVisualRange() {
+ return visualRange;
+ }
+ /**
+ * @param visualRange the visualRange to set
+ */
+ public void setVisualRange(String visualRange) {
+ this.visualRange = visualRange;
+ }
+ /**
+ * @return the nodes
+ */
+ public List<ServiceNodeVo> getNodes() {
+ return nodes;
+ }
+ /**
+ * @param nodes the nodes to set
+ */
+ public void setNodes(List<ServiceNodeVo> nodes) {
+ this.nodes = nodes;
+ }
+
+
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.serviceregister.model;
+
+
+public class ServiceNodeVo {
+ private String ip;
+ private String port;
+ private int ttl;
+
+ /**
+ * @return the ip
+ */
+ public String getIp() {
+ return ip;
+ }
+ /**
+ * @param ip the ip to set
+ */
+ public void setIp(String ip) {
+ this.ip = ip;
+ }
+ /**
+ * @return the port
+ */
+ public String getPort() {
+ return port;
+ }
+ /**
+ * @param port the port to set
+ */
+ public void setPort(String port) {
+ this.port = port;
+ }
+ /**
+ * @return the ttl
+ */
+ public int getTtl() {
+ return ttl;
+ }
+ /**
+ * @param ttl the ttl to set
+ */
+ public void setTtl(int ttl) {
+ this.ttl = ttl;
+ }
+
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.taskscheduler;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.openo.nfvo.emsdriver.commons.constant.Constant;
+import org.openo.nfvo.emsdriver.commons.model.CollectVo;
+import org.openo.nfvo.emsdriver.commons.model.EMSInfo;
+import org.openo.nfvo.emsdriver.commons.utils.DriverThread;
+import org.openo.nfvo.emsdriver.configmgr.ConfigurationInterface;
+import org.quartz.Job;
+
+
+public class CollectManager extends DriverThread{
+
+ private ConfigurationInterface configurationInterface;
+ public void dispose() {
+ if(configurationInterface != null){
+ List<EMSInfo> emsInfos = configurationInterface.getAllEMSInfo();
+ while(isRun() && emsInfos.size() == 0){
+
+ emsInfos = configurationInterface.getAllEMSInfo();
+ if(emsInfos.size() == 0){
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ }
+ }
+
+ }
+
+ List<CollectVo> collectVos = new ArrayList<CollectVo>();
+ for(EMSInfo emsInfo : emsInfos){
+ //cm
+ CollectVo CollectVoCm = emsInfo.getCollectVoByType(Constant.COLLECT_TYPE_CM);
+ if(CollectVoCm != null){
+ CollectVoCm.setEmsName(emsInfo.getName());
+ collectVos.add(CollectVoCm);
+ }
+
+ //pm
+ CollectVo CollectVoPm = emsInfo.getCollectVoByType(Constant.COLLECT_TYPE_PM);
+ if(CollectVoPm != null){
+ CollectVoPm.setEmsName(emsInfo.getName());
+ collectVos.add(CollectVoPm);
+ }
+
+ }
+ this.addCollectJob(collectVos);
+ log.info("addCollectJob is OK ");
+ }else{
+ log.error("configurationInterface = null,check spring.xml");
+ }
+ }
+
+ private void addCollectJob(List<CollectVo> collectVos) {
+ for(CollectVo collectVo : collectVos){
+ try {
+ String jobName = collectVo.getEmsName()+"_"+collectVo.getType()+collectVo.getIP();
+ Job job = new CollectOderJob();
+ String jobClass = job.getClass().getName();
+ String time = collectVo.getCrontab();
+ if(time != null && !"".equals(time)){
+ QuartzManager.addJob(jobName, jobClass, time,collectVo);
+ }else{
+ log.error("type =["+collectVo.getType()+"]ip=["+collectVo.getIP()+"] crontab is null,check EMSInfo.xml");
+ }
+
+ } catch (Exception e) {
+ log.error("addJob is error",e);
+ }
+ }
+ }
+
+ /**
+ * @return the configurationInterface
+ */
+ public ConfigurationInterface getConfigurationInterface() {
+ return configurationInterface;
+ }
+
+ /**
+ * @param configurationInterface the configurationInterface to set
+ */
+ public void setConfigurationInterface(
+ ConfigurationInterface configurationInterface) {
+ this.configurationInterface = configurationInterface;
+ }
+
+
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.taskscheduler;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.openo.nfvo.emsdriver.commons.constant.Constant;
+import org.openo.nfvo.emsdriver.commons.model.CollectMsg;
+import org.openo.nfvo.emsdriver.commons.model.CollectVo;
+import org.openo.nfvo.emsdriver.commons.utils.DriverThread;
+import org.openo.nfvo.emsdriver.messagemgr.MessageChannel;
+import org.openo.nfvo.emsdriver.messagemgr.MessageChannelFactory;
+import org.quartz.Job;
+import org.quartz.JobExecutionContext;
+import org.quartz.JobExecutionException;
+
+
+public class CollectOderJob implements Job {
+
+ private Log log = LogFactory.getLog(DriverThread.class);
+ private MessageChannel collectChannel = MessageChannelFactory.getMessageChannel(Constant.COLLECT_CHANNEL_KEY);
+
+ public void execute(JobExecutionContext context) throws JobExecutionException {
+ // TODO Auto-generated method stub
+ CollectVo collectVo = (CollectVo)context.getJobDetail().getJobDataMap().get("collectVo");
+ if(collectVo != null){
+ CollectMsg collectMsg = new CollectMsg();
+ collectMsg.setEmsName(collectVo.getEmsName());
+ collectMsg.setId(System.nanoTime());
+ collectMsg.setType(collectVo.getType());
+
+ try {
+ collectChannel.put(collectMsg);
+ } catch (InterruptedException e) {
+ log.error("collectChannel.put is error ",e);
+ }
+ }else{
+ log.error("collectVo is null, collectMsg is not created! ");
+ }
+
+
+ }
+
+}
--- /dev/null
+/**
+ * Copyright 2017 BOCO Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openo.nfvo.emsdriver.taskscheduler;
+
+
+import java.text.ParseException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.openo.nfvo.emsdriver.commons.model.CollectVo;
+import org.quartz.CronTrigger;
+import org.quartz.Job;
+import org.quartz.JobDetail;
+import org.quartz.Scheduler;
+import org.quartz.SchedulerException;
+import org.quartz.SchedulerFactory;
+import org.quartz.impl.StdSchedulerFactory;
+
+public class QuartzManager {
+
+
+ private static Log log = LogFactory.getFactory().getInstance(QuartzManager.class);
+ private static SchedulerFactory gSchedulerFactory = new StdSchedulerFactory();
+ private static String JOB_GROUP_NAME = "EXTJWEB_JOBGROUP_NAME";
+ private static String TRIGGER_GROUP_NAME = "EXTJWEB_TRIGGERGROUP_NAME";
+
+ /**
+ * @param jobName
+ * @param job
+ * @param time
+ * @throws SchedulerException
+ * @throws ParseException
+ */
+ public static boolean addJob(String jobName, String jobClass, String time,CollectVo collectVo) {
+ boolean sucess = false;
+ try {
+ Scheduler sched = gSchedulerFactory.getScheduler();
+ JobDetail jobDetail = new JobDetail(jobName, JOB_GROUP_NAME, Class.forName(jobClass));
+
+ CronTrigger trigger = new CronTrigger(jobName, TRIGGER_GROUP_NAME);
+ trigger.setCronExpression(time);
+
+ jobDetail.getJobDataMap().put("collectVo", collectVo);
+ sched.scheduleJob(jobDetail, trigger);
+ if (!sched.isShutdown()){
+ sched.start();
+
+ }
+ sucess = true;
+ } catch (Exception e) {
+ log.error("add job fail cronExpression="+time,e);
+ sucess = false;
+ }
+ return sucess;
+ }
+
+
+
+ /**
+ * @param jobName
+ * @param time
+ */
+ @SuppressWarnings("unchecked")
+ public static boolean modifyJobTime(String jobName, String time,CollectVo collectVo) {
+ boolean sucess = false;
+ try {
+ Scheduler sched = gSchedulerFactory.getScheduler();
+ CronTrigger trigger = (CronTrigger) sched.getTrigger(jobName, TRIGGER_GROUP_NAME);
+ if(trigger == null) {
+ return false;
+ }
+ String oldTime = trigger.getCronExpression();
+ if (!oldTime.equalsIgnoreCase(time)) {
+ JobDetail jobDetail = sched.getJobDetail(jobName, JOB_GROUP_NAME);
+
+ Class<Job> objJobClass = jobDetail.getJobClass();
+ String jobClass = objJobClass.getName();
+ removeJob(jobName);
+
+ addJob(jobName, jobClass, time,collectVo);
+ }
+ sucess = true;
+ } catch (Exception e) {
+ log.error("modifyJobTime fail cronExpression="+time,e);
+ sucess = false;
+ }
+ return sucess ;
+ }
+
+ /**
+
+
+ /**
+ * @param jobName
+ */
+ public static boolean removeJob(String jobName) {
+ boolean sucess = false;
+ try {
+ Scheduler sched = gSchedulerFactory.getScheduler();
+ sched.pauseTrigger(jobName, TRIGGER_GROUP_NAME);
+ sched.unscheduleJob(jobName, TRIGGER_GROUP_NAME);
+ sched.deleteJob(jobName, JOB_GROUP_NAME);
+ sucess = true;
+ } catch (Exception e) {
+ sucess = false;
+ log.error("remove job fail jobName="+jobName,e);
+ }
+ return sucess;
+ }
+
+
+
+ /**
+ *
+ * @return
+ */
+ public static boolean startJobs() {
+ boolean sucess = false;
+ try {
+ Scheduler sched = gSchedulerFactory.getScheduler();
+ sched.start();
+ sucess = true;
+ } catch (Exception e) {
+ sucess = false;
+ log.error("start jobs fail",e);
+ }
+ return sucess;
+ }
+
+ /**
+ *
+ * @return
+ */
+ public static boolean shutdownJobs() {
+ boolean sucess = false;
+ try {
+ Scheduler sched = gSchedulerFactory.getScheduler();
+ if(!sched.isShutdown()) {
+ sched.shutdown();
+ }
+ sucess = true;
+ } catch (Exception e) {
+ sucess = false;
+ log.error("shutdown jobs fail ",e);
+ }
+
+ return sucess;
+ }
+}
Code Review / vfc / nfvo / driver / ems.git / commitdiff