+++ /dev/null
-/**\r
-* Copyright (C) 2016 ZTE, Inc. and others. All rights reserved. (ZTE)\r
-*\r
-* Licensed under the Apache License, Version 2.0 (the "License");\r
-* you may not use this file except in compliance with the License.\r
-* You may obtain a copy of the License at\r
-*\r
-* http://www.apache.org/licenses/LICENSE-2.0\r
-*\r
-* Unless required by applicable law or agreed to in writing, software\r
-* distributed under the License is distributed on an "AS IS" BASIS,\r
-* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
-* See the License for the specific language governing permissions and\r
-* limitations under the License.\r
-*/\r
-\r
-package org.openo.msb;\r
-\r
-import java.net.MalformedURLException;\r
-import java.net.URL;\r
-import java.util.ArrayList;\r
-import java.util.HashSet;\r
-import java.util.List;\r
-import java.util.ListIterator;\r
-import java.util.Map;\r
-import java.util.concurrent.atomic.AtomicReference;\r
-\r
-import org.apache.commons.lang3.StringUtils;\r
-import org.openo.msb.api.MicroServiceFullInfo;\r
-import org.openo.msb.api.MicroServiceInfo;\r
-import org.openo.msb.api.Node;\r
-import org.openo.msb.wrapper.MicroServiceWrapper;\r
-import org.openo.msb.wrapper.consul.CatalogClient;\r
-import org.openo.msb.wrapper.consul.Consul;\r
-import org.openo.msb.wrapper.consul.HealthClient;\r
-import org.openo.msb.wrapper.consul.cache.CatalogCache;\r
-import org.openo.msb.wrapper.consul.cache.ConsulCache;\r
-import org.openo.msb.wrapper.consul.cache.ConsulCache4Map;\r
-import org.openo.msb.wrapper.consul.cache.HealthCache;\r
-import org.openo.msb.wrapper.consul.cache.ServiceCache;\r
-import org.openo.msb.wrapper.consul.model.catalog.CatalogService;\r
-import org.openo.msb.wrapper.consul.model.catalog.ServiceInfo;\r
-import org.openo.msb.wrapper.consul.model.health.Service;\r
-import org.openo.msb.wrapper.consul.model.health.ServiceHealth;\r
-import org.slf4j.Logger;\r
-import org.slf4j.LoggerFactory;\r
-\r
-public class ConsulClientApp {\r
-\r
- private final Consul consul;\r
- private final CatalogClient catalogClient;\r
- private final HealthClient healthClient;\r
- private AtomicReference<List<HealthCache>> cacheList = new AtomicReference<List<HealthCache>>(\r
- new ArrayList<HealthCache>());\r
-\r
-\r
- private static final Logger LOGGER = LoggerFactory.getLogger(ConsulClientApp.class);\r
-\r
- public ConsulClientApp(String ip, int port) {\r
- URL url = null;\r
- try {\r
- url = new URL("http", ip, port, "");\r
- } catch (MalformedURLException e1) {\r
- // TODO Auto-generated catch block\r
- LOGGER.error("start ConsulClientApp throw exception", e1);\r
- throw new RuntimeException(e1);\r
- }\r
- this.consul = Consul.builder().withUrl(url).build(); // connect to Consul on localhost\r
- this.catalogClient = consul.catalogClient();\r
- this.healthClient = consul.healthClient();\r
- }\r
-\r
- public Consul getConsul() {\r
- return consul;\r
- }\r
-\r
- public CatalogClient getCatalogClient() {\r
- return catalogClient;\r
- }\r
-\r
- private void stopNodeListen(String serviceName) {\r
- try {\r
-\r
- ListIterator<HealthCache> cacheListLit = cacheList.get().listIterator();\r
- while (cacheListLit.hasNext()) {\r
- HealthCache cache = (HealthCache) cacheListLit.next();\r
- if (cache.getServiceName().equals(serviceName)) {\r
-\r
- cache.stop();\r
- cacheListLit.remove();\r
- LOGGER.info(cache.getServiceName() + " NodeListen stoped");\r
- break;\r
- }\r
- }\r
-\r
- } catch (Exception e) {\r
- // TODO Auto-generated catch block\r
- LOGGER.error("stop Node:[" + serviceName + "] Listen throw exception", e);\r
- }\r
-\r
-\r
- }\r
-\r
- /**\r
- * @Title startServiceListen\r
- * @Description TODO(Open the consul registration services to monitor)\r
- * @return void\r
- */\r
- public void startServiceListen() {\r
- final ServiceCache serviceCache = ServiceCache.newCache(catalogClient, 30);\r
- serviceCache.addListener(new ConsulCache4Map.Listener<String, Map<String, List<String>>>() {\r
- @Override\r
- public void notify(List<ServiceInfo> oldValues, List<ServiceInfo> newValues) {\r
- // do Something with updated server List\r
- LOGGER.info("--new service notify--");\r
-\r
- List<ServiceInfo> deRegisterServiceList = getDiffrent(oldValues, newValues);\r
-\r
-\r
- for (ServiceInfo serviceInfo : deRegisterServiceList) {\r
- try {\r
- \r
- MicroServiceWrapper.getInstance().deleteMicroService(\r
- serviceInfo.getServiceName(), serviceInfo.getVersion());\r
- \r
-\r
- stopNodeListen(serviceInfo.getServiceName());\r
- LOGGER.info("Cancel MicroServiceInfo and stop node listen successs:"\r
- + serviceInfo.getServiceName());\r
- } catch (Exception e) {\r
- LOGGER.error("Cancel MicroServiceInfo and stop node listen FAIL : ", e);\r
-\r
- }\r
-\r
- }\r
-\r
-\r
- List<ServiceInfo> registerServiceList = getDiffrent(newValues, oldValues);\r
- for (ServiceInfo serviceInfo : registerServiceList) {\r
-\r
- // if (deRegisterServiceList.contains(serviceInfo)) continue;\r
-\r
-\r
- LOGGER.info(" new serviceName:" + serviceInfo.getServiceName() + " version:"\r
- + serviceInfo.getVersion());\r
- // Open Node to monitor new registration service\r
- startHealthNodeListen(serviceInfo.getServiceName(), serviceInfo.getVersion());\r
-\r
- }\r
-\r
-\r
- }\r
-\r
- });\r
-\r
- try {\r
- LOGGER.info("start...consul ... service..Listening.");\r
- serviceCache.start();\r
-\r
- } catch (Exception e) {\r
- // TODO Auto-generated catch block\r
- LOGGER.error("start...service..Listen throw exception", e);\r
- }\r
- }\r
-\r
-\r
- /**\r
- * @Title startHealthNodeListen\r
- * @Description TODO(Open a service node changes to monitor, only to return to health service)\r
- * @param serviceName\r
- * @return\r
- * @return HealthCache\r
- */\r
- private HealthCache startHealthNodeListen(final String serviceName, final String version) {\r
- final HealthCache healthCache = HealthCache.newCache(healthClient, serviceName, 30);\r
- healthCache.addListener(new HealthCache.Listener<String, ServiceHealth>() {\r
- @Override\r
- public void notify(Map<String, ServiceHealth> newValues) {\r
- // do Something with updated server map\r
- LOGGER.info(serviceName + "--new node notify--");\r
-\r
- if (newValues.isEmpty()) {\r
- LOGGER.info(serviceName + "--nodeList is Empty--");\r
-\r
- \r
- MicroServiceWrapper.getInstance().deleteMicroService(serviceName, version);\r
- \r
- // try {\r
- // healthCache.stop();\r
- // } catch (Exception e) {\r
- // LOGGER.equals(serviceName+"-- stop Node error:"+e.getMessage());\r
- // }\r
-\r
- } else {\r
-\r
- MicroServiceInfo microServiceInfo = new MicroServiceInfo();\r
- HashSet<Node> nodes = new HashSet<Node>();\r
- String url = "";\r
- String version = "", visualRange = "", protocol = "",lb_policy="";\r
-\r
- for (Map.Entry<String, ServiceHealth> entry : newValues.entrySet()) {\r
- String nodeName = entry.getKey().toString();\r
- ServiceHealth value = (ServiceHealth) entry.getValue();\r
-\r
- Node node = new Node();\r
- Service service = value.getService();\r
- node.setIp(service.getAddress());\r
- node.setPort(String.valueOf(service.getPort()));\r
-\r
-\r
- try {\r
- List<String> tagList = service.getTags();\r
- for (String tag : tagList) {\r
- if (tag.startsWith("url")) {\r
- if (tag.split(":").length == 2) {\r
- url = tag.split(":")[1];\r
- } else {\r
- url = "";\r
- }\r
-\r
-\r
- continue;\r
- }\r
- if (tag.startsWith("version")) {\r
- if (tag.split(":").length == 2) {\r
- version = tag.split(":")[1];\r
- } else {\r
- version = "";\r
- }\r
- continue;\r
- }\r
- if (tag.startsWith("protocol")) {\r
- protocol = tag.split(":")[1];\r
- continue;\r
- }\r
- if (tag.startsWith("visualRange")) {\r
- visualRange = tag.split(":")[1];\r
- continue;\r
- }\r
- \r
- if (tag.startsWith("lb_policy")) {\r
- lb_policy = tag.split(":")[1];\r
- continue;\r
- }\r
-\r
- }\r
-\r
-\r
- } catch (Exception e) {\r
- LOGGER.error(serviceName + " read tag throw exception", e);\r
- System.out.println(serviceName + " read tag throw exception");\r
- }\r
-\r
- nodes.add(node);\r
- }\r
-\r
- microServiceInfo.setNodes(nodes);\r
- microServiceInfo.setProtocol(protocol);\r
- microServiceInfo.setUrl(url);\r
- microServiceInfo.setServiceName(serviceName);\r
- microServiceInfo.setLb_policy(lb_policy);\r
- if (!visualRange.isEmpty()) {\r
- microServiceInfo.setVisualRange(visualRange);\r
- }\r
- microServiceInfo.setVersion(version);\r
-\r
- try {\r
- MicroServiceFullInfo microServiceFullInfo =\r
- MicroServiceWrapper.getInstance().saveMicroServiceInstance(\r
- microServiceInfo, false, "", "");\r
- LOGGER.info("register MicroServiceInfo successs:"\r
- + microServiceFullInfo.getServiceName());\r
- } catch (Exception e) {\r
- LOGGER.error("register MicroServiceInfo FAIL : " + serviceName, e);\r
-\r
- }\r
- }\r
- }\r
- });\r
- try {\r
- LOGGER.info(serviceName + " Node Listen start");\r
- cacheList.get().add(healthCache);\r
- healthCache.start();\r
-\r
- } catch (Exception e) {\r
- // TODO Auto-generated catch block\r
- LOGGER.error(serviceName + " Node Listen start throw exception", e);\r
- }\r
-\r
- return healthCache;\r
- }\r
-\r
- /**\r
- * @Title startNodeListen\r
- * @Description TODO(Open a service node changes to monitor)\r
- * @param serviceName\r
- * @return\r
- * @return CatalogCache\r
- */\r
- @Deprecated\r
- private CatalogCache startNodeListen(final String serviceName) {\r
- final CatalogCache catalogCache = CatalogCache.newCache(catalogClient, serviceName, 30);\r
- catalogCache.addListener(new ConsulCache.Listener<String, CatalogService>() {\r
- @Override\r
- public void notify(Map<String, CatalogService> newValues) {\r
- // do Something with updated server map\r
- System.out.println(serviceName + "--new node notify--");\r
- LOGGER.info(serviceName + "--new node notify--");\r
-\r
- if (newValues.isEmpty()) {\r
- System.out.println(serviceName + "-- nodeList is Empty--");\r
- LOGGER.info(serviceName + "--nodeList is Empty-stop service[" + serviceName\r
- + "] listen-");\r
- try {\r
- catalogCache.stop();\r
- } catch (Exception e) {\r
- LOGGER.equals(serviceName + "-- stop Node error:" + e.getMessage());\r
- }\r
-\r
- } else {\r
-\r
- MicroServiceInfo microServiceInfo = new MicroServiceInfo();\r
- HashSet<Node> nodes = new HashSet<Node>();\r
- String url = "";\r
- String version = "", visualRange = "", protocol = "";\r
-\r
- for (Map.Entry<String, CatalogService> entry : newValues.entrySet()) {\r
- String nodeName = entry.getKey().toString();\r
- CatalogService value = (CatalogService) entry.getValue();\r
-\r
- Node node = new Node();\r
- node.setIp(value.getServiceAddress());\r
- node.setPort(String.valueOf(value.getServicePort()));\r
-\r
-\r
- try {\r
- List<String> tagList = value.getServiceTags();\r
- for (String tag : tagList) {\r
- if (tag.startsWith("url")) {\r
- if (tag.split(":").length == 2) {\r
- url = tag.split(":")[1];\r
- } else {\r
- url = "";\r
- }\r
-\r
-\r
- continue;\r
- }\r
- if (tag.startsWith("version")) {\r
- if (tag.split(":").length == 2) {\r
- version = tag.split(":")[1];\r
- } else {\r
- version = "";\r
- }\r
- continue;\r
- }\r
- if (tag.startsWith("protocol")) {\r
- protocol = tag.split(":")[1];\r
- continue;\r
- }\r
- if (tag.startsWith("visualRange")) {\r
- visualRange = tag.split(":")[1];\r
- continue;\r
- }\r
- if (tag.startsWith("ttl")) {\r
- int ttl = Integer.parseInt(tag.split(":")[1]);\r
- node.setTtl(ttl);\r
- continue;\r
- }\r
- }\r
-\r
-\r
- } catch (Exception e) {\r
- LOGGER.error(serviceName + " read tag throw exception", e);\r
- System.out.println(serviceName + " read tag throw exception");\r
- }\r
-\r
- nodes.add(node);\r
-\r
-\r
- System.out.println(nodeName + ":" + value.getServiceAddress() + " "\r
- + value.getServicePort() + " " + value.getServiceTags());\r
- }\r
-\r
- microServiceInfo.setNodes(nodes);\r
- microServiceInfo.setProtocol(protocol);\r
- microServiceInfo.setUrl(url);\r
- microServiceInfo.setServiceName(serviceName);\r
- if (!visualRange.isEmpty()) {\r
- microServiceInfo.setVisualRange(visualRange);\r
- }\r
- microServiceInfo.setVersion(version);\r
-\r
- try {\r
- MicroServiceFullInfo microServiceFullInfo =\r
- MicroServiceWrapper.getInstance().saveMicroServiceInstance(\r
- microServiceInfo, false, "", "");\r
- LOGGER.info("register MicroServiceInfo successs:" + microServiceFullInfo);\r
- System.out.println("register MicroServiceInfo successs:" + serviceName);\r
- } catch (Exception e) {\r
- LOGGER.error("register MicroServiceInfo FAIL : ", e);\r
-\r
- }\r
- }\r
- }\r
- });\r
- try {\r
- System.out.println(serviceName + " Node Listen start");\r
- LOGGER.info(serviceName + " Node Listen start");\r
- catalogCache.start();\r
-\r
- } catch (Exception e) {\r
- // TODO Auto-generated catch block\r
- LOGGER.error(serviceName + " Node Listen start throw exception", e);\r
- }\r
-\r
- return catalogCache;\r
- }\r
-\r
-\r
- /**\r
- * @Title getDiffrent\r
- * @Description TODO(Extract the list1 and list2 different data sets)\r
- * @param list1\r
- * @param list2\r
- * @return\r
- * @return List<String>\r
- */\r
- private List<ServiceInfo> getDiffrent(List<ServiceInfo> list1, List<ServiceInfo> list2) {\r
-\r
- List<ServiceInfo> diff = new ArrayList<ServiceInfo>();\r
-\r
-\r
-\r
- for (ServiceInfo serviceInfo : list1) {\r
- if (!list2.contains(serviceInfo)) {\r
- diff.add(serviceInfo);\r
- }\r
- }\r
-\r
- return diff;\r
- }\r
-\r
- public static void main(String[] args) {\r
- ConsulClientApp consulTest = new ConsulClientApp("127.0.0.1", 10081);\r
- consulTest.startServiceListen();\r
-\r
-\r
- }\r
-\r
-\r
-}\r
Code Review / msb / apigateway.git / blobdiff