vongosling closed pull request #217: [ROCKETMQ-353] Add SendMessage Command
URL: https://github.com/apache/rocketmq/pull/217
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java
b/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java
index d3342e818..7d7fce875 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java
+++ b/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 {
diff --git
a/tools/src/main/java/org/apache/rocketmq/tools/command/message/DecodeMessageIdCommond.java
b/tools/src/main/java/org/apache/rocketmq/tools/command/message/DecodeMessageIdCommand.java
similarity index 97%
rename from
tools/src/main/java/org/apache/rocketmq/tools/command/message/DecodeMessageIdCommond.java
rename to
tools/src/main/java/org/apache/rocketmq/tools/command/message/DecodeMessageIdCommand.java
index 532508c0b..c676151b0 100644
---
a/tools/src/main/java/org/apache/rocketmq/tools/command/message/DecodeMessageIdCommond.java
+++
b/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";
diff --git
a/tools/src/main/java/org/apache/rocketmq/tools/command/message/SendMessageCommand.java
b/tools/src/main/java/org/apache/rocketmq/tools/command/message/SendMessageCommand.java
new file mode 100644
index 000000000..23e2d6a1d
--- /dev/null
+++
b/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"
+ );
+ }
+ }
+}
diff --git
a/tools/src/test/java/org/apache/rocketmq/tools/command/broker/SendMessageCommandTest.java
b/tools/src/test/java/org/apache/rocketmq/tools/command/broker/SendMessageCommandTest.java
new file mode 100644
index 000000000..00e9b5e28
--- /dev/null
+++
b/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"));
+ }
+
+}
diff --git a/tools/src/test/resources/conf/logback_tools.xml
b/tools/src/test/resources/conf/logback_tools.xml
new file mode 100644
index 000000000..28283ad1d
--- /dev/null
+++ b/tools/src/test/resources/conf/logback_tools.xml
@@ -0,0 +1,93 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+ -->
+
+<configuration>
+ <appender name="DefaultAppender"
+ class="ch.qos.logback.core.rolling.RollingFileAppender">
+ <file>${user.home}/logs/rocketmqlogs/tools_default.log</file>
+ <append>true</append>
+ <rollingPolicy
class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
+
<fileNamePattern>${user.home}/logs/rocketmqlogs/otherdays/tools_default.%i.log.gz</fileNamePattern>
+ <minIndex>1</minIndex>
+ <maxIndex>5</maxIndex>
+ </rollingPolicy>
+ <triggeringPolicy
class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
+ <maxFileSize>100MB</maxFileSize>
+ </triggeringPolicy>
+ <encoder>
+ <pattern>%d{yyy-MM-dd HH:mm:ss,GMT+8} %p %t - %m%n</pattern>
+ <charset class="java.nio.charset.Charset">UTF-8</charset>
+ </encoder>
+ </appender>
+
+ <appender name="RocketmqToolsAppender_inner"
+ class="ch.qos.logback.core.rolling.RollingFileAppender">
+ <file>${user.home}/logs/rocketmqlogs/tools.log</file>
+ <append>true</append>
+ <rollingPolicy
class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
+
<fileNamePattern>${user.home}/logs/rocketmqlogs/otherdays/tools.%i.log.gz</fileNamePattern>
+ <minIndex>1</minIndex>
+ <maxIndex>5</maxIndex>
+ </rollingPolicy>
+ <triggeringPolicy
+ class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
+ <maxFileSize>100MB</maxFileSize>
+ </triggeringPolicy>
+ <encoder>
+ <pattern>%d{yyy-MM-dd HH:mm:ss,GMT+8} %p %t - %m%n</pattern>
+ <charset class="java.nio.charset.Charset">UTF-8</charset>
+ </encoder>
+ </appender>
+ <appender name="RocketmqToolsAppender"
class="ch.qos.logback.classic.AsyncAppender">
+ <appender-ref ref="RocketmqToolsAppender_inner"/>
+ </appender>
+
+
+ <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+ <append>true</append>
+ <encoder>
+ <pattern>%d{yyy-MM-dd HH\:mm\:ss,GMT+8} %p %t - %m%n</pattern>
+ <charset class="java.nio.charset.Charset">UTF-8</charset>
+ </encoder>
+ </appender>
+
+ <logger name="RocketmqTools" additivity="false">
+ <level value="INFO"/>
+ <appender-ref ref="RocketmqToolsAppender"/>
+ </logger>
+
+ <logger name="RocketmqCommon" additivity="false">
+ <level value="INFO"/>
+ <appender-ref ref="RocketmqBrokerAppender"/>
+ </logger>
+
+ <logger name="RocketmqStore" additivity="false">
+ <level value="INFO"/>
+ <appender-ref ref="RocketmqBrokerAppender"/>
+ </logger>
+
+ <logger name="RocketmqRemoting" additivity="false">
+ <level value="INFO"/>
+ <appender-ref ref="RocketmqBrokerAppender"/>
+ </logger>
+
+ <root>
+ <level value="INFO"/>
+ <appender-ref ref="DefaultAppender"/>
+ </root>
+</configuration>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
