src/main/java/org/onap/dmaap/mr/tools/MessageCommand.java

   1 /*******************************************************************************
   2  *  ============LICENSE_START=======================================================
   3  *  org.onap.dmaap
   4  *  ================================================================================
   5  *  Copyright © 2017 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  *        http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  *  Unless required by applicable law or agreed to in writing, software
  13  *  distributed under the License is distributed on an "AS IS" BASIS,
  14  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  *  See the License for the specific language governing permissions and
  16  *  limitations under the License.
  17  *  ============LICENSE_END=========================================================
  18  *
  19  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  20  *
  21  *******************************************************************************/
  22 package org.onap.dmaap.mr.tools;
  23
  24 import java.io.IOException;
  25 import java.io.PrintStream;
  26 import java.util.List;
  27 import java.util.concurrent.TimeUnit;
  28
  29 import org.slf4j.Logger;
  30 import org.slf4j.LoggerFactory;
  31
  32 import com.att.nsa.cmdtool.Command;
  33 import com.att.nsa.cmdtool.CommandNotReadyException;
  34 import org.onap.dmaap.mr.client.MRBatchingPublisher;
  35 import org.onap.dmaap.mr.client.MRClientFactory;
  36 import org.onap.dmaap.mr.client.MRConsumer;
  37 import org.onap.dmaap.mr.client.MRClientBuilders.PublisherBuilder;
  38 import org.onap.dmaap.mr.client.MRPublisher.message;
  39
  40 public class MessageCommand implements Command<MRCommandContext>
  41 {
  42         final Logger logger = LoggerFactory.getLogger(ApiKeyCommand.class);
  43         @Override
  44         public String[] getMatches ()
  45         {
  46                 return new String[]{
  47                         "(post) (\\S*) (\\S*) (.*)",
  48                         "(read) (\\S*) (\\S*) (\\S*)",
  49                 };
  50         }
  51
  52         @Override
  53         public void checkReady ( MRCommandContext context ) throws CommandNotReadyException
  54         {
  55                 if ( !context.checkClusterReady () )
  56                 {
  57                         throw new CommandNotReadyException ( "Use 'cluster' to specify a cluster to use." );
  58                 }
  59         }
  60
  61         @Override
  62         public void execute ( String[] parts, MRCommandContext context, PrintStream out ) throws CommandNotReadyException
  63         {
  64                 if ( parts[0].equalsIgnoreCase ( "read" ))
  65                 {
  66                         final MRConsumer cc = MRClientFactory.createConsumer ( context.getCluster (), parts[1], parts[2], parts[3],
  67                                 -1, -1, null, context.getApiKey(), context.getApiPwd() );
  68                         context.applyTracer ( cc );
  69                         try
  70                         {
  71                                 for ( String msg : cc.fetch () )
  72                                 {
  73                                         out.println ( msg );
  74                                 }
  75                         }
  76                         catch ( Exception e )
  77                         {
  78                                 out.println ( "Problem fetching messages: " + e.getMessage() );
  79                             logger.error("Problem fetching messages: ", e);
  80                         }
  81                         finally
  82                         {
  83                                 cc.close ();
  84                         }
  85                 }
  86                 else
  87                 {
  88                         final MRBatchingPublisher pub=ToolsUtil.createBatchPublisher(context, parts[1]);
  89                         try
  90                         {
  91                                 pub.send ( parts[2], parts[3] );
  92                         }
  93                         catch ( IOException e )
  94                         {
  95                                 out.println ( "Problem sending message: " + e.getMessage() );
  96                             logger.error("Problem sending message: ", e);
  97                         }
  98                         finally
  99                         {
 100                                 List<message> left = null;
 101                                 try
 102                                 {
 103                                         left = pub.close ( 500, TimeUnit.MILLISECONDS );
 104                                 }
 105                                 catch ( IOException e )
 106                                 {
 107                                         out.println ( "Problem sending message: " + e.getMessage() );
 108                                     logger.error("Problem sending message: ", e);
 109                                 }
 110                                 catch ( InterruptedException e )
 111                                 {
 112                                         out.println ( "Problem sending message: " + e.getMessage() );
 113                                     logger.error("Problem sending message: ", e);
 114                                     Thread.currentThread().interrupt();
 115                                 }
 116                                 if ( left != null && left.isEmpty() )
 117                                 {
 118                                         out.println ( left.size() + " messages not sent." );
 119                                 }
 120                         }
 121                 }
 122         }
 123
 124         @Override
 125         public void displayHelp ( PrintStream out )
 126         {
 127                 out.println ( "post <topicName> <partition> <message>" );
 128                 out.println ( "read <topicName> <consumerGroup> <consumerId>" );
 129         }
 130
 131 }