Sync Integ to Master

Change-Id: I71e3acc26fa612127756ac04073a522b9cc6cd74
Issue-ID: SDC-977
Signed-off-by: Gitelman, Tal (tg851x) <tg851x@intl.att.com>

6 files changed, 395 insertions, 0 deletions

diff --git a/utils/DmaapPublisher/src/main/java/org/openecomp/sdc/dmaap/CliArgs.java b/utils/DmaapPublisher/src/main/java/org/openecomp/sdc/dmaap/CliArgs.java
new file mode 100644
index 0000000000..38e53c824f
--- /dev/null
+++ b/utils/DmaapPublisher/src/main/java/org/openecomp/sdc/dmaap/CliArgs.java
@@ -0,0 +1,59 @@
+package org.openecomp.sdc.dmaap;
+
+import org.kohsuke.args4j.Option;
+import com.google.common.base.MoreObjects;
+
+public class CliArgs {
+
+    @Option(name="yml",aliases = {"-YML","YML","-yml","-YAML","YAML","-yaml"}, usage="mandatory arg. YAML filename", required=true)
+    private String yamlFilename;
+
+    @Option(name="path",aliases = {"-path","PATH","-PATH"}, usage="mandatory arg. path to the yaml file which contains topic config (publisher data + messages)", required=true)
+    private String yamlPath;
+
+    @Option(name="cr",aliases = {"CR","-cr","-CR"}, usage="optional arg. concurrent requests", required=false)
+    private String concurrentRequests;
+
+    @Option(name="notification",aliases = {"NOTIFICATION","-NOTIFICATION","-notification"}, usage="optional load dynamic messages", required=false)
+    private String notificationData;
+
+    public String getYamlPath() {
+        return yamlPath;
+    }
+
+    public String getYamlFilename() {
+        return yamlFilename;
+    }
+
+    public void setYamlPath(String yamlPath) {
+        this.yamlPath = yamlPath;
+    }
+
+
+    public String getConcurrentRequests() {
+        return concurrentRequests;
+    }
+
+    public void setConcurrentRequests(String concurrentRequests) {
+        this.concurrentRequests = concurrentRequests;
+    }
+
+    public String getNotificationData() {
+        return notificationData;
+    }
+
+
+    public void setYamlFilename(String yamlFilename) {
+        this.yamlFilename = yamlFilename;
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(this)
+                .add("yamlPath", yamlPath)
+                .add("concurrentRequests", concurrentRequests)
+                .toString();
+    }
+    
+    
+}
diff --git a/utils/DmaapPublisher/src/main/java/org/openecomp/sdc/dmaap/DmaapPublishTool.java b/utils/DmaapPublisher/src/main/java/org/openecomp/sdc/dmaap/DmaapPublishTool.java
new file mode 100644
index 0000000000..61e48fa50e
--- /dev/null
+++ b/utils/DmaapPublisher/src/main/java/org/openecomp/sdc/dmaap/DmaapPublishTool.java
@@ -0,0 +1,97 @@
+package org.openecomp.sdc.dmaap;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.yaml.snakeyaml.Yaml;
+
+import com.att.nsa.mr.client.MRBatchingPublisher;
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.client.MRPublisher.message;
+
+public class DmaapPublishTool {
+
+    private static final Logger logger = LoggerFactory.getLogger(DmaapPublishTool.class);   
+    final private TopicConfig topicConfig;
+
+    public DmaapPublishTool(String yamlPath) throws FileNotFoundException {
+        topicConfig = loadTopicConfig(yamlPath);
+        System.out.println("yaml file loaded.");
+    }
+    public DmaapPublishTool(String yamlPath , String notifications) throws FileNotFoundException {
+        topicConfig = loadTopicConfig(yamlPath);
+        if (StringUtils.isNotBlank(notifications) )
+            topicConfig.add( notifications );
+        System.out.println("yaml file loaded.");
+    }
+
+    public void addNotifications(Collection<String> notification){
+        topicConfig.addAll( notification );
+    }
+
+    //safe stream doesn't throw null pointer exception
+    public <T> Collection<T> safe(Collection<T> obj){
+        return Optional.ofNullable(obj).orElse(Collections.emptySet());
+    }
+    public <T> List<T> safe(List<T> obj){
+        return Optional.ofNullable(obj).orElse(Collections.emptyList());
+    }
+
+    public void publish(String path) throws IOException, InterruptedException {
+        MRBatchingPublisher pub = createPublisher( topicConfig, path );
+        System.out.println( "pending message count -> "+pub.getPendingMessageCount() );
+        List<String> list = this.topicConfig.getIncomingTopicMessages();
+        for(String msg : safe(list) ){
+                publishOne( pub , msg );
+        }
+        closePublisher(pub);
+    }
+
+    private MRBatchingPublisher createPublisher(TopicConfig topicConfig,String path) throws IOException {
+        MRBatchingPublisher publisher = MRClientFactory.createBatchingPublisher(Objects.requireNonNull(Util.toPath(path,topicConfig.getPublisherPropertiesFilePath())));
+        System.out.println("publisher created.");
+        return publisher;
+    }
+
+    private TopicConfig loadTopicConfig(String yamlPath) throws FileNotFoundException {
+        File yamlFile = new File(Objects.requireNonNull(yamlPath));
+        InputStream input = new FileInputStream(yamlFile);
+        Yaml yamlHelper = new Yaml();
+        return yamlHelper.loadAs(input, TopicConfig.class);
+    }
+
+    private void publishOne(MRBatchingPublisher pub, String msg) throws IOException, InterruptedException {
+        System.out.println("sending:    " + msg);
+        pub.send(msg);
+        System.out.println("message sent.");
+    }
+
+    private void closePublisher(MRBatchingPublisher pub) throws IOException, InterruptedException {
+        System.out.println("closing publisher...");
+        // close the publisher to make sure everything's sent before exiting. The batching
+        // publisher interface allows the app to get the set of unsent messages. It could
+        // write them to disk, for example, to try to send them later.
+        final List<message> stuck = pub.close(20, TimeUnit.SECONDS);
+        if(!stuck.isEmpty())
+        {
+            final String errMsg = stuck.size() + " messages unsent";
+            logger.error(errMsg);
+            System.err.println(errMsg);
+        }
+        else
+        {
+            final String successMsg = "Clean exit; all messages sent.";
+            logger.info(successMsg);
+            System.out.println(successMsg);
+        }
+    }
+}
diff --git a/utils/DmaapPublisher/src/main/java/org/openecomp/sdc/dmaap/DmaapPublisher.java b/utils/DmaapPublisher/src/main/java/org/openecomp/sdc/dmaap/DmaapPublisher.java
new file mode 100644
index 0000000000..fd558356ed
--- /dev/null
+++ b/utils/DmaapPublisher/src/main/java/org/openecomp/sdc/dmaap/DmaapPublisher.java
@@ -0,0 +1,149 @@
+package org.openecomp.sdc.dmaap;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.math.NumberUtils;
+import org.kohsuke.args4j.CmdLineException;
+import org.kohsuke.args4j.CmdLineParser;
+import org.kohsuke.args4j.OptionHandlerFilter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.function.Consumer;
+import java.util.stream.IntStream;
+
+
+import static org.openecomp.sdc.dmaap.Util.*;
+
+public class DmaapPublisher {
+    private static final Logger logger = LoggerFactory.getLogger(DmaapPublisher.class);    
+    private static RequestManager requestManager ;
+    private static final ConcurrentLinkedDeque notificationBuffer = new ConcurrentLinkedDeque();
+
+
+    private static final List<Long> registeredTasks = new CopyOnWriteArrayList<>();
+    private DmaapPublisher() {}
+
+    public static void add(String notification){
+        notificationBuffer.add( notification );
+    }
+    public static void addAll(List<String> notifications){
+        notificationBuffer.addAll( notifications );
+    }
+    public static void main(String[] args) {
+        doPublish(args);
+    }
+
+    private static void doPublish( String[] args ) {
+        CliArgs cliArgs = new CliArgs();
+        CmdLineParser parser = new CmdLineParser(cliArgs);
+
+        try {
+            // parse the arguments.
+            parser.parseArgument( args );
+            doPublish( cliArgs );
+        }
+        catch(CmdLineException e) {
+            logger.error("#doPublish - failed to parse arguments.", e);
+            printUsage(parser, e);
+            return;
+        }
+    }
+
+    public static void doPublish( CliArgs cliArgs ){
+        try {
+            // parse the arguments.
+            DmaapPublishTool tool = new DmaapPublishTool( toPath(cliArgs.getYamlPath() , cliArgs.getYamlFilename()) , cliArgs.getNotificationData()  );
+            Collection<String> notifications = new ArrayList<String>( notificationBuffer );
+            tool.addNotifications( notifications );
+            notificationBuffer.removeAll(notifications);
+            Integer concurrentRequestCount = 1;
+            if ( StringUtils.isNotBlank( cliArgs.getConcurrentRequests() ) )
+                concurrentRequestCount = Integer.parseInt( cliArgs.getConcurrentRequests() );
+            requestManager = new RequestManager( concurrentRequestCount );
+
+            IntStream.range(0,concurrentRequestCount).forEach( it -> {
+                                        //region -  report upon finish mechanishem
+                                        long ticket = System.nanoTime();
+                                        registeredTasks.add( ticket );
+                                        Consumer callback = ( uniqueTicket ) -> {
+                                            synchronized ( registeredTasks ){
+                                                registeredTasks.remove( (long)uniqueTicket );
+                                                registeredTasks.notifyAll();
+                                            }};
+
+                                        RunnableReporter task = new RunnableReporter( ticket , tool , cliArgs , callback );
+                                        requestManager.getExecutor().execute( task ) ;
+            });
+        }
+        catch(NumberFormatException e) {
+            logger.error("#doPublish - failed to parse argument CR.", e);
+            return;
+        }
+        catch(Exception e) {
+            logger.error("#doPublish - failed to publish.", e);
+        }
+    }
+
+    public static class RunnableReporter implements Runnable{
+
+            final private long ticket ;
+            final private DmaapPublishTool tool;
+            final private CliArgs cliArgs;
+            final Consumer reporter;
+
+            public RunnableReporter(final long ticket , final DmaapPublishTool tool , final CliArgs args ,  Consumer reporter){
+                this.ticket = ticket ;
+                this.tool = tool ;
+                this.cliArgs = args ;
+                this.reporter = reporter;
+            }
+            @Override
+            public void run() {
+                try {
+                    tool.publish( cliArgs.getYamlPath() );
+                    reporter.accept(ticket);
+                }catch(IOException e){
+                    logger.error("#doPublish - failed to publish.", e);
+                }catch(InterruptedException e){
+                    logger.error("#doPublish - cannot complete publish, thread interuppted.", e);
+                    Thread.currentThread().interrupt();
+                }
+            }
+    }
+
+
+    public static List<Long> getRegisteredTasks() {
+        return registeredTasks;
+    }
+
+    public static void preparePublish( String path,  String filename , String concurrentRequests ){
+
+            CliArgs cliArgs = new CliArgs();
+            if ( StringUtils.isNotBlank( filename ) )
+                cliArgs.setYamlFilename( filename );
+            if ( StringUtils.isNotBlank( path ) )
+                cliArgs.setYamlPath( path );
+            if ( NumberUtils.isCreatable( concurrentRequests ) )
+                cliArgs.setConcurrentRequests(  concurrentRequests );
+
+            doPublish( cliArgs );
+
+    }
+
+
+    private static void printUsage(CmdLineParser parser, CmdLineException e) {
+        System.err.println( e.getMessage() );
+        System.err.println("java DmaapPublisher [options...] arguments...");
+        // print the list of available options
+        parser.printUsage(System.err);
+        System.err.println();
+        // print option sample. This is useful some time
+        System.err.println("  Example: java DmaapPublisher " + parser.printExample(OptionHandlerFilter.ALL));
+        
+    }
+}
diff --git a/utils/DmaapPublisher/src/main/java/org/openecomp/sdc/dmaap/RequestManager.java b/utils/DmaapPublisher/src/main/java/org/openecomp/sdc/dmaap/RequestManager.java
new file mode 100644
index 0000000000..597baac2bd
--- /dev/null
+++ b/utils/DmaapPublisher/src/main/java/org/openecomp/sdc/dmaap/RequestManager.java
@@ -0,0 +1,19 @@
+package org.openecomp.sdc.dmaap;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+
+public class RequestManager {
+
+    private Executor executor;
+
+    public RequestManager(int poolSize ){
+        int sz = Math.max( poolSize , 1);
+        int recommendedMaxSz = Runtime.getRuntime().availableProcessors() * 2;
+        executor = Executors.newFixedThreadPool( Math.min( sz , recommendedMaxSz ) );
+    }
+
+    public Executor getExecutor() {
+        return executor;
+    }
+}
diff --git a/utils/DmaapPublisher/src/main/java/org/openecomp/sdc/dmaap/TopicConfig.java b/utils/DmaapPublisher/src/main/java/org/openecomp/sdc/dmaap/TopicConfig.java
new file mode 100644
index 0000000000..a5b43ad0d5
--- /dev/null
+++ b/utils/DmaapPublisher/src/main/java/org/openecomp/sdc/dmaap/TopicConfig.java
@@ -0,0 +1,52 @@
+package org.openecomp.sdc.dmaap;
+
+import com.google.common.base.MoreObjects;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+public class TopicConfig {
+    
+    private String publisherPropertiesFilePath;
+    private String[] topicMessages; //messages from file
+    private final List<String> incomingTopicMessages = Collections.synchronizedList( new ArrayList<String>() );  //incoming messages from network stream|Main
+
+    public String getPublisherPropertiesFilePath() {
+        return publisherPropertiesFilePath;
+    }
+    public void setPublisherPropertiesFilePath(String publisherPropertiesFilePath) {
+        this.publisherPropertiesFilePath = publisherPropertiesFilePath;
+    }
+
+    public List<String> getIncomingTopicMessages() {
+        return incomingTopicMessages;
+    }
+    public String[] getTopicMessages() {
+        return topicMessages;
+    }
+    //add incoming message
+    public TopicConfig add( String notifications ){
+        incomingTopicMessages.add( notifications);
+        return this;
+    }
+
+    public TopicConfig addAll( Collection<String> notifications ){
+        incomingTopicMessages.addAll( notifications );
+        return this;
+    }
+
+    public void setTopicMessages(String[] topicMessages) {
+        this.topicMessages = topicMessages;
+    }
+    
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(this)
+                .add("publisherPropertiesFilePath", publisherPropertiesFilePath)
+                .add("topicMessages", topicMessages)
+                .toString();
+    }   
+    
+}
diff --git a/utils/DmaapPublisher/src/main/java/org/openecomp/sdc/dmaap/Util.java b/utils/DmaapPublisher/src/main/java/org/openecomp/sdc/dmaap/Util.java
new file mode 100644
index 0000000000..491b07abdd
--- /dev/null
+++ b/utils/DmaapPublisher/src/main/java/org/openecomp/sdc/dmaap/Util.java
@@ -0,0 +1,19 @@
+package org.openecomp.sdc.dmaap;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.nio.file.InvalidPathException;
+
+public class Util {
+
+    public static String toPath(String path , String filename) throws InvalidPathException{
+        if (StringUtils.isNotBlank(path) ){
+            if (path.trim().endsWith("/") || path.trim().endsWith("/")){
+                return path+(filename!=null ? filename : "");
+            }
+            return path+"/"+(filename!=null ? filename : "");
+
+        }
+        throw new InvalidPathException("wrong path configuration cannot find path -> ",path);
+    }
+}