This is an automated email from the ASF dual-hosted git repository. liubao pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-java-chassis.git
commit 0e227f6afb180200c3e886a2de89245b92f4a678 Author: weichao666 <[email protected]> AuthorDate: Thu Oct 18 17:36:56 2018 +0800 [SCB-967] do not affect original logic, and add configuration to take effect --- .../loadbalance/LoadbalanceHandler.java | 52 +++++++++++++--------- .../loadbalance/TestLoadBalanceHandler2.java | 8 +++- 2 files changed, 36 insertions(+), 24 deletions(-) diff --git a/handlers/handler-loadbalance/src/main/java/org/apache/servicecomb/loadbalance/LoadbalanceHandler.java b/handlers/handler-loadbalance/src/main/java/org/apache/servicecomb/loadbalance/LoadbalanceHandler.java index d8ae0e1..eab708c 100644 --- a/handlers/handler-loadbalance/src/main/java/org/apache/servicecomb/loadbalance/LoadbalanceHandler.java +++ b/handlers/handler-loadbalance/src/main/java/org/apache/servicecomb/loadbalance/LoadbalanceHandler.java @@ -30,10 +30,10 @@ import java.util.concurrent.atomic.AtomicInteger; import javax.ws.rs.core.Response.Status; import org.apache.commons.lang3.StringUtils; -import org.apache.servicecomb.core.CseContext; import org.apache.servicecomb.core.Endpoint; import org.apache.servicecomb.core.Handler; import org.apache.servicecomb.core.Invocation; +import org.apache.servicecomb.core.SCBEngine; import org.apache.servicecomb.core.Transport; import org.apache.servicecomb.core.exception.ExceptionUtils; import org.apache.servicecomb.core.provider.consumer.SyncResponseExecutor; @@ -69,6 +69,9 @@ public class LoadbalanceHandler implements Handler { public static final String SERVICECOMB_SERVER_ENDPOINT = "scb-endpoint"; + public static final boolean supportDefinedEndpoint = + DynamicPropertyFactory.getInstance().getBooleanProperty("servicecomb.loadbalance.userDefinedEndpoint.enabled", false).get(); + // just a wrapper to make sure in retry mode to choose a different server. class RetryLoadBalancer implements ILoadBalancer { // Enough times to make sure to choose a different server in high volume. @@ -176,23 +179,12 @@ public class LoadbalanceHandler implements Handler { @Override public void handle(Invocation invocation, AsyncResponse asyncResp) throws Exception { - String endpointUri = invocation.getLocalContext(SERVICECOMB_SERVER_ENDPOINT); - if (endpointUri != null) { - String endpointRule = "[a-z]+://.+"; - if (!endpointUri.matches(endpointRule)) { - throw new InvocationException(Status.BAD_REQUEST, - "the endpoint's format of the configuration is incorrect, e.g rest://127.0.0.1:8080"); + if (supportDefinedEndpoint) { + if (defineEndpointAndHandle(invocation, asyncResp)) { + return; } - URI formatUri = new URI(endpointUri); - Transport transport = CseContext.getInstance().getTransportManager().findTransport(formatUri.getScheme()); - if (transport == null) { - LOGGER.error("not deployed transport {}, ignore {}.", formatUri.getScheme(), endpointUri); - throw new InvocationException(Status.BAD_REQUEST, - "the endpoint's transport is not found."); - } - Endpoint endpoint = new Endpoint(transport, endpointUri); - invocation.setEndpoint(endpoint); } + String strategy = Configuration.INSTANCE.getRuleStrategyName(invocation.getMicroserviceName()); if (!isEqual(strategy, this.strategy)) { //配置变化，需要重新生成所有的lb实例 @@ -211,6 +203,26 @@ public class LoadbalanceHandler implements Handler { } } + private boolean defineEndpointAndHandle(Invocation invocation, AsyncResponse asyncResp) throws Exception { + String endpointUri = invocation.getLocalContext(SERVICECOMB_SERVER_ENDPOINT); + if (endpointUri == null) { + return false; + } + URI formatUri = new URI(endpointUri); + Transport transport = SCBEngine.getInstance().getTransportManager().findTransport(formatUri.getScheme()); + if (transport == null) { + LOGGER.error("not deployed transport {}, ignore {}.", formatUri.getScheme(), endpointUri); + throw new InvocationException(Status.BAD_REQUEST, + "the endpoint's transport is not found."); + } + Endpoint endpoint = new Endpoint(transport, endpointUri); + invocation.setEndpoint(endpoint); + invocation.next(resp -> { + asyncResp.handle(resp); + }); + return true; + } + private void clearLoadBalancer() { loadBalancerMap.clear(); } @@ -225,9 +237,7 @@ public class LoadbalanceHandler implements Handler { return; } chosenLB.getLoadBalancerStats().incrementNumRequests(server); - if (invocation.getEndpoint() == null) { - invocation.setEndpoint(server.getEndpoint()); - } + invocation.setEndpoint(server.getEndpoint()); invocation.next(resp -> { // this stats is for WeightedResponseTimeRule chosenLB.getLoadBalancerStats().noteResponseTime(server, (System.currentTimeMillis() - time)); @@ -335,9 +345,7 @@ public class LoadbalanceHandler implements Handler { ServiceCombServer server = (ServiceCombServer) s; chosenLB.getLoadBalancerStats().incrementNumRequests(s); invocation.setHandlerIndex(currentHandler); // for retry - if (invocation.getEndpoint() == null) { - invocation.setEndpoint(server.getEndpoint()); - } + invocation.setEndpoint(server.getEndpoint()); invocation.next(resp -> { if (isFailedResponse(resp)) { LOGGER.error("service {}, call error, msg is {}, server is {} ", diff --git a/handlers/handler-loadbalance/src/test/java/org/apache/servicecomb/loadbalance/TestLoadBalanceHandler2.java b/handlers/handler-loadbalance/src/test/java/org/apache/servicecomb/loadbalance/TestLoadBalanceHandler2.java index 51cb333..8ac41a7 100644 --- a/handlers/handler-loadbalance/src/test/java/org/apache/servicecomb/loadbalance/TestLoadBalanceHandler2.java +++ b/handlers/handler-loadbalance/src/test/java/org/apache/servicecomb/loadbalance/TestLoadBalanceHandler2.java @@ -27,6 +27,7 @@ import java.util.concurrent.TimeUnit; import org.apache.servicecomb.core.CseContext; import org.apache.servicecomb.core.Invocation; +import org.apache.servicecomb.core.SCBEngine; import org.apache.servicecomb.core.Transport; import org.apache.servicecomb.core.definition.MicroserviceMeta; import org.apache.servicecomb.core.definition.OperationMeta; @@ -54,6 +55,8 @@ import org.mockito.Mockito; public class TestLoadBalanceHandler2 { @BeforeClass public static void beforeClass() { + //prepare for defineEndpointAndHandle + ArchaiusUtils.setProperty("servicecomb.loadbalance.userDefinedEndpoint.enabled", "true"); // avoid mock ServiceCombLoadBalancerStats.INSTANCE.init(); } @@ -200,7 +203,7 @@ public class TestLoadBalanceHandler2 { } @Test - public void testConfiguredEndpoint() { + public void testConfigEndpoint() { ReferenceConfig referenceConfig = Mockito.mock(ReferenceConfig.class); OperationMeta operationMeta = Mockito.mock(OperationMeta.class); SchemaMeta schemaMeta = Mockito.mock(SchemaMeta.class); @@ -232,6 +235,7 @@ public class TestLoadBalanceHandler2 { Map<String, MicroserviceInstance> data = new HashMap<>(); DiscoveryTreeNode parent = new DiscoveryTreeNode().name("parent").data(data); CseContext.getInstance().setTransportManager(transportManager); + SCBEngine.getInstance().setTransportManager(transportManager); RegistryUtils.setServiceRegistry(serviceRegistry); @@ -270,7 +274,7 @@ public class TestLoadBalanceHandler2 { Assert.assertEquals("endpoint's format is not correct, throw exception", " but not throw exception"); } catch (Exception e) { Assert.assertTrue(e.getMessage() - .contains("the endpoint's format of the configuration is incorrect, e.g rest://127.0.0.1:8080")); + .contains("Illegal character in scheme name")); } //transport is not find
