Merge pull request #147 from nacos-group/develop

Release 0.3.7

nacossync-console/pom.xml

@@ -16,7 +16,7 @@
     <parent>
         <artifactId>nacossync-parent</artifactId>
         <groupId>com.alibaba.nacossync</groupId>
-        <version>0.3.6</version>
+        <version>0.3.7</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

nacossync-distribution/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>nacossync-parent</artifactId>
         <groupId>com.alibaba.nacossync</groupId>
-        <version>0.3.6</version>
+        <version>0.3.7</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <packaging>pom</packaging>

nacossync-test/pom.xml

@@ -17,7 +17,7 @@
     <parent>
         <artifactId>nacossync-parent</artifactId>
         <groupId>com.alibaba.nacossync</groupId>
-        <version>0.3.6</version>
+        <version>0.3.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
@@ -39,7 +39,7 @@
         <dependency>
             <groupId>com.alibaba.nacossync</groupId>
             <artifactId>nacossync-worker</artifactId>
-            <version>0.3.6</version>
+            <version>0.3.7</version>
         </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>

nacossync-worker/pom.xml

@@ -16,11 +16,11 @@
     <parent>
         <artifactId>nacossync-parent</artifactId>
         <groupId>com.alibaba.nacossync</groupId>
-        <version>0.3.6</version>
+        <version>0.3.7</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>nacossync-worker</artifactId>
-    <version>0.3.6</version>
+    <version>0.3.7</version>
     <properties>
         <zookeeper.version>3.4.9</zookeeper.version>
         <curator.version>4.1.0</curator.version>

nacossync-worker/src/main/java/com/alibaba/nacossync/api/TaskApi.java

@@ -48,6 +48,9 @@ public class TaskApi {
 
     @Autowired
     private TaskDeleteProcessor taskDeleteProcessor;
+
+    @Autowired
+    private TaskDeleteInBatchProcessor taskDeleteInBatchProcessor;
 
     @Autowired
     private TaskListQueryProcessor taskListQueryProcessor;
@@ -74,6 +77,16 @@ public BaseResult deleteTask(TaskDeleteRequest taskDeleteRequest) {
 
         return SkyWalkerTemplate.run(taskDeleteProcessor, taskDeleteRequest, new BaseResult());
     }
+
+    /**
+     * @author yongchao9 
+     * @param taskBatchDeleteRequest
+     * @return
+     */
+    @RequestMapping(path = "/v1/task/deleteInBatch", method = RequestMethod.DELETE)
+    public BaseResult batchDeleteTask(TaskDeleteInBatchRequest taskBatchDeleteRequest) {
+        return SkyWalkerTemplate.run(taskDeleteInBatchProcessor, taskBatchDeleteRequest, new BaseResult());
+    }
 
     @RequestMapping(path = "/v1/task/add", method = RequestMethod.POST)
     public BaseResult taskAdd(@RequestBody TaskAddRequest addTaskRequest) {

nacossync-worker/src/main/java/com/alibaba/nacossync/dao/TaskAccessService.java

@@ -52,6 +52,16 @@ public TaskDO findByTaskId(String taskId) {
     public void deleteTaskById(String taskId) {
         taskRepository.deleteByTaskId(taskId);
     }
+
+    /**
+     * batch delete tasks by taskIds
+     * @author yongchao9
+     * @param taskIds
+     */
+    public void deleteTaskInBatch(List<String> taskIds) {
+    	List<TaskDO> tds=taskRepository.findAllByTaskIdIn(taskIds);
+        taskRepository.deleteInBatch(tds);
+    }
 
     public Iterable<TaskDO> findAll() {
 

nacossync-worker/src/main/java/com/alibaba/nacossync/dao/repository/TaskRepository.java

@@ -37,7 +37,9 @@ public interface TaskRepository extends CrudRepository<TaskDO, Integer>, JpaRepo
 
     @Transactional
     int deleteByTaskId(String taskId);
-
+
+    List<TaskDO> findAllByTaskIdIn(List<String> taskIds);
+
     List<TaskDO> getAllByWorkerIp(String workerIp);
 
 }

nacossync-worker/src/main/java/com/alibaba/nacossync/extension/impl/EurekaSyncToNacosServiceImpl.java

@@ -28,14 +28,12 @@
 import com.alibaba.nacossync.monitor.MetricsManager;
 import com.alibaba.nacossync.pojo.model.TaskDO;
 import com.netflix.appinfo.InstanceInfo;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.util.CollectionUtils;
-
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.util.CollectionUtils;
 
 /**
  * eureka
@@ -89,27 +87,18 @@ public boolean sync(TaskDO taskDO) {
         try {
             EurekaNamingService eurekaNamingService = eurekaServerHolder.get(taskDO.getSourceClusterId(), null);
             NamingService destNamingService = nacosServerHolder.get(taskDO.getDestClusterId(), null);
-            List<InstanceInfo> instanceInfos = eurekaNamingService.getApplications(taskDO.getServiceName());
-            List<Instance> allInstances = destNamingService.getAllInstances(taskDO.getServiceName());
-
-            if (Objects.nonNull(instanceInfos)) {
-                for (InstanceInfo instanceInfo : instanceInfos) {
-                    if (needSync(instanceInfo.getMetadata())) {
-                        if (CollectionUtils.isEmpty(allInstances)
-                            || isExistInNacosInstance(allInstances, instanceInfo)) {
-                            destNamingService.registerInstance(taskDO.getServiceName(),
-                                buildSyncInstance(instanceInfo, taskDO));
-                        } else {
-                            log.info("Remove invalid service instance from Nacos, serviceName={}, Ip={}, port={}",
-                                instanceInfo.getAppName(), instanceInfo.getIPAddr(), instanceInfo.getPort());
-                            destNamingService.deregisterInstance(instanceInfo.getAppName(), instanceInfo.getIPAddr(),
-                                instanceInfo.getPort());
-                        }
-                    }
+            List<InstanceInfo> eurekaInstances = eurekaNamingService.getApplications(taskDO.getServiceName());
+            List<Instance> nacosInstances = destNamingService.getAllInstances(taskDO.getServiceName());
 
-                }
+            if (CollectionUtils.isEmpty(eurekaInstances)) {
+                // Clear all instance from Nacos
+                deleteAllInstance(taskDO, destNamingService, nacosInstances);
             } else {
-                deleteAllInstance(taskDO, destNamingService, allInstances);
+                if (!CollectionUtils.isEmpty(nacosInstances)) {
+                    // Remove invalid instance from Nacos
+                    removeInvalidInstance(taskDO, destNamingService, eurekaInstances, nacosInstances);
+                }
+                addValidInstance(taskDO, destNamingService, eurekaInstances);
             }
             specialSyncEventBus.subscribe(taskDO, this::sync);
         } catch (Exception e) {
@@ -120,9 +109,32 @@ public boolean sync(TaskDO taskDO) {
         return true;
     }
 
-    private boolean isExistInNacosInstance(List<Instance> allInstances, InstanceInfo instanceInfo) {
-        return allInstances.stream().anyMatch(instance -> instance.getIp().equals(instanceInfo.getIPAddr())
-            && instance.getPort() == instanceInfo.getPort());
+    private void addValidInstance(TaskDO taskDO, NamingService destNamingService, List<InstanceInfo> eurekaInstances)
+        throws NacosException {
+        for (InstanceInfo instance : eurekaInstances) {
+            if (needSync(instance.getMetadata())) {
+                log.info("Add service instance from Eureka, serviceName={}, Ip={}, port={}",
+                    instance.getAppName(), instance.getIPAddr(), instance.getPort());
+                destNamingService.registerInstance(taskDO.getServiceName(), buildSyncInstance(instance, taskDO));
+            }
+        }
+    }
+
+    private void removeInvalidInstance(TaskDO taskDO, NamingService destNamingService,
+        List<InstanceInfo> eurekaInstances, List<Instance> nacosInstances) throws NacosException {
+        for (Instance instance : nacosInstances) {
+            if (!isExistInEurekaInstance(eurekaInstances, instance) && needDelete(instance.getMetadata(), taskDO)) {
+                log.info("Remove invalid service instance from Nacos, serviceName={}, Ip={}, port={}",
+                    instance.getServiceName(), instance.getIp(), instance.getPort());
+                destNamingService.deregisterInstance(taskDO.getServiceName(), instance.getIp(), instance.getPort());
+            }
+        }
+    }
+
+    private boolean isExistInEurekaInstance(List<InstanceInfo> eurekaInstances, Instance nacosInstance) {
+
+        return eurekaInstances.stream().anyMatch(instance -> instance.getIPAddr().equals(nacosInstance.getIp())
+            && instance.getPort() == nacosInstance.getPort());
     }
 
     private void deleteAllInstance(TaskDO taskDO, NamingService destNamingService, List<Instance> allInstances)

nacossync-worker/src/main/java/com/alibaba/nacossync/pojo/request/TaskDeleteInBatchRequest.java

@@ -0,0 +1,31 @@
+/*
+ * 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 com.alibaba.nacossync.pojo.request;
+
+import java.util.List;
+
+import lombok.Data;
+
+/**
+ * @author yongchao9
+ * @version $Id: TaskBatchDeleteRequest.java, v 0.3.1 2019-06-27 PM14:03 NacosSync Exp $$
+ */
+
+@Data
+public class TaskDeleteInBatchRequest extends BaseRequest {
+    private List<String> taskIds;
+}

nacossync-worker/src/main/java/com/alibaba/nacossync/template/processor/TaskDeleteInBatchProcessor.java

@@ -0,0 +1,59 @@
+/*
+ * 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 com.alibaba.nacossync.template.processor;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import com.alibaba.nacossync.dao.TaskAccessService;
+import com.alibaba.nacossync.pojo.model.TaskDO;
+import com.alibaba.nacossync.pojo.request.TaskDeleteInBatchRequest;
+import com.alibaba.nacossync.pojo.result.BaseResult;
+import com.alibaba.nacossync.template.Processor;
+
+import lombok.extern.slf4j.Slf4j;
+
+
+/**
+ * @author yongchao9
+ * @version $Id: TaskBatchDeleteProcessor.java, v 0.3.1 2019-06-27 PM14:33 NacosSync Exp $$
+ */
+
+@Slf4j
+@Service
+public class TaskDeleteInBatchProcessor implements Processor<TaskDeleteInBatchRequest, BaseResult> {
+
+    @Autowired
+    private TaskAccessService taskAccessService;
+
+    @Override
+    public void process(TaskDeleteInBatchRequest taskBatchDeleteRequest, BaseResult baseResult,
+                        Object... others) {
+//    	
+//    	String[] taskIds= taskBatchDeleteRequest.getTaskIds();
+//    	List<TaskDO> taskDOs = new ArrayList<TaskDO>();
+//    	for (String taskId : taskIds) {
+//    		TaskDO taskDO = new TaskDO();
+//    		taskDO.setTaskId(taskId);
+//    		taskDOs.add(taskDO);
+//		}
+        taskAccessService.deleteTaskInBatch(taskBatchDeleteRequest.getTaskIds());
+    }
+}

nacossync-worker/src/main/resources/application.properties

@@ -3,6 +3,8 @@ server.servlet.context-path=/
 
 spring.jpa.properties.hibernate.dialect=org.hibernate.dialect.MySQL5Dialect
 spring.jpa.hibernate.ddl-auto=update
+spring.jpa.properties.hibernate.show_sql=false
+
 
 
 spring.datasource.url=jdbc:mysql://127.0.0.1:3306/nacos_sync?characterEncoding=utf8

pom.xml

@@ -18,7 +18,7 @@
 
     <groupId>com.alibaba.nacossync</groupId>
     <artifactId>nacossync-parent</artifactId>
-    <version>0.3.6</version>
+    <version>0.3.7</version>
     <modules>
         <module>nacossync-console</module>
         <module>nacossync-worker</module>
@@ -37,7 +37,7 @@
         <url>[email protected]:nacos-group/nacos-sync.git</url>
         <connection>scm:[email protected]:nacos-group/nacos-sync.git</connection>
         <developerConnection>scm:[email protected]:nacos-group/nacos-sync.git</developerConnection>
-        <tag>nacossync-0.3.6</tag>
+        <tag>nacossync-0.3.7</tag>
     </scm>
     <developers>
         <developer>
@@ -73,7 +73,7 @@
             <dependency>
                 <groupId>com.alibaba.nacossync</groupId>
                 <artifactId>nacossync-worker</artifactId>
-                <version>0.3.6</version>
+                <version>0.3.7</version>
             </dependency>
         </dependencies>
     </dependencyManagement>