/*
 * 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.console.controller;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import java.util.List;
import org.apache.rocketmq.console.support.annotation.JsonBody;
import org.apache.rocketmq.console.util.JsonUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;

@Controller
@RequestMapping("/test")
public class TestController {
    private Logger logger = LoggerFactory.getLogger(TestController.class);
    private String nameSvr = "120.55.113.35:9876";
    private String testTopic = "TestTopic";

    @RequestMapping(value = "/runTask.query", method = RequestMethod.GET)
    @JsonBody
    public Object list() throws MQClientException, RemotingException, InterruptedException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(testTopic + "Group");
        consumer.setNamesrvAddr(nameSvr);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe(testTopic, "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                ConsumeConcurrentlyContext context) {
                logger.info("receiveMessage msgSize={}", msgs.size());
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        DefaultMQProducer producer = new DefaultMQProducer(testTopic + "ProducerGroup");
        producer.setInstanceName(String.valueOf(System.currentTimeMillis()));
        producer.setNamesrvAddr(nameSvr);
        producer.start();
        int i = 0;
        while (true) {
            try {
                Message msg = new Message(testTopic,
                    "TagA" + i,
                    "KEYS" + i,
                    ("Hello RocketMQ " + i).getBytes()
                );
                Thread.sleep(1000L);
                SendResult sendResult = producer.send(msg);
                logger.info("sendMessage={}", JsonUtil.obj2String(sendResult));
            }
            catch (Exception e) {
                e.printStackTrace();
                Thread.sleep(1000);
            }
        }

    }
}

@@ -1600,16 +1600,80 @@
 l.List;%0A
+import org.apache.rocketmq.console.config.ConfigureInitializer;%0A
 import o
@@ -1835,16 +1835,79 @@
 actory;%0A
+import org.springframework.beans.factory.annotation.Autowired;%0A
 import o
@@ -2239,82 +2239,101 @@
 ing 
-nameSvr = %22120.55.113.35:9876%22;%0A    private String testTopic = %22TestTopic%22
+testTopic = %22TestTopic%22;%0A    @Autowired%0A    private ConfigureInitializer configureInitializer
 ;%0A%0A 
@@ -2373,13 +2373,10 @@
 ask.
-query
+do
 %22, m
@@ -2622,39 +2622,69 @@
 .setNamesrvAddr(
-nameSvr
+configureInitializer.getNameSrvAddr()
 );%0A        consu
@@ -3222,32 +3222,38 @@
 tart();%0A        
+final 
 DefaultMQProduce
@@ -3300,24 +3300,16 @@
 opic + %22
-Producer
 Group%22);
@@ -3423,44 +3423,163 @@
 ddr(
-nameSvr);%0A        producer.start();%0A
+configureInitializer.getNameSrvAddr());%0A        producer.start();%0A%0A        new Thread(new Runnable() %7B%0A%0A            @Override public void run() %7B%0A%0A        
     
@@ -3593,16 +3593,24 @@
  i = 0;%0A
+        
         
@@ -3640,14 +3640,30 @@
     
+   
+     
 try %7B%0A
+        
     
@@ -3731,16 +3731,24 @@
         
+        
 %22TagA%22 +
@@ -3771,16 +3771,24 @@
         
+       
+ 
 %22KEYS%22 +
@@ -3787,24 +3787,32 @@
 %22KEYS%22 + i,%0A
+        
             
@@ -3870,19 +3870,35 @@
         
+      
+  
 );%0A
+        
         
@@ -3942,16 +3942,24 @@
         
+    
+    
 SendResu
@@ -3994,16 +3994,24 @@
 d(msg);%0A
+        
         
@@ -4086,34 +4086,50 @@
 ));%0A            
-%7D%0A
+        %7D%0A        
             catc
@@ -4162,16 +4162,24 @@
         
+        
 e.printS
@@ -4187,24 +4187,66 @@
 ackTrace();%0A
+                        try %7B%0A            
             
@@ -4273,32 +4273,44 @@
 0);%0A            
+       
+     
 %7D%0A        %7D%0A%0A   
@@ -4303,18 +4303,184 @@
         
-%7D%0A
+                catch (Exception ignore) %7B%0A                        %7D%0A                    %7D%0A                %7D%0A            %7D%0A        %7D).start();%0A        return %22started%22;
 %0A    %7D%0A%7D

add testTask