utils/DmaapPublisher/src/main/java/org/openecomp/sdc/dmaap/DmaapPublisher.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * SDC
   4  * ================================================================================
   5  * Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.openecomp.sdc.dmaap;
  22
  23 import org.apache.commons.lang3.StringUtils;
  24 import org.apache.commons.lang3.math.NumberUtils;
  25 import org.kohsuke.args4j.CmdLineException;
  26 import org.kohsuke.args4j.CmdLineParser;
  27 import org.kohsuke.args4j.OptionHandlerFilter;
  28 import org.slf4j.Logger;
  29 import org.slf4j.LoggerFactory;
  30
  31 import java.io.IOException;
  32 import java.util.ArrayList;
  33 import java.util.Collection;
  34 import java.util.List;
  35 import java.util.concurrent.ConcurrentLinkedDeque;
  36 import java.util.concurrent.CopyOnWriteArrayList;
  37 import java.util.function.Consumer;
  38 import java.util.stream.IntStream;
  39
  40 import static org.openecomp.sdc.dmaap.Util.*;
  41
  42 public class DmaapPublisher {
  43     private static final Logger logger = LoggerFactory.getLogger(DmaapPublisher.class);
  44     private static RequestManager requestManager ;
  45     private static final ConcurrentLinkedDeque notificationBuffer = new ConcurrentLinkedDeque();
  46
  47
  48     private static final List<Long> registeredTasks = new CopyOnWriteArrayList<>();
  49     private DmaapPublisher() {}
  50
  51     public static void add(String notification){
  52         notificationBuffer.add( notification );
  53     }
  54     public static void addAll(List<String> notifications){
  55         notificationBuffer.addAll( notifications );
  56     }
  57     public static void main(String[] args) {
  58         doPublish(args);
  59     }
  60
  61     private static void doPublish( String[] args ) {
  62         CliArgs cliArgs = new CliArgs();
  63         CmdLineParser parser = new CmdLineParser(cliArgs);
  64
  65         try {
  66             // parse the arguments.
  67             parser.parseArgument( args );
  68             doPublish( cliArgs );
  69         }
  70         catch(CmdLineException e) {
  71             logger.error("#doPublish - failed to parse arguments.", e);
  72             printUsage(parser, e);
  73             return;
  74         }
  75     }
  76
  77     public static void doPublish( CliArgs cliArgs ){
  78         try {
  79             // parse the arguments.
  80             DmaapPublishTool tool = new DmaapPublishTool( toPath(cliArgs.getYamlPath() , cliArgs.getYamlFilename()) , cliArgs.getNotificationData()  );
  81             Collection<String> notifications = new ArrayList<String>( notificationBuffer );
  82             tool.addNotifications( notifications );
  83             notificationBuffer.removeAll(notifications);
  84             Integer concurrentRequestCount = 1;
  85             if ( StringUtils.isNotBlank( cliArgs.getConcurrentRequests() ) )
  86                 concurrentRequestCount = Integer.parseInt( cliArgs.getConcurrentRequests() );
  87             requestManager = new RequestManager( concurrentRequestCount );
  88
  89             IntStream.range(0,concurrentRequestCount).forEach( it -> {
  90                                         //region -  report upon finish mechanishem
  91                                         long ticket = System.nanoTime();
  92                                         registeredTasks.add( ticket );
  93                                         Consumer callback = ( uniqueTicket ) -> {
  94                                             synchronized ( registeredTasks ){
  95                                                 registeredTasks.remove( (long)uniqueTicket );
  96                                                 registeredTasks.notifyAll();
  97                                             }};
  98
  99                                         RunnableReporter task = new RunnableReporter( ticket , tool , cliArgs , callback );
 100                                         requestManager.getExecutor().execute( task ) ;
 101             });
 102         }
 103         catch(NumberFormatException e) {
 104             logger.error("#doPublish - failed to parse argument CR.", e);
 105             return;
 106         }
 107         catch(Exception e) {
 108             logger.error("#doPublish - failed to publish.", e);
 109         }
 110     }
 111
 112     public static class RunnableReporter implements Runnable{
 113
 114             final private long ticket ;
 115             final private DmaapPublishTool tool;
 116             final private CliArgs cliArgs;
 117             final Consumer reporter;
 118
 119             public RunnableReporter(final long ticket , final DmaapPublishTool tool , final CliArgs args ,  Consumer reporter){
 120                 this.ticket = ticket ;
 121                 this.tool = tool ;
 122                 this.cliArgs = args ;
 123                 this.reporter = reporter;
 124             }
 125             @Override
 126             public void run() {
 127                 try {
 128                     tool.publish( cliArgs.getYamlPath() );
 129                     reporter.accept(ticket);
 130                 }catch(IOException e){
 131                     logger.error("#doPublish - failed to publish.", e);
 132                 }catch(InterruptedException e){
 133                     logger.error("#doPublish - cannot complete publish, thread interuppted.", e);
 134                     Thread.currentThread().interrupt();
 135                 }
 136             }
 137     }
 138
 139
 140     public static List<Long> getRegisteredTasks() {
 141         return registeredTasks;
 142     }
 143
 144     public static void preparePublish( String path,  String filename , String concurrentRequests ){
 145
 146             CliArgs cliArgs = new CliArgs();
 147             if ( StringUtils.isNotBlank( filename ) )
 148                 cliArgs.setYamlFilename( filename );
 149             if ( StringUtils.isNotBlank( path ) )
 150                 cliArgs.setYamlPath( path );
 151             if ( NumberUtils.isCreatable( concurrentRequests ) )
 152                 cliArgs.setConcurrentRequests(  concurrentRequests );
 153
 154             doPublish( cliArgs );
 155
 156     }
 157
 158
 159     private static void printUsage(CmdLineParser parser, CmdLineException e) {
 160         System.err.println( e.getMessage() );
 161         System.err.println("java DmaapPublisher [options...] arguments...");
 162         // print the list of available options
 163         parser.printUsage(System.err);
 164         System.err.println();
 165         // print option sample. This is useful some time
 166         System.err.println("  Example: java DmaapPublisher " + parser.printExample(OptionHandlerFilter.ALL));
 167
 168     }
 169 }