<artifactId>mail</artifactId>
<version>1.4.7</version>
</dependency>
+ <dependency>
+ <groupId>io.vavr</groupId>
+ <artifactId>vavr</artifactId>
+ <version>0.9.2</version>
+ </dependency>
+
<!-- TESTING -->
<dependency>
*/
package org.onap.dcae.commonFunction;
+import static io.vavr.API.Set;
+
+import io.vavr.collection.List;
+import io.vavr.collection.Set;
+import io.vavr.control.Option;
+import java.util.stream.StreamSupport;
import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;
-import org.json.JSONTokener;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.FileNotFoundException;
-import java.io.FileReader;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-import java.util.stream.StreamSupport;
/**
- * This class is a wrapper for 2 most used entities of org.json lib: JSONArray and JSONObject and
- * comprises utility methods for fast access of json structures without need to explicitly coerce between them.
- * While using this, bear in mind it does not contain exception handling - it is assumed that when using, the parsed json structure is known.
+ * This class is a wrapper for 2 most used entities of org.json lib: JSONArray and JSONObject and comprises utility
+ * methods for fast access of json structures without need to explicitly coerce between them. While using this, bear in
+ * mind it does not contain exception handling - it is assumed that when using, the parsed json structure is known.
*
* @author koblosz
*/
public class AnyNode {
- private final Object obj;
- private static final Logger log = LoggerFactory.getLogger(AnyNode.class);
- public static AnyNode parse(String filePath) throws IOException {
- try (FileReader fr = new FileReader(filePath)) {
- return new AnyNode(new JSONObject(new JSONTokener(fr)));
- } catch (FileNotFoundException | JSONException e1) {
- log.error("Could not find or parse file under path %s due to: %s", filePath, e1.toString());
- e1.printStackTrace();
- throw e1;
- }
- }
+ private Object obj;
- /**
- * Returns key set of underlying object. It is assumed that underlying object is of type org.json.JSONObject.
- *
- * @return Set of string keys present in underlying JSONObject
- */
- public Set<String> getKeys() {
- return asJsonObject().keySet();
+ private AnyNode(Object object) {
+ this.obj = object;
}
- /**
- * Returns value associated with specified key wrapped with AnyValue object. It is assumed that this is of type org.json.JSONObject.
- *
- * @param key A key string
- * @return The AnyNode object associated with given key.
- */
- public AnyNode get(String key) {
- return new AnyNode(asJsonObject().get(key));
+ public static AnyNode fromString(String content) {
+ return new AnyNode(new JSONObject(content));
}
/**
- * Returns value under specified index wrapped with AnyValue object. It is assumed that this is of type org.json.JSONArray.
- *
- * @param idx An index of JSONArray
- * @return The AnyNode object associated with given index.
+ * Returns key set of underlying object. It is assumed that underlying object is of type org.json.JSONObject.
*/
- public AnyNode get(int idx) {
- return new AnyNode(asJsonArray().get(idx));
+ public Set<String> keys() {
+ return Set(asJsonObject().keySet().toArray(new String[]{}));
}
/**
- * Returns int assuming this can be coerced to int.
+ * Returns value associated with specified key wrapped with AnyValue object. It is assumed that this is of type
+ * org.json.JSONObject.
*/
- public int asInt() {
- return (int) this.obj;
+ public AnyNode get(String key) {
+ return new AnyNode(asJsonObject().get(key));
}
/**
- * Returns string representation of this. If it happens to have null, the value is treated as org.json.JSONObject.NULL and "null" string is returned then.
- *
- * @return A String
+ * Returns string representation of this. If it happens to have null, the value is treated as
+ * org.json.JSONObject.NULL and "null" string is returned then.
*/
- public String asString() {
- return this.obj != JSONObject.NULL ? (String) this.obj : JSONObject.NULL.toString();
- }
-
public String toString() {
return this.obj.toString();
}
/**
- * Converts underlying object to String-to-Object map. It is assumed that underlying object is of type org.json.JSONObject.
- *
- * @return A map.
+ * Returns optional of object under specified key, wrapped with AnyNode object.
+ * If underlying object is not of type org.json.JSONObject
+ * or underlying object has no given key
+ * or given key is null
+ * then Optional.empty will be returned.
*/
- public Map<String, Object> asRawMap() {
- return asJsonObject().toMap();
- }
-
- /**
- * Returns optional of object under specified key, wrapped with AnyNode object. If underlying object is not of type org.json.JSONObject, then Optional.empty will be returned.
- *
- * @param key A key string
- */
- public Optional<AnyNode> getAsOptional(String key) {
- AnyNode result = null;
+ public Option<AnyNode> getAsOption(String key) {
try {
- result = get(key);
- } catch (JSONException ignored) {
+ AnyNode value = get(key);
+ if (value.toString().equals("null")) {
+ return Option.none();
+ }
+ return Option.some(value);
+ } catch (JSONException ex) {
+ return Option.none();
}
- return Optional.ofNullable(result);
- }
-
- private JSONObject asJsonObject() {
- return (JSONObject) this.obj;
- }
-
- /**
- * Converts underlying object to map representation with map values wrapped with AnyNode object. It is assumed that underlying object is of type org.json.JSONObject.
- */
- public Map<String, AnyNode> asMap() {
- Map<String, AnyNode> map = new HashMap<>();
- getKeys().forEach(key -> map.put(key, get(key)));
- return map;
- }
-
- /**
- * Converts underlying object to map representation with map values wrapped with AnyNode object. It is assumed that underlying object is of type org.json.JSONObject.
- */
- public java.util.List<AnyNode> asList() {
- return asStream().collect(Collectors.toList());
}
/**
- * Converts this object to stream of underlying objects wrapped with AnyNode class. It is assumed that this is of type JSONArray.
+ * Converts underlying object to map representation with map values wrapped with AnyNode object. It is assumed that
+ * underlying object is of type org.json.JSONObject.
*/
- private Stream<AnyNode> asStream() {
- return StreamSupport.stream(((JSONArray) this.obj).spliterator(), false).map(AnyNode::new);
+ public List<AnyNode> toList() {
+ return List.ofAll(StreamSupport.stream(((JSONArray) this.obj).spliterator(), false).map(AnyNode::new));
}
/**
* Checks if specified key is present in this. It is assumed that this is of type JSONObject.
*/
- boolean hasKey(String key) {
- return getAsOptional(key).isPresent();
- }
-
- /**
- * Returns empty AnyNode (with null inside)
- */
- public static AnyNode nullValue() {
- return new AnyNode(JSONObject.NULL.toString());
+ public boolean has(String key) {
+ return !getAsOption(key).isEmpty();
}
- private JSONArray asJsonArray() {
- return (JSONArray) this.obj;
+ private JSONObject asJsonObject() {
+ return (JSONObject) this.obj;
}
- private AnyNode(Object object) {
- this.obj = object;
- }
}
+++ /dev/null
-package org.onap.dcae.commonFunction;
-/*-
- * ============LICENSE_START=======================================================
- * org.onap.dcaegen2.collectors.ves
- * ================================================================================
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
- * Copyright (C) 2018 Nokia. 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=========================================================
- */
-import com.att.nsa.cambria.client.CambriaBatchingPublisher;
-import com.att.nsa.cambria.client.CambriaClientBuilders;
-import java.net.MalformedURLException;
-import java.security.GeneralSecurityException;
-import java.util.Map;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-class CambriaPublisherFactory {
-
- private final static Logger log = LoggerFactory.getLogger(CambriaPublisherFactory.class);
-
- CambriaBatchingPublisher createCambriaPublisher(String streamId)
- throws MalformedURLException, GeneralSecurityException {
- String authpwd = null;
- DmaapPropertyReader reader = DmaapPropertyReader.getInstance(CommonStartup.cambriaConfigFile);
- Map<String, String> dMaaPProperties = reader.getDmaapProperties();
- String ueburl = dMaaPProperties.get(streamId + ".cambria.url");
-
- if (ueburl == null) {
- ueburl = dMaaPProperties.get(streamId + ".cambria.hosts");
- }
- String topic = reader.getKeyValue(streamId + ".cambria.topic");
- String authuser = reader.getKeyValue(streamId + ".basicAuthUsername");
-
- if (authuser != null) {
- authpwd = dMaaPProperties.get(streamId + ".basicAuthPassword");
- }
-
- if ((authuser != null) && (authpwd != null)) {
- log.debug(String.format("URL:%sTOPIC:%sAuthUser:%sAuthpwd:%s", ueburl, topic, authuser, authpwd));
- return new CambriaClientBuilders.PublisherBuilder().usingHosts(ueburl).onTopic(topic).usingHttps()
- .authenticatedByHttp(authuser, authpwd).logSendFailuresAfter(5)
- .build();
- } else {
- log.debug(String.format("URL:%sTOPIC:%s", ueburl, topic));
- return new CambriaClientBuilders.PublisherBuilder().usingHosts(ueburl).onTopic(topic)
- .logSendFailuresAfter(5)
- .build();
- }
- }
-}
* Licensed under the Apache License, Version 2.0 (the "License");\r
* you may not use this file except in compliance with the License.\r
* You may obtain a copy of the License at\r
- * \r
+ *\r
* http://www.apache.org/licenses/LICENSE-2.0\r
- * \r
+ *\r
* Unless required by applicable law or agreed to in writing, software\r
* distributed under the License is distributed on an "AS IS" BASIS,\r
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
import com.github.fge.jsonschema.core.report.ProcessingReport;\r
import com.github.fge.jsonschema.main.JsonSchema;\r
import com.github.fge.jsonschema.main.JsonSchemaFactory;\r
-\r
-import org.apache.catalina.LifecycleException;\r
-import org.json.JSONArray;\r
-import org.json.JSONException;\r
-import org.json.JSONObject;\r
-import org.onap.dcae.restapi.RestfulCollectorServlet;\r
-import org.slf4j.Logger;\r
-import org.slf4j.LoggerFactory;\r
-\r
import java.io.IOException;\r
import java.net.URL;\r
import java.nio.file.Files;\r
import java.util.concurrent.ExecutorService;\r
import java.util.concurrent.Executors;\r
import java.util.concurrent.LinkedBlockingQueue;\r
+import org.apache.catalina.LifecycleException;\r
+import org.json.JSONArray;\r
+import org.json.JSONException;\r
+import org.json.JSONObject;\r
+import org.onap.dcae.commonFunction.event.publishing.DMaaPConfigurationParser;\r
+import org.onap.dcae.commonFunction.event.publishing.EventPublisher;\r
+import org.onap.dcae.restapi.RestfulCollectorServlet;\r
+import org.slf4j.Logger;\r
+import org.slf4j.LoggerFactory;\r
\r
public class CommonStartup extends NsaBaseEndpoint implements Runnable {\r
\r
- private static final String KCONFIG = "c";\r
- private static final String KSETTING_PORT = "collector.service.port";\r
- private static final int KDEFAULT_PORT = 8080;\r
- private static final String KSETTING_SECUREPORT = "collector.service.secure.port";\r
- private static final int KDEFAULT_SECUREPORT = -1;\r
- private static final String KSETTING_KEYSTOREPASSFILE = "collector.keystore.passwordfile";\r
- private static final String KDEFAULT_KEYSTOREPASSFILE = "../etc/passwordfile";\r
- private static final String KSETTING_KEYSTOREFILE = "collector.keystore.file.location";\r
- private static final String KDEFAULT_KEYSTOREFILE = "../etc/keystore";\r
- private static final String KSETTING_KEYALIAS = "collector.keystore.alias";\r
- private static final String KDEFAULT_KEYALIAS = "tomcat";\r
- private static final String KSETTING_DMAAPCONFIGS = "collector.dmaapfile";\r
- private static final String[] KDEFAULT_DMAAPCONFIGS = new String[] { "/etc/DmaapConfig.json" };\r
- private static final String KSETTING_SCHEMAVALIDATOR = "collector.schema.checkflag";\r
- private static final int KDEFAULT_SCHEMAVALIDATOR = -1;\r
- private static final String KSETTING_SCHEMAFILE = "collector.schema.file";\r
- private static final String KDEFAULT_SCHEMAFILE = "{\"v5\":\"./etc/CommonEventFormat_28.3.json\"}";\r
- private static final String KSETTING_DMAAPSTREAMID = "collector.dmaap.streamid";\r
- private static final String KSETTING_AUTHFLAG = "header.authflag";\r
- private static final int KDEFAULT_AUTHFLAG = 0;\r
- private static final String KSETTING_EVENTTRANSFORMFLAG = "event.transform.flag";\r
- private static final int KDEFAULT_EVENTTRANSFORMFLAG = 1;\r
- private static final Logger metriclog = LoggerFactory.getLogger("com.att.ecomp.metrics");\r
- public static final Logger inlog = LoggerFactory.getLogger("org.onap.dcae.commonFunction.input");\r
- static final Logger oplog = LoggerFactory.getLogger("org.onap.dcae.commonFunction.output");\r
- public static final Logger eplog = LoggerFactory.getLogger("org.onap.dcae.commonFunction.error");\r
-\r
- public static final String KSETTING_AUTHLIST = "header.authlist";\r
- static final int KDEFAULT_MAXQUEUEDEVENTS = 1024 * 4;\r
- public static int schemaValidatorflag = -1;\r
- public static int authflag = 1;\r
- static int eventTransformFlag = 1;\r
- public static JSONObject schemaFileJson;\r
- static String cambriaConfigFile;\r
- public static String streamID;\r
-\r
- static LinkedBlockingQueue<JSONObject> fProcessingInputQueue;\r
- private static ApiServer fTomcatServer = null;\r
- private static final Logger log = LoggerFactory.getLogger(CommonStartup.class);\r
-\r
- private CommonStartup(rrNvReadable settings) throws loadException, IOException, rrNvReadable.missingReqdSetting {\r
- final List<ApiServerConnector> connectors = new LinkedList<>();\r
-\r
- if (settings.getInt(KSETTING_PORT, KDEFAULT_PORT) > 0) {\r
- connectors.add(new ApiServerConnector.Builder(settings.getInt(KSETTING_PORT, KDEFAULT_PORT)).secure(false)\r
- .build());\r
- }\r
-\r
- final int securePort = settings.getInt(KSETTING_SECUREPORT, KDEFAULT_SECUREPORT);\r
- final String keystoreFile = settings.getString(KSETTING_KEYSTOREFILE, KDEFAULT_KEYSTOREFILE);\r
- final String keystorePasswordFile = settings.getString(KSETTING_KEYSTOREPASSFILE, KDEFAULT_KEYSTOREPASSFILE);\r
- final String keyAlias = settings.getString(KSETTING_KEYALIAS, KDEFAULT_KEYALIAS);\r
-\r
- if (securePort > 0) {\r
- String keystorePassword = readFile(keystorePasswordFile);\r
- connectors.add(new ApiServerConnector.Builder(securePort).secure(true)\r
- .keystorePassword(keystorePassword).keystoreFile(keystoreFile).keyAlias(keyAlias).build());\r
-\r
- }\r
-\r
- schemaValidatorflag = settings.getInt(KSETTING_SCHEMAVALIDATOR, KDEFAULT_SCHEMAVALIDATOR);\r
- if (schemaValidatorflag > 0) {\r
- String schemaFile = settings.getString(KSETTING_SCHEMAFILE, KDEFAULT_SCHEMAFILE);\r
- schemaFileJson = new JSONObject(schemaFile);\r
-\r
- }\r
- authflag = settings.getInt(CommonStartup.KSETTING_AUTHFLAG, CommonStartup.KDEFAULT_AUTHFLAG);\r
- String[] currentConfigFile = settings.getStrings(KSETTING_DMAAPCONFIGS, KDEFAULT_DMAAPCONFIGS);\r
- cambriaConfigFile = currentConfigFile[0];\r
- streamID = settings.getString(KSETTING_DMAAPSTREAMID, null);\r
- eventTransformFlag = settings.getInt(KSETTING_EVENTTRANSFORMFLAG, KDEFAULT_EVENTTRANSFORMFLAG);\r
-\r
- fTomcatServer = new ApiServer.Builder(connectors, new RestfulCollectorServlet(settings)).encodeSlashes(true)\r
- .name("collector").build();\r
- }\r
-\r
- public static void main(String[] args) {\r
- try {\r
- final Map<String, String> argMap = NsaCommandLineUtil.processCmdLine(args, true);\r
- final String config = NsaCommandLineUtil.getSetting(argMap, KCONFIG, "collector.properties");\r
- final URL settingStream = DrumlinServlet.findStream(config, CommonStartup.class);\r
-\r
- final nvReadableStack settings = new nvReadableStack();\r
- settings.push(new nvPropertiesFile(settingStream));\r
- settings.push(new nvReadableTable(argMap));\r
-\r
- fProcessingInputQueue = new LinkedBlockingQueue<>(CommonStartup.KDEFAULT_MAXQUEUEDEVENTS);\r
-\r
- VESLogger.setUpEcompLogging();\r
-\r
- CommonStartup cs = new CommonStartup(settings);\r
-\r
- Thread commonStartupThread = new Thread(cs);\r
- commonStartupThread.start();\r
-\r
- EventProcessor ep = new EventProcessor();\r
- ExecutorService executor = Executors.newFixedThreadPool(20);\r
- for (int i = 0; i < 20; ++i) {\r
- executor.execute(ep);\r
- }\r
-\r
- } catch (loadException | missingReqdSetting | IOException e) {\r
- CommonStartup.eplog.error("FATAL_STARTUP_ERROR" + e.getMessage());\r
- throw new RuntimeException(e);\r
- } catch (Exception e) {\r
- System.err.println("Uncaught exception - " + e.getMessage());\r
- CommonStartup.eplog.error("FATAL_ERROR" + e.getMessage());\r
- e.printStackTrace(System.err);\r
- }\r
- }\r
-\r
- public void run() {\r
- try {\r
- fTomcatServer.start();\r
- fTomcatServer.await();\r
- } catch (LifecycleException | IOException e) {\r
- throw new RuntimeException(e);\r
- }\r
- }\r
-\r
- public static class QueueFullException extends Exception {\r
- private static final long serialVersionUID = 1L;\r
- }\r
-\r
- public static void handleEvents(JSONArray a) throws QueueFullException, JSONException {\r
- CommonStartup.metriclog.info("EVENT_PUBLISH_START");\r
- for (int i = 0; i < a.length(); i++) {\r
- if (!fProcessingInputQueue.offer(a.getJSONObject(i))) {\r
- throw new QueueFullException();\r
- }\r
- }\r
- log.debug("CommonStartup.handleEvents:EVENTS has been published successfully!");\r
- CommonStartup.metriclog.info("EVENT_PUBLISH_END");\r
- }\r
-\r
- private static String readFile(String path) throws IOException {\r
- byte[] encoded = Files.readAllBytes(Paths.get(path));\r
- String pwd = new String(encoded);\r
- return pwd.substring(0, pwd.length() - 1);\r
- }\r
-\r
- public static String validateAgainstSchema(String jsonData, String jsonSchema) {\r
- ProcessingReport report;\r
- String result = "false";\r
-\r
- try {\r
- log.trace("Schema validation for event:" + jsonData);\r
- JsonNode schemaNode = JsonLoader.fromString(jsonSchema);\r
- JsonNode data = JsonLoader.fromString(jsonData);\r
- JsonSchemaFactory factory = JsonSchemaFactory.byDefault();\r
- JsonSchema schema = factory.getJsonSchema(schemaNode);\r
- report = schema.validate(data);\r
- } catch (JsonParseException e) {\r
- log.error("validateAgainstSchema:JsonParseException for event:" + jsonData);\r
- return e.getMessage();\r
- } catch (ProcessingException e) {\r
- log.error("validateAgainstSchema:Processing exception for event:" + jsonData);\r
- return e.getMessage();\r
- } catch (IOException e) {\r
- log.error(\r
- "validateAgainstSchema:IO exception; something went wrong trying to read json data for event:" + jsonData);\r
- return e.getMessage();\r
- }\r
- if (report != null) {\r
- for (ProcessingMessage pm : report) {\r
- log.trace("Processing Message: " + pm.getMessage());\r
- }\r
- result = String.valueOf(report.isSuccess());\r
- }\r
- try {\r
- log.debug("Validation Result:" + result + " Validation report:" + report);\r
- } catch (NullPointerException e) {\r
- log.error("validateAgainstSchema:NullpointerException on report");\r
- }\r
- return result;\r
- }\r
+ private static final String KCONFIG = "c";\r
+ private static final String KSETTING_PORT = "collector.service.port";\r
+ private static final int KDEFAULT_PORT = 8080;\r
+ private static final String KSETTING_SECUREPORT = "collector.service.secure.port";\r
+ private static final int KDEFAULT_SECUREPORT = -1;\r
+ private static final String KSETTING_KEYSTOREPASSFILE = "collector.keystore.passwordfile";\r
+ private static final String KDEFAULT_KEYSTOREPASSFILE = "../etc/passwordfile";\r
+ private static final String KSETTING_KEYSTOREFILE = "collector.keystore.file.location";\r
+ private static final String KDEFAULT_KEYSTOREFILE = "../etc/keystore";\r
+ private static final String KSETTING_KEYALIAS = "collector.keystore.alias";\r
+ private static final String KDEFAULT_KEYALIAS = "tomcat";\r
+ private static final String KSETTING_DMAAPCONFIGS = "collector.dmaapfile";\r
+ private static final String[] KDEFAULT_DMAAPCONFIGS = new String[]{"/etc/DmaapConfig.json"};\r
+ private static final String KSETTING_SCHEMAVALIDATOR = "collector.schema.checkflag";\r
+ private static final int KDEFAULT_SCHEMAVALIDATOR = -1;\r
+ private static final String KSETTING_SCHEMAFILE = "collector.schema.file";\r
+ private static final String KDEFAULT_SCHEMAFILE = "{\"v5\":\"./etc/CommonEventFormat_28.3.json\"}";\r
+ private static final String KSETTING_DMAAPSTREAMID = "collector.dmaap.streamid";\r
+ private static final String KSETTING_AUTHFLAG = "header.authflag";\r
+ private static final int KDEFAULT_AUTHFLAG = 0;\r
+ private static final String KSETTING_EVENTTRANSFORMFLAG = "event.transform.flag";\r
+ private static final int KDEFAULT_EVENTTRANSFORMFLAG = 1;\r
+ private static final Logger metriclog = LoggerFactory.getLogger("com.att.ecomp.metrics");\r
+ public static final Logger inlog = LoggerFactory.getLogger("org.onap.dcae.commonFunction.input");\r
+ static final Logger oplog = LoggerFactory.getLogger("org.onap.dcae.commonFunction.output");\r
+ public static final Logger eplog = LoggerFactory.getLogger("org.onap.dcae.commonFunction.error");\r
+\r
+ public static final String KSETTING_AUTHLIST = "header.authlist";\r
+ static final int KDEFAULT_MAXQUEUEDEVENTS = 1024 * 4;\r
+ public static int schemaValidatorflag = -1;\r
+ public static int authflag = 1;\r
+ static int eventTransformFlag = 1;\r
+ public static JSONObject schemaFileJson;\r
+ static String cambriaConfigFile;\r
+ public static String streamID;\r
+\r
+ static LinkedBlockingQueue<JSONObject> fProcessingInputQueue;\r
+ private static ApiServer fTomcatServer = null;\r
+ private static final Logger log = LoggerFactory.getLogger(CommonStartup.class);\r
+\r
+ private CommonStartup(rrNvReadable settings) throws loadException, IOException, rrNvReadable.missingReqdSetting {\r
+ final List<ApiServerConnector> connectors = new LinkedList<>();\r
+\r
+ if (settings.getInt(KSETTING_PORT, KDEFAULT_PORT) > 0) {\r
+ connectors.add(new ApiServerConnector.Builder(settings.getInt(KSETTING_PORT, KDEFAULT_PORT)).secure(false)\r
+ .build());\r
+ }\r
+\r
+ final int securePort = settings.getInt(KSETTING_SECUREPORT, KDEFAULT_SECUREPORT);\r
+ final String keystoreFile = settings.getString(KSETTING_KEYSTOREFILE, KDEFAULT_KEYSTOREFILE);\r
+ final String keystorePasswordFile = settings.getString(KSETTING_KEYSTOREPASSFILE, KDEFAULT_KEYSTOREPASSFILE);\r
+ final String keyAlias = settings.getString(KSETTING_KEYALIAS, KDEFAULT_KEYALIAS);\r
+\r
+ if (securePort > 0) {\r
+ String keystorePassword = readFile(keystorePasswordFile);\r
+ connectors.add(new ApiServerConnector.Builder(securePort).secure(true)\r
+ .keystorePassword(keystorePassword).keystoreFile(keystoreFile).keyAlias(keyAlias).build());\r
+\r
+ }\r
+\r
+ schemaValidatorflag = settings.getInt(KSETTING_SCHEMAVALIDATOR, KDEFAULT_SCHEMAVALIDATOR);\r
+ if (schemaValidatorflag > 0) {\r
+ String schemaFile = settings.getString(KSETTING_SCHEMAFILE, KDEFAULT_SCHEMAFILE);\r
+ schemaFileJson = new JSONObject(schemaFile);\r
+\r
+ }\r
+ authflag = settings.getInt(CommonStartup.KSETTING_AUTHFLAG, CommonStartup.KDEFAULT_AUTHFLAG);\r
+ String[] currentConfigFile = settings.getStrings(KSETTING_DMAAPCONFIGS, KDEFAULT_DMAAPCONFIGS);\r
+ cambriaConfigFile = currentConfigFile[0];\r
+ streamID = settings.getString(KSETTING_DMAAPSTREAMID, null);\r
+ eventTransformFlag = settings.getInt(KSETTING_EVENTTRANSFORMFLAG, KDEFAULT_EVENTTRANSFORMFLAG);\r
+\r
+ fTomcatServer = new ApiServer.Builder(connectors, new RestfulCollectorServlet(settings)).encodeSlashes(true)\r
+ .name("collector").build();\r
+ }\r
+\r
+ public static void main(String[] args) {\r
+ try {\r
+ final Map<String, String> argMap = NsaCommandLineUtil.processCmdLine(args, true);\r
+ final String config = NsaCommandLineUtil.getSetting(argMap, KCONFIG, "collector.properties");\r
+ final URL settingStream = DrumlinServlet.findStream(config, CommonStartup.class);\r
+\r
+ final nvReadableStack settings = new nvReadableStack();\r
+ settings.push(new nvPropertiesFile(settingStream));\r
+ settings.push(new nvReadableTable(argMap));\r
+\r
+ fProcessingInputQueue = new LinkedBlockingQueue<>(CommonStartup.KDEFAULT_MAXQUEUEDEVENTS);\r
+\r
+ VESLogger.setUpEcompLogging();\r
+\r
+ CommonStartup cs = new CommonStartup(settings);\r
+\r
+ Thread commonStartupThread = new Thread(cs);\r
+ commonStartupThread.start();\r
+\r
+ EventProcessor ep = new EventProcessor(EventPublisher.createPublisher(oplog,\r
+ DMaaPConfigurationParser\r
+ .parseToDomainMapping(Paths.get(cambriaConfigFile))\r
+ .get()));\r
+ ExecutorService executor = Executors.newFixedThreadPool(20);\r
+ for (int i = 0; i < 20; ++i) {\r
+ executor.execute(ep);\r
+ }\r
+ } catch (Exception e) {\r
+ CommonStartup.eplog.error("Fatal error during application startup", e);\r
+ throw new RuntimeException(e);\r
+ }\r
+ }\r
+\r
+ public void run() {\r
+ try {\r
+ fTomcatServer.start();\r
+ fTomcatServer.await();\r
+ } catch (LifecycleException | IOException e) {\r
+ throw new RuntimeException(e);\r
+ }\r
+ }\r
+\r
+ public static class QueueFullException extends Exception {\r
+\r
+ private static final long serialVersionUID = 1L;\r
+ }\r
+\r
+ public static void handleEvents(JSONArray a) throws QueueFullException, JSONException {\r
+ CommonStartup.metriclog.info("EVENT_PUBLISH_START");\r
+ for (int i = 0; i < a.length(); i++) {\r
+ if (!fProcessingInputQueue.offer(a.getJSONObject(i))) {\r
+ throw new QueueFullException();\r
+ }\r
+ }\r
+ log.debug("CommonStartup.handleEvents:EVENTS has been published successfully!");\r
+ CommonStartup.metriclog.info("EVENT_PUBLISH_END");\r
+ }\r
+\r
+ private static String readFile(String path) throws IOException {\r
+ byte[] encoded = Files.readAllBytes(Paths.get(path));\r
+ String pwd = new String(encoded);\r
+ return pwd.substring(0, pwd.length() - 1);\r
+ }\r
+\r
+ public static String validateAgainstSchema(String jsonData, String jsonSchema) {\r
+ ProcessingReport report;\r
+ String result = "false";\r
+\r
+ try {\r
+ log.trace("Schema validation for event:" + jsonData);\r
+ JsonNode schemaNode = JsonLoader.fromString(jsonSchema);\r
+ JsonNode data = JsonLoader.fromString(jsonData);\r
+ JsonSchemaFactory factory = JsonSchemaFactory.byDefault();\r
+ JsonSchema schema = factory.getJsonSchema(schemaNode);\r
+ report = schema.validate(data);\r
+ } catch (JsonParseException e) {\r
+ log.error("validateAgainstSchema:JsonParseException for event:" + jsonData);\r
+ return e.getMessage();\r
+ } catch (ProcessingException e) {\r
+ log.error("validateAgainstSchema:Processing exception for event:" + jsonData);\r
+ return e.getMessage();\r
+ } catch (IOException e) {\r
+ log.error(\r
+ "validateAgainstSchema:IO exception; something went wrong trying to read json data for event:" + jsonData);\r
+ return e.getMessage();\r
+ }\r
+ if (report != null) {\r
+ for (ProcessingMessage pm : report) {\r
+ log.trace("Processing Message: " + pm.getMessage());\r
+ }\r
+ result = String.valueOf(report.isSuccess());\r
+ }\r
+ try {\r
+ log.debug("Validation Result:" + result + " Validation report:" + report);\r
+ } catch (NullPointerException e) {\r
+ log.error("validateAgainstSchema:NullpointerException on report");\r
+ }\r
+ return result;\r
+ }\r
\r
\r
}\r
+++ /dev/null
-/*-
- * ============LICENSE_START=======================================================
- * PROJECT
- * ================================================================================
- * Copyright (C) 2017 AT&T 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.onap.dcae.commonFunction;
-
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.stream.Collectors;
-
-public class DmaapPropertyReader {
-
-
- private static final Logger log = LoggerFactory.getLogger(DmaapPropertyReader.class);
- private static final String CAMBRIA_TOPIC_KEY = "cambria.topic";
- private static final String CAMBRIA_HOSTS_KEY = "cambria.hosts";
- private static final String CAMBRIA_URL_KEY = "cambria.url";
- private static final List<String> LEGACY_CHANNEL_MANDATORY_PARAMS = Lists.newArrayList(CAMBRIA_TOPIC_KEY, CAMBRIA_HOSTS_KEY, CAMBRIA_URL_KEY, "basicAuthPassword", "basicAuthUsername");
- private static DmaapPropertyReader instance = null;
- private final Map<String, String> dmaapProperties;
-
- public DmaapPropertyReader(String cambriaConfigFilePath) {
- this.dmaapProperties = DmaapPropertyReader.getProcessedDmaapProperties(cambriaConfigFilePath);
- }
-
- public Map<String, String> getDmaapProperties() {
- return dmaapProperties;
- }
-
- public static synchronized DmaapPropertyReader getInstance(String channelConfig) {
- if (instance == null) {
- instance = new DmaapPropertyReader(channelConfig);
- }
- return instance;
- }
-
- public String getKeyValue(String hashKey) {
- return dmaapProperties.get(hashKey);
- }
-
- private static Map<String, String> getProcessedDmaapProperties(String configFilePath) {
- Map<String, String> transformedDmaapProperties = new HashMap<>();
- try {
- AnyNode root = AnyNode.parse(configFilePath);
- if (isInLegacyFormat(root)) {
- transformedDmaapProperties = getLegacyDmaapPropertiesWithChannels(root.get("channels"));
- } else {
- transformedDmaapProperties = getDmaapPropertiesWithInfoData(root);
- }
- } catch (IOException e) {
- e.printStackTrace();
- }
- return transformedDmaapProperties;
- }
-
- private static boolean isInLegacyFormat(AnyNode root) {
- return root.hasKey("channels");
- }
-
- private static Map<String, String> getLegacyDmaapPropertiesWithChannels(AnyNode channelsNode) {
- return channelsNode.asList().stream()
- .map(DmaapPropertyReader::getTransformedMandatoryChannelProperties)
- .flatMap(m -> m.entrySet().stream())
- .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
- }
-
- private static Map<String, String> getTransformedMandatoryChannelProperties(AnyNode channel) {
- String prefix = channel.get("name").asString() + ".";
- return channel.asMap().entrySet().stream().filter(el -> LEGACY_CHANNEL_MANDATORY_PARAMS.contains(el.getKey()) && !Objects.equals(el.getKey(), "name"))
- .collect(Collectors.toMap(k -> prefix + k.getKey(), v -> v.getValue().asString().replace("\"", "")));
- }
-
- private static Map<String, String> getDmaapPropertiesWithInfoData(AnyNode root) {
- return root.asMap().entrySet().stream()
- .map(DmaapPropertyReader::getTransformedMandatoryInfoProperties).flatMap(m -> m.entrySet().stream())
- .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
- }
-
- private static Map<String, String> getTransformedMandatoryInfoProperties(Map.Entry<String, AnyNode> el) {
- String prefix = el.getKey() + ".";
- AnyNode val = el.getValue();
- Map<String, String> map = Maps.newHashMap();
- map.put(prefix + "basicAuthUsername", val.getAsOptional("aaf_username").orElse(AnyNode.nullValue()).asString().replace("\"", ""));
- map.put(prefix + "basicAuthPassword", val.getAsOptional("aaf_password").orElse(AnyNode.nullValue()).asString().replace("\"", ""));
- map.putAll(getParamsFromDmaapInfoTopicUrl(prefix, val.get("dmaap_info").get("topic_url").asString().replace("\"", "")));
- return map;
- }
-
- /***
- * Dmaap url structure pub - https://<dmaaphostname>:<port>/events/
- * <namespace>.<dmaapcluster>.<topic>, sub - https://<dmaaphostname>:
- * <port>/events/<namespace>.<dmaapcluster>.<topic>/G1/u1";
- *
- * Onap url structure pub - http://<dmaaphostname>:<port>/<unauthenticated>.
- * <topic>,
- */
- private static Map<String, String> getParamsFromDmaapInfoTopicUrl(String keyPrefix, String topicUrl) {
- Map<String, String> topicUrlParts = Maps.newHashMap();
- try {
- URL url = new URL(topicUrl);
- topicUrlParts.put(keyPrefix + CAMBRIA_URL_KEY, url.getAuthority());
- String[] pathParts = url.getPath().split("/");
- if (pathParts.length > 2 && "events".equals(pathParts[1])) {
- // DCAE internal dmaap topic convention
- topicUrlParts.put(keyPrefix + CAMBRIA_TOPIC_KEY, pathParts[2]);
- } else {
- // ONAP dmaap topic convention
- topicUrlParts.put(keyPrefix + CAMBRIA_TOPIC_KEY, pathParts[2]);
- topicUrlParts.put(keyPrefix + CAMBRIA_HOSTS_KEY, url.getHost());
- }
- } catch (MalformedURLException e) {
- log.error("Invalid URL found under topic_url key!", e);
- e.printStackTrace();
- }
- return topicUrlParts;
- }
-}
\ No newline at end of file
+++ /dev/null
-/*-
- * ============LICENSE_START=======================================================
- * org.onap.dcaegen2.collectors.ves
- * ================================================================================
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
- * Copyright (C) 2018 Nokia. 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.onap.dcae.commonFunction;
-
-import com.att.nsa.cambria.client.CambriaBatchingPublisher;
-import com.google.common.cache.CacheBuilder;
-import com.google.common.cache.CacheLoader;
-import com.google.common.cache.LoadingCache;
-import com.google.common.cache.RemovalListener;
-import java.io.IOException;
-import java.net.MalformedURLException;
-import java.security.GeneralSecurityException;
-import java.util.List;
-import java.util.concurrent.TimeUnit;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-class DmaapPublishers {
-
- private static final Logger log = LoggerFactory.getLogger(DmaapPublishers.class);
- private final LoadingCache<String, CambriaBatchingPublisher> publishers;
-
- private DmaapPublishers(
- LoadingCache<String, CambriaBatchingPublisher> publishers) {
- this.publishers = publishers;
- }
-
- static DmaapPublishers create() {
- return create(new CambriaPublisherFactory());
- }
-
- static DmaapPublishers create(final CambriaPublisherFactory publisherFactory) {
- final LoadingCache<String, CambriaBatchingPublisher> cache = CacheBuilder.<String, CambriaBatchingPublisher>newBuilder()
- .removalListener((RemovalListener<String, CambriaBatchingPublisher>) notification -> {
- if (notification.getValue() != null) {
- onCacheItemInvalidated(notification.getValue());
- }
- })
- .build(new CacheLoader<String, CambriaBatchingPublisher>() {
- @Override
- public CambriaBatchingPublisher load(String streamId)
- throws MalformedURLException, GeneralSecurityException {
- try {
- return publisherFactory.createCambriaPublisher(streamId);
- } catch (MalformedURLException | GeneralSecurityException e) {
- log.error("CambriaClientBuilders connection reader exception : streamID - " + streamId + " "
- + e.getMessage());
- throw e;
- }
- }
- });
- return new DmaapPublishers(cache);
- }
-
- CambriaBatchingPublisher getByStreamId(String streamId) {
- return publishers.getUnchecked(streamId);
- }
-
- void closeByStreamId(String streamId) {
- publishers.invalidate(streamId);
- }
-
- private static void onCacheItemInvalidated(CambriaBatchingPublisher pub) {
- try {
- final List<?> stuck = pub.close(20, TimeUnit.SECONDS);
- if (!stuck.isEmpty()) {
- log.error(stuck.size() + " messages unsent");
- }
- } catch (InterruptedException | IOException e) {
- log.error("Caught Exception on Close event: {}", e);
- }
- }
-}
import com.google.gson.Gson;\r
import java.util.Map;\r
import org.json.JSONObject;\r
+import org.onap.dcae.commonFunction.event.publishing.EventPublisher;\r
import org.slf4j.Logger;\r
import org.slf4j.LoggerFactory;\r
\r
\r
static Map<String, String[]> streamidHash = new HashMap<>();\r
public JSONObject event;\r
+ private EventPublisher eventPublisher;\r
\r
- EventProcessor() {\r
+ public EventProcessor(EventPublisher eventPublisher) {\r
+ this.eventPublisher = eventPublisher;\r
streamidHash = parseStreamIdToStreamHashMapping(CommonStartup.streamID);\r
}\r
\r
for (String aStreamIdList : streamIdList) {\r
log.info("Invoking publisher for streamId:" + aStreamIdList);\r
this.overrideEvent();\r
- EventPublisherHash.getInstance().sendEvent(event, aStreamIdList);\r
+ eventPublisher.sendEvent(event, aStreamIdList);\r
\r
}\r
}\r
+++ /dev/null
-/*-
- * ============LICENSE_START=======================================================
- * org.onap.dcaegen2.collectors.ves
- * ================================================================================
- * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
- * Copyright (C) 2018 Nokia. 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.onap.dcae.commonFunction;
-
-import com.att.nsa.cambria.client.CambriaBatchingPublisher;
-import com.att.nsa.clock.SaClock;
-import com.att.nsa.logging.LoggingContext;
-import com.att.nsa.logging.log4j.EcompFields;
-import com.google.common.annotations.VisibleForTesting;
-import java.io.IOException;
-import org.json.JSONObject;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class EventPublisherHash {
-
- private static final String VES_UNIQUE_ID = "VESuniqueId";
- private static final Logger log = LoggerFactory.getLogger(EventPublisherHash.class);
- private static volatile EventPublisherHash instance = new EventPublisherHash(DmaapPublishers.create());
- private final DmaapPublishers dmaapPublishers;
-
- public static EventPublisherHash getInstance() {
- return instance;
- }
-
- @VisibleForTesting
- EventPublisherHash(DmaapPublishers dmaapPublishers) {
- this.dmaapPublishers = dmaapPublishers;
- }
-
- void sendEvent(JSONObject event, String streamid) {
- log.debug("EventPublisher.sendEvent: instance for publish is ready");
- clearVesUniqueId(event);
-
- try {
- sendEventUsingCachedPublisher(streamid, event);
- } catch (IOException | IllegalArgumentException e) {
- log.error("Unable to publish event: {} streamID: {}. Exception: {}", event, streamid, e);
- dmaapPublishers.closeByStreamId(streamid);
- }
- }
-
- private void clearVesUniqueId(JSONObject event) {
- if (event.has(VES_UNIQUE_ID)) {
- String uuid = event.get(VES_UNIQUE_ID).toString();
- LoggingContext localLC = VESLogger.getLoggingContextForThread(uuid);
- localLC.put(EcompFields.kBeginTimestampMs, SaClock.now());
- log.debug("Removing VESuniqueid object from event");
- event.remove(VES_UNIQUE_ID);
- }
- }
-
- private void sendEventUsingCachedPublisher(String streamid, JSONObject event) throws IOException {
- int pendingMsgs = dmaapPublishers.getByStreamId(streamid).send("MyPartitionKey", event.toString());
- if (pendingMsgs > 100) {
- log.info("Pending Message Count=" + pendingMsgs);
- }
- log.info("pub.send invoked - no error");
- CommonStartup.oplog.info(String.format("StreamID:%s Event Published:%s ", streamid, event));
- }
-
- @VisibleForTesting
- CambriaBatchingPublisher getDmaapPublisher(String streamId) {
- return dmaapPublishers.getByStreamId(streamId);
- }
-}
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2018 Nokia. 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.onap.dcae.commonFunction.event.publishing;
+
+import static io.vavr.API.List;
+import static io.vavr.API.Try;
+import static io.vavr.API.Tuple;
+import static io.vavr.API.unchecked;
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.enhanceError;
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f;
+
+import io.vavr.collection.List;
+import io.vavr.collection.Map;
+import io.vavr.control.Option;
+import io.vavr.control.Try;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import org.onap.dcae.commonFunction.AnyNode;
+
+/**
+ * @author Pawel Szalapski (pawel.szalapski@nokia.com)
+ */
+@SuppressWarnings("mapFailure takes a generic varargs, unchecked because of Javas type system limitation, actually safe to do")
+public final class DMaaPConfigurationParser {
+
+ public static Try<Map<String, PublisherConfig>> parseToDomainMapping(Path configLocation) {
+ return readFromFile(configLocation)
+ .flatMap(DMaaPConfigurationParser::toJSON)
+ .flatMap(DMaaPConfigurationParser::toConfigMap);
+ }
+
+ private static Try<String> readFromFile(Path configLocation) {
+ return Try(() -> new String(Files.readAllBytes(configLocation)))
+ .mapFailure(enhanceError(f("Could not read DMaaP configuration from location: '%s'", configLocation)));
+ }
+
+ private static Try<AnyNode> toJSON(String config) {
+ return Try(() -> AnyNode.fromString(config))
+ .mapFailure(enhanceError(f("DMaaP configuration '%s' is not a valid JSON document", config)));
+ }
+
+ private static Try<Map<String, PublisherConfig>> toConfigMap(AnyNode config) {
+ return Try(() -> usesLegacyFormat(config) ? parseLegacyFormat(config) : parseNewFormat(config))
+ .mapFailure(enhanceError(
+ f("Parsing DMaaP configuration: '%s' failed, probably it is in unexpected format", config)));
+ }
+
+ private static boolean usesLegacyFormat(AnyNode dMaaPConfig) {
+ return dMaaPConfig.has("channels");
+ }
+
+ private static Map<String, PublisherConfig> parseLegacyFormat(AnyNode root) {
+ return root.get("channels").toList().toMap(
+ channel -> channel.get("name").toString(),
+ channel -> {
+ String destinationsStr = channel.getAsOption("cambria.url")
+ .getOrElse(channel.getAsOption("cambria.hosts").get())
+ .toString();
+ String topic = channel.get("cambria.topic").toString();
+ Option<String> maybeUser = channel.getAsOption("basicAuthUsername").map(AnyNode::toString);
+ Option<String> maybePassword = channel.getAsOption("basicAuthPassword").map(AnyNode::toString);
+ List<String> destinations = List(destinationsStr.split(","));
+ return buildBasedOnAuth(maybeUser, maybePassword, topic, destinations);
+ });
+ }
+
+ private static Map<String, PublisherConfig> parseNewFormat(AnyNode root) {
+ return root.keys().toMap(
+ channelName -> channelName,
+ channelName -> {
+ AnyNode channelConfig = root.get(channelName);
+ Option<String> maybeUser = channelConfig.getAsOption("aaf_username").map(AnyNode::toString);
+ Option<String> maybePassword = channelConfig.getAsOption("aaf_password").map(AnyNode::toString);
+ URL topicURL = unchecked(
+ () -> new URL(channelConfig.get("dmaap_info").get("topic_url").toString())).apply();
+ String[] pathSegments = topicURL.getPath().substring(1).split("/");
+ String topic = pathSegments[1];
+ String destination = "events".equals(pathSegments[0]) ? topicURL.getAuthority() : topicURL.getHost();
+ List<String> destinations = List(destination);
+ return buildBasedOnAuth(maybeUser, maybePassword, topic, destinations);
+ });
+ }
+
+ private static PublisherConfig buildBasedOnAuth(Option<String> maybeUser, Option<String> maybePassword,
+ String topic, List<String> destinations) {
+ return maybeUser.flatMap(user -> maybePassword.map(password -> Tuple(user, password)))
+ .map(credentials -> new PublisherConfig(destinations, topic, credentials._1, credentials._2))
+ .getOrElse(new PublisherConfig(destinations, topic));
+ }
+}
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2018 Nokia. 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.onap.dcae.commonFunction.event.publishing;
+
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f;
+
+import com.att.nsa.cambria.client.CambriaBatchingPublisher;
+import com.att.nsa.clock.SaClock;
+import com.att.nsa.logging.LoggingContext;
+import com.att.nsa.logging.log4j.EcompFields;
+import io.vavr.collection.Map;
+import io.vavr.control.Try;
+import java.io.IOException;
+import org.json.JSONObject;
+import org.onap.dcae.commonFunction.VESLogger;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author Pawel Szalapski (pawel.szalapski@nokia.com)
+ */
+class DMaaPEventPublisher implements EventPublisher {
+ private static final int PENDING_MESSAGE_LOG_THRESHOLD = 100;
+ private static final String VES_UNIQUE_ID = "VESuniqueId";
+ private static final Logger log = LoggerFactory.getLogger(DMaaPEventPublisher.class);
+ private final DMaaPPublishersCache publishersCache;
+ private final Logger outputLogger;
+
+ DMaaPEventPublisher(DMaaPPublishersCache DMaaPPublishersCache,
+ Logger outputLogger) {
+ this.publishersCache = DMaaPPublishersCache;
+ this.outputLogger = outputLogger;
+ }
+
+ @Override
+ public void sendEvent(JSONObject event, String domain) {
+ clearVesUniqueIdFromEvent(event);
+ publishersCache.getPublisher(domain)
+ .onEmpty(() ->
+ log.warn(f("Could not find event publisher for domain: '%s', dropping message: '%s'", domain, event)))
+ .forEach(publisher -> sendEvent(event, domain, publisher));
+ }
+
+ @Override
+ public void reconfigure(Map<String, PublisherConfig> dMaaPConfig) {
+ publishersCache.reconfigure(dMaaPConfig);
+ }
+
+ private void sendEvent(JSONObject event, String domain, CambriaBatchingPublisher publisher) {
+ Try.run(() -> uncheckedSendEvent(event, domain, publisher))
+ .onFailure(exc -> closePublisher(event, domain, exc));
+ }
+
+ private void uncheckedSendEvent(JSONObject event, String domain, CambriaBatchingPublisher publisher)
+ throws IOException {
+ int pendingMsgs = publisher.send("MyPartitionKey", event.toString());
+ if (pendingMsgs > PENDING_MESSAGE_LOG_THRESHOLD) {
+ log.info("Pending messages count: " + pendingMsgs);
+ }
+ String infoMsg = f("Event: '%s' scheduled to be send asynchronously on domain: '%s'", event, domain);
+ log.info(infoMsg);
+ outputLogger.info(infoMsg);
+ }
+
+ private void closePublisher(JSONObject event, String domain, Throwable e) {
+ log.error(f("Unable to schedule event: '%s' on domain: '%s'. Closing publisher and dropping message.",
+ event, domain), e);
+ publishersCache.closePublisherFor(domain);
+ }
+
+ private void clearVesUniqueIdFromEvent(JSONObject event) {
+ if (event.has(VES_UNIQUE_ID)) {
+ String uuid = event.get(VES_UNIQUE_ID).toString();
+ LoggingContext localLC = VESLogger.getLoggingContextForThread(uuid);
+ localLC.put(EcompFields.kBeginTimestampMs, SaClock.now());
+ log.debug("Removing VESuniqueid object from event");
+ event.remove(VES_UNIQUE_ID);
+ }
+ }
+}
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2018 Nokia. 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.onap.dcae.commonFunction.event.publishing;
+
+import static io.vavr.API.Try;
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.enhanceError;
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f;
+
+import com.att.nsa.cambria.client.CambriaBatchingPublisher;
+import com.att.nsa.cambria.client.CambriaClientBuilders;
+import com.att.nsa.cambria.client.CambriaClientBuilders.PublisherBuilder;
+import io.vavr.control.Try;
+
+/**
+ * @author Pawel Szalapski (pawel.szalapski@nokia.com)
+ */
+final class DMaaPPublishersBuilder {
+
+ @SuppressWarnings("mapFailure takes a generic varargs, unchecked because of Javas type system limitation, actually safe to do")
+ static Try<CambriaBatchingPublisher> buildPublisher(PublisherConfig config) {
+ return Try(() -> builder(config).build())
+ .mapFailure(enhanceError(f("DMaaP client builder throws exception for this configuration: '%s'", config)));
+ }
+
+ private static PublisherBuilder builder(PublisherConfig config) {
+ if (config.isSecured()) {
+ return authenticatedBuilder(config);
+ } else {
+ return unAuthenticatedBuilder(config);
+ }
+ }
+
+ private static PublisherBuilder authenticatedBuilder(PublisherConfig config) {
+ return unAuthenticatedBuilder(config)
+ .usingHttps()
+ .authenticatedByHttp(config.userName().get(), config.password().get());
+ }
+
+ private static PublisherBuilder unAuthenticatedBuilder(PublisherConfig config) {
+ return new CambriaClientBuilders.PublisherBuilder()
+ .usingHosts(config.destinations().mkString(","))
+ .onTopic(config.topic())
+ .logSendFailuresAfter(5);
+ }
+}
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2018 Nokia. 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.onap.dcae.commonFunction.event.publishing;
+
+import static io.vavr.API.Option;
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f;
+
+import com.att.nsa.cambria.client.CambriaBatchingPublisher;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.cache.RemovalListener;
+import com.google.common.cache.RemovalNotification;
+import io.vavr.collection.Map;
+import io.vavr.control.Option;
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.annotation.Nonnull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author Pawel Szalapski (pawel.szalapski@nokia.com)
+ */
+class DMaaPPublishersCache {
+
+ private static final Logger log = LoggerFactory.getLogger(DMaaPPublishersCache.class);
+ private final LoadingCache<String, CambriaBatchingPublisher> publishersCache;
+ private AtomicReference<Map<String, PublisherConfig>> dMaaPConfiguration;
+
+ DMaaPPublishersCache(Map<String, PublisherConfig> dMaaPConfiguration) {
+ this.dMaaPConfiguration = new AtomicReference<>(dMaaPConfiguration);
+ this.publishersCache = CacheBuilder.newBuilder()
+ .removalListener(new OnPublisherRemovalListener())
+ .build(new CambriaPublishersCacheLoader());
+ }
+
+ DMaaPPublishersCache(CambriaPublishersCacheLoader dMaaPPublishersCacheLoader,
+ OnPublisherRemovalListener onPublisherRemovalListener,
+ Map<String, PublisherConfig> dMaaPConfiguration) {
+ this.dMaaPConfiguration = new AtomicReference<>(dMaaPConfiguration);
+ this.publishersCache = CacheBuilder.newBuilder()
+ .removalListener(onPublisherRemovalListener)
+ .build(dMaaPPublishersCacheLoader);
+ }
+
+ Option<CambriaBatchingPublisher> getPublisher(String streamID) {
+ try {
+ return Option(publishersCache.getUnchecked(streamID));
+ } catch (Exception e) {
+ log.warn("Could not create / load Cambria Publisher for streamID", e);
+ return Option.none();
+ }
+ }
+
+ void closePublisherFor(String streamId) {
+ publishersCache.invalidate(streamId);
+ }
+
+ synchronized void reconfigure(Map<String, PublisherConfig> newConfig) {
+ Map<String, PublisherConfig> currentConfig = dMaaPConfiguration.get();
+ Map<String, PublisherConfig> removedConfigurations = currentConfig
+ .filterKeys(domain -> !newConfig.containsKey(domain));
+ Map<String, PublisherConfig> changedConfigurations = newConfig
+ .filterKeys(e -> currentConfig.containsKey(e) && !currentConfig.get(e).equals(newConfig.get(e)));
+ dMaaPConfiguration.set(newConfig);
+ removedConfigurations.merge(changedConfigurations).forEach(e -> publishersCache.invalidate(e._1));
+ }
+
+ static class OnPublisherRemovalListener implements RemovalListener<String, CambriaBatchingPublisher> {
+
+ @Override
+ public void onRemoval(@Nonnull RemovalNotification<String, CambriaBatchingPublisher> notification) {
+ CambriaBatchingPublisher publisher = notification.getValue();
+ if (publisher != null) { // The value might get Garbage Collected at this moment, regardless of @Nonnull
+ try {
+ int timeout = 20;
+ TimeUnit unit = TimeUnit.SECONDS;
+ java.util.List<?> stuck = publisher.close(timeout, unit);
+ if (!stuck.isEmpty()) {
+ log.error(f("Publisher got stuck and did not manage to close in '%s' '%s', "
+ + "%s messages were dropped", stuck.size(), timeout, unit));
+ }
+ } catch (InterruptedException | IOException e) {
+ log.error("Could not close Cambria publisher, some messages might have been dropped", e);
+ }
+ }
+ }
+ }
+
+ class CambriaPublishersCacheLoader extends CacheLoader<String, CambriaBatchingPublisher> {
+
+ @Override
+ public CambriaBatchingPublisher load(@Nonnull String domain) {
+ return dMaaPConfiguration.get()
+ .get(domain)
+ .toTry(() -> new RuntimeException(
+ f("DMaaP configuration contains no configuration for domain: '%s'", domain)))
+ .flatMap(DMaaPPublishersBuilder::buildPublisher)
+ .get();
+ }
+ }
+
+}
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2018 Nokia. 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.onap.dcae.commonFunction.event.publishing;
+
+import io.vavr.collection.Map;
+import org.json.JSONObject;
+import org.slf4j.Logger;
+
+/**
+ * @author Pawel Szalapski (pawel.szalapski@nokia.com)
+ */
+public interface EventPublisher {
+
+ static EventPublisher createPublisher(Logger outputLogger, Map<String, PublisherConfig> dMaaPConfig) {
+ return new DMaaPEventPublisher(new DMaaPPublishersCache(dMaaPConfig), outputLogger);
+ }
+
+ void sendEvent(JSONObject event, String domain);
+
+ void reconfigure(Map<String, PublisherConfig> dMaaPConfig);
+}
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2018 Nokia. 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.onap.dcae.commonFunction.event.publishing;
+
+import io.vavr.collection.List;
+import io.vavr.control.Option;
+import java.util.Objects;
+
+/**
+ * @author Pawel Szalapski (pawel.szalapski@nokia.com)
+ */
+public final class PublisherConfig {
+
+ private final List<String> destinations;
+ private final String topic;
+ private String userName;
+ private String password;
+
+ PublisherConfig(List<String> destinations, String topic) {
+ this.destinations = destinations;
+ this.topic = topic;
+ }
+
+ PublisherConfig(List<String> destinations, String topic, String userName, String password) {
+ this.destinations = destinations;
+ this.topic = topic;
+ this.userName = userName;
+ this.password = password;
+ }
+
+ List<String> destinations() {
+ return destinations;
+ }
+
+ String topic() {
+ return topic;
+ }
+
+ Option<String> userName() {
+ return Option.of(userName);
+ }
+
+ Option<String> password() {
+ return Option.of(password);
+ }
+
+ boolean isSecured() {
+ return userName().isDefined() && password().isDefined();
+ }
+
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ PublisherConfig that = (PublisherConfig) o;
+ return Objects.equals(destinations, that.destinations) &&
+ Objects.equals(topic, that.topic) &&
+ Objects.equals(userName, that.userName) &&
+ Objects.equals(password, that.password);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(destinations, topic, userName, password);
+ }
+
+ @Override
+ public String toString() {
+ return "PublisherConfig{" +
+ "destinations=" + destinations +
+ ", topic='" + topic + '\'' +
+ ", userName='" + userName + '\'' +
+ ", password='" + password + '\'' +
+ '}';
+ }
+}
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2018 Nokia. 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.onap.dcae.commonFunction.event.publishing;
+
+import static io.vavr.API.$;
+
+import io.vavr.API;
+import io.vavr.API.Match.Case;
+
+/**
+ * @author Pawel Szalapski (pawel.szalapski@nokia.com)
+ */
+final class VavrUtils {
+
+ private VavrUtils() {
+ // utils aggregator
+ }
+
+ /**
+ * Shortcut for 'string interpolation'
+ */
+ static String f(String msg, Object... args) {
+ return String.format(msg, args);
+ }
+
+ /**
+ * Wrap failure with a more descriptive message of what has failed and chain original cause. Used to provide a
+ * context for errors instead of raw exception.
+ */
+ static Case<Throwable, Throwable> enhanceError(String msg) {
+ return API.Case($(), e -> new RuntimeException(msg, e));
+ }
+
+}
+++ /dev/null
-/*-
- * ============LICENSE_START=======================================================
- * org.onap.dcaegen2.collectors.ves
- * ================================================================================
- * Copyright (C) 2018 Nokia. 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.onap.dcae.commonFunction;
-
-import static org.hamcrest.CoreMatchers.allOf;
-import static org.hamcrest.CoreMatchers.instanceOf;
-import static org.junit.Assert.assertSame;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyLong;
-import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.BDDMockito.given;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-
-import com.att.nsa.cambria.client.CambriaBatchingPublisher;
-import com.att.nsa.cambria.client.CambriaPublisher;
-import com.google.common.collect.ImmutableList;
-import com.google.common.util.concurrent.UncheckedExecutionException;
-import java.io.IOException;
-import java.net.MalformedURLException;
-import java.security.GeneralSecurityException;
-import java.util.concurrent.TimeUnit;
-import org.hamcrest.BaseMatcher;
-import org.hamcrest.Description;
-import org.hamcrest.Matcher;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
-import org.junit.runner.RunWith;
-import org.mockito.Mock;
-import org.mockito.junit.MockitoJUnitRunner;
-
-
-@RunWith(MockitoJUnitRunner.class)
-public class DmaapPublishersTest {
-
- @Mock
- private CambriaPublisherFactory publisherFactory;
- @Mock
- private CambriaBatchingPublisher cambriaPublisher;
- private DmaapPublishers cut;
-
- @Rule
- public final ExpectedException expectedException = ExpectedException.none();
-
- @Before
- public void setUp() throws MalformedURLException, GeneralSecurityException {
- given(publisherFactory.createCambriaPublisher(anyString())).willReturn(cambriaPublisher);
- cut = DmaapPublishers.create(publisherFactory);
- }
-
- @Test
- public void getByStreamIdShouldUseCachedItem() throws IOException, GeneralSecurityException {
- // given
- String streamId = "sampleStream";
-
- // when
- CambriaBatchingPublisher firstPublisher = cut.getByStreamId(streamId);
- CambriaBatchingPublisher secondPublisher = cut.getByStreamId(streamId);
-
- // then
- verify(publisherFactory, times(1)).createCambriaPublisher(streamId);
- assertSame("should return same instance", firstPublisher, secondPublisher);
- }
-
- @Test
- public void getByStreamIdShouldHandleErrors() throws MalformedURLException, GeneralSecurityException {
- // given
- MalformedURLException exception = new MalformedURLException();
- given(publisherFactory.createCambriaPublisher(anyString())).willThrow(exception);
- expectedException.expect(allOf(
- instanceOf(UncheckedExecutionException.class),
- causeIsInstanceOf(exception.getClass())));
-
- // when
- cut.getByStreamId("a stream");
-
- // then
- // exception should have been thrown
- }
-
- @Test
- public void closeByStreamIdShouldCloseConnection() throws IOException, InterruptedException {
- // given
- String streamId = "sampleStream";
- given(cambriaPublisher.close(anyLong(), any(TimeUnit.class)))
- .willReturn(ImmutableList.of(new CambriaPublisher.message("p", "msg")));
-
- // when
- CambriaBatchingPublisher cachedPublisher = cut.getByStreamId(streamId);
- cut.closeByStreamId(streamId);
-
- // then
- assertSame("should return proper publisher", cambriaPublisher, cachedPublisher);
- verify(cambriaPublisher).close(20, TimeUnit.SECONDS);
- }
-
- @Test
- public void closeByStreamIdShouldHandleErrors() throws IOException, InterruptedException {
- // given
- String streamId = "sampleStream";
- given(cambriaPublisher.close(anyLong(), any(TimeUnit.class))).willThrow(IOException.class);
-
- // when
- CambriaBatchingPublisher cachedPublisher = cut.getByStreamId(streamId);
- cut.closeByStreamId(streamId);
-
- // then
- assertSame("should return proper publisher", cambriaPublisher, cachedPublisher);
- verify(cambriaPublisher).close(20, TimeUnit.SECONDS);
- }
-
- private Matcher<Exception> causeIsInstanceOf(final Class<?> clazz) {
- return new BaseMatcher<Exception>() {
- @Override
- public boolean matches(Object o) {
- return o instanceof Throwable && clazz.isInstance(((Throwable) o).getCause());
- }
-
- @Override
- public void describeTo(Description description) {
- description.appendText("exception cause should be an instance of " + clazz.getName());
- }
- };
- }
-}
\ No newline at end of file
*/
package org.onap.dcae.commonFunction;
-import com.att.nsa.cambria.client.CambriaBatchingPublisher;
import com.google.gson.Gson;
+import java.util.concurrent.atomic.AtomicReference;
import org.json.JSONObject;
import org.junit.Before;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import java.util.List;
+import org.onap.dcae.commonFunction.event.publishing.EventPublisher;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
@Test
public void testLoad() {
//given
- EventProcessor ec = new EventProcessor();
+ EventProcessor ec = new EventProcessor(mock(EventPublisher.class));
ec.event = new org.json.JSONObject(ev);
//when
ec.overrideEvent();
@Test
public void shouldParseJsonEvents() throws ReflectiveOperationException {
//given
- EventProcessor eventProcessor = new EventProcessor();
+ EventProcessor eventProcessor = new EventProcessor(mock(EventPublisher.class));
String event_json = "[{ \"filter\": {\"event.commonEventHeader.domain\":\"heartbeat\",\"VESversion\":\"v4\"},\"processors\":[" +
"{\"functionName\": \"concatenateValue\",\"args\":{\"field\":\"event.commonEventHeader.eventName\",\"concatenate\": [\"$event.commonEventHeader.domain\",\"$event.commonEventHeader.eventType\",\"$event.faultFields.alarmCondition\"], \"delimiter\":\"_\"}}" +
",{\"functionName\": \"addAttribute\",\"args\":{\"field\": \"event.heartbeatFields.heartbeatFieldsVersion\",\"value\": \"1.0\",\"fieldType\": \"number\"}}" +
assertThat(stringArgumentCaptor.getAllValues()).contains("concatenateValue", "addAttribute", "map");
}
- @Test
- public void shouldCreateDmaapPublisher() {
-
- //given
- EventPublisherHash eph = EventPublisherHash.getInstance();
- EventProcessor ec = new EventProcessor();
- ec.event = new org.json.JSONObject(ev);
- CommonStartup.cambriaConfigFile = "src/test/resources/testDmaapConfig_ip.json";
-
- //when
- CambriaBatchingPublisher pub = eph.getDmaapPublisher("sec_fault_ueb");
-
- //then
- assertNotNull(pub);
- }
-
- @Test
- public void shouldSendEventWithNoError() {
-
- EventPublisherHash eph = EventPublisherHash.getInstance();
- EventProcessor eventProcessor = new EventProcessor();
- eventProcessor.event = new org.json.JSONObject(ev);
- CommonStartup.cambriaConfigFile = "src/test/resources/testDmaapConfig_ip.json";
-
- //when
- eph.sendEvent(eventProcessor.event, "sec_fault_ueb");
- }
}
import static java.util.Base64.getDecoder;
import static java.util.Base64.getEncoder;
import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import com.att.nsa.cmdLine.NsaCommandLineUtil;
import org.junit.Test;
import org.mockito.Mockito;
import org.onap.dcae.commonFunction.CommonStartup.QueueFullException;
+import org.onap.dcae.commonFunction.event.publishing.EventPublisher;
import org.onap.dcae.restapi.RestfulCollectorServlet;
public void testParseStreamIdToStreamHashMapping() {
// given
CommonStartup.streamID = "fault=sec_fault|syslog=sec_syslog|heartbeat=sec_heartbeat|measurementsForVfScaling=sec_measurement|mobileFlow=sec_mobileflow|other=sec_other|stateChange=sec_statechange|thresholdCrossingAlert=sec_thresholdCrossingAlert|voiceQuality=ves_voicequality|sipSignaling=ves_sipsignaling";
- EventProcessor eventProcessor = new EventProcessor();
+ EventProcessor eventProcessor = new EventProcessor(mock(EventPublisher.class));
// when
Map<String, String[]> streamHashMapping = EventProcessor.streamidHash;
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2018 Nokia. 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.onap.dcae.commonFunction.event.publishing;
+
+import static io.vavr.API.List;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.onap.dcae.commonFunction.event.publishing.DMaaPConfigurationParser.parseToDomainMapping;
+
+import io.vavr.collection.Map;
+import io.vavr.control.Try;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import org.junit.Test;
+
+/**
+ * @author Pawel Szalapski (pawel.szalapski@nokia.com)
+ */
+public class DMaaPConfigurationParserTest {
+
+ @Test
+ public void testParseCredentialsForGen2() {
+ Path path = Paths.get("src/test/resources/testParseDMaaPCredentialsGen2.json");
+ Try<Map<String, PublisherConfig>> publisherConfigs = parseToDomainMapping(path);
+
+ PublisherConfig authCredentialsNulls = publisherConfigs.get().get("auth-credentials-null").getOrNull();
+ assertThat(authCredentialsNulls.userName().isEmpty()).isTrue();
+ assertThat(authCredentialsNulls.password().isEmpty()).isTrue();
+ assertThat(authCredentialsNulls.isSecured()).isFalse();
+
+ PublisherConfig authCredentialsPresent = publisherConfigs.get().get("auth-credentials-present").getOrNull();
+ assertThat(authCredentialsPresent.userName().getOrNull()).isEqualTo("sampleUser");
+ assertThat(authCredentialsPresent.password().getOrNull()).isEqualTo("samplePassword");
+ assertThat(authCredentialsPresent.isSecured()).isTrue();
+
+ PublisherConfig authCredentialsKeysMissing = publisherConfigs.get().get("auth-credentials-missing").getOrNull();
+ assertThat(authCredentialsKeysMissing.userName().isEmpty()).isTrue();
+ assertThat(authCredentialsKeysMissing.password().isEmpty()).isTrue();
+ assertThat(authCredentialsKeysMissing.isSecured()).isFalse();
+ }
+
+
+ @Test
+ public void testParseCredentialsForLegacy() {
+ Path path = Paths.get("src/test/resources/testParseDMaaPCredentialsLegacy.json");
+ Try<Map<String, PublisherConfig>> publisherConfigs = parseToDomainMapping(path);
+
+ PublisherConfig authCredentialsNull = publisherConfigs.get().get("auth-credentials-null").getOrNull();
+ assertThat(authCredentialsNull.userName().isEmpty()).isTrue();
+ assertThat(authCredentialsNull.password().isEmpty()).isTrue();
+ assertThat(authCredentialsNull.isSecured()).isFalse();
+
+ PublisherConfig authCredentialsPresent = publisherConfigs.get().get("auth-credentials-present").getOrNull();
+ assertThat(authCredentialsPresent.userName().getOrNull()).isEqualTo("sampleUser");
+ assertThat(authCredentialsPresent.password().getOrNull()).isEqualTo("samplePassword");
+ assertThat(authCredentialsPresent.isSecured()).isTrue();
+
+ PublisherConfig authCredentialsMissing = publisherConfigs.get().get("auth-credentials-missing").getOrNull();
+ assertThat(authCredentialsMissing.userName().isEmpty()).isTrue();
+ assertThat(authCredentialsMissing.password().isEmpty()).isTrue();
+ assertThat(authCredentialsMissing.isSecured()).isFalse();
+ }
+
+
+ @Test
+ public void testParseGen2() {
+ Path path = Paths.get("src/test/resources/testParseDMaaPGen2.json");
+ Try<Map<String, PublisherConfig>> publisherConfigs = parseToDomainMapping(path);
+
+ PublisherConfig withEventsSegment = publisherConfigs.get().get("event-segments-with-port").getOrNull();
+ assertThat(withEventsSegment.destinations()).isEqualTo(List("UEBHOST:3904"));
+ assertThat(withEventsSegment.topic()).isEqualTo("DCAE-SE-COLLECTOR-EVENTS-DEV");
+
+ PublisherConfig withOtherSegment = publisherConfigs.get().get("other-segments-without-ports").getOrNull();
+ assertThat(withOtherSegment.destinations()).isEqualTo(List("UEBHOST"));
+ assertThat(withOtherSegment.topic()).isEqualTo("DCAE-SE-COLLECTOR-EVENTS-DEV");
+ }
+
+ @Test
+ public void testParseLegacy() {
+ Path exemplaryConfig = Paths.get("src/test/resources/testParseDMaaPLegacy.json");
+ Try<Map<String, PublisherConfig>> publisherConfigs = DMaaPConfigurationParser
+ .parseToDomainMapping(exemplaryConfig);
+
+ PublisherConfig urlFirstThenHosts = publisherConfigs.get().get("url-precedes-hosts").getOrNull();
+ assertThat(urlFirstThenHosts.destinations()).isEqualTo(List("127.0.0.1:3904"));
+ assertThat(urlFirstThenHosts.topic()).isEqualTo("DCAE-SE-COLLECTOR-EVENTS-DEV");
+
+ PublisherConfig urlKeyMissing = publisherConfigs.get().get("url-key-missing").getOrNull();
+ assertThat(urlKeyMissing.destinations()).isEqualTo(List("h1.att.com", "h2.att.com"));
+ assertThat(urlKeyMissing.topic()).isEqualTo("DCAE-SE-COLLECTOR-EVENTS-DEV");
+
+ PublisherConfig urlIsMissing = publisherConfigs.get().get("url-is-null").getOrNull();
+ assertThat(urlIsMissing.destinations()).isEqualTo(List("h1.att.com", "h2.att.com"));
+ assertThat(urlIsMissing.topic()).isEqualTo("DCAE-SE-COLLECTOR-EVENTS-DEV");
+ }
+}
\ No newline at end of file
* limitations under the License.
* ============LICENSE_END=========================================================
*/
-package org.onap.dcae.commonFunction;
+package org.onap.dcae.commonFunction.event.publishing;
+import static io.vavr.API.Option;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
import com.att.nsa.cambria.client.CambriaBatchingPublisher;
import java.io.IOException;
import org.json.JSONObject;
import org.junit.Before;
import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.Mock;
-import org.mockito.junit.MockitoJUnitRunner;
+import org.slf4j.Logger;
-@RunWith(MockitoJUnitRunner.class)
-public class EventPublisherHashTest {
- private EventPublisherHash cut;
+public class DMaaPEventPublisherTest {
- @Mock
- private DmaapPublishers dmaapPublishers;
- @Mock
+ private static final String STREAM_ID = "sampleStreamId";
+
+ private DMaaPEventPublisher eventPublisher;
private CambriaBatchingPublisher cambriaPublisher;
+ private DMaaPPublishersCache DMaaPPublishersCache;
@Before
public void setUp() {
- given(dmaapPublishers.getByStreamId(anyString())).willReturn(cambriaPublisher);
-
- cut = new EventPublisherHash(dmaapPublishers);
+ cambriaPublisher = mock(CambriaBatchingPublisher.class);
+ DMaaPPublishersCache = mock(DMaaPPublishersCache.class);
+ when(DMaaPPublishersCache.getPublisher(anyString())).thenReturn(Option(cambriaPublisher));
+ eventPublisher = new DMaaPEventPublisher(DMaaPPublishersCache, mock(Logger.class));
}
@Test
- public void sendEventShouldSendEventToATopic() throws Exception {
+ public void shouldSendEventToTopic() throws Exception {
// given
JSONObject event = new JSONObject("{}");
- final String streamId = "sampleStreamId";
// when
- cut.sendEvent(event, streamId);
+ eventPublisher.sendEvent(event, STREAM_ID);
// then
verify(cambriaPublisher).send("MyPartitionKey", event.toString());
}
@Test
- public void sendEventShouldRemoveUuid() throws Exception {
+ public void shouldRemoveInternalVESUIDBeforeSending() throws Exception {
// given
- JSONObject event = new JSONObject("{\"VESuniqueId\": \"362e0146-ec5f-45f3-8d8f-bfe877c3f58e\", \"another\": 8}");
- final String streamId = "sampleStreamId";
+ JSONObject event = new JSONObject(
+ "{\"VESuniqueId\": \"362e0146-ec5f-45f3-8d8f-bfe877c3f58e\", \"another\": 8}");
// when
- cut.sendEvent(event, streamId);
+ eventPublisher.sendEvent(event, STREAM_ID);
// then
verify(cambriaPublisher).send("MyPartitionKey", new JSONObject("{\"another\": 8}").toString());
}
@Test
- public void sendEventShouldCloseConnectionWhenExceptionOccurred() throws Exception {
+ public void shouldCloseConnectionWhenExceptionOccurred() throws Exception {
// given
JSONObject event = new JSONObject("{}");
- final String streamId = "sampleStreamId";
given(cambriaPublisher.send(anyString(), anyString())).willThrow(new IOException("epic fail"));
// when
- cut.sendEvent(event, streamId);
+ eventPublisher.sendEvent(event, STREAM_ID);
// then
- verify(dmaapPublishers).closeByStreamId(streamId);
+ verify(DMaaPPublishersCache).closePublisherFor(STREAM_ID);
}
}
\ No newline at end of file
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2018 Nokia. 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.onap.dcae.commonFunction.event.publishing;
+
+import static io.vavr.API.List;
+import static io.vavr.API.Map;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.mockito.Mockito.when;
+
+import com.att.nsa.cambria.client.CambriaBatchingPublisher;
+import io.vavr.collection.Map;
+import io.vavr.control.Option;
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.dcae.commonFunction.event.publishing.DMaaPPublishersCache.CambriaPublishersCacheLoader;
+import org.onap.dcae.commonFunction.event.publishing.DMaaPPublishersCache.OnPublisherRemovalListener;
+
+
+public class DMaaPPublishersCacheTest {
+
+ private String streamId1;
+ private Map<String, PublisherConfig> dMaaPConfigs;
+
+ @Before
+ public void setUp() {
+ streamId1 = "sampleStream1";
+ dMaaPConfigs = Map("sampleStream1", new PublisherConfig(List("destination1"), "topic1"));
+ }
+
+ @Test
+ public void shouldReturnTheSameCachedInstanceOnConsecutiveRetrievals() {
+ // given
+ DMaaPPublishersCache dMaaPPublishersCache = new DMaaPPublishersCache(dMaaPConfigs);
+
+ // when
+ Option<CambriaBatchingPublisher> firstPublisher = dMaaPPublishersCache.getPublisher(streamId1);
+ Option<CambriaBatchingPublisher> secondPublisher = dMaaPPublishersCache.getPublisher(streamId1);
+
+ // then
+ assertSame("should return same instance", firstPublisher.get(), secondPublisher.get());
+ }
+
+ @Test
+ public void shouldCloseCambriaPublisherOnCacheInvalidate() throws IOException, InterruptedException {
+ // given
+ CambriaBatchingPublisher cambriaPublisherMock1 = mock(CambriaBatchingPublisher.class);
+ CambriaPublishersCacheLoader cacheLoaderMock = mock(CambriaPublishersCacheLoader.class);
+ DMaaPPublishersCache dMaaPPublishersCache = new DMaaPPublishersCache(cacheLoaderMock,
+ new OnPublisherRemovalListener(),
+ dMaaPConfigs);
+ when(cacheLoaderMock.load(streamId1)).thenReturn(cambriaPublisherMock1);
+
+ // when
+ dMaaPPublishersCache.getPublisher(streamId1);
+ dMaaPPublishersCache.closePublisherFor(streamId1);
+
+ // then
+ verify(cambriaPublisherMock1).close(20, TimeUnit.SECONDS);
+
+ }
+
+ @Test
+ public void shouldReturnNoneIfThereIsNoDMaaPConfigurationForGivenStreamID() {
+ // given
+ DMaaPPublishersCache dMaaPPublishersCache = new DMaaPPublishersCache(dMaaPConfigs);
+
+ // then
+ assertTrue("should not exist", dMaaPPublishersCache.getPublisher("non-existing").isEmpty());
+ }
+
+
+ @Test
+ public void shouldCloseOnlyChangedPublishers() throws IOException, InterruptedException {
+ // given
+ CambriaBatchingPublisher cambriaPublisherMock1 = mock(CambriaBatchingPublisher.class);
+ CambriaBatchingPublisher cambriaPublisherMock2 = mock(CambriaBatchingPublisher.class);
+ CambriaPublishersCacheLoader cacheLoaderMock = mock(CambriaPublishersCacheLoader.class);
+ String firstDomain = "domain1";
+ String secondDomain = "domain2";
+ Map<String, PublisherConfig> oldConfig = Map(firstDomain,
+ new PublisherConfig(List("destination1"), "topic1"),
+ secondDomain,
+ new PublisherConfig(List("destination2"), "topic2",
+ "user", "pass"));
+ Map<String, PublisherConfig> newConfig = Map(firstDomain, new PublisherConfig(List("destination1"), "topic1"),
+ secondDomain, new PublisherConfig(List("destination2"), "topic2"));
+ DMaaPPublishersCache dMaaPPublishersCache = new DMaaPPublishersCache(cacheLoaderMock,
+ new OnPublisherRemovalListener(),
+ oldConfig);
+ when(cacheLoaderMock.load(firstDomain)).thenReturn(cambriaPublisherMock1);
+ when(cacheLoaderMock.load(secondDomain)).thenReturn(cambriaPublisherMock2);
+
+ dMaaPPublishersCache.getPublisher(firstDomain);
+ dMaaPPublishersCache.getPublisher(secondDomain);
+
+ // when
+ dMaaPPublishersCache.reconfigure(newConfig);
+
+ // then
+ verify(cambriaPublisherMock2).close(20, TimeUnit.SECONDS);
+ verifyZeroInteractions(cambriaPublisherMock1);
+ }
+}
\ No newline at end of file
*/
package org.onap.dcae.vestest;
-import com.google.common.collect.ImmutableMap;
+import static org.assertj.core.api.Assertions.assertThat;
+
import com.google.common.collect.Sets;
-import org.json.JSONObject;
+import java.util.Set;
import org.junit.BeforeClass;
import org.junit.Test;
import org.onap.dcae.commonFunction.AnyNode;
-import java.io.IOException;
-import java.util.Map;
-import java.util.Set;
-import java.util.stream.Collectors;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
/**
* Created by koblosz on 07.06.18.
*/
public class AnyNodeTest {
- private static final String SAMPLE_JSON_FILEPATH = "src/test/resources/test_anynode_class.json";
- private static final Map<String, Object> EXPECTED_RAW_MAP = ImmutableMap.<String, Object>builder().put("a", 1).put("b", 2).build();
- private static final Set<String> EXPECTED_JSON_KEYS = Sets.newHashSet("channels", "sampleStrList", "sampleNestedObject", "sampleInt", "sampleString", "sampleNull");
+ private static final String SAMPLE_JSON_FILEPATH = "{\n"
+ + " \"channels\": [{\n"
+ + " \"one\": \"number1\", \"two\": \"number2\", \"three\": \"number3\"}],\n"
+ + " \"sampleStrList\": [\"1\", \"2\", \"3\", \"4\", \"5\"],\n"
+ + " \"sampleNestedObject\": {\"a\": 1, \"b\": 2},\n"
+ + " \"sampleInt\": 1,\n"
+ + " \"sampleString\": \"str\",\n"
+ + " \"sampleNull\": null\n"
+ + "}\n";
+ private static final Set<String> EXPECTED_JSON_KEYS = Sets
+ .newHashSet("channels", "sampleStrList", "sampleNestedObject", "sampleInt", "sampleString", "sampleNull");
private static AnyNode node;
@BeforeClass
- public static void setUpClass() throws IOException {
- node = AnyNode.parse(SAMPLE_JSON_FILEPATH);
- }
-
- @Test(expected = IOException.class)
- public void testShouldRethrowExceptionWhenFileNotFound() throws IOException {
- AnyNode.parse("not/existing/path");
+ public static void setUpClass() {
+ node = AnyNode.fromString(SAMPLE_JSON_FILEPATH);
}
@Test
public void testShouldReturnJsonObjectKeySet() {
- assertThat(node.getKeys()).containsOnlyElementsOf(EXPECTED_JSON_KEYS);
- }
-
- @Test
- public void testShouldGetElementAsString() {
- assertThat(node.get("sampleStrList").get(0).asString()).isEqualTo("1");
- }
-
- @Test
- public void testShouldGetElementAsInt() {
- assertThat(node.get("sampleInt").asInt()).isSameAs(1);
- }
-
- @Test
- public void testWhenNullValuePresentShouldReturnJsonObjectNullAsString() {
- assertThat(node.get("sampleNull").asString()).isSameAs(JSONObject.NULL.toString());
+ assertThat(node.keys()).containsOnlyElementsOf(EXPECTED_JSON_KEYS);
}
- @Test
- public void testShouldGetJsonObjectAsStringToObjectMap() {
- assertThat(node.get("sampleNestedObject").asRawMap()).containsAllEntriesOf(EXPECTED_RAW_MAP);
- }
-
- @Test
- public void testShouldGetAsMap() {
- assertThat(node.asMap().keySet()).containsOnlyElementsOf(EXPECTED_JSON_KEYS);
- }
-
- @Test
- public void testShouldGetAsList() {
- assertThat(node.get("sampleStrList").asList().stream().map(AnyNode::asString).collect(Collectors.toList())).containsExactly("1", "2", "3", "4", "5");
- }
-
- @Test
- public void testShouldGetAsOptional() {
- assertThat(node.getAsOptional("absentKey")).isNotPresent();
- }
-
- @Test
- public void testWhenChainMethodsShouldReturnValue() {
- assertThat(node.get("channels").get(0).get("two").asString()).isEqualTo("number2");
- }
-
-
@Test(expected = ClassCastException.class)
public void whenInvokedOnJsonObjInsteadOfJsonArrShouldRaiseRuntimeEx() {
- node.asList();
+ node.toList();
}
}
\ No newline at end of file
+++ /dev/null
-/*-\r
- * ============LICENSE_START=======================================================\r
- * org.onap.dcaegen2.collectors.ves\r
- * ================================================================================\r
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.\r
- * Copyright (C) 2018 Nokia. All rights reserved.\r
- * ================================================================================\r
- * Licensed under the Apache License, Version 2.0 (the "License");\r
- * you may not use this file except in compliance with the License.\r
- * You may obtain a copy of the License at\r
- * \r
- * http://www.apache.org/licenses/LICENSE-2.0\r
- * \r
- * Unless required by applicable law or agreed to in writing, software\r
- * distributed under the License is distributed on an "AS IS" BASIS,\r
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
- * See the License for the specific language governing permissions and\r
- * limitations under the License.\r
- * ============LICENSE_END=========================================================\r
- */\r
-package org.onap.dcae.vestest;\r
-\r
-import com.google.common.collect.ImmutableMap;\r
-import org.junit.Test;\r
-import org.onap.dcae.commonFunction.DmaapPropertyReader;\r
-\r
-import java.util.Map;\r
-\r
-import static org.assertj.core.api.AssertionsForInterfaceTypes.assertThat;\r
-\r
-public class DmaapPropertyReaderTest {\r
-\r
-\r
- private static final String legacyConfigFilePath = "src/test/resources/testDmaapConfig_ip.json";\r
- private static final String dmaapInputConfigFilePath = "src/test/resources/testDmaapConfig_gen2.json";\r
- private static final String fullDmaapConfigWithChannels = "src/test/resources/testFullDmaapConfig_channels.json";\r
- private static final String fullGen2DmaapConfig = "src/test/resources/testFullDmaapConfig_gen2.json";\r
-\r
- private static final String FAULT_UEB_KEY_PREFIX = "sec_fault_ueb";\r
- private static final String VES_ALERT_SND_KEY_PREFIX = "ves-thresholdCrossingAlert-secondary";\r
- private static final String VES_FAULT_SECONDARY = "ves-fault-secondary";\r
-\r
- private static final String FAULT_BASIC_AUTH_USERNAME_KEY = VES_FAULT_SECONDARY + ".basicAuthUsername";\r
- private static final String ALERT_BASIC_AUTH_PWD_KEY = VES_ALERT_SND_KEY_PREFIX + ".basicAuthPassword";\r
-\r
- private static final String VES_ALERT_CAMBRIA_TOPIC_KEY = VES_ALERT_SND_KEY_PREFIX + ".cambria.topic";\r
- private static final String VES_ALERT_CAMBRIA_URL_KEY = VES_ALERT_SND_KEY_PREFIX + ".cambria.url";\r
- private static final String VES_FAULT_SND_CAMBRIA_URL_KEY = VES_FAULT_SECONDARY + ".cambria.url";\r
- private static final String VES_FAULT_SND_AUTH_PWD_KEY = VES_FAULT_SECONDARY + ".basicAuthPassword";\r
- private static final String VES_FAULT_SND_CAMBRIA_TOPIC_KEY = VES_FAULT_SECONDARY + ".cambria.topic";\r
- private static final String FAULT_UEB_CAMBRIA_HOSTS_KEY = FAULT_UEB_KEY_PREFIX + ".cambria.hosts";\r
- private static final String FAULT_UEB_CAMBRIA_TOPIC_KEY = FAULT_UEB_KEY_PREFIX + ".cambria.topic";\r
- private static final String VES_ALERT_SND_AUTH_USERNAME_KEY = VES_ALERT_SND_KEY_PREFIX + ".basicAuthUsername";\r
-\r
- private static final String NULL_TOSTRING = "null";\r
-\r
- private static final Map<String, String> expectedCompleteGen2DmaapConfig = ImmutableMap.<String, String>builder()\r
- .put(ALERT_BASIC_AUTH_PWD_KEY, "SamplePassWD2")\r
- .put(VES_ALERT_CAMBRIA_TOPIC_KEY, "DCAE-SE-COLLECTOR-EVENTS-DEV")\r
- .put(FAULT_BASIC_AUTH_USERNAME_KEY, "sampleUsername")\r
- .put(VES_ALERT_CAMBRIA_URL_KEY, "UEBHOST:3904")\r
- .put(VES_FAULT_SND_CAMBRIA_URL_KEY, "UEBHOST:3904")\r
- .put(VES_FAULT_SND_AUTH_PWD_KEY, "SamplePasswd")\r
- .put(VES_FAULT_SND_CAMBRIA_TOPIC_KEY, "DCAE-SE-COLLECTOR-EVENTS-DEV")\r
- .put(VES_ALERT_SND_AUTH_USERNAME_KEY, "sampleUsername2")\r
- .build();\r
-\r
- private static final Map<String, String> expectedIncompleteGen2DmaapConfig = ImmutableMap.<String, String>builder()\r
- .put(VES_ALERT_SND_AUTH_USERNAME_KEY, NULL_TOSTRING)\r
- .put(FAULT_BASIC_AUTH_USERNAME_KEY, NULL_TOSTRING)\r
- .put(VES_ALERT_CAMBRIA_TOPIC_KEY, "DCAE-SE-COLLECTOR-EVENTS-DEV")\r
- .put(VES_ALERT_CAMBRIA_URL_KEY, "UEBHOST:3904")\r
- .put(VES_FAULT_SND_CAMBRIA_URL_KEY, "UEBHOST:3904")\r
- .put(ALERT_BASIC_AUTH_PWD_KEY, NULL_TOSTRING)\r
- .put(VES_FAULT_SND_AUTH_PWD_KEY, NULL_TOSTRING)\r
- .put(VES_FAULT_SND_CAMBRIA_TOPIC_KEY, "DCAE-SE-COLLECTOR-EVENTS-DEV")\r
- .build();\r
-\r
- private static final Map<String, String> expectedCompleteChannelsDmaapConfig = ImmutableMap.<String, String>builder()\r
- .put(FAULT_UEB_CAMBRIA_HOSTS_KEY, "uebsb91kcdc.it.att.com,uebsb92kcdc.it.att.com,uebsb93kcdc.it.att.com")\r
- .put(FAULT_UEB_CAMBRIA_TOPIC_KEY, "DCAE-SE-COLLECTOR-EVENTS-DEV")\r
- .put(FAULT_UEB_KEY_PREFIX + ".basicAuthPassword", "S0mEPassWD")\r
- .put(FAULT_UEB_KEY_PREFIX + ".basicAuthUsername", "sampleUser")\r
- .put(FAULT_UEB_KEY_PREFIX + ".cambria.url", "127.0.0.1:3904")\r
- .build();\r
-\r
- private static final Map<String, String> expectedIncompleteChannelsDmaapConfig = ImmutableMap.<String, String>builder()\r
- .put(FAULT_UEB_CAMBRIA_HOSTS_KEY, "uebsb91kcdc.it.att.com,uebsb92kcdc.it.att.com,uebsb93kcdc.it.att.com")\r
- .put(FAULT_UEB_CAMBRIA_TOPIC_KEY, "DCAE-SE-COLLECTOR-EVENTS-DEV")\r
- .build();\r
-\r
- @Test\r
- public void testShouldCreateReaderWithAbsentParamsOmittedBasedOnChannelDmaapConfig() {\r
- assertReaderPreservedAllEntriesAfterTransformation(legacyConfigFilePath, expectedIncompleteChannelsDmaapConfig);\r
- }\r
-\r
- @Test\r
- public void testShouldCreateReaderWithAbsentParamsOmittedBasedOnGen2DmaapConfig() {\r
- assertReaderPreservedAllEntriesAfterTransformation(dmaapInputConfigFilePath, expectedIncompleteGen2DmaapConfig);\r
- }\r
-\r
- @Test\r
- public void shouldCreateReaderWithCompleteChannelDmaapConfig() {\r
- assertReaderPreservedAllEntriesAfterTransformation(fullDmaapConfigWithChannels, expectedCompleteChannelsDmaapConfig);\r
- }\r
-\r
- @Test\r
- public void shouldCreateReaderWithCompleteGen2DmaapConfig() {\r
- assertReaderPreservedAllEntriesAfterTransformation(fullGen2DmaapConfig, expectedCompleteGen2DmaapConfig);\r
- }\r
-\r
- private void assertReaderPreservedAllEntriesAfterTransformation(String dmaapConfigFilePath, Map<String, String> expectedMap) {\r
- DmaapPropertyReader reader = new DmaapPropertyReader(dmaapConfigFilePath);\r
-\r
- assertThat(reader.getDmaapProperties()).containsAllEntriesOf(expectedMap);\r
- assertThat(expectedMap).containsAllEntriesOf(reader.getDmaapProperties());\r
- }\r
-\r
-}\r
-\r
+++ /dev/null
-{
- "ves-fault-secondary": {
- "aaf_username": null,
- "dmaap_info": {
- "topic_url": "http://UEBHOST:3904/events/DCAE-SE-COLLECTOR-EVENTS-DEV",
- "location": "mtl5",
- "client_id": null,
- "client_role": null
- },
- "type": "message_router",
- "aaf_password": null
- },
- "ves-thresholdCrossingAlert-secondary": {
- "aaf_username": null,
- "dmaap_info": {
- "topic_url": "http://UEBHOST:3904/events/DCAE-SE-COLLECTOR-EVENTS-DEV",
- "location": "mtl5",
- "client_id": null,
- "client_role": null
- },
- "type": "message_router",
- "aaf_password": null
- }
- }
\ No newline at end of file
+++ /dev/null
-{
- "channels": [
- {
- "name": "sec_fault_ueb",
- "cambria.url": "127.0.0.1:3904",
- "cambria.hosts": "uebsb91kcdc.it.att.com,uebsb92kcdc.it.att.com,uebsb93kcdc.it.att.com",
- "cambria.topic": "DCAE-SE-COLLECTOR-EVENTS-DEV",
- "basicAuthPassword": "S0mEPassWD",
- "basicAuthUsername": "sampleUser",
- "stripHpId": "true"
- }
- ]
-}
\ No newline at end of file
--- /dev/null
+{
+ "auth-credentials-null": {
+ "aaf_username": null,
+ "dmaap_info": {
+ "topic_url": "http://UEBHOST:3904/events/DCAE-SE-COLLECTOR-EVENTS-DEV",
+ },
+ "aaf_password": null
+ },
+ "auth-credentials-present": {
+ "aaf_username": "sampleUser",
+ "dmaap_info": {
+ "topic_url": "http://UEBHOST:3904/events/DCAE-SE-COLLECTOR-EVENTS-DEV",
+ },
+ "aaf_password": "samplePassword"
+ },
+ "auth-credentials-missing": {
+ "dmaap_info": {
+ "topic_url": "http://UEBHOST:3904/events/DCAE-SE-COLLECTOR-EVENTS-DEV",
+ }
+ }
+}
\ No newline at end of file
--- /dev/null
+{
+ "channels": [
+ {
+ "name": "auth-credentials-null",
+ "cambria.url": "127.0.0.1:3904",
+ "cambria.hosts": "uebsb91kcdc.it.att.com,uebsb92kcdc.it.att.com,uebsb93kcdc.it.att.com",
+ "cambria.topic": "DCAE-SE-COLLECTOR-EVENTS-DEV",
+ "basicAuthPassword": null,
+ "basicAuthUsername": null,
+ },
+ {
+ "name": "auth-credentials-present",
+ "cambria.url": "127.0.0.1:3904",
+ "cambria.hosts": "uebsb91kcdc.it.att.com,uebsb92kcdc.it.att.com,uebsb93kcdc.it.att.com",
+ "cambria.topic": "DCAE-SE-COLLECTOR-EVENTS-DEV",
+ "basicAuthPassword": "samplePassword",
+ "basicAuthUsername": "sampleUser",
+ },
+ {
+ "name": "auth-credentials-missing",
+ "cambria.url": "127.0.0.1:3904",
+ "cambria.hosts": "uebsb91kcdc.it.att.com,uebsb92kcdc.it.att.com,uebsb93kcdc.it.att.com",
+ "cambria.topic": "DCAE-SE-COLLECTOR-EVENTS-DEV",
+ }
+ ]
+}
\ No newline at end of file
--- /dev/null
+{
+ "event-segments-with-port": {
+ "dmaap_info": {
+ "topic_url": "http://UEBHOST:3904/events/DCAE-SE-COLLECTOR-EVENTS-DEV",
+ }
+ },
+ "other-segments-without-ports": {
+ "dmaap_info": {
+ "topic_url": "http://UEBHOST:3904/somethingHere/DCAE-SE-COLLECTOR-EVENTS-DEV",
+ }
+ }
+}
\ No newline at end of file
--- /dev/null
+{
+ "channels": [
+ {
+ "name": "url-precedes-hosts",
+ "cambria.url": "127.0.0.1:3904",
+ "cambria.hosts": "h1.att.com,h2.att.com",
+ "cambria.topic": "DCAE-SE-COLLECTOR-EVENTS-DEV",
+ },
+ {
+ "name": "url-key-missing",
+ "cambria.hosts": "h1.att.com,h2.att.com",
+ "cambria.topic": "DCAE-SE-COLLECTOR-EVENTS-DEV",
+ },
+ {
+ "name": "url-is-null",
+ "cambria.url": null,
+ "cambria.hosts": "h1.att.com,h2.att.com",
+ "cambria.topic": "DCAE-SE-COLLECTOR-EVENTS-DEV"
+ }
+ ]
+}
\ No newline at end of file
Code Review / dcaegen2 / collectors / ves.git / commitdiff