1 package org.openecomp.sdc.dmaap;
3 import org.apache.commons.lang3.StringUtils;
4 import org.apache.commons.lang3.math.NumberUtils;
5 import org.kohsuke.args4j.CmdLineException;
6 import org.kohsuke.args4j.CmdLineParser;
7 import org.kohsuke.args4j.OptionHandlerFilter;
8 import org.slf4j.Logger;
9 import org.slf4j.LoggerFactory;
11 import java.io.IOException;
12 import java.util.ArrayList;
13 import java.util.Collection;
14 import java.util.List;
15 import java.util.concurrent.ConcurrentLinkedDeque;
16 import java.util.concurrent.CopyOnWriteArrayList;
17 import java.util.function.Consumer;
18 import java.util.stream.IntStream;
20 import static org.openecomp.sdc.dmaap.Util.*;
22 public class DmaapPublisher {
23 private static final Logger logger = LoggerFactory.getLogger(DmaapPublisher.class);
24 private static RequestManager requestManager ;
25 private static final ConcurrentLinkedDeque notificationBuffer = new ConcurrentLinkedDeque();
28 private static final List<Long> registeredTasks = new CopyOnWriteArrayList<>();
29 private DmaapPublisher() {}
31 public static void add(String notification){
32 notificationBuffer.add( notification );
34 public static void addAll(List<String> notifications){
35 notificationBuffer.addAll( notifications );
37 public static void main(String[] args) {
41 private static void doPublish( String[] args ) {
42 CliArgs cliArgs = new CliArgs();
43 CmdLineParser parser = new CmdLineParser(cliArgs);
46 // parse the arguments.
47 parser.parseArgument( args );
50 catch(CmdLineException e) {
51 logger.error("#doPublish - failed to parse arguments.", e);
52 printUsage(parser, e);
57 public static void doPublish( CliArgs cliArgs ){
59 // parse the arguments.
60 DmaapPublishTool tool = new DmaapPublishTool( toPath(cliArgs.getYamlPath() , cliArgs.getYamlFilename()) , cliArgs.getNotificationData() );
61 Collection<String> notifications = new ArrayList<String>( notificationBuffer );
62 tool.addNotifications( notifications );
63 notificationBuffer.removeAll(notifications);
64 Integer concurrentRequestCount = 1;
65 if ( StringUtils.isNotBlank( cliArgs.getConcurrentRequests() ) )
66 concurrentRequestCount = Integer.parseInt( cliArgs.getConcurrentRequests() );
67 requestManager = new RequestManager( concurrentRequestCount );
69 IntStream.range(0,concurrentRequestCount).forEach( it -> {
70 //region - report upon finish mechanishem
71 long ticket = System.nanoTime();
72 registeredTasks.add( ticket );
73 Consumer callback = ( uniqueTicket ) -> {
74 synchronized ( registeredTasks ){
75 registeredTasks.remove( (long)uniqueTicket );
76 registeredTasks.notifyAll();
79 RunnableReporter task = new RunnableReporter( ticket , tool , cliArgs , callback );
80 requestManager.getExecutor().execute( task ) ;
83 catch(NumberFormatException e) {
84 logger.error("#doPublish - failed to parse argument CR.", e);
88 logger.error("#doPublish - failed to publish.", e);
92 public static class RunnableReporter implements Runnable{
94 final private long ticket ;
95 final private DmaapPublishTool tool;
96 final private CliArgs cliArgs;
97 final Consumer reporter;
99 public RunnableReporter(final long ticket , final DmaapPublishTool tool , final CliArgs args , Consumer reporter){
100 this.ticket = ticket ;
102 this.cliArgs = args ;
103 this.reporter = reporter;
108 tool.publish( cliArgs.getYamlPath() );
109 reporter.accept(ticket);
110 }catch(IOException e){
111 logger.error("#doPublish - failed to publish.", e);
112 }catch(InterruptedException e){
113 logger.error("#doPublish - cannot complete publish, thread interuppted.", e);
114 Thread.currentThread().interrupt();
120 public static List<Long> getRegisteredTasks() {
121 return registeredTasks;
124 public static void preparePublish( String path, String filename , String concurrentRequests ){
126 CliArgs cliArgs = new CliArgs();
127 if ( StringUtils.isNotBlank( filename ) )
128 cliArgs.setYamlFilename( filename );
129 if ( StringUtils.isNotBlank( path ) )
130 cliArgs.setYamlPath( path );
131 if ( NumberUtils.isCreatable( concurrentRequests ) )
132 cliArgs.setConcurrentRequests( concurrentRequests );
134 doPublish( cliArgs );
139 private static void printUsage(CmdLineParser parser, CmdLineException e) {
140 System.err.println( e.getMessage() );
141 System.err.println("java DmaapPublisher [options...] arguments...");
142 // print the list of available options
143 parser.printUsage(System.err);
144 System.err.println();
145 // print option sample. This is useful some time
146 System.err.println(" Example: java DmaapPublisher " + parser.printExample(OptionHandlerFilter.ALL));