[SCB-322] merge config center modification

dynamic-config/config-cc/src/main/java/org/apache/servicecomb/config/client/ConfigCenterClient.java

@@ -39,6 +39,7 @@
 import org.apache.servicecomb.config.archaius.sources.ConfigCenterConfigurationSourceImpl;
 import org.apache.servicecomb.foundation.auth.AuthHeaderProvider;
 import org.apache.servicecomb.foundation.auth.SignRequest;
+import org.apache.servicecomb.foundation.common.event.EventManager;
 import org.apache.servicecomb.foundation.common.net.IpPort;
 import org.apache.servicecomb.foundation.common.net.NetUtils;
 import org.apache.servicecomb.foundation.common.utils.JsonUtils;
@@ -67,6 +68,7 @@
 import io.vertx.core.http.WebSocket;
 import io.vertx.core.http.impl.FrameType;
 import io.vertx.core.http.impl.ws.WebSocketFrameImpl;
+import io.vertx.core.net.ProxyOptions;
 
 /**
  * Created by on 2016/5/17.
@@ -126,8 +128,10 @@ public void connectServer() {
     }
     MemberDiscovery memberdis = new MemberDiscovery(serverUri);
     refreshMembers(memberdis);
-    EXECUTOR.scheduleWithFixedDelay(new ConfigRefresh(parseConfigUtils, memberdis), firstRefreshInterval,
-        refreshInterval, TimeUnit.MILLISECONDS);
+    EXECUTOR.scheduleWithFixedDelay(new ConfigRefresh(parseConfigUtils, memberdis),
+        firstRefreshInterval,
+        refreshInterval,
+        TimeUnit.MILLISECONDS);
   }
 
   private void refreshMembers(MemberDiscovery memberDiscovery) {
@@ -167,9 +171,17 @@ private void deployConfigClient() throws InterruptedException {
 
   private HttpClientOptions createHttpClientOptions() {
     HttpClientOptions httpClientOptions = new HttpClientOptions();
+    if (ConfigCenterConfig.INSTANCE.isProxyEnable()) {
+      ProxyOptions proxy = new ProxyOptions()
+          .setHost(ConfigCenterConfig.INSTANCE.getProxyHost())
+          .setPort(ConfigCenterConfig.INSTANCE.getProxyPort())
+          .setUsername(ConfigCenterConfig.INSTANCE.getProxyUsername())
+          .setPassword(ConfigCenterConfig.INSTANCE.getProxyPasswd());
+      httpClientOptions.setProxyOptions(proxy);
+    }
     httpClientOptions.setConnectTimeout(CONFIG_CENTER_CONFIG.getConnectionTimeout());
     if (serverUri.get(0).toLowerCase().startsWith("https")) {
-      LOGGER.debug("service center client performs requests over TLS");
+      LOGGER.debug("config center client performs requests over TLS");
       SSLOptionFactory factory = SSLOptionFactory.createSSLOptionFactory(SSL_KEY,
           ConfigCenterConfig.INSTANCE.getConcurrentCompositeConfiguration());
       SSLOption sslOption;
@@ -216,7 +228,7 @@ public void run() {
 
     // create watch and wait for done
     public void doWatch(String configCenter)
-        throws URISyntaxException, UnsupportedEncodingException, InterruptedException {
+        throws UnsupportedEncodingException, InterruptedException {
       CountDownLatch waiter = new CountDownLatch(1);
       IpPort ipPort = NetUtils.parseIpPortFromURI(configCenter);
       String url = URIConst.REFRESH_ITEMS + "?dimensionsInfo="
@@ -285,7 +297,9 @@ private void stopHeartBeatThread() {
     private void sendHeartbeat(WebSocket ws) {
       try {
         ws.writeFrame(new WebSocketFrameImpl(FrameType.PING));
+        EventManager.post(new ConnSuccEvent());
       } catch (IllegalStateException e) {
+        EventManager.post(new ConnFailEvent("heartbeat fail, " + e.getMessage()));
         LOGGER.error("heartbeat fail", e);
       }
     }
@@ -302,11 +316,17 @@ public void refreshConfig(String configcenter) {
                     .refreshConfigItems(JsonUtils.OBJ_MAPPER.readValue(buf.toString(),
                         new TypeReference<LinkedHashMap<String, Map<String, String>>>() {
                         }));
+                EventManager.post(new ConnSuccEvent());
               } catch (IOException e) {
+                EventManager.post(new ConnFailEvent("config refresh result parse fail " + e.getMessage()));
                 LOGGER.error("config refresh result parse fail", e);
               }
             });
           } else {
+            rsp.bodyHandler(buf -> {
+              LOGGER.error("fetch config fail: " + buf);
+            });
+            EventManager.post(new ConnFailEvent("fetch config fail"));
             LOGGER.error("fetch config fail");
           }
         });
@@ -333,7 +353,7 @@ public static SignRequest createSignRequest(String method, String endpoint, Map<
     try {
       signReq.setEndpoint(new URI(endpoint));
     } catch (URISyntaxException e) {
-      LOGGER.warn("set uri failed, uri is {}, message: {}", endpoint.toString(), e.getMessage());
+      LOGGER.warn("set uri failed, uri is {}, message: {}", endpoint, e.getMessage());
     }
 
     Map<String, String[]> queryParams = new HashMap<>();

dynamic-config/config-cc/src/main/java/org/apache/servicecomb/config/client/ConfigCenterConfig.java

@@ -18,7 +18,6 @@
 package org.apache.servicecomb.config.client;
 
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.List;
 
 import com.google.common.base.Joiner;
@@ -57,6 +56,18 @@ public final class ConfigCenterConfig {
 
   private static final String INSTANCE_TAGS = "instance_description.properties.tags";
 
+  public static final String PROXY_PRE_NAME = "cse.proxy.";
+
+  public static final String PROXY_ENABLE = PROXY_PRE_NAME + "enable";
+
+  public static final String PROXY_HOST = PROXY_PRE_NAME + "host";
+
+  public static final String PROXY_PORT = PROXY_PRE_NAME + "port";
+
+  public static final String PROXY_USERNAME = PROXY_PRE_NAME + "username";
+
+  public static final String PROXY_PASSWD = PROXY_PRE_NAME + "passwd";
+
   private static final int DEFAULT_REFRESH_MODE = 0;
 
   private static final int DEFAULT_REFRESH_PORT = 30104;
@@ -71,7 +82,7 @@ private ConfigCenterConfig() {
   }
 
   public static void setConcurrentCompositeConfiguration(ConcurrentCompositeConfiguration config) {
-    finalConfig = (ConcurrentCompositeConfiguration) config;
+    finalConfig = config;
   }
 
   public ConcurrentCompositeConfiguration getConcurrentCompositeConfiguration() {
@@ -110,6 +121,26 @@ public int getFirstRefreshInterval() {
     return finalConfig.getInt(FIRST_REFRESH_INTERVAL, DEFAULT_FIRST_REFRESH_INTERVAL);
   }
 
+  public Boolean isProxyEnable() {
+    return finalConfig.getBoolean(PROXY_ENABLE, false);
+  }
+
+  public String getProxyHost() {
+    return finalConfig.getString(PROXY_HOST, "127.0.0.1");
+  }
+
+  public int getProxyPort() {
+    return finalConfig.getInt(PROXY_PORT, 8080);
+  }
+
+  public String getProxyUsername() {
+    return finalConfig.getString(PROXY_USERNAME, "");
+  }
+
+  public String getProxyPasswd() {
+    return finalConfig.getString(PROXY_PASSWD, "");
+  }
+
   @SuppressWarnings("unchecked")
   public String getServiceName() {
     String service = finalConfig.getString(SERVICE_NAME);
@@ -140,7 +171,9 @@ public String getServiceName() {
   public List<String> getServerUri() {
     String[] result = finalConfig.getStringArray(SERVER_URL_KEY);
     List<String> configCenterUris = new ArrayList<>(result.length);
-    configCenterUris.addAll(Arrays.asList(result));
+    for (int i = 0; i < result.length; i++) {
+      configCenterUris.add(result[i]);
+    }
     return configCenterUris;
   }
 

dynamic-config/config-cc/src/main/java/org/apache/servicecomb/config/client/ConnFailEvent.java

@@ -0,0 +1,35 @@
+/*
+ * 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.servicecomb.config.client;
+
+public class ConnFailEvent {
+
+  private String msg;
+
+  public ConnFailEvent(String msg) {
+    this.msg = msg;
+  }
+
+  public String getMsg() {
+    return msg;
+  }
+
+  public void setMsg(String msg) {
+    this.msg = msg;
+  }
+}

dynamic-config/config-cc/src/main/java/org/apache/servicecomb/config/client/ConnSuccEvent.java

@@ -0,0 +1,22 @@
+/*
+ * 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.servicecomb.config.client;
+
+public class ConnSuccEvent {
+
+}

dynamic-config/config-cc/src/test/java/org/apache/servicecomb/config/client/TestConfigCenterClient.java

@@ -18,19 +18,23 @@
 package org.apache.servicecomb.config.client;
 
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.Map;
 
 import org.apache.servicecomb.config.ConfigUtil;
 import org.apache.servicecomb.config.archaius.sources.ConfigCenterConfigurationSourceImpl;
 import org.apache.servicecomb.config.archaius.sources.ConfigCenterConfigurationSourceImpl.UpdateHandler;
 import org.apache.servicecomb.config.client.ConfigCenterClient.ConfigRefresh;
+import org.apache.servicecomb.foundation.common.event.EventManager;
 import org.apache.servicecomb.foundation.vertx.client.http.HttpClientWithContext;
 import org.apache.servicecomb.foundation.vertx.client.http.HttpClientWithContext.RunHandler;
 import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.mockito.Mockito;
 
+import com.google.common.eventbus.Subscribe;
+
 import io.vertx.core.Handler;
 import io.vertx.core.MultiMap;
 import io.vertx.core.buffer.Buffer;
@@ -168,4 +172,63 @@ public void runOnContext(RunHandler handler) {
     Deencapsulation.setField(cc, "refreshMode", 0);
     refresh.run();
   }
+
+  @SuppressWarnings("unchecked")
+  @Test
+  public void testConfigRefreshException() {
+    ConfigCenterConfigurationSourceImpl impl = new ConfigCenterConfigurationSourceImpl();
+    Map<String, String> map = new HashMap<>();
+    EventManager.register(new Object() {
+      @Subscribe
+      public void testMsg(Object event) {
+        if (event instanceof ConnFailEvent) {
+          map.put("result", "Fail event trigger");
+        }
+        if (event instanceof ConnSuccEvent) {
+          map.put("result", "Succ event trigger");
+        }
+      }
+    });
+    impl.init(ConfigUtil.createLocalConfig());
+    UpdateHandler updateHandler = impl.new UpdateHandler();
+    HttpClientRequest request = Mockito.mock(HttpClientRequest.class);
+    Mockito.when(request.headers()).thenReturn(MultiMap.caseInsensitiveMultiMap());
+    Buffer rsp = Mockito.mock(Buffer.class);
+    Mockito.when(rsp.toString())
+        .thenReturn("{\"application\":{\"2\":\"2\",\"aa\":\"1\"},\"vmalledge\":{\"aa\":\"3\"}}");
+
+    HttpClientResponse event = Mockito.mock(HttpClientResponse.class);
+    Mockito.when(event.bodyHandler(Mockito.any(Handler.class))).then(invocation -> {
+      Handler<Buffer> handler = invocation.getArgumentAt(0, Handler.class);
+      handler.handle(rsp);
+      return null;
+    });
+    Mockito.when(event.statusCode()).thenReturn(400);
+    Buffer buf = Mockito.mock(Buffer.class);
+    Mockito.when(buf.toJsonObject()).thenReturn(new JsonObject(
+        "{\"action\":\"UPDATE\",\"key\":\"vmalledge\",\"value\":\"{\\\"aa\\\":\\\"3\\\"}\"}"));
+    HttpClient httpClient = Mockito.mock(HttpClient.class);
+    Mockito.when(
+        httpClient.get(Mockito.anyInt(), Mockito.anyString(), Mockito.anyString(), Mockito.any(Handler.class)))
+        .then(invocation -> {
+          Handler<HttpClientResponse> handler = invocation.getArgumentAt(3, Handler.class);
+          handler.handle(event);
+          return request;
+        });
+    new MockUp<HttpClientWithContext>() {
+      @Mock
+      public void runOnContext(RunHandler handler) {
+        handler.run(httpClient);
+      }
+    };
+    ConfigCenterClient cc = new ConfigCenterClient(updateHandler);
+    ParseConfigUtils parseConfigUtils = new ParseConfigUtils(updateHandler);
+    MemberDiscovery memberdis = new MemberDiscovery(Arrays.asList("http://configcentertest:30103"));
+    ConfigRefresh refresh = cc.new ConfigRefresh(parseConfigUtils, memberdis);
+    refresh.run();
+    Assert.assertEquals("Fail event trigger", map.get("result"));
+    Mockito.when(event.statusCode()).thenReturn(200);
+    refresh.run();
+    Assert.assertEquals("Succ event trigger", map.get("result"));
+  }
 }

dynamic-config/config-cc/src/test/java/org/apache/servicecomb/config/client/TestURIConst.java

@@ -17,22 +17,26 @@
 
 package org.apache.servicecomb.config.client;
 
-import org.apache.servicecomb.config.ConfigUtil;
 import org.junit.Assert;
-import org.junit.BeforeClass;
 import org.junit.Test;
 
-public class TestURIConst {
-  @BeforeClass
-  public static void setUpClass() {
-    ConfigCenterConfig.setConcurrentCompositeConfiguration(ConfigUtil.createLocalConfig());
-  }
+import mockit.Expectations;
+import mockit.Mocked;
 
+public class TestURIConst {
   @Test
-  public void testURIV3() {
-    Assert.assertEquals(URIConst.MEMBERS, "/v3/default/configuration/members");
-    Assert.assertEquals(URIConst.REFRESH_ITEMS, "/v3/default/configuration/refresh/items");
-    Assert.assertEquals(URIConst.ITEMS, "/v3/default/configuration/items");
-  }
+  public void testURI(final @Mocked ConfigCenterConfig config) {
+    new Expectations() {
+      {
+        config.getDomainName();
+        result = "mytenant";
+        config.getApiVersion();
+        result = "v3";
+      }
+    };
 
+    Assert.assertEquals(URIConst.MEMBERS, "/v3/mytenant/configuration/members");
+    Assert.assertEquals(URIConst.REFRESH_ITEMS, "/v3/mytenant/configuration/refresh/items");
+    Assert.assertEquals(URIConst.ITEMS, "/v3/mytenant/configuration/items");
+  }
 }