complete lookup method of consul registry and add integration test (#…

…3906)

* add test case

* fix bug

* add something

* complete lookup method of consul registry and add test

* clean code

* correct dependency of embedded-consul

dubbo-dependencies-bom/pom.xml

@@ -103,6 +103,7 @@
         <curator_test_version>2.12.0</curator_test_version>
         <jedis_version>2.9.0</jedis_version>
         <consul_version>1.4.2</consul_version>
+        <consul_process_version>2.0.0</consul_process_version>
         <xmemcached_version>1.3.6</xmemcached_version>
         <cxf_version>3.1.15</cxf_version>
         <thrift_version>0.8.0</thrift_version>
@@ -233,6 +234,11 @@
                 <artifactId>consul-api</artifactId>
                 <version>${consul_version}</version>
             </dependency>
+            <dependency>
+                <groupId>com.pszymczyk.consul</groupId>
+                <artifactId>embedded-consul</artifactId>
+                <version>${consul_process_version}</version>
+            </dependency>
             <dependency>
                 <groupId>com.googlecode.xmemcached</groupId>
                 <artifactId>xmemcached</artifactId>

dubbo-registry/dubbo-registry-consul/pom.xml

@@ -36,6 +36,11 @@
             <groupId>com.ecwid.consul</groupId>
             <artifactId>consul-api</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.pszymczyk.consul</groupId>
+            <artifactId>embedded-consul</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>

dubbo-registry/dubbo-registry-consul/src/main/java/org/apache/dubbo/registry/consul/ConsulRegistry.java

@@ -32,10 +32,12 @@
 import com.ecwid.consul.v1.catalog.CatalogServicesRequest;
 import com.ecwid.consul.v1.health.HealthServicesRequest;
 import com.ecwid.consul.v1.health.model.HealthService;
+import org.apache.dubbo.rpc.RpcException;
 
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.ArrayList;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
@@ -155,6 +157,24 @@ public void doUnsubscribe(URL url, NotifyListener listener) {
         notifier.stop();
     }
 
+    @Override
+    public List<URL> lookup(URL url) {
+        if (url == null) {
+            throw new IllegalArgumentException("lookup url == null");
+        }
+        try {
+            String service = url.getServiceKey();
+            Response<List<HealthService>> result = client.getHealthServices(service, HealthServicesRequest.newBuilder().setTag(SERVICE_TAG).build());
+            if (result == null || result.getValue() == null || result.getValue().isEmpty()) {
+                return new ArrayList<>();
+            } else {
+                return  convert(result.getValue());
+            }
+        } catch (Throwable e) {
+            throw new RpcException("Failed to lookup " + url + " from consul " + getUrl() + ", cause: " + e.getMessage(), e);
+        }
+    }
+
     @Override
     public boolean isAvailable() {
         return client.getAgentSelf() != null;

dubbo-registry/dubbo-registry-consul/src/test/java/org/apache/dubbo/registry/consul/ConsulRegistryTest.java

@@ -0,0 +1,134 @@
+/*
+ * 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.dubbo.registry.consul;
+
+import com.pszymczyk.consul.ConsulProcess;
+import com.pszymczyk.consul.ConsulStarterBuilder;
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.status.Status;
+import org.apache.dubbo.registry.NotifyListener;
+import org.apache.dubbo.registry.Registry;
+import org.apache.dubbo.registry.status.RegistryStatusChecker;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.nullValue;
+import static org.hamcrest.CoreMatchers.not;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.Mockito.mock;
+
+public class ConsulRegistryTest {
+
+    private static ConsulProcess consul;
+    private ConsulRegistry consulRegistry;
+    private String service = "org.apache.dubbo.test.injvmServie";
+    private URL serviceUrl = URL.valueOf("consul://127.0.0.1:8012/" + service + "?notify=false&methods=test1,test2");
+    private URL registryUrl;
+    private ConsulRegistryFactory consulRegistryFactory;
+
+    @BeforeEach
+    public void setUp() throws Exception {
+        this.consul = ConsulStarterBuilder.consulStarter()
+                .build()
+                .start();
+        this.registryUrl = URL.valueOf("consul://localhost:" + consul.getHttpPort());
+
+        consulRegistryFactory = new ConsulRegistryFactory();
+        this.consulRegistry = (ConsulRegistry) consulRegistryFactory.createRegistry(registryUrl);
+    }
+
+    @AfterEach
+    public void tearDown() throws Exception {
+        consul.close();
+        this.consulRegistry.destroy();
+    }
+
+    @Test
+    public void testRegister() {
+        Set<URL> registered;
+
+        for (int i = 0; i < 2; i++) {
+            consulRegistry.register(serviceUrl);
+            registered = consulRegistry.getRegistered();
+            assertThat(registered.contains(serviceUrl), is(true));
+        }
+
+        registered = consulRegistry.getRegistered();
+
+        assertThat(registered.size(), is(1));
+    }
+
+    @Test
+    public void testSubscribe() {
+        NotifyListener listener = mock(NotifyListener.class);
+        consulRegistry.subscribe(serviceUrl, listener);
+
+        Map<URL, Set<NotifyListener>> subscribed = consulRegistry.getSubscribed();
+        assertThat(subscribed.size(), is(1));
+        assertThat(subscribed.get(serviceUrl).size(), is(1));
+
+        consulRegistry.unsubscribe(serviceUrl, listener);
+        subscribed = consulRegistry.getSubscribed();
+        assertThat(subscribed.size(), is(1));
+        assertThat(subscribed.get(serviceUrl).size(), is(0));
+    }
+
+    @Test
+    public void testAvailable() {
+        consulRegistry.register(serviceUrl);
+        assertThat(consulRegistry.isAvailable(), is(true));
+
+//        consulRegistry.destroy();
+//        assertThat(consulRegistry.isAvailable(), is(false));
+    }
+
+    @Test
+    public void testLookup() throws InterruptedException {
+        List<URL> lookup = consulRegistry.lookup(serviceUrl);
+        assertThat(lookup.size(), is(0));
+
+        consulRegistry.register(serviceUrl);
+        Thread.sleep(5000);
+        lookup = consulRegistry.lookup(serviceUrl);
+        assertThat(lookup.size(), is(1));
+    }
+
+    @Test
+    public void testStatusChecker() {
+        RegistryStatusChecker registryStatusChecker = new RegistryStatusChecker();
+        Status status = registryStatusChecker.check();
+        assertThat(status.getLevel(), is(Status.Level.UNKNOWN));
+
+        Registry registry = consulRegistryFactory.getRegistry(registryUrl);
+        assertThat(registry, not(nullValue()));
+
+        status = registryStatusChecker.check();
+        assertThat(status.getLevel(), is(Status.Level.OK));
+
+        registry.register(serviceUrl);
+        status = registryStatusChecker.check();
+        assertThat(status.getLevel(), is(Status.Level.OK));
+    }
+
+}