review

In the Soul Request Processing Overview overview article, we learned that Soul’s request-specific processing repository is excute in DefaultSoulPluginChain, which implements a plugin chain pattern to complete the processing of the request.

We summarized the plugins injected into the plugins in general, but even then, we still couldn’t see the whole picture. For this purpose, we specifically combed the classes involved in soul plug-in. The overall combing results are shown in the following figure.

In the carding article, we can see that the core classes are SoulPlugin, PluginEnum, PluginDataHandler and MetaDataSubscriber. In the articles related to carding request, we only need to focus on SoulPlugin and PluginEnum classes.

Now that we know about the SoulPlugin class, what is the main purpose of the PluginEnum enum class?

PluginEnum: An enumerated class of plug-ins

In fact, it is not difficult to find that the plugins of DefaultSoulPluginChain have a fixed execution order. Then, where is the execution order of this plug-in defined?

It can be traced back to the SoulConfiguration class

    public SoulWebHandler soulWebHandler(final ObjectProvider<List<SoulPlugin>> plugins) {
        / / to omit
        final List<SoulPlugin> soulPlugins = pluginList.stream()
               .sorted(Comparator.comparingInt(SoulPlugin::getOrder)).collect(Collectors.toList());
        return new SoulWebHandler(soulPlugins);
    }
Copy the code

After sorting out the entire PluginEnum class references, the following table shows the sequential relationship between plug-ins

In this review we’ve seen the general flow of soul processing requests

• 1. The GloBalPlugin plugin performs global initialization
• 2. Some plug-ins process requests based on rules such as authentication, traffic limiting, and fusing
• 3. Select an appropriate call method to assemble parameters and initiate the call.
• 4. Monitor
• 5. Process the result of the call

Request process combing

The following demo code screenshots from the soul – examples of HTTP demo, call interface address to http://127.0.0.1:9195/http/test/findByUserId? userId=10

Check the excute method in DefaultSoulPluginChain to see what classes an HTTP request call passed through.

public Mono<Void> execute(final ServerWebExchange exchange) {
            return Mono.defer(() -> {
                if (this.index < plugins.size()) {
                    SoulPlugin plugin = plugins.get(this.index++);
                    Boolean skip = plugin.skip(exchange);
                    if (skip) {
                        System.out.println("Skipped plug-in is"+plugin.getClass().getName().replace("org.dromara.soul.plugin.".""));
                        return this.execute(exchange);
                    }
                    System.out.println("Unskipped plug-ins are"+plugin.getClass().getName().replace("org.dromara.soul.plugin.".""));
                    return plugin.execute(exchange, this);
                }
                return Mono.empty();
            });
        }
Copy the code

The final output of the unskipped plug-in is as follows:

Did not skip the plugin for global. GlobalPlugin did not skip the plug-in to sign SignPlugin did not skip the plugin for the waf. WafPlugin did not skip the plugin for ratelimiter. RateLimiterPlugin Did not skip the plugin for hystrix. HystrixPlugin did not skip the plugin for resilience4j. Resilience4JPlugin did not skip the plugin to divide DividePlugin Did not skip the plugin for httpclient. WebClientPlugin did not skip the plugin for alibaba. Dubbo. Param. BodyParamPlugin did not skip the plugin for the monitor. The MonitorPlugin Did not skip the plugin for httpclient. Response. WebClientResponsePlugin

. Here is a little confused, why the alibaba dubbo. Param. BodyParamPlugin plug-ins can be executed, ignored, late tracking.

We found that the general flow of the plug-in for a gateway call to an HTTP request was consistent with what we expected. At present, we only choose the key points, namely GlobalPlugin, DividePlugin, WebClientPlugin, WebClientResponsePlugin.

Issue a Debug call to trace the action of the four plug-ins in turn.

The GlobalPlugin SoulContext object encapsulates the plug-in

The Excute method for the GlobalPlugin plugin is shown below

public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
        final ServerHttpRequest request = exchange.getRequest();
        final HttpHeaders headers = request.getHeaders();
        final String upgrade = headers.getFirst("Upgrade");
        SoulContext soulContext;
        if (StringUtils.isBlank(upgrade) || !"websocket".equals(upgrade)) {
            soulContext = builder.build(exchange);
        } else {
            final MultiValueMap<String, String> queryParams = request.getQueryParams();
            soulContext = transformMap(queryParams);
        }
        exchange.getAttributes().put(Constants.CONTEXT, soulContext);
        return chain.execute(exchange);
    }
Copy the code

As you can see, the main purpose of GlobalPlugin’s Excute method is to encapsulate a SoulContext object and place it in Exchange. (Exchange objects are shared objects along the chain of plug-ins, and one plug-in executes and passes them to the next. I understand it as a ThreadLocal object.

So what are the properties in the SoulContext object?

After the GlobalPlugin plugin is executed, the final wrapped SoulContext object is shown below.

The parameters of other RPCType SoulContext encapsulation can view DefaultSoulContextBuilder build method of tracking, because this title this paper traces the HTTP calls, so here not discuss excess.

DividePlugin A routing plugin

After executing the GlobalPlugin plugin, a SoulContext object is finally encapsulated and placed in ServerWebExchange for use by the downstream call chain.

Now let’s see what role the DividePlugin plays in the chain call process.

AbstractSoulPlugin

Tracing the source code shows that the DividePlugin extends from the AbstractSoulPlugin class, which implements the SoulPlugin interface.

So what kind of extensions does AbstractSoulPlugin make? Let’s comb through the methods of this class.

Take a look at the Excute method in action

public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
        String pluginName = named();
        // Get the corresponding plug-in
        final PluginData pluginData = BaseDataCache.getInstance().obtainPluginData(pluginName);
        // Check whether the plug-in is enabled
        if(pluginData ! =null && pluginData.getEnabled()) {
            // Get all the selectors under the plug-in
            final Collection<SelectorData> selectors = BaseDataCache.getInstance().obtainSelectorData(pluginName);
            if (CollectionUtils.isEmpty(selectors)) {
                return handleSelectorIsNull(pluginName, exchange, chain);
            }
            // Match selector
            final SelectorData selectorData = matchSelector(exchange, selectors);
            if (Objects.isNull(selectorData)) {
                return handleSelectorIsNull(pluginName, exchange, chain);
            }
            // Prints the selector log
            selectorLog(selectorData, pluginName);
            final List<RuleData> rules = BaseDataCache.getInstance().obtainRuleData(selectorData.getId());
            if (CollectionUtils.isEmpty(rules)) {
                return handleRuleIsNull(pluginName, exchange, chain);
            }
            RuleData rule;
            if (selectorData.getType() == SelectorTypeEnum.FULL_FLOW.getCode()) {
                rule = rules.get(rules.size() - 1);
            } else {
                // Match rules
                rule = matchRule(exchange, rules);
            }
            if (Objects.isNull(rule)) {
                returnhandleRu! [](https://p9-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/f523f655f0014d288b7a4502cc6a08d1~tplv-k3u1fbpfcp-watermark.image)leIsNull(pl uginName, exchange, chain);
            }
            // Prints rule logs
            ruleLog(rule, pluginName);
            // Implement the subclass implementation
            return doExecute(exchange, chain, selectorData, rule);
        }
        return chain.execute(exchange);
    }
Copy the code

The final flow chart is as follows:

Ps: No specific method-level processing is detailed in the above flowchart.

But there are a few points that need to be highlighted:

• 1. Plug-in data, selector data, and rule data are all obtained from BaseDataCache, which is the class that is ultimately affected during data synchronization.
• In the global proxy mode, only one selector rule (which refers to all interfaces of the proxy) will be registered, so the corresponding processing here is rule-.size ()-1.
• 3. The actual processing of selector and rule selection is much more complicated. Considering that it is to introduce the general logic of a request process, it will not be elaborated here. The corresponding page is as follows:

AbstractSoulPlugin exeute method AbstractSoulPlugin exeute method AbstractSoulPlugin exeute method AbstractSoulPlugin exexcute method

Let’s take a look at what DividePlugin’s Doexcute method does.

DividePlugin

protected Mono<Void> doExecute(final ServerWebExchange exchange, final SoulPluginChain chain, final SelectorData selector, final RuleData rule) {
        final SoulContext soulContext = exchange.getAttribute(Constants.CONTEXT);
        assertsoulContext ! =null;
        // Get rule processing data
        final DivideRuleHandle ruleHandle = GsonUtils.getInstance().fromJson(rule.getHandle(), DivideRuleHandle.class);
        // Gets the address of the injection under this selector
        final List<DivideUpstream> upstreamList = UpstreamCacheManager.getInstance().findUpstreamListBySelectorId(selector.getId());
        if (CollectionUtils.isEmpty(upstreamList)) {
            log.error("divide upstream configuration error： {}", rule.toString());
            Object error = SoulResultWrap.error(SoulResultEnum.CANNOT_FIND_URL.getCode(), SoulResultEnum.CANNOT_FIND_URL.getMsg(), null);
            return WebFluxResultUtils.result(exchange, error);
        }
        final String ip = Objects.requireNonNull(exchange.getRequest().getRemoteAddress()).getAddress().getHostAddress();
        // Select an ADDRESS based on the load balancing policy corresponding to the rule
        DivideUpstream divideUpstream = LoadBalanceUtils.selector(upstreamList, ruleHandle.getLoadBalance(), ip);
        if (Objects.isNull(divideUpstream)) {
            log.error("divide has no upstream");
            Object error = SoulResultWrap.error(SoulResultEnum.CANNOT_FIND_URL.getCode(), SoulResultEnum.CANNOT_FIND_URL.getMsg(), null);
            return WebFluxResultUtils.result(exchange, error);
        }
        // set the http url
        String domain = buildDomain(divideUpstream);
        // Assemble the real call address
        String realURL = buildRealURL(domain, soulContext, exchange);
        exchange.getAttributes().put(Constants.HTTP_URL, realURL);
        // Set the timeout period and retry times
        exchange.getAttributes().put(Constants.HTTP_TIME_OUT, ruleHandle.getTimeout());
        exchange.getAttributes().put(Constants.HTTP_RETRY, ruleHandle.getRetry());
        return chain.execute(exchange);
    }
Copy the code

The general logic after combing through the above code is as follows:

• 1. Obtain the registration address of the selector

• 2. Obtain the load balancing policy based on the Handle field of the rule, select the actual call address (LoadBalanceUtils), retry times, and timeout period.

• 3. Send the actual call address, timeout period, and retry times to ServerWebExchange for use by the downstream call chain.

The debug demo:Ps: Where do we not see the parameters in the theme logic above? So where is this parameter encapsulated? The answer inBuildRealURL methodFrom theexchangeRetrieved from the context.

WebClientPlugin Http request invocation plug-in

Now let’s see how does Soul initiate a request call

public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
        final SoulContext soulContext = exchange.getAttribute(Constants.CONTEXT);
        assertsoulContext ! =null;
        // Get the real address
        String urlPath = exchange.getAttribute(Constants.HTTP_URL);
        if (StringUtils.isEmpty(urlPath)) {
            Object error = SoulResultWrap.error(SoulResultEnum.CANNOT_FIND_URL.getCode(), SoulResultEnum.CANNOT_FIND_URL.getMsg(), null);
            return WebFluxResultUtils.result(exchange, error);
        }
        // Get the timeout period
        long timeout = (long) Optional.ofNullable(exchange.getAttribute(Constants.HTTP_TIME_OUT)).orElse(3000L);
        // Get the number of retries
        int retryTimes = (int) Optional.ofNullable(exchange.getAttribute(Constants.HTTP_RETRY)).orElse(0);
        log.info("The request urlPath is {}, retryTimes is {}", urlPath, retryTimes);
        HttpMethod method = HttpMethod.valueOf(exchange.getRequest().getMethodValue());
        WebClient.RequestBodySpec requestBodySpec = webClient.method(method).uri(urlPath);
        return handleRequestBody(requestBodySpec, exchange, timeout, retryTimes, chain);
    }
Copy the code

There are three main things you do in webClient’s Excute method

• 1. Pull out the properties put into Exchange from Divide plug-in, the real address of the call, timeout, number of retries.
• 2. Encapsulates a RequestBodySpec object (not aware of this reactive programming thing)
• 3. A handleRequestBody method is called

See the handleRequestBody method first

private Mono<Void> handleRequestBody(final WebClient.RequestBodySpec requestBodySpec,
                                         final ServerWebExchange exchange,
                                         final long timeout,
                                         final int retryTimes,
                                         final SoulPluginChain chain) {
        return requestBodySpec.headers(httpHeaders -> {
            httpHeaders.addAll(exchange.getRequest().getHeaders());
            httpHeaders.remove(HttpHeaders.HOST);
        })
                .contentType(buildMediaType(exchange))
                .body(BodyInserters.fromDataBuffers(exchange.getRequest().getBody()))
                .exchange()
                // Failed to print logs
                .doOnError(e -> log.error(e.getMessage()))
                // Set the timeout period
                .timeout(Duration.ofMillis(timeout))
                // Set the actual request retry
                .retryWhen(Retry.onlyIf(x -> x.exception() instanceof ConnectTimeoutException)
                    .retryMax(retryTimes)
                    .backoff(Backoff.exponential(Duration.ofMillis(200), Duration.ofSeconds(20), 2.true)))
                // Process the request after it ends
                .flatMap(e -> doNext(e, exchange, chain));

    }
Copy the code

In this method, it can be roughly interpreted as

• The request header in Exchange is placed in the request header for this call
• Set the contentType
• Setting timeout
• Setting failure response
• Set the retry scenario and retry times
• Processing of final results.

You also need to look at the doNext method in the process

The general logic is to determine whether the request was successful and put the result of the request into Exchange for the downstream plug-in to process.

private Mono<Void> doNext(final ClientResponse res, final ServerWebExchange exchange, final SoulPluginChain chain) {
        if (res.statusCode().is2xxSuccessful()) {
            exchange.getAttributes().put(Constants.CLIENT_RESPONSE_RESULT_TYPE, ResultEnum.SUCCESS.getName());
        } else {
            exchange.getAttributes().put(Constants.CLIENT_RESPONSE_RESULT_TYPE, ResultEnum.ERROR.getName());
        }
        exchange.getAttributes().put(Constants.CLIENT_RESPONSE_ATTR, res);
        return chain.execute(exchange);
    }
Copy the code

Ps: The fact that we don’t understand responsive programming doesn’t stop us from reading code.

WebClientResponsePlugin Http result handling plug-in

The excute method of this implementation has no core logic except to determine the request status code and return it to the front end in different data formats based on the status code.

public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
        return chain.execute(exchange).then(Mono.defer(() -> {
            ServerHttpResponse response = exchange.getResponse();
            ClientResponse clientResponse = exchange.getAttribute(Constants.CLIENT_RESPONSE_ATTR);
            if (Objects.isNull(clientResponse)
                    || response.getStatusCode() == HttpStatus.BAD_GATEWAY
                    || response.getStatusCode() == HttpStatus.INTERNAL_SERVER_ERROR) {
                Object error = SoulResultWrap.error(SoulResultEnum.SERVICE_RESULT_ERROR.getCode(), SoulResultEnum.SERVICE_RESULT_ERROR.getMsg(), null);
                return WebFluxResultUtils.result(exchange, error);
            }
            if (response.getStatusCode() == HttpStatus.GATEWAY_TIMEOUT) {
                Object error = SoulResultWrap.error(SoulResultEnum.SERVICE_TIMEOUT.getCode(), SoulResultEnum.SERVICE_TIMEOUT.getMsg(), null);
                return WebFluxResultUtils.result(exchange, error);
            }
            response.setStatusCode(clientResponse.statusCode());
            response.getCookies().putAll(clientResponse.cookies());
            response.getHeaders().putAll(clientResponse.headers().asHttpHeaders());
            return response.writeWith(clientResponse.body(BodyExtractors.toDataBuffers()));
        }));
    }
Copy the code

conclusion

At this point, the process of invoking an Http request based on the Soul Gateway is basically over.

Sort out the HTTP request invocation process

• The Global plug-in encapsulates the SoulContext object
• The front plug-in handles fuses and current limiting authentication.
• Divide plug-in selects the actual address for the call, number of retries, and timeout.
• The WebClient plug-in makes the actual Http call
• The WebClientResponse plug-in processes the result and returns to the foreground.

Based on the general flow of Http calls, we can roughly guess the process based on other RPC calls, which is to replace the plug-in that initiates the request and the plug-in that returns the result processing.

In the above article we also mentioned the selection of routing rules LoadBalanceUtils, selector and rule processing MatchStrategy.

After that, a new chapter will be opened to reveal the mystery of RPC generalization call, routing, selector and rule matching step by step.