Merge bd159cdcc8 into 2fb0fce0b1

src/main/java/org/apache/rocketmq/dashboard/config/AuthWebMVCConfigurerAdapter.java

@@ -24,6 +24,8 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.MethodParameter;
+import org.springframework.http.converter.HttpMessageConverter;
+import org.springframework.http.converter.StringHttpMessageConverter;
 import org.springframework.web.bind.support.WebDataBinderFactory;
 import org.springframework.web.context.request.NativeWebRequest;
 import org.springframework.web.method.support.HandlerMethodArgumentResolver;
@@ -62,7 +64,8 @@ public class AuthWebMVCConfigurerAdapter extends WebMvcConfigurerAdapter {
                 "/producer/**",
                 "/test/**",
                 "/topic/**",
-                "/acl/**");
+                "/acl/**",
+                "/connect/**");
         }
     }
 
@@ -94,4 +97,9 @@ public class AuthWebMVCConfigurerAdapter extends WebMvcConfigurerAdapter {
     public void addViewControllers(ViewControllerRegistry registry) {
         registry.addViewController("*.htm").setViewName("forward:/app.html");
     }
+
+    @Override
+    public void configureMessageConverters(List<HttpMessageConverter<?>> converters) {
+        converters.removeIf(e -> e.getClass().isAssignableFrom(StringHttpMessageConverter.class));
+    }
 }

src/main/java/org/apache/rocketmq/dashboard/controller/ConnectController.java

@@ -0,0 +1,102 @@
+/*
+ * 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.dashboard.controller;
+
+import org.apache.rocketmq.dashboard.permisssion.Permission;
+import org.apache.rocketmq.dashboard.service.ConnectService;
+import org.springframework.stereotype.Controller;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.ResponseBody;
+
+import javax.annotation.Resource;
+import java.util.Map;
+
+@Controller
+@RequestMapping("/connect")
+@Permission
+public class ConnectController {
+
+    @Resource
+    private ConnectService connectService;
+
+    @RequestMapping(value = "/WorkerConnectors.query", method = RequestMethod.GET)
+    @ResponseBody
+    public Object workerConnectorList() {
+        return connectService.queryWorkerConnectorList();
+    }
+
+    @RequestMapping(value = "/worker.query", method = RequestMethod.GET)
+    @ResponseBody
+    public Object workerList() {
+        return connectService.queryWorkerList();
+    }
+
+    @RequestMapping(value = "/workerTasks.query", method = RequestMethod.GET)
+    @ResponseBody
+    public Object workerTaskList() {
+        return connectService.queryWorkerTaskList();
+    }
+
+
+    @RequestMapping(value = "/createConnector.do", method = RequestMethod.POST)
+    @ResponseBody
+    public String createConnector(@RequestBody Map<String, String> workerConnector) {
+        return connectService.createConnector(workerConnector);
+    }
+
+    @RequestMapping(value = "/stopConnector.do", method = RequestMethod.GET)
+    @ResponseBody
+    public String stopConnector(@RequestParam(value = "name") String connectorName) {
+        return connectService.stopConnector(connectorName);
+    }
+
+    @RequestMapping(value = "/stopAllConnectors.do", method = RequestMethod.GET)
+    @ResponseBody
+    public String stopAllConnectors() {
+        return connectService.stopAllConnectors();
+    }
+
+    @RequestMapping(value = "/reloadConnector.do", method = RequestMethod.GET)
+    @ResponseBody
+    public String reloadConnector() {
+        return connectService.reloadAllConnectors();
+    }
+
+    @RequestMapping(value = "/connectorStatus.query", method = RequestMethod.GET)
+    @ResponseBody
+    public Object getConnectorStatus(@RequestParam(value = "name") String connectorName) {
+
+        String connectorStatus = connectService.getConnectorStatus(connectorName);
+        return connectorStatus;
+    }
+
+    @RequestMapping(value = "/allocatedConnectors.query", method = RequestMethod.GET)
+    @ResponseBody
+    public void allocatedConnectors(@RequestParam(value = "ipAddr") String ipAddr) {
+    }
+
+    @RequestMapping(value = "/allocatedTasks.query", method = RequestMethod.GET)
+    @ResponseBody
+    public void allocatedTasks(@RequestParam(value = "ipAddr") String ipAddr) {
+    }
+
+
+
+}

src/main/java/org/apache/rocketmq/dashboard/model/connect/WorkerConnector.java

@@ -0,0 +1,122 @@
+/*
+ * 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.dashboard.model.connect;
+
+import java.util.Map;
+
+public class WorkerConnector {
+
+    private String connectorName;
+
+    private String connectTopicname;
+
+    private String connectorClass;
+
+    private String namesrvAddr;
+
+    private String workerPort;
+
+    private Long updateTimestamp;
+
+    private Map<String, String> properties;
+
+
+    public WorkerConnector() {
+    }
+
+    public WorkerConnector(String connectorName, String connectTopicname, String connectorClass, String namesrvAddr, String workerPort, Long updateTimestamp, Map<String, String> properties) {
+        this.connectorName = connectorName;
+        this.connectTopicname = connectTopicname;
+        this.connectorClass = connectorClass;
+        this.namesrvAddr = namesrvAddr;
+        this.workerPort = workerPort;
+        this.updateTimestamp = updateTimestamp;
+        this.properties = properties;
+    }
+
+    public String getConnectorName() {
+        return connectorName;
+    }
+
+    public void setConnectorName(String connectorName) {
+        this.connectorName = connectorName;
+    }
+
+    public String getConnectTopicname() {
+        return connectTopicname;
+    }
+
+    public void setConnectTopicname(String connectTopicname) {
+        this.connectTopicname = connectTopicname;
+    }
+
+    public String getConnectorClass() {
+        return connectorClass;
+    }
+
+    public void setConnectorClass(String connectorClass) {
+        this.connectorClass = connectorClass;
+    }
+
+    public Long getUpdateTimestamp() {
+        return updateTimestamp;
+    }
+
+    public void setUpdateTimestamp(Long updateTimestamp) {
+        this.updateTimestamp = updateTimestamp;
+    }
+
+    public String getNamesrvAddr() {
+        return namesrvAddr;
+    }
+
+    public void setNamesrvAddr(String namesrvAddr) {
+        this.namesrvAddr = namesrvAddr;
+    }
+
+    public String getWorkerPort() {
+        return workerPort;
+    }
+
+    public void setWorkerPort(String workerPort) {
+        this.workerPort = workerPort;
+    }
+
+    public Map<String, String> getProperties() {
+        return properties;
+    }
+
+    public void setProperties(Map<String, String> properties) {
+        this.properties = properties;
+    }
+
+    public void set(String key, String value) {
+        properties.put(key, value);
+    }
+
+
+    @Override
+    public String toString() {
+        return "WorkerConnector{" +
+                "connectorName='" + connectorName + '\'' +
+                ", connectTopicname='" + connectTopicname + '\'' +
+                ", connectorClass='" + connectorClass + '\'' +
+                ", updateTimestamp=" + updateTimestamp +
+                ", properties=" + properties +
+                '}';
+    }
+}

src/main/java/org/apache/rocketmq/dashboard/model/connect/WorkerInfo.java

@@ -0,0 +1,90 @@
+/*
+ * 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.dashboard.model.connect;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+public class WorkerInfo {
+    private static final Logger log = LoggerFactory.getLogger(WorkerInfo.class);
+
+    private String ipAddr;
+
+    private String namesrvAddr;
+
+    private List<WorkerConnector> allocatedConnectors;
+
+    private List<WorkerTask> allocatedTasks;
+
+    public WorkerInfo(String ipAddr, String namesrvAddr, List<WorkerConnector> allocatedConnectors, List<WorkerTask> allocatedTasks) {
+        this.ipAddr = ipAddr;
+        this.namesrvAddr = namesrvAddr;
+        this.allocatedConnectors = allocatedConnectors;
+        this.allocatedTasks = allocatedTasks;
+    }
+
+    public WorkerInfo() {
+    }
+
+    public static Logger getLog() {
+        return log;
+    }
+
+    public String getIpAddr() {
+        return ipAddr;
+    }
+
+    public void setIpAddr(String ipAddr) {
+        this.ipAddr = ipAddr;
+    }
+
+    public String getNamesrvAddr() {
+        return namesrvAddr;
+    }
+
+    public void setNamesrvAddr(String namesrvAddr) {
+        this.namesrvAddr = namesrvAddr;
+    }
+
+    public List<WorkerConnector> getAllocatedConnectors() {
+        return allocatedConnectors;
+    }
+
+    public void setAllocatedConnectors(List<WorkerConnector> allocatedConnectors) {
+        this.allocatedConnectors = allocatedConnectors;
+    }
+
+    public List<WorkerTask> getAllocatedTasks() {
+        return allocatedTasks;
+    }
+
+    public void setAllocatedTasks(List<WorkerTask> allocatedTasks) {
+        this.allocatedTasks = allocatedTasks;
+    }
+
+    @Override
+    public String toString() {
+        return "WorkerInfo{" +
+                "ipAddr='" + ipAddr + '\'' +
+                ", namesrvAddr='" + namesrvAddr + '\'' +
+                ", workingConnectors=" + allocatedConnectors +
+                ", existingTasks=" + allocatedTasks +
+                '}';
+    }
+}

src/main/java/org/apache/rocketmq/dashboard/model/connect/WorkerTask.java

@@ -0,0 +1,106 @@
+/*
+ * 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.dashboard.model.connect;
+
+import java.util.Map;
+
+public class WorkerTask {
+
+    private String connectorName;
+
+    private String connectTopicname;
+
+    private String taskClass;
+
+    private String connectorClass;
+
+    private Long updateTimestamp;
+
+    private Map<String, String> properties;
+
+    public WorkerTask() {
+    }
+
+    public WorkerTask(String connectorName, String connectTopicName, String taskClass, String connectorClass, Long updateTimestamp, Map<String, String> properties) {
+        this.connectorName = connectorName;
+        this.connectTopicname = connectTopicName;
+        this.taskClass = taskClass;
+        this.connectorClass = connectorClass;
+        this.updateTimestamp = updateTimestamp;
+        this.properties = properties;
+    }
+
+    public String getConnectorName() {
+        return connectorName;
+    }
+
+    public void setConnectorName(String connectorName) {
+        this.connectorName = connectorName;
+    }
+
+    public String getConnectTopicname() {
+        return connectTopicname;
+    }
+
+    public void setConnectTopicname(String connectTopicname) {
+        this.connectTopicname = connectTopicname;
+    }
+
+    public String getTaskClass() {
+        return taskClass;
+    }
+
+    public void setTaskClass(String taskClass) {
+        this.taskClass = taskClass;
+    }
+
+    public String getConnectorClass() {
+        return connectorClass;
+    }
+
+    public void setConnectorClass(String connectorClass) {
+        this.connectorClass = connectorClass;
+    }
+
+    public Long getUpdateTimestamp() {
+        return updateTimestamp;
+    }
+
+    public void setUpdateTimestamp(Long updateTimestamp) {
+        this.updateTimestamp = updateTimestamp;
+    }
+
+    public Map<String, String> getProperties() {
+        return properties;
+    }
+
+    public void setProperties(Map<String, String> properties) {
+        this.properties = properties;
+    }
+
+    @Override
+    public String toString() {
+        return "WorkerTask{" +
+                "connectorName='" + connectorName + '\'' +
+                ", connectTopicName='" + connectTopicname + '\'' +
+                ", taskClass='" + taskClass + '\'' +
+                ", connectorClass='" + connectorClass + '\'' +
+                ", updateTimestamp=" + updateTimestamp +
+                ", properties=" + properties +
+                '}';
+    }
+}

src/main/java/org/apache/rocketmq/dashboard/service/ConnectService.java

@@ -0,0 +1,51 @@
+/*
+ * 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.dashboard.service;
+
+
+import org.apache.rocketmq.dashboard.model.connect.WorkerConnector;
+import org.apache.rocketmq.dashboard.model.connect.WorkerInfo;
+import org.apache.rocketmq.dashboard.model.connect.WorkerTask;
+
+import java.util.List;
+import java.util.Map;
+
+public interface ConnectService {
+
+    List<WorkerConnector> queryWorkerConnectorList();
+
+    List<WorkerInfo> queryWorkerList();
+
+    List<WorkerTask> queryWorkerTaskList();
+
+
+    String stopConnector(String connectorName);
+
+    String stopAllConnectors();
+
+    String reloadAllConnectors();
+
+    String createConnector(Map<String, String> workerConnector);
+
+
+    String getAllocatedConnectors(String ipAddr);
+
+    String getAllocatedTasks(String ipAddr);
+
+    String getConnectorStatus(String connectorName);
+
+}

src/main/java/org/apache/rocketmq/dashboard/service/impl/ConnectServiceImpl.java

@@ -0,0 +1,162 @@
+/*
+ * 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.dashboard.service.impl;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import org.apache.rocketmq.dashboard.model.connect.WorkerConnector;
+import org.apache.rocketmq.dashboard.model.connect.WorkerInfo;
+import org.apache.rocketmq.dashboard.model.connect.WorkerTask;
+import org.apache.rocketmq.dashboard.service.ConnectService;
+import org.apache.rocketmq.dashboard.util.HttpRequestUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+@Service
+public class ConnectServiceImpl implements ConnectService {
+    private static final Logger log = LoggerFactory.getLogger(ConnectServiceImpl.class);
+
+    @Value("${rocketmq.config.connectAPIAddr}")
+    private static String ipAddr;
+
+    @Override
+    public List<WorkerInfo> queryWorkerList() {
+        String status = HttpRequestUtil.requestString("/getClusterInfo");
+        JSONArray jsonArray = JSON.parseArray(status);
+        List<WorkerInfo> workerInfoList = new ArrayList<>();
+        for (Iterator<Object> iterator = jsonArray.iterator(); iterator.hasNext(); ) {
+            WorkerInfo workerInfo = new WorkerInfo();
+            String next = (String) iterator.next();
+            int seperator = next.indexOf("@", 0);
+            String worker = next.substring(0, seperator);
+            String namesrv = next.substring(seperator + 1, next.length());
+
+            workerInfo.setIpAddr(worker);
+            workerInfo.setNamesrvAddr(namesrv);
+
+            workerInfoList.add(workerInfo);
+        }
+        return workerInfoList;
+    }
+
+    @Override
+    public List<WorkerConnector> queryWorkerConnectorList() {
+        String status = HttpRequestUtil.requestString("/getConfigInfo");
+        JSONObject jsonObject = JSON.parseObject(status).getJSONObject("connectorConfigs");
+        List<WorkerConnector> workerConnectorList = new ArrayList<>();
+        for (Map.Entry<String, Object> entry : jsonObject.entrySet()) {
+            JSONObject properties = ((JSONObject) entry.getValue()).getJSONObject("properties");
+            properties.put("connector-name", entry.getKey());
+
+            WorkerConnector workerConnector = JSON.parseObject(properties.toJSONString(), WorkerConnector.class);
+
+            Map<String, String> finalProperties = (Map<String, String>) JSONObject.parse(properties.toJSONString());
+            workerConnector.setProperties(finalProperties);
+
+            workerConnectorList.add(workerConnector);
+        }
+        return workerConnectorList;
+    }
+
+    @Override
+    public List<WorkerTask> queryWorkerTaskList() {
+        String status = HttpRequestUtil.requestString("/getConfigInfo");
+        JSONObject jsonObject = JSON.parseObject(status).getJSONObject("taskConfigs");
+        List<WorkerTask> workerTaskList = new ArrayList<>();
+        for (Map.Entry<String, Object> entry : jsonObject.entrySet()) {
+            JSONObject properties = ((JSONArray) entry.getValue()).getJSONObject(0).getJSONObject("properties");
+            properties.put("connector-name", entry.getKey());
+
+            WorkerTask workerTask = JSON.parseObject(properties.toJSONString(), WorkerTask.class);
+
+            Map<String, String> finalProperties = (Map<String, String>) JSONObject.parse(properties.toJSONString());
+            workerTask.setProperties(finalProperties);
+
+            workerTaskList.add(workerTask);
+        }
+        return workerTaskList;
+    }
+
+
+    @Override
+    public String getConnectorStatus(String connectorName) {
+        return HttpRequestUtil.requestString("/connectors/" + connectorName + "/status");
+    }
+
+    @Override
+    public String createConnector(Map<String, String> workerConnector) {
+        String url = null;
+        JSONObject jsonObject = null;
+        String ipAddr = workerConnector.get("clusterAddr").trim();
+        String workerPort = workerConnector.get("workerPort").trim();
+        String connectorName = workerConnector.get("connectorName").trim();
+        url = "http://" + ipAddr + ":" + workerPort + "/connectors/" + connectorName;
+
+        String properties = workerConnector.get("Properties").trim();
+        if (properties != null) {
+            jsonObject = JSONObject.parseObject(properties);
+        }
+
+        try {
+            jsonObject.put("connect-topicname", workerConnector.get("connectTopicname").trim());
+            jsonObject.put("connector-class", workerConnector.get("connectorClass").trim());
+        } catch (Exception e) {
+            throw new RuntimeException("BAD PARAMS,         " + e);
+        }
+
+        return HttpRequestUtil.postRequest(url, jsonObject);
+    }
+
+    @Override
+    public String stopConnector(String connectorName) {
+        return HttpRequestUtil.requestString("/connectors/" + connectorName + "/stop");
+    }
+
+    @Override
+    public String stopAllConnectors() {
+        return HttpRequestUtil.requestString("/connectors/stopAll");
+    }
+
+    @Override
+    public String reloadAllConnectors() {
+        return HttpRequestUtil.requestString("/plugin/reload");
+    }
+
+    @Override
+    public String getAllocatedConnectors(String ipAddr) {
+        String status = HttpRequestUtil.requestString("/getAllocatedConnectors");
+
+        return null;
+    }
+
+    @Override
+    public String getAllocatedTasks(String ipAddr) {
+        String status = HttpRequestUtil.requestString("/getAllocatedTasks");
+
+        return null;
+    }
+
+
+}

src/main/java/org/apache/rocketmq/dashboard/util/HttpRequestUtil.java

@@ -0,0 +1,167 @@
+/*
+ * 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.dashboard.util;
+
+import com.alibaba.fastjson.JSONObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import java.io.BufferedReader;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.PrintWriter;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.net.URLConnection;
+
+
+@Component
+public class HttpRequestUtil {
+
+    private static final Logger log = LoggerFactory.getLogger(HttpRequestUtil.class);
+
+    private static String ipAddr;
+
+    @Value("${rocketmq.config.connectAPIAddr}")
+    public void setIpAddr(String ipAddr) {
+        HttpRequestUtil.ipAddr = ipAddr;
+    }
+
+    public static String requestString(String urlSuffix) {
+        String urlString = getUrl(urlSuffix);
+
+        return requestMsg(urlString);
+    }
+
+    public static <T> T requestJSON(String urlSuffix, Class<T> targetClass) {
+        String urlString = getUrl(urlSuffix);
+        String responseByte = requestMsg(urlString);
+
+        return JsonUtil.string2Obj(responseByte, targetClass);
+    }
+
+    public static String getUrl(String urlSuffix) {
+        if (ipAddr == null || ipAddr.length() == 0) {
+            throw new RuntimeException("Failed to get url! Please edit it in application.properties!");
+        }
+
+        return "http://" + ipAddr + urlSuffix;
+    }
+
+    public static String requestMsg(String urlString) {
+
+        HttpURLConnection conn = null;
+        InputStream is = null;
+
+        try {
+
+            URL mURL = new URL(urlString);
+            conn = (HttpURLConnection) mURL.openConnection();
+
+            conn.setRequestMethod("GET");
+            conn.setReadTimeout(5000);
+            conn.setConnectTimeout(10000);
+
+            int responseCode = conn.getResponseCode();
+            if (responseCode == 200) {
+
+                is = conn.getInputStream();
+                String state = getStringFromInputStream(is);
+                return state;
+            } else {
+                log.info("Failed to request! Code: " + responseCode);
+
+            }
+
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            if (conn != null) {
+                conn.disconnect();
+            }
+        }
+        return null;
+    }
+
+    private static String getStringFromInputStream(InputStream is)
+            throws IOException {
+        ByteArrayOutputStream os = new ByteArrayOutputStream();
+
+        byte[] buffer = new byte[1024];
+        int len = -1;
+
+        while ((len = is.read(buffer)) != -1) {
+            os.write(buffer, 0, len);
+        }
+        is.close();
+        String state = os.toString();
+        os.close();
+        return state;
+    }
+
+    public static String postRequest(String url, JSONObject requestParam) {
+
+        PrintWriter out = null;
+        BufferedReader in = null;
+        String result = "";
+        try {
+            URL realUrl = new URL(url);
+            URLConnection conn = realUrl.openConnection();
+
+            conn.setDoOutput(true);
+            conn.setDoInput(true);
+
+            conn.setReadTimeout(5000);
+            conn.setConnectTimeout(5000);
+
+            conn.setRequestProperty("Content-type", "application/json;charset=UTF-8");
+
+            out = new PrintWriter(conn.getOutputStream());
+            out.print(requestParam);
+            out.flush();
+
+            in = new BufferedReader(
+                    new InputStreamReader(conn.getInputStream()));
+
+            String line;
+            while ((line = in.readLine()) != null) {
+                line = new String(line.getBytes(), "utf-8");
+                result += line;
+            }
+        } catch (Exception e) {
+            throw new RuntimeException("Error connecting to: " + url + "," + e);
+        } finally {
+            try {
+                if (out != null) {
+                    out.close();
+                }
+                if (in != null) {
+                    in.close();
+                }
+            } catch (IOException ex) {
+                throw new RuntimeException("Error closing connection");
+            }
+        }
+        return result;
+
+
+    }
+}

src/main/resources/application.yml

@@ -60,7 +60,8 @@ rocketmq:
     # set the accessKey and secretKey if you used acl
     accessKey: # if version > 4.4.0
     secretKey: # if version > 4.4.0
-
+    # set if use rocketmq-connect
+    connectAPIAddr: 127.0.0.1:8082
 threadpool:
   config:
     coreSize: 10

src/main/resources/role-permission.yml

@@ -38,3 +38,5 @@ rolePerms:
     - /dlqMessage/exportDlqMessage.do
     - /dlqMessage/batchResendDlqMessage.do
     - /acl/*.query
+    - /connect/*.query
+    - /connect/*.do

src/main/resources/static/index.html

@@ -114,5 +114,6 @@
 <script type="text/javascript" src="vendor/preLoading/main.js"></script>
 <script type="text/javascript" src="src/login.js"></script>
 <script type="text/javascript" src="src/acl.js"></script>
+<script type="text/javascript" src="src/connect.js"></script>
 </body>
 </html>

src/main/resources/static/src/app.js

@@ -216,6 +216,9 @@ app.config(['$routeProvider', '$httpProvider','$cookiesProvider','getDictNamePro
         }).when('/acl', {
             templateUrl: 'view/pages/acl.html',
             controller: 'aclController'
+        }).when('/connect', {
+            templateUrl: 'view/pages/connect.html',
+            controller: 'connectController'
         }).when('/404', {
             templateUrl: 'view/pages/404.html'
         }).otherwise('/404');

src/main/resources/static/src/connect.js

@@ -0,0 +1,295 @@
+/*
+ * 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.
+ */
+
+var module = app;
+
+module.controller('connectController', ['$scope', 'ngDialog', '$http', 'Notification', function ($scope, ngDialog, $http, Notification) {
+    $scope.allTopicList = [];
+    $scope.selectedTopic = [];
+
+    $scope.connectorList = [];
+    $scope.workerTaskList = [];
+    $scope.allWorkerList = [];
+
+    $scope.respConnectors = [];
+    $scope.respTasks = [];
+    $scope.respWorkers = [];
+
+    $scope.connectorPaginationConf  = {
+        currentPage: 1,
+        totalItems: 0,
+        itemsPerPage: 10,
+        pagesLength: 15,
+        perPageOptions: [10],
+        rememberPerPage: 'perPageItems',
+        onChange: function () {
+            $scope.queryWorkerConnectorList(this.currentPage, this.totalItems);
+        }
+    };
+
+    $scope.taskPaginationConf = {
+        currentPage: 1,
+        totalItems: 0,
+        itemsPerPage: 10,
+        pagesLength: 15,
+        perPageOptions: [10],
+        rememberPerPage: 'perPageItems',
+        onChange: function () {
+            $scope.queryWorkerTaskList(this.currentPage, this.totalItems);
+        }
+    };
+
+    $scope.workerPaginationConf = {
+        currentPage: 1,
+        totalItems: 0,
+        itemsPerPage: 10,
+        pagesLength: 15,
+        perPageOptions: [10],
+        rememberPerPage: 'perPageItems',
+        onChange: function () {
+            $scope.queryWorkerList(this.currentPage, this.totalItems);
+        }
+    };
+
+    $scope.queryWorkerConnectorList = function (currentPage, totalItem) {
+        var perPage = $scope.connectorPaginationConf.itemsPerPage;
+        var from = (currentPage - 1) * perPage;
+        var to = (from + perPage) > totalItem ? totalItem : from + perPage;
+        $scope.connectorList = $scope.respConnectors.slice(from, to);
+        $scope.connectorPaginationConf.totalItems = totalItem;
+
+    };
+
+    $scope.queryWorkerTaskList = function (currentPage, totalItem) {
+        var perPage = $scope.taskPaginationConf.itemsPerPage;
+        var from = (currentPage - 1) * perPage;
+        var to = (from + perPage) > totalItem ? totalItem : from + perPage;
+        $scope.workerTaskList = $scope.respTasks.slice(from, to);
+        $scope.taskPaginationConf.totalItems = totalItem;
+
+    };
+
+    $scope.queryWorkerList = function (currentPage, totalItem) {
+        var perPage = $scope.workerPaginationConf.itemsPerPage;
+        var from = (currentPage - 1) * perPage;
+        var to = (from + perPage) > totalItem ? totalItem : from + perPage;
+        $scope.allWorkerList = $scope.respWorkers.slice(from, to);
+        $scope.workerPaginationConf.totalItems = totalItem;
+
+    };
+
+    function queryTopicName() {
+        $http({
+            method: "GET",
+            url: "topic/list.query",
+        }).success(function (resp) {
+            if (resp.status == 0) {
+                $scope.allTopicList = resp.data.topicList.sort();
+                console.log($scope.allTopicList);
+            } else {
+                Notification.error({message: resp.errMsg, delay: 2000});
+            }
+        });
+    }
+
+
+
+    $scope.showWorkerTaskList = function () {
+        $http({
+            method: "GET",
+            url: "connect/workerTasks.query",
+        }).success(function (resp) {
+            if (resp.status === 0) {
+                console.log(resp);
+
+                $scope.respTasks = resp.data;
+                $scope.queryWorkerTaskList($scope.taskPaginationConf.currentPage, $scope.respTasks.length);
+
+            } else {
+                Notification.error({message: resp.errMsg, delay: 2000});
+            }
+        });
+    };
+
+    $scope.showWorkerConnectorList = function () {
+        $http({
+            method: "GET",
+            url: "connect/WorkerConnectors.query",
+        }).success(function (resp) {
+            if (resp.status === 0) {
+                console.log(resp);
+                $scope.respConnectors = resp.data;
+
+                $scope.queryWorkerConnectorList($scope.connectorPaginationConf.currentPage, $scope.respConnectors.length);
+            } else {
+                Notification.error({message: resp.errMsg, delay: 2000});
+            }
+        });
+    };
+
+    $scope.showWorkerList = function () {
+        $http({
+            method: "GET",
+            url: "connect/worker.query",
+        }).success(function (resp) {
+            if (resp.status === 0) {
+                console.log(resp);
+
+                $scope.respWorkers = resp.data;
+                $scope.queryWorkerList($scope.workerPaginationConf.currentPage, $scope.respWorkers.length);
+
+            } else {
+                Notification.error({message: resp.errMsg, delay: 2000});
+            }
+        });
+    };
+
+
+    $scope.queryConnectorStatus = function (name) {
+        $http({
+            method: "GET",
+            url: "connect/connectorStatus.query",
+            params: {
+                name: name
+            }
+        }).success(function (resp) {
+            if (resp.data === "running") {
+                Notification.info(resp.data);
+            } else {
+                Notification.error("not running");
+            }
+            console.log("Connector: " + name + ", Status:" + resp.data);
+        })
+    };
+
+
+    $scope.stopThisConnector = function (name) {
+        $http({
+            method: "GET",
+            url: "connect/stopConnector.do",
+            params: {
+                name: name
+            }
+        }).success(function (resp) {
+            if (resp.status == 0) {
+                if (resp.data === "success") {
+                    Notification.info({message: "success!", delay: 2000});
+                } else {
+                    Notification.error(resp.data);
+                }
+            } else {
+                Notification.error({message: resp.errMsg, delay: 2000});
+            }
+            $scope.showWorkerConnectorList();
+        })
+    };
+
+    $scope.stopAllConnectors = function () {
+        $http({
+            method: "GET",
+            url: "connect/stopAllConnectors.do"
+        }).success(function (resp) {
+            if (resp.status == 0) {
+                if (resp.data === "success") {
+                    Notification.info({message: "success!", delay: 2000});
+                } else {
+                    Notification.error(resp.data);
+                }
+            } else {
+                Notification.error({message: resp.errMsg, delay: 2000});
+            }
+            $scope.queryWorkerConnectorList();
+        })
+    };
+
+    $scope.reloadAllConnectors = function () {
+        $http({
+            method: "GET",
+            url: "connect/reloadConnector.do"
+        }).success(function (resp) {
+            if (resp.status == 0) {
+                if (resp.data === "success") {
+                    Notification.info({message: "success!", delay: 2000});
+                } else {
+                    Notification.error(resp.data);
+                }
+            } else {
+                Notification.error({message: resp.errMsg, delay: 2000});
+            }
+        })
+
+    };
+
+
+    $scope.connectorDetail = function (item) {
+        ngDialog.open({
+            template: 'ConnectorViewDialog',
+            data: item
+        });
+    };
+
+    $scope.taskDetail = function (item) {
+        ngDialog.open({
+            template: 'TaskViewDialog',
+            data: item
+        });
+    };
+
+    $scope.workerDetail = function (item) {
+        ngDialog.open({
+            template: 'WorkerViewDialog',
+            data: item
+        });
+    };
+
+
+    $scope.openCreationDialog = function () {
+        $scope.showWorkerList();
+        queryTopicName();
+        ngDialog.open({
+            template: 'connectorCreationDialog',
+            scope: $scope
+
+        });
+    };
+
+
+    $scope.postConnectorRequest = function (connectRequestItem) {
+        console.log(connectRequestItem);
+        connectRequestItem.clusterAddr = connectRequestItem.clusterAddr.ipAddr;
+        var request = JSON.parse(JSON.stringify(connectRequestItem));
+        console.log(request);
+        $http({
+            method: "POST",
+            url: "connect/createConnector.do",
+            data: request
+        }).success(function (resp) {
+            if (resp.status == 0) {
+                if (resp.data === "success") {
+                    Notification.info({message: "success!", delay: 2000});
+                    $scope.queryWorkerConnectorList();
+                    ngDialog.close(this);
+                } else {
+                    Notification.error(resp.data);
+                }
+            } else {
+                Notification.error({message: resp.errMsg, delay: 2000});
+            }
+        });
+    }
+
+}]);

src/main/resources/static/src/i18n/en.js

@@ -32,16 +32,16 @@ var en = {
     "CLUSTER": "Cluster",
     "CLUSTER_DETAIL": "Cluster Detail",
     "TOPIC": "Topic",
-    "SUBSCRIPTION_GROUP":"SubscriptionGroup",
-    "PRODUCER_GROUP":"ProducerGroup",
-    "CONSUMER":"Consumer",
-    "PRODUCER":"Producer",
-    "MESSAGE":"Message",
-    "MESSAGE_DETAIL":"Message Detail",
-    "RESEND_MESSAGE":"Resend Message",
-    "VIEW_EXCEPTION":"View Exception",
-    "MESSAGETRACE":"MessageTrace",
-    "DLQ_MESSAGE":"DLQMessage",
+    "SUBSCRIPTION_GROUP": "SubscriptionGroup",
+    "PRODUCER_GROUP": "ProducerGroup",
+    "CONSUMER": "Consumer",
+    "PRODUCER": "Producer",
+    "MESSAGE": "Message",
+    "MESSAGE_DETAIL": "Message Detail",
+    "RESEND_MESSAGE": "Resend Message",
+    "VIEW_EXCEPTION": "View Exception",
+    "MESSAGETRACE": "MessageTrace",
+    "DLQ_MESSAGE": "DLQMessage",
     "COMMIT": "Commit",
     "OPERATION": "Operation",
     "ADD": "Add",
@@ -58,70 +58,78 @@ var en = {
     "NORMAL": "NORMAL",
     "RETRY": "RETRY",
     "DLQ": "DLQ",
-    "QUANTITY":"Quantity",
-    "TYPE":"Type",
-    "MODE":"Mode",
-    "DELAY":"Delay",
-    "DASHBOARD":"Dashboard",
-    "CONSUME_DETAIL":"CONSUME DETAIL",
-    "CLIENT":"CLIENT",
-    "LAST_CONSUME_TIME":"LastConsumeTime",
-    "TIME":"Time",
-    "RESET":"RESET",
-    "DATE":"Date",
-    "NO_DATA":"NO DATA",
-    "SEARCH":"Search",
-    "BEGIN":"Begin",
-    "END":"End",
-    "TOPIC_CHANGE":"Topic Change",
-    "SEND":"Send",
-    "SUBSCRIPTION_CHANGE":"Subscription Change",
-	"QUEUE":"Queue",
-    "MIN_OFFSET":"minOffset",
-    "MAX_OFFSET":"maxOffset",
-    "LAST_UPDATE_TIME_STAMP":"lastUpdateTimeStamp",
-    "QUEUE_DATAS":"queueDatas",
-    "READ_QUEUE_NUMS":"readQueueNums",
-    "WRITE_QUEUE_NUMS":"writeQueueNums",
-    "PERM":"perm",
-    "TAG":"Tag",
-    "KEY":"Key",
-    "MESSAGE_BODY":"Message Body",
-    "TOPIC_NAME":"topicName",
-    "ORDER":"order",
-    "CONSUMER_CLIENT":"consumerClient",
-    "BROKER_OFFSET":"brokerOffset",
-    "CONSUMER_OFFSET":"consumerOffset",
-    "DIFF_TOTAL":"diffTotal",
-    "LAST_TIME_STAMP":"lastTimeStamp",
-    "RESET_OFFSET":"resetOffset",
-    "CLUSTER_NAME":"clusterName",
-    "OPS":"OPS",
-    "AUTO_REFRESH":"AUTO_REFRESH",
-    "REFRESH":"REFRESH",
-    "LOGOUT":"Logout",
-    "LOGIN":"Login",
-    "USER_NAME":"Username",
-    "PASSWORD":"Password",
-    "NO_DATA":"Don't have ",
-    "SYSTEM":"SYSTEM",
-    "WELCOME":"Hi, welcome using RocketMQ Dashboard",
-    "ENABLE_MESSAGE_TRACE":"Enable Message Trace",
-    "MESSAGE_TRACE_DETAIL":"Message Trace Detail",
-    "TRACE_TOPIC":"TraceTopic",
-    "SELECT_TRACE_TOPIC":"selectTraceTopic",
+    "QUANTITY": "Quantity",
+    "TYPE": "Type",
+    "MODE": "Mode",
+    "DELAY": "Delay",
+    "DASHBOARD": "Dashboard",
+    "CONSUME_DETAIL": "CONSUME DETAIL",
+    "CLIENT": "CLIENT",
+    "LAST_CONSUME_TIME": "LastConsumeTime",
+    "TIME": "Time",
+    "RESET": "RESET",
+    "DATE": "Date",
+    "NO_DATA": "NO DATA",
+    "SEARCH": "Search",
+    "BEGIN": "Begin",
+    "END": "End",
+    "TOPIC_CHANGE": "Topic Change",
+    "SEND": "Send",
+    "SUBSCRIPTION_CHANGE": "Subscription Change",
+    "QUEUE": "Queue",
+    "MIN_OFFSET": "minOffset",
+    "MAX_OFFSET": "maxOffset",
+    "LAST_UPDATE_TIME_STAMP": "lastUpdateTimeStamp",
+    "QUEUE_DATAS": "queueDatas",
+    "READ_QUEUE_NUMS": "readQueueNums",
+    "WRITE_QUEUE_NUMS": "writeQueueNums",
+    "PERM": "perm",
+    "TAG": "Tag",
+    "KEY": "Key",
+    "MESSAGE_BODY": "Message Body",
+    "TOPIC_NAME": "topicName",
+    "ORDER": "order",
+    "CONSUMER_CLIENT": "consumerClient",
+    "BROKER_OFFSET": "brokerOffset",
+    "CONSUMER_OFFSET": "consumerOffset",
+    "DIFF_TOTAL": "diffTotal",
+    "LAST_TIME_STAMP": "lastTimeStamp",
+    "RESET_OFFSET": "resetOffset",
+    "CLUSTER_NAME": "clusterName",
+    "OPS": "OPS",
+    "AUTO_REFRESH": "AUTO_REFRESH",
+    "REFRESH": "REFRESH",
+    "LOGOUT": "Logout",
+    "LOGIN": "Login",
+    "USER_NAME": "Username",
+    "PASSWORD": "Password",
+    "NO_DATA": "Don't have ",
+    "SYSTEM": "SYSTEM",
+    "WELCOME": "Hi, welcome using RocketMQ Dashboard",
+    "ENABLE_MESSAGE_TRACE": "Enable Message Trace",
+    "MESSAGE_TRACE_DETAIL": "Message Trace Detail",
+    "TRACE_TOPIC": "TraceTopic",
+    "SELECT_TRACE_TOPIC": "selectTraceTopic",
     "EXPORT": "export",
     "NO_MATCH_RESULT": "no match result",
     "BATCH_RESEND": "batchReSend",
     "BATCH_EXPORT": "batchExport",
-    "WHITE_LIST":"White List",
-    "ACCOUNT_INFO":"Account Info",
-    "IS_ADMIN":"Is Admin",
-    "DEFAULT_TOPIC_PERM":"Default Topic Permission",
-    "DEFAULT_GROUP_PERM":"Default Group Permission",
-    "TOPIC_PERM":"Topic Permission",
-    "GROUP_PERM":"Group Permission",
-    "SYNCHRONIZE":"Synchronize Data",
-    "SHOW":"Show",
-    "HIDE":"Hide"
+    "WHITE_LIST": "White List",
+    "ACCOUNT_INFO": "Account Info",
+    "IS_ADMIN": "Is Admin",
+    "DEFAULT_TOPIC_PERM": "Default Topic Permission",
+    "DEFAULT_GROUP_PERM": "Default Group Permission",
+    "TOPIC_PERM": "Topic Permission",
+    "GROUP_PERM": "Group Permission",
+    "SYNCHRONIZE": "Synchronize Data",
+    "SHOW": "Show",
+    "HIDE": "Hide",
+    "CONNECT": "Connect",
+    "CREATE_CONNECTOR": "Create Connector",
+    "RELOAD_ALL_CONNECTORS": "Reload All Connectors",
+    "STOP_ALL_CONNECTORS": "Stop All Connectors",
+    "CONNECTOR_DETAIL": "Connector Detail",
+    "STOP": "Stop",
+    "TASK_DETAIL": "Task Detail",
+    "DETAIL": "Detail"
 }

src/main/resources/static/src/i18n/zh.js

@@ -124,5 +124,13 @@ var zh = {
     "GROUP_PERM":"消费组权限",
     "SYNCHRONIZE":"同步",
     "SHOW":"显示",
-    "HIDE":"隐藏"
+    "HIDE":"隐藏",
+    "CONNECT":"连接",
+    "CREATE_CONNECTOR":"创建连接器",
+    "RELOAD_ALL_CONNECTORS": "重新载入连接器",
+    "STOP_ALL_CONNECTORS": "停止所有连接器",
+    "CONNECTOR_DETAIL": "连接器详情",
+    "STOP": "停止",
+    "TASK_DETAIL": "任务详情",
+    "DETAIL": "详情"
 }

src/main/resources/static/view/layout/_header.html

@@ -36,6 +36,7 @@
                 <li ng-class="path =='message' ? 'active':''"><a ng-href="#/message">{{'MESSAGE' | translate}}</a></li>
                 <li ng-class="path =='dlqMessage' ? 'active':''"><a ng-href="#/dlqMessage">{{'DLQ_MESSAGE' | translate}}</a></li>
                 <li ng-class="path =='messageTrace' ? 'active':''"><a ng-href="#/messageTrace">{{'MESSAGETRACE' | translate}}</a></li>
+                <li ng-class="path =='connect' ? 'active':''"><a ng-href="#/connect">{{'CONNECT' | translate}}</a></li>
                 <li ng-show="{{ show }}" ng-class="path =='acl' ? 'active':''"><a ng-href="#/acl">Acl</a></li>
             </ul>
             <ul class="nav navbar-nav navbar-right">

src/main/resources/static/view/pages/connect.html

@@ -0,0 +1,385 @@
+<!--
+  ~ 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.
+  -->
+
+<!--<div class="container-fluid" id="deployHistoryList">-->
+    <div class="modal-body">
+        <div ng-cloak="" class="tabsdemoDynamicHeight">
+            <md-content>
+                <md-tabs md-dynamic-height="" md-border-bottom="">
+                    <md-tab label="Connectors">
+                        <md-content class="md-padding" style="min-height:600px">
+                            <div class="row">
+                                <button id="reloadAppButton" type="button"
+                                        class="btn btn-raised btn-sm  btn-primary"
+                                        data-toggle="modal"
+                                        ng-click="showWorkerConnectorList()">
+                                    <span class="glyphicon glyphicon-search"></span>{{ 'REFRESH' | translate}}
+                                </button>
+                                <button id="createConnector" type="button"
+                                        class="btn btn-raised btn-sm  btn-primary"
+                                        data-toggle="modal"
+                                        ng-click="openCreationDialog()">
+                                    <span class="glyphicon glyphicon-search"></span>{{ 'CREATE_CONNECTOR' | translate}}
+                                </button>
+                                <button class="btn btn-raised btn-sm btn-danger ng-binding"
+                                        ng-confirm-click="Are you sure to reload?"
+                                        confirmed-click="reloadAllConnectors(item.connectorName)"> {{ 'RELOAD_ALL_CONNECTORS' | translate }}
+                                </button>
+                                <button class="btn btn-raised btn-sm btn-danger ng-binding"
+                                        ng-confirm-click="Are you sure to stop all?"
+                                        confirmed-click="stopAllConnectors(item.connectorName)"> {{ 'STOP_ALL_CONNECTORS' | translate }}
+                                </button>
+                                <table class="table table-bordered">
+                                    <tr>
+                                        <th class="text-center">Connector Name</th>
+                                        <th class="text-center">Connector Class</th>
+                                        <th class="text-center">Connect Topicname</th>
+                                        <th class="text-center">Update Time</th>
+                                        <th class="text-center">Operation</th>
+                                    </tr>
+                                    <tr ng-repeat="item in connectorList">
+                                        <td class="text-center">{{item.connectorName}}</td>
+                                        <td class="text-center">{{item.connectorClass}}</td>
+                                        <td class="text-center">{{item.connectTopicname}}</td>
+                                        <td class="text-center">{{item.updateTimestamp | date:'yyyy-MM-dd HH:mm:ss'}}
+                                        </td>
+                                        <td class="text-center">
+                                            <button class="btn btn-raised btn-sm btn-primary" type="button"
+                                                    ng-click="queryConnectorStatus(item.connectorName)">{{'STATUS' | translate}}
+                                            </button>
+                                            <button class="btn btn-raised btn-sm btn-primary" type="button"
+                                                    ng-click="connectorDetail(item)">{{'CONNECTOR_DETAIL' | translate}}
+                                            </button>
+                                            <button class="btn btn-raised btn-sm btn-danger"
+                                                    ng-confirm-click="Are you sure to stop?"
+                                                    confirmed-click="stopThisConnector(item.connectorName)"> {{ 'STOP' | translate }}
+                                            </button>
+                                        </td>
+                                    </tr>
+                                </table>
+                                <tm-pagination conf="connectorPaginationConf"></tm-pagination>
+                            </div>
+                        </md-content>
+                    </md-tab>
+
+                    <md-tab label="Tasks">
+                        <md-content class="md-padding" style="min-height:600px">
+                            <button id="reloadTaskButton" type="button"
+                                    class="btn btn-raised btn-sm  btn-primary"
+                                    data-toggle="modal"
+                                    ng-click="showWorkerTaskList()">
+                                <span class="glyphicon glyphicon-search"></span>{{ 'REFRESH' | translate}}
+                            </button>
+                            <table class="table table-bordered">
+                                <tr>
+                                    <th class="text-center">Task Class</th>
+                                    <th class="text-center">Connector Name</th>
+                                    <th class="text-center">Connector Class</th>
+                                    <th class="text-center">Connect Topicname</th>
+                                    <th class="text-center">Update Time</th>
+                                    <th class="text-center">Operation</th>
+                                </tr>
+                                <tr ng-repeat="item in workerTaskList">
+                                    <td class="text-center">{{item.taskClass}}</td>
+                                    <td class="text-center">{{item.connectorName}}</td>
+                                    <td class="text-center">{{item.connectorClass}}</td>
+                                    <td class="text-center">{{item.connectTopicname}}</td>
+                                    <td class="text-center">{{item.updateTimestamp | date:'yyyy-MM-dd HH:mm:ss'}}</td>
+                                    <td class="text-center">
+                                        <button class="btn btn-raised btn-sm btn-primary" type="button"
+                                                ng-click="taskDetail(item)">{{ 'DETAIL' | translate}}
+                                        </button>
+                                    </td>
+                                </tr>
+                            </table>
+                            <tm-pagination conf="taskPaginationConf"></tm-pagination>
+                        </md-content>
+                    </md-tab>
+
+                    <md-tab label="Clusters">
+                        <md-content class="md-padding" style="min-height:600px">
+                            <button id="reloadWorkerButton" type="button"
+                                    class="btn btn-raised btn-sm  btn-primary"
+                                    data-toggle="modal"
+                                    ng-click="showWorkerList()">
+                                <span class="glyphicon glyphicon-search"></span>{{ 'REFRESH' | translate}}
+                            </button>
+                            <table class="table table-bordered">
+                                <tr>
+                                    <th class="text-center">Cluster IP</th>
+                                    <th class="text-center">Namesrv Addr</th>
+                                    <th class="text-center">Operation</th>
+
+                                </tr>
+                                <tr ng-repeat="item in allWorkerList">
+                                    <td class="text-center">{{item.ipAddr}}</td>
+                                    <td class="text-center">{{item.namesrvAddr}}</td>
+                                    <td class="text-center">
+                                        <button class="btn btn-raised btn-sm btn-primary" type="button"
+                                                ng-click="workerDetail(item)">{{ 'DETAIL' | translate}}
+                                        </button>
+                                    </td>
+                                </tr>
+                            </table>
+                            <tm-pagination conf="workerPaginationConf"></tm-pagination>
+                        </md-content>
+                    </md-tab>
+                </md-tabs>
+            </md-content>
+        </div>
+
+    </div>
+<!--</div>-->
+
+
+<script type="text/ng-template" id="ConnectorViewDialog">
+    <md-content class="md-padding">
+        <div>
+            <form id="addAppForm" name="addAppForm" class="form-horizontal" novalidate>
+                <div class="form-group">
+                    <label class="control-label col-sm-2">Connector Name:</label>
+                    <div class="col-sm-10">
+                        <label class="form-control">{{ngDialogData.connectorName}}</label>
+                    </div>
+                </div>
+                <div class="form-group">
+                    <label class="control-label col-sm-2">Connector Class:</label>
+                    <div class="col-sm-10">
+                        <label class="form-control">{{ngDialogData.connectorClass}}</label>
+                    </div>
+                </div>
+                <div class="form-group">
+                    <label class="control-label col-sm-2">Connect Topicname:</label>
+                    <div class="col-sm-10">
+                        <label class="form-control">{{ngDialogData.connectTopicname}}</label>
+                    </div>
+                </div>
+                <div class="form-group">
+                    <label class="control-label col-sm-2">Update Time:</label>
+                    <div class="col-sm-10">
+                        <label class="form-control">{{ngDialogData.updateTimestamp | date:'yyyy-MM-dd HH:mm:ss'}}</label>
+                    </div>
+                </div>
+                <div class="form-group">
+                    <label class="control-label col-sm-2">Properties:</label>
+                    <div class="col-sm-10">
+                        <p ng-repeat="(x,y) in ngDialogData.properties">"{{x}}"  :  "{{y}}"</p>
+                    </div>
+                </div>
+            </form>
+        </div>
+    <div class="modal-footer">
+        <div class="ngdialog-buttons">
+            <button type="button" class="ngdialog-button ngdialog-button-secondary"
+                    ng-click="closeThisDialog('Cancel')">{{ 'CLOSE' | translate }}
+            </button>
+        </div>
+    </div>
+</script>
+
+<script type="text/ng-template" id="TaskViewDialog">
+    <md-content class="md-padding">
+        <div>
+            <form id="addAppForm1" name="addAppForm" class="form-horizontal" novalidate>
+                <div class="form-group">
+                    <label class="control-label col-sm-2">Connector Clas:</label>
+                    <div class="col-sm-10">
+                        <label class="form-control">{{ngDialogData.connectorClass}}</label>
+                    </div>
+                </div>
+                <div class="form-group">
+                    <label class="control-label col-sm-2">Connector Name:</label>
+                    <div class="col-sm-10">
+                        <label class="form-control">{{ngDialogData.connectorName}}</label>
+                    </div>
+                </div>
+                <div class="form-group">
+                    <label class="control-label col-sm-2">Connector Class:</label>
+                    <div class="col-sm-10">
+                        <label class="form-control">{{ngDialogData.connectorClass}}</label>
+                    </div>
+                </div>
+                <div class="form-group">
+                    <label class="control-label col-sm-2">Connect Topicname:</label>
+                    <div class="col-sm-10">
+                        <label class="form-control">{{ngDialogData.connectTopicname}}</label>
+                    </div>
+                </div>
+                <div class="form-group">
+                    <label class="control-label col-sm-2">Update Time:</label>
+                    <div class="col-sm-10">
+                        <label class="form-control">{{ngDialogData.updateTimestamp | date:'yyyy-MM-dd HH:mm:ss'}}</label>
+                    </div>
+                </div>
+                <div class="form-group">
+                    <label class="control-label col-sm-2">Properties:</label>
+                    <div class="col-sm-10">
+                        <p ng-repeat="(x,y) in ngDialogData.properties">"{{x}}"  :  "{{y}}"</p>
+                    </div>
+                </div>
+            </form>
+        </div>
+        <div class="modal-footer">
+            <div class="ngdialog-buttons">
+                <button type="button" class="ngdialog-button ngdialog-button-secondary"
+                        ng-click="closeThisDialog('Cancel')">{{ 'CLOSE' | translate }}
+                </button>
+            </div>
+        </div>
+</script>
+
+<script type="text/ng-template" id="WorkerViewDialog">
+    <md-content class="md-padding">
+        <div>
+            <form id="addAppForm2" name="addAppForm" class="form-horizontal" novalidate>
+                <div class="form-group">
+                    <label class="control-label col-sm-2">IP Address:</label>
+                    <div class="col-sm-10">
+                        <label class="form-control">{{ngDialogData.ipAddr}}</label>
+                    </div>
+                </div>
+                <div class="form-group">
+                    <label class="control-label col-sm-2">Namesrv Address:</label>
+                    <div class="col-sm-10">
+                        <label class="form-control">{{ngDialogData.namesrvAddr}}</label>
+                    </div>
+                </div>
+<!--                <div class="form-group">-->
+<!--                    <label class="control-label col-sm-2">Allocated Connectors:</label>-->
+<!--                    <div class="col-sm-10">-->
+<!--                        <label class="form-control">{{ngDialogData.connectors}}</label>-->
+<!--                    </div>-->
+<!--                </div>-->
+<!--                <div class="form-group">-->
+<!--                    <label class="control-label col-sm-2">Allocated Tasks:</label>-->
+<!--                    <div class="col-sm-10">-->
+<!--                        <label class="form-control">{{ngDialogData.tasks}}</label>-->
+<!--                    </div>-->
+<!--                </div>-->
+            </form>
+        </div>
+
+        <div class="modal-footer">
+            <div class="ngdialog-buttons">
+                <button type="button" class="ngdialog-button ngdialog-button-secondary"
+                        ng-click="closeThisDialog('Cancel')">{{ 'CLOSE' | translate }}
+                </button>
+            </div>
+        </div>
+</script>
+
+<script type="text/ng-template" id="operationResultDialog">
+    <div class="modal-header">
+        <h4 class="modal-title">Result</h4>
+    </div>
+    <div class="modal-body ">
+        <form class="form-horizontal" novalidate>
+            {{ngDialogData.result}}
+        </form>
+    </div>
+    <div class="modal-footer">
+        <div class="ngdialog-buttons">
+
+            <button type="button" class="ngdialog-button ngdialog-button-secondary"
+                    ng-click="closeThisDialog('Cancel')">{{ 'CLOSE' | translate }}
+            </button>
+        </div>
+    </div>
+</script>
+
+<script type="text/ng-template" id="connectorCreationDialog">
+    <div>
+        <div>
+            <div class="modal-header">
+                <h4 class="modal-title">{{'CREATE_CONNECTOR'|translate}}</h4>
+            </div>
+            <div class="modal-body ">
+                <form id="addAppForm3" name="addAppForm" class="form-horizontal" novalidate>
+                    <div class="form-group">
+                        <label class="control-label col-sm-4">connectorName:</label>
+                        <div class="col-sm-8">
+                            <input class="form-control" ng-model="item.connectorName"
+                                   type="text" name="connectorName"
+                                   required/>
+                            <span class="text-danger" ng-show="addAppForm.connectorName.$error.required">connectorName不能为空.</span>
+                        </div>
+                    </div>
+                    <div class="form-group">
+                        <label class="control-label col-sm-4">clusterAddr:</label>
+                        <div class="col-sm-8">
+                            <select name="mySelectClusterNameList" chosen
+                                    ng-model="item.clusterAddr"
+                                    ng-options="x.ipAddr for x in allWorkerList">
+                                <option value="" name="clusterAddr"></option>
+                            </select>
+                            <span class="text-danger" ng-show="addAppForm.clusterAddr.$error.required">clusterAddr不能为空.</span>
+                        </div>
+                    </div>
+                    <div class="form-group">
+                        <label class="control-label col-sm-4">Port:</label>
+                        <div class="col-sm-8">
+                            <input class="form-control" ng-model="item.workerPort"
+                                   ng-init="item.workerPort='8082'" value=""
+                                   type="text" name="workerPort"
+                                   required/>
+                            <span class="text-danger" ng-show="addAppForm.workerPort.$error.required">Port不能为空.</span>
+                        </div>
+                    </div>
+                    <div class="form-group">
+                        <label class="control-label col-sm-4">connectorClass:</label>
+                        <div class="col-sm-8">
+                            <input class="form-control" ng-model="item.connectorClass"
+                                   type="text" name="connectorClass"
+                                   required/>
+                            <span class="text-danger" ng-show="addAppForm.connectorClass.$error.required">connectorClass不能为空.</span>
+                        </div>
+                    </div>
+                    <div class="form-group">
+                        <label class="control-label col-sm-4">connectTopicname:</label>
+                        <div class="col-sm-8">
+                            <select name="mySelectTopicNameList" chosen
+                                    ng-model="item.connectTopicname"
+                                    ng-options="y for y in allTopicList">
+                                <option value="" name="connectTopicname"></option>
+                            </select>
+                            <span class="text-danger" ng-show="addAppForm.connectTopicname.$error.required">connectTopicname不能为空.</span>
+                        </div>
+                    </div>
+                    <div class="form-group">
+                        <label class="control-label col-sm-4">Properties:</label>
+                        <div class="col-sm-8">
+                            <input class="form-control" ng-model="item.Properties"
+                                   type="text" ng-init="item.Properties='{}'" value=""
+                                   required/>
+                            <span class="text" >sample:  {"filename":"/usr/test/test-source-file.txt",......}</span>
+                        </div>
+                    </div>
+                </form>
+                <div class="modal-footer">
+                    <div class="ngdialog-buttons">
+                        <button type="button" class="ngdialog-button ngdialog-button-primary"
+                                ng-click="postConnectorRequest(item)">{{ 'COMMIT' | translate }}
+                        </button>
+                        <button type="button" class="ngdialog-button ngdialog-button-secondary"
+                                ng-click="closeThisDialog('Cancel')">{{ 'CLOSE' | translate }}
+                        </button>
+                    </div>
+                </div>
+            </div>
+        </div>
+    </div>
+</script>