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.MRPublisher.message;
  38
  39 public class MessageCommand implements Command<MRCommandContext>
  40 {
  41         final Logger logger = LoggerFactory.getLogger(ApiKeyCommand.class);
  42         @Override
  43         public String[] getMatches ()
  44         {
  45                 return new String[]{
  46                         "(post) (\\S*) (\\S*) (.*)",
  47                         "(read) (\\S*) (\\S*) (\\S*)",
  48                 };
  49         }
  50
  51         @Override
  52         public void checkReady ( MRCommandContext context ) throws CommandNotReadyException
  53         {
  54                 if ( !context.checkClusterReady () )
  55                 {
  56                         throw new CommandNotReadyException ( "Use 'cluster' to specify a cluster to use." );
  57                 }
  58         }
  59
  60         @Override
  61         public void execute ( String[] parts, MRCommandContext context, PrintStream out ) throws CommandNotReadyException
  62         {
  63                 if ( parts[0].equalsIgnoreCase ( "read" ))
  64                 {
  65                         final MRConsumer cc = MRClientFactory.createConsumer ( context.getCluster (), parts[1], parts[2], parts[3],
  66                                 -1, -1, null, context.getApiKey(), context.getApiPwd() );
  67                         context.applyTracer ( cc );
  68                         try
  69                         {
  70                                 for ( String msg : cc.fetch () )
  71                                 {
  72                                         out.println ( msg );
  73                                 }
  74                         }
  75                         catch ( Exception e )
  76                         {
  77                                 out.println ( "Problem fetching messages: " + e.getMessage() );
  78                             logger.error("Problem fetching messages: ", e);
  79                         }
  80                         finally
  81                         {
  82                                 cc.close ();
  83                         }
  84                 }
  85                 else
  86                 {
  87                         final MRBatchingPublisher pub=ToolsUtil.createBatchPublisher(context, parts[1]);
  88                         try
  89                         {
  90                                 pub.send ( parts[2], parts[3] );
  91                         }
  92                         catch ( IOException e )
  93                         {
  94                                 out.println ( "Problem sending message: " + e.getMessage() );
  95                             logger.error("Problem sending message: ", e);
  96                         }
  97                         finally
  98                         {
  99                                 List<message> left = null;
 100                                 try
 101                                 {
 102                                         left = pub.close ( 500, TimeUnit.MILLISECONDS );
 103                                 }
 104                                 catch ( IOException e )
 105                                 {
 106                                         out.println ( "Problem sending message: " + e.getMessage() );
 107                                     logger.error("Problem sending message: ", e);
 108                                 }
 109                                 catch ( InterruptedException e )
 110                                 {
 111                                         out.println ( "Problem sending message: " + e.getMessage() );
 112                                     logger.error("Problem sending message: ", e);
 113                                     Thread.currentThread().interrupt();
 114                                 }
 115                                 if ( left != null && left.isEmpty() )
 116                                 {
 117                                         out.println ( left.size() + " messages not sent." );
 118                                 }
 119                         }
 120                 }
 121         }
 122
 123         @Override
 124         public void displayHelp ( PrintStream out )
 125         {
 126                 out.println ( "post <topicName> <partition> <message>" );
 127                 out.println ( "read <topicName> <consumerGroup> <consumerId>" );
 128         }
 129
 130 }