ShenYu Gateway source code reading (six) Sofa request processing overview

Introduction to the

Today let’s explore the Sofa request processing process and see the similarities and differences between the previous HTTP and Dubbo

Sofa Example run

PS: If the request with parameters fails to run, please update the latest version. This problem has been fixed in the new version: blog.csdn.net/baidu_27627… Add sofa param resolve service

First run the official Sofa example, first start mysql and ZooKeeper, here using docker:

docker run -dit --name zk -p 2181:2181 zookeepe
docker run --name mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=123456 -d mysql:latest
Copy the code

Then run soul-admin, soul-bootst, and use sofa plugin in admin diagram

Run official example: soul-examples –> soul-examples-sofa

There is a pit here. It should be noted that there is no SOFA plug-in in the bootstrap print log after startup, and the request has been failed

o.d.s.w.configuration.SoulConfiguration : load plugin:[global] [org.dromara.soul.plugin.global.GlobalPlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[sign] [org.dromara.soul.plugin.sign.SignPlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[waf] [org.dromara.soul.plugin.waf.WafPlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[rate_limiter] [org.dromara.soul.plugin.ratelimiter.RateLimiterPlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[hystrix] [org.dromara.soul.plugin.hystrix.HystrixPlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[resilience4j] [org.dromara.soul.plugin.resilience4j.Resilience4JPlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[divide] [org.dromara.soul.plugin.divide.DividePlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[webClient] [org.dromara.soul.plugin.httpclient.WebClientPlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[divide] [org.dromara.soul.plugin.divide.websocket.WebSocketPlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[alibaba-dubbo-body-param] [org.dromara.soul.plugin.alibaba.dubbo.param.BodyParamPlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[dubbo] [org.dromara.soul.plugin.alibaba.dubbo.AlibabaDubboPlugin] o.d.s.w.configuration.SoulConfiguration :  load plugin:[monitor] [org.dromara.soul.plugin.monitor.MonitorPlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[response] [org.dromara.soul.plugin.httpclient.response.WebClientResponsePlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[response] [org.dromara.soul.plugin.alibaba.dubbo.response.DubboResponsePlugin] ```xml&ensp;&ensp;&ensp;&ensp;Java public SoulWebHandler SoulWebHandler (final ObjectProvider<List<SoulPlugin>> plugins) {
        List<SoulPlugin> pluginList = plugins.getIfAvailable(Collections::emptyList);
        final List<SoulPlugin> soulPlugins = pluginList.stream()
                .sorted(Comparator.comparingInt(SoulPlugin::getOrder)).collect(Collectors.toList());
        soulPlugins.forEach(soulPlugin -> log.info("load plugin:[{}] [{}]", soulPlugin.named(), soulPlugin.getClass().getName()));
        return new SoulWebHandler(soulPlugins);
    }
Copy the code

After exploration and elder brother’s discussion, it was found that sofa is used every day

We add the following dependencies to the pom.xml file of Bootstrap and restart it

        <! -- sofa plugin start-->
        <dependency>
            <groupId>com.alipay.sofa</groupId>
            <artifactId>sofa-rpc-all</artifactId>
            <version>5.7.6</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-client</artifactId>
            <version>4.0.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>4.0.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>4.0.1</version>
        </dependency>
        <dependency>
            <groupId>org.dromara</groupId>
            <artifactId>soul-spring-boot-starter-plugin-sofa</artifactId>
            <version>${project.version}</version>
        </dependency>
        <! -- sofa plugin end-->
Copy the code

Then I looked at the log print and found a SOFA related plug-in

o.d.s.w.configuration.SoulConfiguration : load plugin:[global] [org.dromara.soul.plugin.global.GlobalPlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[sign] [org.dromara.soul.plugin.sign.SignPlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[waf] [org.dromara.soul.plugin.waf.WafPlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[rate_limiter] [org.dromara.soul.plugin.ratelimiter.RateLimiterPlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[hystrix] [org.dromara.soul.plugin.hystrix.HystrixPlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[resilience4j] [org.dromara.soul.plugin.resilience4j.Resilience4JPlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[divide] [org.dromara.soul.plugin.divide.DividePlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[webClient] [org.dromara.soul.plugin.httpclient.WebClientPlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[divide] [org.dromara.soul.plugin.divide.websocket.WebSocketPlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[sofa-body-param] [org.dromara.soul.plugin.sofa.param.BodyParamPlugin] o.d.s.w.configuration.SoulConfiguration : The load plugins: [dubbo] [org. Dromara. Soul. Plugin. Alibaba. Dubbo. AlibabaDubboPlugin] / / new sofa o.d.s.w.configuration.SoulConfiguration : load plugin:[sofa] [org.dromara.soul.plugin.sofa.SofaPlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[monitor] [org.dromara.soul.plugin.monitor.MonitorPlugin] o.d.s.w.configuration.SoulConfiguration : load plugin:[response] [org.dromara.soul.plugin.alibaba.dubbo.response.DubboResponsePlugin] o.d.s.w.configuration.SoulConfiguration : The load plugins: [response] [org. Dromara. Soul. Plugin. Httpclient. Response. WebClientResponsePlugin] / / new sofa o.d.s.w.configuration.SoulConfiguration : load plugin:[response] [org.dromara.soul.plugin.sofa.response.SofaResponsePlugin]Copy the code

Metadata related to the successful loading of SOFA is also printed in the log

o.d.s.p.s.cache.ApplicationConfigCache   : init sofa reference success there meteData is :MetaData
o.d.s.p.s.cache.ApplicationConfigCache   : init sofa reference success there meteData is :MetaData
o.d.s.p.s.cache.ApplicationConfigCache   : init sofa reference success there meteData is :MetaData
Copy the code

Access the link: http://localhost:9195/sofa/findAll, successfully returns the following request

{
    "code": 200."message": "Access to success!"."data": {
        "name": "hello world Soul Sofa , findAll"."id": "998932133"}}Copy the code

The source code parsing

PS:Debug time process. This is normal

Start with the familiar pointcut function SoulWebHandler, set a breakpoint in the method below, and then step into each plugin to observe its behavior

        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) {
                        return this.execute(exchange);
                    }
                    return plugin.execute(exchange, this);
                }
                return Mono.empty();
            });
        }
Copy the code

GlobalPlugin

Go in there and perform the processing logic, which we learned from the previous analysis is basically to put the request type into Exchange

SignPlugin/WafPlugin/RateLimiterPlugin/HystrixPlugin/Resilience4JPlugin

Enter, but the plugin is not in use and does not execute logic

DividePlugin/WebClientPlugin/WebSocketPlugin

Skip the operation based on the type

BodyParamPlugin

This plugin is also executed in Dubbo, so let’s see what it does. It can be seen from the following logic that the request address should be replaced with the real back-end address after determining whether the execution condition is met

    public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
        final ServerHttpRequest request = exchange.getRequest();
        final SoulContext soulContext = exchange.getAttribute(Constants.CONTEXT);
        // Check whether the type is SOFA
        if (Objects.nonNull(soulContext) && RpcTypeEnum.SOFA.getName().equals(soulContext.getRpcType())) {
            MediaType mediaType = request.getHeaders().getContentType();
            ServerRequest serverRequest = ServerRequest.create(exchange, messageReaders);
            if (MediaType.APPLICATION_JSON.isCompatibleWith(mediaType)) {
                return body(exchange, serverRequest, chain);
            }
            if (MediaType.APPLICATION_FORM_URLENCODED.isCompatibleWith(mediaType)) {
                return formData(exchange, serverRequest, chain);
            }
            // Replace the path with that of the back-end server
            return query(exchange, serverRequest, chain);
        }
        return chain.execute(exchange);
    }

    private Mono<Void> query(final ServerWebExchange exchange, final ServerRequest serverRequest, final SoulPluginChain chain) {
        exchange.getAttributes().put(Constants.SOFA_PARAMS,
                HttpParamConverter.ofString(() -> serverRequest.uri().getQuery()));
        return chain.execute(exchange);
    }
Copy the code

Here is a very interesting phenomenon, in our fourth analysis, Dubbo also walks the same class, in the above function logic, we can see that it is not compatible with Dubbo, so how dubbo walks this class?

Through debugging we found that when starting dubbo and SOFA at the same time, will generate two BodyParamPlugin, the name is exactly the same, but the inside of the judgment type changed, very magical, guess this class is a means of dynamic generation, here first do not explore, can be studied behind

AlibabaDubboPlugin

Based on the type, skip

SofaPlugin

This is a core class by definition, so let’s see what it does. In the comments below, you can see that this is very similar to dubbo’s request. Route matching, successful RPC call, after the result is put into exchange

    # AbstractSoulPlugin
    public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
        String pluginName = named();
        final PluginData pluginData = BaseDataCache.getInstance().obtainPluginData(pluginName);
        if(pluginData ! =null && pluginData.getEnabled()) {
            final Collection<SelectorData> selectors = BaseDataCache.getInstance().obtainSelectorData(pluginName);
            if (CollectionUtils.isEmpty(selectors)) {
                return handleSelectorIsNull(pluginName, exchange, chain);
            }
            final SelectorData selectorData = matchSelector(exchange, selectors);
            if (Objects.isNull(selectorData)) {
                return handleSelectorIsNull(pluginName, exchange, chain);
            }
            selectorLog(selectorData, pluginName);
            final List<RuleData> rules = BaseDataCache.getInstance().obtainRuleData(selectorData.getId());
            if (CollectionUtils.isEmpty(rules)) {
                return handleRuleIsNull(pluginName, exchange, chain);
            }
            // Check whether any routing rules can be matched
            RuleData rule;
            if (selectorData.getType() == SelectorTypeEnum.FULL_FLOW.getCode()) {
                //get last
                rule = rules.get(rules.size() - 1);
            } else {
                rule = matchRule(exchange, rules);
            }
            if (Objects.isNull(rule)) {
                return handleRuleIsNull(pluginName, exchange, chain);
            }
            ruleLog(rule, pluginName);
            // Execute processing logic after matching
            return doExecute(exchange, chain, selectorData, rule);
        }
        return chain.execute(exchange);
    }

    # SofaPlugin
    protected Mono<Void> doExecute(final ServerWebExchange exchange, final SoulPluginChain chain, final SelectorData selector, final RuleData rule) {
        String body = exchange.getAttribute(Constants.SOFA_PARAMS);
        SoulContext soulContext = exchange.getAttribute(Constants.CONTEXT);
        assertsoulContext ! =null;
        MetaData metaData = exchange.getAttribute(Constants.META_DATA);
        if(! checkMetaData(metaData)) {assertmetaData ! =null;
            log.error(" path is :{}, meta data have error.... {}", soulContext.getPath(), metaData.toString());
            exchange.getResponse().setStatusCode(HttpStatus.INTERNAL_SERVER_ERROR);
            Object error = SoulResultWrap.error(SoulResultEnum.META_DATA_ERROR.getCode(), SoulResultEnum.META_DATA_ERROR.getMsg(), null);
            return WebFluxResultUtils.result(exchange, error);
        }
        if (StringUtils.isNoneBlank(metaData.getParameterTypes()) && StringUtils.isBlank(body)) {
            exchange.getResponse().setStatusCode(HttpStatus.INTERNAL_SERVER_ERROR);
            Object error = SoulResultWrap.error(SoulResultEnum.SOFA_HAVE_BODY_PARAM.getCode(), SoulResultEnum.SOFA_HAVE_BODY_PARAM.getMsg(), null);
            return WebFluxResultUtils.result(exchange, error);
        }
        // Get the result here, follow along
        final Mono<Object> result = sofaProxyService.genericInvoker(body, metaData, exchange);
        return result.then(chain.execute(exchange));
    }

    # SofaProxyService
    public Mono<Object> genericInvoker(final String body, final MetaData metaData, final ServerWebExchange exchange) throws SoulException {
        // Get the consumer in RPC based on the request path
        ConsumerConfig<GenericService> reference = ApplicationConfigCache.getInstance().get(metaData.getPath());
        if (Objects.isNull(reference) || StringUtils.isEmpty(reference.getInterfaceId())) {
            ApplicationConfigCache.getInstance().invalidate(metaData.getServiceName());
            reference = ApplicationConfigCache.getInstance().initRef(metaData);
        }
        GenericService genericService = reference.refer();
        Pair<String[], Object[]> pair;
        if (null == body || "".equals(body) || "{}".equals(body) || "null".equals(body)) {
            pair = new ImmutablePair<>(new String[]{}, new Object[]{});
        } else {
            pair = sofaParamResolveService.buildParameter(body, metaData.getParameterTypes());
        }
        CompletableFuture<Object> future = new CompletableFuture<>();
        RpcInvokeContext.getContext().setResponseCallback(new SofaResponseCallback<Object>() {
            @Override
            public void onAppResponse(final Object o, final String s, final RequestBase requestBase) {
                future.complete(o);
            }

            @Override
            public void onAppException(final Throwable throwable, final String s, final RequestBase requestBase) {
                future.completeExceptionally(throwable);
            }

            @Override
            public void onSofaException(final SofaRpcException e, final String s, final RequestBase requestBase) { future.completeExceptionally(e); }});// From the function name, you can guess the RPC call, then get the result, and put the result into exchange
        genericService.$invoke(metaData.getMethodName(), pair.getLeft(), pair.getRight());
        return Mono.fromFuture(future.thenApply(ret -> {
            if (Objects.isNull(ret)) {
                ret = Constants.SOFA_RPC_RESULT_EMPTY;
            }
            exchange.getAttributes().put(Constants.SOFA_RPC_RESULT, ret);
            exchange.getAttributes().put(Constants.CLIENT_RESPONSE_RESULT_TYPE, ResultEnum.SUCCESS.getName());
            return ret;
        })).onErrorMap(SoulException::new);
    }
Copy the code

MonitorPlugin

No skip, but the plug-in is not enabled

DubboResponsePlugin/WebClientResponsePlugin

Skip execution based on the type

SofaResponsePlugin

It can be guessed from the previous analysis and name that the response is returned to the client, as can be seen from the logic of the following code

    public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
        return chain.execute(exchange).then(Mono.defer(() -> {
            // Get the result from exchange
            final Object result = exchange.getAttribute(Constants.SOFA_RPC_RESULT);
            if (Objects.isNull(result)) {
                Object error = SoulResultWrap.error(SoulResultEnum.SERVICE_RESULT_ERROR.getCode(), SoulResultEnum.SERVICE_RESULT_ERROR.getMsg(), null);
                return WebFluxResultUtils.result(exchange, error);
            }
            Object success = SoulResultWrap.success(SoulResultEnum.SUCCESS.getCode(), SoulResultEnum.SUCCESS.getMsg(), JsonUtils.removeClass(result));
            // Familiar function that returns a response
            return WebFluxResultUtils.result(exchange, success);
        }));
    }
Copy the code

conclusion

The plugin flow is as follows:

• GlobalPlugin: Put the request type in
• SignPlugin: Skip not executing logic
• WafPlugin: Skip not executing logic
• RateLimiterPlugin: Skip not executing logic
• HystrixPlugin: Skip not executing logic
• Resilience4JPlugin: Skip logic execution
• DividePlugin: Skip not executing logic
• WebClientPlugin: Skip not executing logic
• WebSocketPlugin: Skip not executing logic
• BodyParamPlugin: Performs RPC request path substitution with the real server back-end path, acting like the dividePlugin; Different RPCS have this plug-in name associated with them, that is, there will be multiple BodyParamplugins
• AlibabaDubboPlugin: Skip not executing logic
• SofaPlugin: send the request to the backend server, get the result, write to Exchange
• MonitorPlugin: Skip not executing logic
• DubboResponsePlugin: Skip not executing logic
• WebClientResponsePlugin: Skip not executing logic
• SofaResponsePlugin: Takes the response from the Exchange and sends it to the client

After analyzing these articles, we further optimized our request flow for the Soul gateway as follows:

Updated our understanding of some of the classes in the process:

• From the previous analysis, the specific role of the GlobalPlugin is to put in the request type
• BodyParamPlugin is similar to dividePlugin in that it can match routes and then modify the path to the real back-end server path. It can also dynamically generate plugins with the same name for different RPC implementations

Soul Gateway source code analysis article list

Github

• Soul source reading (a) overview
• Soul source code reading (two) the initial run of the code
• HTTP request processing overview
• Dubbo Request Overview
• Soul Gateway source code reading (five) request type exploration
• Soul Gateway source code reading (6) Sofa request processing overview

The Denver nuggets

• Soul Gateway source code read (a) overview # nuggets article # juejin.cn/post/691786…
• Soul Gateway source code reading (two) initial run # nuggets article # juejin.cn/post/691786…
• Soul Gateway source code read (three) request processing overview # nuggets article # juejin.cn/post/691786…
• Dubbo request Overview # Nuggets article # juejin.cn/post/691786…
• Soul Gateway source code read (five) request types explore # nuggets article # juejin.cn/post/691857…