[ROCKETMQ-353] Add SendMessage Command

tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java

@@ -54,6 +54,7 @@
 import org.apache.rocketmq.tools.command.message.QueryMsgByKeySubCommand;
 import org.apache.rocketmq.tools.command.message.QueryMsgByOffsetSubCommand;
 import org.apache.rocketmq.tools.command.message.QueryMsgByUniqueKeySubCommand;
+import org.apache.rocketmq.tools.command.message.SendMessageCommand;
 import org.apache.rocketmq.tools.command.namesrv.DeleteKvConfigCommand;
 import org.apache.rocketmq.tools.command.namesrv.GetNamesrvConfigCommand;
 import org.apache.rocketmq.tools.command.namesrv.UpdateKvConfigCommand;
@@ -193,6 +194,7 @@ public static void initCommand() {
         initCommand(new GetBrokerConfigCommand());
 
         initCommand(new QueryConsumeQueueCommand());
+        initCommand(new SendMessageCommand());
     }
 
     private static void initLogback() throws JoranException {

tools/src/main/java/org/apache/rocketmq/tools/command/message/DecodeMessageIdCommond.java → tools/src/main/java/org/apache/rocketmq/tools/command/message/DecodeMessageIdCommand.java

@@ -25,7 +25,7 @@
 import org.apache.rocketmq.tools.command.SubCommand;
 import org.apache.rocketmq.tools.command.SubCommandException;
 
-public class DecodeMessageIdCommond implements SubCommand {
+public class DecodeMessageIdCommand implements SubCommand {
     @Override
     public String commandName() {
         return "DecodeMessageId";

tools/src/main/java/org/apache/rocketmq/tools/command/message/SendMessageCommand.java

@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.tools.command.message;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.tools.command.SubCommand;
+import org.apache.rocketmq.tools.command.SubCommandException;
+
+public class SendMessageCommand implements SubCommand {
+
+    private DefaultMQProducer producer;
+
+    @Override
+    public String commandName() {
+        return "sendMessage";
+    }
+
+    @Override
+    public String commandDesc() {
+        return "Send Message to a topic";
+    }
+
+    @Override
+    public Options buildCommandlineOptions(Options options) {
+        Option opt = new Option("t", "topic", true, "topic name");
+        opt.setRequired(true);
+        options.addOption(opt);
+
+        opt = new Option("b", "body", true, "utf-8 string format of the message body");
+        opt.setRequired(true);
+        options.addOption(opt);
+
+        opt = new Option("k", "key", true, "message keys");
+        opt.setRequired(false);
+        options.addOption(opt);
+
+        opt = new Option("g", "tags", true, "message tags");
+        opt.setRequired(false);
+        options.addOption(opt);
+
+        opt = new Option("o", "qbroker", true, "send message to which broker");
+        opt.setRequired(false);
+        options.addOption(opt);
+
+        opt = new Option("i", "qid", true, "send message to which queue id");
+        opt.setRequired(false);
+        options.addOption(opt);
+
+        return options;
+    }
+
+    public DefaultMQProducer createProducer(RPCHook rpcHook) {
+        if (this.producer != null) {
+            return producer;
+        } else {
+            producer = new DefaultMQProducer(rpcHook);
+            producer.setProducerGroup(Long.toString(System.currentTimeMillis()));
+            return producer;
+        }
+    }
+
+    @Override
+    public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) throws SubCommandException {
+        Message msg = null;
+        String topic = commandLine.getOptionValue('t').trim();
+        String body = commandLine.getOptionValue('b').trim();
+        String tag = null;
+        String keys = null;
+        String brokerName = null;
+        int queueId = -1;
+        try {
+            if (commandLine.hasOption('k')) {
+                keys = commandLine.getOptionValue('k').trim();
+            }
+            if (commandLine.hasOption('g')) {
+                tag = commandLine.getOptionValue('g').trim();
+            }
+            if (commandLine.hasOption('q')) {
+                brokerName = commandLine.getOptionValue('q').trim();
+            }
+            if (commandLine.hasOption('i')) {
+                queueId = Integer.parseInt(commandLine.getOptionValue('i').trim());
+            }
+            msg = new Message(topic, tag, keys, body.getBytes("utf-8"));
+        } catch (Exception e) {
+            throw new RuntimeException(this.getClass().getSimpleName() + " command failed", e);
+        }
+
+        DefaultMQProducer producer = this.createProducer(rpcHook);
+        SendResult result = null;
+        try {
+            producer.start();
+            if (brokerName != null && queueId > -1) {
+                MessageQueue messageQueue = new MessageQueue(topic, brokerName, queueId);
+                result = producer.send(msg, messageQueue);
+            } else {
+                result = producer.send(msg);
+            }
+
+        } catch (Exception e) {
+            throw new RuntimeException(this.getClass().getSimpleName() + " command failed", e);
+        } finally {
+            producer.shutdown();
+        }
+
+        System.out.printf("%-32s  %-4s  %-20s    %s%n",
+            "#Broker Name",
+            "#QID",
+            "#Send Result",
+            "#MsgId"
+        );
+
+        if (result != null) {
+            System.out.printf("%-32s  %-4s  %-20s    %s%n",
+                result.getMessageQueue().getBrokerName(),
+                result.getMessageQueue().getQueueId(),
+                result.getSendStatus(),
+                result.getMsgId()
+            );
+        } else {
+            System.out.printf("%-32s  %-4s  %-20s    %s%n",
+                "Unknown",
+                "Unknown",
+                "Failed",
+                "None"
+            );
+        }
+    }
+}

tools/src/test/java/org/apache/rocketmq/tools/command/broker/SendMessageCommandTest.java

@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.tools.command.broker;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.PosixParser;
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.client.producer.SendStatus;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.exception.RemotingException;
+import org.apache.rocketmq.srvutil.ServerUtil;
+import org.apache.rocketmq.tools.command.MQAdminStartup;
+import org.apache.rocketmq.tools.command.SubCommandException;
+import org.apache.rocketmq.tools.command.message.SendMessageCommand;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.PrintStream;
+import java.lang.reflect.Field;
+import java.net.URL;
+
+import static org.mockito.Mockito.doAnswer;
+
+@RunWith(org.mockito.junit.MockitoJUnitRunner.class)
+public class SendMessageCommandTest {
+
+    private SendMessageCommand sendMessageCommand = new SendMessageCommand();
+    @Mock
+    private DefaultMQProducer defaultMQProducer;
+
+    private SendResult sendResult;
+
+    private SendResult sendResultByQueue;
+
+    @Before
+    public void init() throws MQClientException, SubCommandException, RemotingException, InterruptedException, MQBrokerException, NoSuchFieldException, IllegalAccessException {
+        sendResult = new SendResult();
+        sendResult.setMessageQueue(new MessageQueue());
+        sendResult.getMessageQueue().setBrokerName("broker1");
+        sendResult.getMessageQueue().setQueueId(1);
+        sendResult.setSendStatus(SendStatus.SEND_OK);
+        sendResult.setMsgId("fgwejigherughwueyutyu4t4343t43");
+
+        sendResultByQueue = sendResult;
+
+
+        doAnswer(new Answer() {
+            @Override
+            public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
+                return sendResult;
+            }
+        }).when(defaultMQProducer).send(Mockito.any(Message.class));
+
+        doAnswer(new Answer() {
+            @Override
+            public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
+                return sendResultByQueue;
+            }
+        }).when(defaultMQProducer).send(Mockito.any(Message.class),Mockito.any(MessageQueue.class));
+
+        Field producerField = SendMessageCommand.class.getDeclaredField("producer");
+        producerField.setAccessible(true);
+        producerField.set(sendMessageCommand,defaultMQProducer);
+    }
+
+    @After
+    public void terminate() {
+
+    }
+
+    @Test
+    public void testExecute() throws SubCommandException {
+        PrintStream out = System.out;
+        ByteArrayOutputStream bos = new ByteArrayOutputStream();
+        System.setOut(new PrintStream(bos));
+        Options options = ServerUtil.buildCommandlineOptions(new Options());
+        String[] subargs = new String[] {"-t mytopic","-b 'send message test'","-g tagA","-k order-16546745756"};
+        CommandLine commandLine =
+            ServerUtil.parseCmdLine("mqadmin " + sendMessageCommand.commandName(), subargs, sendMessageCommand.buildCommandlineOptions(options), new PosixParser());
+        sendMessageCommand.execute(commandLine, options, null);
+
+        subargs = new String[] {"-t mytopic","-b 'send message test'","-g tagA","-k order-16546745756","-o brokera","-i 1"};
+        commandLine = ServerUtil.parseCmdLine("mqadmin " + sendMessageCommand.commandName(), subargs, sendMessageCommand.buildCommandlineOptions(options), new PosixParser());
+        sendMessageCommand.execute(commandLine, options, null);
+        System.setOut(out);
+        String s = new String(bos.toByteArray());
+        Assert.assertTrue(s.contains("SEND_OK"));
+    }
+
+    @Test
+    public void testSendMessageCommandAdd() {
+        URL resource = SendMessageCommandTest.class.getClassLoader().getResource("conf/logback_tools.xml");
+        String file = resource.getFile();
+        File file1 = new File(file);
+        String path = file1.getParentFile().getParentFile().getPath();
+        System.setProperty(MixAll.ROCKETMQ_HOME_PROPERTY, path);
+        PrintStream out = System.out;
+        ByteArrayOutputStream bos = new ByteArrayOutputStream();
+        System.setOut(new PrintStream(bos));
+        MQAdminStartup.main(new String[]{"sendMessage"});
+        System.setOut(out);
+        String s = new String(bos.toByteArray());
+        Assert.assertTrue(s.contains("utf-8 string format of the message body"));
+    }
+
+}