5226

Committed 16 Aug 2020 - 23:44 coverage: 86.277% (-0.01%) from 86.288%

Build # 5226

Build Type

Pull #1920

travis-ci

Committed by

web-flow

Commit Message

Merge 296890e57 into e84f4c8c1

Pull Request Pull Request #1920: Fix word grammar errors

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

5 existing lines in 1 file now uncovered.

33070 of 38330 relevant lines covered (86.28%)

1.44 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.38

/dynamic-config/config-kie/src/main/java/org/apache/servicecomb/config/kie/client/KieClient.java

/*
 * 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.kie.client;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.http.HttpStatus;
import org.apache.servicecomb.config.kie.archaius.sources.KieConfigurationSourceImpl.UpdateHandler;
import org.apache.servicecomb.config.kie.model.KVResponse;
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;
import org.apache.servicecomb.foundation.vertx.client.http.HttpClients;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.vertx.core.http.HttpClientRequest;

public class KieClient {

  private static final Logger LOGGER = LoggerFactory.getLogger(KieClient.class);

  private ScheduledExecutorService EXECUTOR = Executors.newScheduledThreadPool(1, (r) -> {
    Thread thread = new Thread(r);
    thread.setName("org.apache.servicecomb.config.kie");
    thread.setDaemon(true);
    return thread;
  });

  private static final long PULL_REQUEST_TIME_OUT_IN_MILLIS = 10000;

  private static final long LONG_POLLING_REQUEST_TIME_OUT_IN_MILLIS = 60000;

  private static AtomicBoolean IS_FIRST_PULL = new AtomicBoolean(true);

  private static final int LONG_POLLING_WAIT_TIME_IN_SECONDS = 30;

  private static String revision = "0";

  private static final KieConfig KIE_CONFIG = KieConfig.INSTANCE;

  private final int refreshInterval = KIE_CONFIG.getRefreshInterval();

  private final int firstRefreshInterval = KIE_CONFIG.getFirstRefreshInterval();

  private final boolean enableLongPolling = KIE_CONFIG.enableLongPolling();

  private final String serviceUri = KIE_CONFIG.getServerUri();

  public KieClient(UpdateHandler updateHandler) {
    HttpClients.addNewClientPoolManager(new ConfigKieHttpClientOptionsSPI());
    KieWatcher.INSTANCE.setUpdateHandler(updateHandler);
  }

  public void refreshKieConfig() {
    if (enableLongPolling) {
      EXECUTOR.execute(new ConfigRefresh(serviceUri));
    } else {
      EXECUTOR.scheduleWithFixedDelay(new ConfigRefresh(serviceUri), firstRefreshInterval,
          refreshInterval, TimeUnit.MILLISECONDS);
    }
  }

  public void destroy() {
    if (EXECUTOR != null) {
      EXECUTOR.shutdown();
      EXECUTOR = null;
    }
  }

  class ConfigRefresh implements Runnable {

    private final String serviceUri;

    ConfigRefresh(String serviceUris) {
      this.serviceUri = serviceUris;
    }

    @Override
    public void run() {
      try {
        CountDownLatch latch = new CountDownLatch(1);
        refreshConfig(latch);
        latch.await();
      } catch (Throwable e) {
        LOGGER.error("client refresh thread exception ", e);
      }
      if (enableLongPolling) {
        EXECUTOR.execute(this);
      }
    }

    @SuppressWarnings("deprecation")
    void refreshConfig(CountDownLatch latch) {
      String path = "/v1/"
          + KieConfig.INSTANCE.getDomainName()
          + "/kie/kv?label=app:"
          + KieConfig.INSTANCE.getAppName()
          + "&revision=" + revision;
      long timeout;
      if (enableLongPolling && !IS_FIRST_PULL.get()) {
        path += "&wait=" + LONG_POLLING_WAIT_TIME_IN_SECONDS + "s";
        timeout = LONG_POLLING_REQUEST_TIME_OUT_IN_MILLIS;
      } else {
        IS_FIRST_PULL.compareAndSet(true, false);
        timeout = PULL_REQUEST_TIME_OUT_IN_MILLIS;
      }
      String finalPath = path;
      HttpClients.getClient(ConfigKieHttpClientOptionsSPI.CLIENT_NAME).runOnContext(client -> {
        IpPort ipPort = NetUtils.parseIpPortFromURI(serviceUri);
        HttpClientRequest request = client
            .get(ipPort.getPort(), ipPort.getHostOrIp(), finalPath, rsp -> {
              if (rsp.statusCode() == HttpStatus.SC_OK) {
                revision = rsp.getHeader("X-Kie-Revision");
                rsp.bodyHandler(buf -> {
                  try {
                    Map<String, Object> resMap = KieUtil.getConfigByLabel(JsonUtils.OBJ_MAPPER
                        .readValue(buf.toString(), KVResponse.class));
                    KieWatcher.INSTANCE.refreshConfigItems(resMap);
                    EventManager.post(new ConnSuccEvent());
                  } catch (IOException e) {
                    EventManager.post(new ConnFailEvent(
                        "config update result parse fail " + e.getMessage()));
                    LOGGER.error("Config update from {} failed. Error message is [{}].",
                        serviceUri,
                        e.getMessage());
                  }
                  latch.countDown();
                });
              } else if (rsp.statusCode() == HttpStatus.SC_NOT_MODIFIED) {
                EventManager.post(new ConnSuccEvent());
                latch.countDown();
              } else {
                EventManager.post(new ConnFailEvent("fetch config fail"));
                LOGGER.error("Config update from {} failed. Error code is {}, error message is [{}].",
                    serviceUri,
                    rsp.statusCode(),
                    rsp.statusMessage());
                latch.countDown();
              }
            }).setTimeout(timeout);

        request.exceptionHandler(e -> {
          EventManager.post(new ConnFailEvent("fetch config fail"));
          LOGGER.error("Config update from {} failed. Error message is [{}].",
              serviceUri,
              e.getMessage());
          latch.countDown();
        });
        request.end();
      });
    }
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.servicecomb.config.kie.client;
19
20	import java.io.IOException;
21	import java.util.Map;
22	import java.util.concurrent.CountDownLatch;
23	import java.util.concurrent.Executors;
24	import java.util.concurrent.ScheduledExecutorService;
25	import java.util.concurrent.TimeUnit;
26	import java.util.concurrent.atomic.AtomicBoolean;
27
28	import org.apache.http.HttpStatus;
29	import org.apache.servicecomb.config.kie.archaius.sources.KieConfigurationSourceImpl.UpdateHandler;
30	import org.apache.servicecomb.config.kie.model.KVResponse;
31	import org.apache.servicecomb.foundation.common.event.EventManager;
32	import org.apache.servicecomb.foundation.common.net.IpPort;
33	import org.apache.servicecomb.foundation.common.net.NetUtils;
34	import org.apache.servicecomb.foundation.common.utils.JsonUtils;
35	import org.apache.servicecomb.foundation.vertx.client.http.HttpClients;
36	import org.slf4j.Logger;
37	import org.slf4j.LoggerFactory;
38
39	import io.vertx.core.http.HttpClientRequest;
40
41	public class KieClient {
42
43	private static final Logger LOGGER = LoggerFactory.getLogger(KieClient.class);	2×
44
45	private ScheduledExecutorService EXECUTOR = Executors.newScheduledThreadPool(1, (r) -> {	2×
46	Thread thread = new Thread(r);	2×
47	thread.setName("org.apache.servicecomb.config.kie");	2×
48	thread.setDaemon(true);	2×
49	return thread;	2×
50	});
51
52	private static final long PULL_REQUEST_TIME_OUT_IN_MILLIS = 10000;
53
54	private static final long LONG_POLLING_REQUEST_TIME_OUT_IN_MILLIS = 60000;
55
56	private static AtomicBoolean IS_FIRST_PULL = new AtomicBoolean(true);	2×
57
58	private static final int LONG_POLLING_WAIT_TIME_IN_SECONDS = 30;
59
60	private static String revision = "0";	2×
61
62	private static final KieConfig KIE_CONFIG = KieConfig.INSTANCE;	2×
63
64	private final int refreshInterval = KIE_CONFIG.getRefreshInterval();	2×
65
66	private final int firstRefreshInterval = KIE_CONFIG.getFirstRefreshInterval();	2×
67
68	private final boolean enableLongPolling = KIE_CONFIG.enableLongPolling();	2×
69
70	private final String serviceUri = KIE_CONFIG.getServerUri();	2×
71
72	public KieClient(UpdateHandler updateHandler) {	2×
73	HttpClients.addNewClientPoolManager(new ConfigKieHttpClientOptionsSPI());	2×
74	KieWatcher.INSTANCE.setUpdateHandler(updateHandler);	2×
75	}	2×
76
77	public void refreshKieConfig() {
78	if (enableLongPolling) {	2×
79	EXECUTOR.execute(new ConfigRefresh(serviceUri));	2×
80	} else {
81	EXECUTOR.scheduleWithFixedDelay(new ConfigRefresh(serviceUri), firstRefreshInterval,	!
82	refreshInterval, TimeUnit.MILLISECONDS);
83	}
84	}	2×
85
86	public void destroy() {
87	if (EXECUTOR != null) {	!
88	EXECUTOR.shutdown();	!
89	EXECUTOR = null;	!
90	}
91	}	!
92
93	class ConfigRefresh implements Runnable {
94
95	private final String serviceUri;
96
97	ConfigRefresh(String serviceUris) {	2×
98	this.serviceUri = serviceUris;	2×
99	}	2×
100
101	@Override
102	public void run() {
103	try {
104	CountDownLatch latch = new CountDownLatch(1);	2×
105	refreshConfig(latch);	2×
106	latch.await();	!
107	} catch (Throwable e) {	2×
108	LOGGER.error("client refresh thread exception ", e);	2×
109	}	!
110	if (enableLongPolling) {	2×
111	EXECUTOR.execute(this);	2×
112	}
113	}	2×
114
115	@SuppressWarnings("deprecation")
116	void refreshConfig(CountDownLatch latch) {
117	String path = "/v1/"	2×
118	+ KieConfig.INSTANCE.getDomainName()	2×
119	+ "/kie/kv?label=app:"
120	+ KieConfig.INSTANCE.getAppName()	2×
121	+ "&revision=" + revision;	2×
122	long timeout;
123	if (enableLongPolling && !IS_FIRST_PULL.get()) {	2×
124	path += "&wait=" + LONG_POLLING_WAIT_TIME_IN_SECONDS + "s";	2×
125	timeout = LONG_POLLING_REQUEST_TIME_OUT_IN_MILLIS;	2×
126	} else {
127	IS_FIRST_PULL.compareAndSet(true, false);	2×
128	timeout = PULL_REQUEST_TIME_OUT_IN_MILLIS;	2×
129	}
130	String finalPath = path;	2×
131	HttpClients.getClient(ConfigKieHttpClientOptionsSPI.CLIENT_NAME).runOnContext(client -> {	2×
132	IpPort ipPort = NetUtils.parseIpPortFromURI(serviceUri);	2×
133	HttpClientRequest request = client	2×
134	.get(ipPort.getPort(), ipPort.getHostOrIp(), finalPath, rsp -> {	2×
135	if (rsp.statusCode() == HttpStatus.SC_OK) {	2×
136	revision = rsp.getHeader("X-Kie-Revision");	2×
137	rsp.bodyHandler(buf -> {	2×
138	try {
139	Map<String, Object> resMap = KieUtil.getConfigByLabel(JsonUtils.OBJ_MAPPER	2×
140	.readValue(buf.toString(), KVResponse.class));	2×
141	KieWatcher.INSTANCE.refreshConfigItems(resMap);	2×
142	EventManager.post(new ConnSuccEvent());	2×
UNCOV 143	} catch (IOException e) {	!
UNCOV 144	EventManager.post(new ConnFailEvent(	!
UNCOV 145	"config update result parse fail " + e.getMessage()));	!
UNCOV 146	LOGGER.error("Config update from {} failed. Error message is [{}].",	!
147	serviceUri,
UNCOV 148	e.getMessage());	!
149	}	2×
150	latch.countDown();	2×
151	});	2×
152	} else if (rsp.statusCode() == HttpStatus.SC_NOT_MODIFIED) {	2×
153	EventManager.post(new ConnSuccEvent());	!
154	latch.countDown();	!
155	} else {
156	EventManager.post(new ConnFailEvent("fetch config fail"));	2×
157	LOGGER.error("Config update from {} failed. Error code is {}, error message is [{}].",	2×
158	serviceUri,
159	rsp.statusCode(),	2×
160	rsp.statusMessage());	2×
161	latch.countDown();	2×
162	}
163	}).setTimeout(timeout);	2×
164
165	request.exceptionHandler(e -> {	2×
166	EventManager.post(new ConnFailEvent("fetch config fail"));	!
167	LOGGER.error("Config update from {} failed. Error message is [{}].",	!
168	serviceUri,
169	e.getMessage());	!
170	latch.countDown();	!
171	});	!
172	request.end();	2×
173	});	2×
174	}	2×
175	}
176	}

apache / servicecomb-java-chassis / 5226

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous