From ea5bf0a1c8a4e525d2cee03841b8e8f9b3563ed0 Mon Sep 17 00:00:00 2001 From: Dan Timoney Date: Tue, 18 Jul 2017 20:32:15 -0400 Subject: [CCSDK-6] Populate seed code Add seed code for sli/northbound repository Update groupId to org.onap.ccsdk.sli.northbound Update to use CCSDK version of sli core Change-Id: Id3a154a53150a74f4b65060544e76f3e0cad932e Signed-off-by: Dan Timoney --- .../openecomp/sdnc/dmaapclient/DmaapListener.java | 165 +++++++++++++++++ .../sdnc/dmaapclient/DummyDmaapConsumer.java | 37 ++++ .../sdnc/dmaapclient/InvalidMessageException.java | 37 ++++ .../sdnc/dmaapclient/SdncDmaapConsumer.java | 146 +++++++++++++++ .../dmaapclient/SdncFlatJsonDmaapConsumer.java | 196 +++++++++++++++++++++ .../sdnc/dmaapclient/SdncOdlConnection.java | 159 +++++++++++++++++ 6 files changed, 740 insertions(+) create mode 100644 dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/DmaapListener.java create mode 100644 dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/DummyDmaapConsumer.java create mode 100644 dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/InvalidMessageException.java create mode 100644 dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/SdncDmaapConsumer.java create mode 100644 dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/SdncFlatJsonDmaapConsumer.java create mode 100644 dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/SdncOdlConnection.java (limited to 'dmaap-listener/src/main/java/org') diff --git a/dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/DmaapListener.java b/dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/DmaapListener.java new file mode 100644 index 000000000..7096d83bf --- /dev/null +++ b/dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/DmaapListener.java @@ -0,0 +1,165 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : SDN-C + * ================================================================================ + * Copyright (C) 2017 ONAP Intellectual Property. All rights + * reserved. + * ================================================================================ + * 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. + * ============LICENSE_END========================================================= + */ + +package org.openecomp.sdnc.dmaapclient; + +import java.io.File; +import java.io.FileInputStream; +import java.util.LinkedList; +import java.util.List; +import java.util.Properties; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class DmaapListener { + + private static final String DMAAP_LISTENER_PROPERTIES = "dmaap-listener.properties"; + private static final String SDNC_CONFIG_DIR = "SDNC_CONFIG_DIR"; + private static final Logger LOG = LoggerFactory + .getLogger(DmaapListener.class); + + public static void main(String[] args) { + + Properties properties = new Properties(); + + + String propFileName = DMAAP_LISTENER_PROPERTIES; + + if (args.length > 0) { + propFileName = args[0]; + } + + String propPath = null; + String propDir = System.getenv(SDNC_CONFIG_DIR); + + List consumers = new LinkedList(); + + if (propDir == null) { + + propDir = "/opt/sdnc/data/properties"; + } + + if (!propFileName.startsWith("/")) { + propPath = propDir + "/" + propFileName; + } + + File propFile = new File(propPath); + + if (!propFile.canRead()) { + LOG.error("Cannot read properties file "+propPath); + System.exit(1); + } + + try { + properties.load(new FileInputStream(propFile)); + } catch (Exception e) { + LOG.error("Caught exception loading properties from "+propPath, e); + System.exit(1); + } + + String subscriptionStr = properties.getProperty("subscriptions"); + + boolean threadsRunning = false; + + LOG.debug("Dmaap subscriptions : "+subscriptionStr); + + if (subscriptionStr != null) { + String[] subscriptions = subscriptionStr.split(";"); + + for (int i = 0; i < subscriptions.length; i++) { + String[] subscription = subscriptions[i].split(":"); + String consumerClassName = subscription[0]; + String propertyPath = subscription[1]; + + LOG.debug("Handling subscription [" + consumerClassName + "," + propertyPath + "]"); + + if (propertyPath == null) { + LOG.error("Invalid subscription (" + subscriptions[i] + ") property file missing"); + continue; + } + + if (!propertyPath.startsWith("/")) { + propertyPath = propDir + "/" + propertyPath; + } + + Class consumerClass = null; + + try { + consumerClass = Class.forName(consumerClassName); + } catch (Exception e) { + LOG.error("Could not find DMaap consumer class " + consumerClassName); + } + + if (consumerClass != null) { + + SdncDmaapConsumer consumer = null; + + try { + consumer = (SdncDmaapConsumer) consumerClass.newInstance(); + } catch (Exception e) { + LOG.error("Could not create consumer from class " + consumerClassName, e); + } + + if (consumer != null) { + LOG.debug("Initializing consumer " + consumerClassName + "(" + propertyPath + ")"); + consumer.init(properties, propertyPath); + + if (consumer.isReady()) { + Thread consumerThread = new Thread(consumer); + consumerThread.start(); + consumers.add(consumer); + threadsRunning = true; + LOG.info("Started consumer thread (" + consumerClassName + " : " + propertyPath + ")"); + } else { + LOG.debug("Consumer " + consumerClassName + " is not ready"); + } + } + + } + + } + } + + while (threadsRunning) { + + threadsRunning = false; + for (SdncDmaapConsumer consumer : consumers) { + if (consumer.isRunning()) { + threadsRunning = true; + } + } + + if (!threadsRunning) { + break; + } + + try { + Thread.sleep(10000); + } catch (InterruptedException e) { + + } + } + + LOG.info("No listener threads running - exitting"); + + } +} diff --git a/dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/DummyDmaapConsumer.java b/dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/DummyDmaapConsumer.java new file mode 100644 index 000000000..5edc1c908 --- /dev/null +++ b/dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/DummyDmaapConsumer.java @@ -0,0 +1,37 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : SDN-C + * ================================================================================ + * Copyright (C) 2017 ONAP Intellectual Property. All rights + * reserved. + * ================================================================================ + * 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. + * ============LICENSE_END========================================================= + */ + +package org.openecomp.sdnc.dmaapclient; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class DummyDmaapConsumer extends SdncDmaapConsumer { + + private static final Logger LOG = LoggerFactory + .getLogger(DummyDmaapConsumer.class); + + @Override + public void processMsg(String msg) { + LOG.info("Consumed message: \n"+msg); + } + +} diff --git a/dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/InvalidMessageException.java b/dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/InvalidMessageException.java new file mode 100644 index 000000000..3de5df6c2 --- /dev/null +++ b/dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/InvalidMessageException.java @@ -0,0 +1,37 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : SDN-C + * ================================================================================ + * Copyright (C) 2017 ONAP Intellectual Property. All rights + * reserved. + * ================================================================================ + * 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. + * ============LICENSE_END========================================================= + */ + +package org.openecomp.sdnc.dmaapclient; + +public class InvalidMessageException extends Exception { + + public InvalidMessageException() { + super(); + } + + public InvalidMessageException(String msg) { + super(msg); + } + + public InvalidMessageException(String msg, Throwable t) { + super(msg, t); + } +} diff --git a/dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/SdncDmaapConsumer.java b/dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/SdncDmaapConsumer.java new file mode 100644 index 000000000..5b55bda5b --- /dev/null +++ b/dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/SdncDmaapConsumer.java @@ -0,0 +1,146 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : SDN-C + * ================================================================================ + * Copyright (C) 2017 ONAP Intellectual Property. All rights + * reserved. + * ================================================================================ + * 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. + * ============LICENSE_END========================================================= + */ + +package org.openecomp.sdnc.dmaapclient; + +import java.io.File; +import java.io.FileInputStream; +import java.util.Properties; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.att.nsa.mr.client.MRClientFactory; +import com.att.nsa.mr.client.MRConsumer; +import com.att.nsa.mr.client.response.MRConsumerResponse; + +public abstract class SdncDmaapConsumer implements Runnable { + + private static final Logger LOG = LoggerFactory + .getLogger(SdncDmaapConsumer.class); + + private String propertiesPath = ""; + private Properties properties = null; + MRConsumer consumer = null; + MRConsumerResponse consumerResponse = null; + boolean running = false; + boolean ready = false; + int fetchPause = 5000; // Default pause between fetchs - 5 seconds + + public boolean isReady() { + return ready; + } + + int timeout = 15000; // Default timeout - 15 seconds + + public boolean isRunning() { + return running; + } + + public SdncDmaapConsumer() { + + } + + public SdncDmaapConsumer(Properties properties, String propertiesPath) { + init(properties, propertiesPath); + } + + public String getProperty(String name) { + return(properties.getProperty(name, "")); + } + + public void init(Properties properties, String propertiesPath) { + + this.propertiesPath = propertiesPath; + + try { + + this.properties = (Properties) properties.clone(); + + this.properties.load(new FileInputStream(new File(propertiesPath))); + + String timeoutStr = properties.getProperty("timeout"); + + if ((timeoutStr != null) && (timeoutStr.length() > 0)) { + try { + timeout = Integer.parseInt(timeoutStr); + } catch (NumberFormatException e) { + LOG.error("Non-numeric value specified for timeout ("+timeoutStr+")"); + } + } + + String fetchPauseStr = properties.getProperty("fetchPause"); + if ((fetchPauseStr != null) && (fetchPauseStr.length() > 0)) { + try { + fetchPause = Integer.parseInt(fetchPauseStr); + } catch (NumberFormatException e) { + LOG.error("Non-numeric valud specified for fetchPause ("+fetchPauseStr+")"); + } + } + + this.consumer = MRClientFactory.createConsumer(propertiesPath); + ready = true; + } catch (Exception e) { + LOG.error("Error initializing DMaaP consumer from file "+propertiesPath, e); + } + } + + + @Override + public void run() { + if (ready) { + + running = true; + + while (running) { + + try { + boolean noData = true; + consumerResponse = consumer.fetchWithReturnConsumerResponse(timeout, -1); + for (String msg : consumerResponse.getActualMessages()) { + noData = false; + LOG.info("Received message from DMaaP:\n"+msg); + processMsg(msg); + } + + if (noData) { + if (fetchPause > 0) { + + LOG.info("No data received from fetch. Pausing "+fetchPause+" ms before retry"); + Thread.sleep(fetchPause); + } else { + + LOG.info("No data received from fetch. No fetch pause specified - retrying immediately"); + } + } + } catch (Exception e) { + LOG.error("Caught exception reading from DMaaP", e); + running = false; + } + + + } + } + + } + + abstract public void processMsg(String msg) throws InvalidMessageException; +} diff --git a/dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/SdncFlatJsonDmaapConsumer.java b/dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/SdncFlatJsonDmaapConsumer.java new file mode 100644 index 000000000..7e9c069ac --- /dev/null +++ b/dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/SdncFlatJsonDmaapConsumer.java @@ -0,0 +1,196 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : SDN-C + * ================================================================================ + * Copyright (C) 2017 ONAP Intellectual Property. All rights + * reserved. + * ================================================================================ + * 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. + * ============LICENSE_END========================================================= + */ + +package org.openecomp.sdnc.dmaapclient; + +import java.io.BufferedReader; +import java.io.DataOutputStream; +import java.io.File; +import java.io.FileReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.net.Authenticator; +import java.net.HttpURLConnection; +import java.net.PasswordAuthentication; +import java.net.URL; +import java.util.HashMap; +import java.util.Iterator; +import java.util.Map; + +import javax.net.ssl.HostnameVerifier; +import javax.net.ssl.HttpsURLConnection; +import javax.net.ssl.SSLSession; + +import org.apache.commons.codec.binary.Base64; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; + + +public class SdncFlatJsonDmaapConsumer extends SdncDmaapConsumer { + + private static final Logger LOG = LoggerFactory + .getLogger(SdncFlatJsonDmaapConsumer.class); + + private static final String DMAAPLISTENERROOT = "DMAAPLISTENERROOT"; + private static final String SDNC_ENDPOINT = "SDNC.endpoint"; + + + + @Override + public void processMsg(String msg) throws InvalidMessageException { + + processMsg(msg, null); + } + + public void processMsg(String msg, String mapDirName) throws InvalidMessageException { + + if (msg == null) { + throw new InvalidMessageException("Null message"); + } + + ObjectMapper oMapper = new ObjectMapper(); + JsonNode instarRootNode = null; + ObjectNode sdncRootNode = null; + + String instarMsgName = null; + + try { + instarRootNode = oMapper.readTree(msg); + } catch (Exception e) { + throw new InvalidMessageException("Cannot parse json object", e); + } + + Iterator> instarFields = instarRootNode.fields(); + + while (instarFields.hasNext()) { + Map.Entry entry = instarFields.next(); + + instarMsgName = entry.getKey(); + instarRootNode = entry.getValue(); + break; + } + + Map fieldMap = loadMap(instarMsgName, mapDirName); + + if (fieldMap == null) { + throw new InvalidMessageException("Unable to process message - cannot load field mappings"); + } + + if (!fieldMap.containsKey(SDNC_ENDPOINT)) { + throw new InvalidMessageException("No SDNC endpoint known for message "+instarMsgName); + } + + String sdncEndpoint = fieldMap.get(SDNC_ENDPOINT); + + sdncRootNode = oMapper.createObjectNode(); + ObjectNode inputNode = oMapper.createObjectNode(); + + + for (String fromField : fieldMap.keySet()) { + + if (!SDNC_ENDPOINT.equals(fromField)) { + JsonNode curNode = instarRootNode.get(fromField); + if (curNode != null) { + String fromValue = curNode.textValue(); + + inputNode.put(fieldMap.get(fromField), fromValue); + } + } + } + sdncRootNode.put("input", inputNode); + + try { + String rpcMsgbody = oMapper.writeValueAsString(sdncRootNode); + String odlUrlBase = getProperty("sdnc.odl.url-base"); + String odlUser = getProperty("sdnc.odl.user"); + String odlPassword = getProperty("sdnc.odl.password"); + + if ((odlUrlBase != null) && (odlUrlBase.length() > 0)) { + SdncOdlConnection conn = SdncOdlConnection.newInstance(odlUrlBase + sdncEndpoint, odlUser, odlPassword); + + conn.send("POST", "application/json", rpcMsgbody); + } else { + LOG.info("POST message body would be:\n"+rpcMsgbody); + } + } catch (Exception e) { + + } + + } + + private Map loadMap(String msgType, String mapDirName) { + Map results = new HashMap(); + + + if (mapDirName == null) { + String rootdir = System.getenv(DMAAPLISTENERROOT); + + if ((rootdir == null) || (rootdir.length() == 0)) { + rootdir = "/opt/app/dmaap-listener"; + } + + mapDirName = rootdir + "/lib"; + + } + + String mapFilename = mapDirName + "/" + msgType + ".map"; + + File mapFile = new File(mapFilename); + + if (!mapFile.canRead()) { + LOG.error("Cannot read map file ("+mapFilename+")"); + return(null); + } + + try { + BufferedReader mapReader = new BufferedReader(new FileReader(mapFile)); + + String curLine = null; + + while ((curLine = mapReader.readLine()) != null) { + curLine = curLine.trim(); + + if ((curLine.length() > 0) && (!curLine.startsWith("#"))) { + + if (curLine.contains("=>")) { + String[] entry = curLine.split("=>"); + if (entry.length == 2) { + results.put(entry[0].trim(), entry[1].trim()); + } + } + } + } + mapReader.close(); + } catch (Exception e) { + LOG.error("Caught exception reading map "+mapFilename, e); + return(null); + } + + return(results); + } + + + +} diff --git a/dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/SdncOdlConnection.java b/dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/SdncOdlConnection.java new file mode 100644 index 000000000..6c6f308fb --- /dev/null +++ b/dmaap-listener/src/main/java/org/openecomp/sdnc/dmaapclient/SdncOdlConnection.java @@ -0,0 +1,159 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : SDN-C + * ================================================================================ + * Copyright (C) 2017 ONAP Intellectual Property. All rights + * reserved. + * ================================================================================ + * 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. + * ============LICENSE_END========================================================= + */ + +package org.openecomp.sdnc.dmaapclient; + +import java.io.BufferedReader; +import java.io.DataOutputStream; +import java.io.IOException; +import java.io.InputStreamReader; +import java.net.Authenticator; +import java.net.HttpURLConnection; +import java.net.PasswordAuthentication; +import java.net.ProtocolException; +import java.net.URL; + +import javax.net.ssl.HostnameVerifier; +import javax.net.ssl.HttpsURLConnection; +import javax.net.ssl.SSLSession; + +import org.apache.commons.codec.binary.Base64; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +public class SdncOdlConnection { + + private static final Logger LOG = LoggerFactory + .getLogger(SdncOdlConnection.class); + + private HttpURLConnection httpConn = null; + + private String url = null; + private String user = null; + private String password = null; + + private class SdncAuthenticator extends Authenticator { + + private String user; + private String passwd; + + SdncAuthenticator(String user, String passwd) { + this.user = user; + this.passwd = passwd; + } + @Override + protected PasswordAuthentication getPasswordAuthentication() { + return new PasswordAuthentication(user, passwd.toCharArray()); + } + + } + + private SdncOdlConnection() { + + } + + private SdncOdlConnection(String url, String user, String password) { + this.url = url; + this.user = user; + this.password = password; + + try { + URL sdncUrl = new URL(url); + Authenticator.setDefault(new SdncAuthenticator(user, password)); + + this.httpConn = (HttpURLConnection) sdncUrl.openConnection(); + } catch (Exception e) { + LOG.error("Unable to create http connection", e); + } + } + + public static SdncOdlConnection newInstance(String url, String user, String password) throws IOException + { + return (new SdncOdlConnection(url, user, password)); + } + + + + public String send(String method, String contentType, String msg) throws IOException { + + LOG.info("Sending REST " + method + " to " + url); + LOG.info("Message body:\n" + msg); + String authStr = user + ":" + password; + String encodedAuthStr = new String(Base64.encodeBase64(authStr.getBytes())); + + httpConn.addRequestProperty("Authentication", "Basic " + encodedAuthStr); + + httpConn.setRequestMethod(method); + httpConn.setRequestProperty("Content-Type", contentType); + httpConn.setRequestProperty("Accept", contentType); + + httpConn.setDoInput(true); + httpConn.setDoOutput(true); + httpConn.setUseCaches(false); + + if (httpConn instanceof HttpsURLConnection) { + HostnameVerifier hostnameVerifier = new HostnameVerifier() { + @Override + public boolean verify(String hostname, SSLSession session) { + return true; + } + }; + ((HttpsURLConnection) httpConn).setHostnameVerifier(hostnameVerifier); + } + + // Write message + httpConn.setRequestProperty("Content-Length", "" + msg.length()); + DataOutputStream outStr = new DataOutputStream(httpConn.getOutputStream()); + outStr.write(msg.getBytes()); + outStr.close(); + + // Read response + BufferedReader respRdr; + + LOG.info("Response: " + httpConn.getResponseCode() + " " + httpConn.getResponseMessage()); + + if (httpConn.getResponseCode() < 300) { + + respRdr = new BufferedReader(new InputStreamReader(httpConn.getInputStream())); + } else { + respRdr = new BufferedReader(new InputStreamReader(httpConn.getErrorStream())); + } + + StringBuffer respBuff = new StringBuffer(); + + String respLn; + + while ((respLn = respRdr.readLine()) != null) { + respBuff.append(respLn + "\n"); + } + respRdr.close(); + + String respString = respBuff.toString(); + + LOG.info("Response body :\n" + respString); + + return (respString); + + } + + +} -- cgit 1.2.3-korg