Soul源码阅读 sofa【第十七天】

2021-02-03  本文已影响0人  cutieagain

sofa环境配置查看

Soul源码阅读 体验sofa代理【第四天】

sofa调用流程

GlobalPlugin -> BodyParamPlugin[org.dromara.soul.plugin.sofa.param] -> AbstractSoulPlugin -> SofaPlugin -> SofaResponsePlugin -> 结束

代码解析

GlobalPlugin 全局插件,调用所有插件之前使用该插件封装soulContext

@Override
    // 全局插件,构造SoulContext
    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;
        // 升级不为空,或者websocket不等于升级
        if (StringUtils.isBlank(upgrade) || !"websocket".equals(upgrade)) {
            // 封装元数据信息
            soulContext = builder.build(exchange);
        } else {
            // 获取请求参数
            final MultiValueMap<String, String> queryParams = request.getQueryParams();
            // 封装参数进soulContext
            soulContext = transformMap(queryParams);
        }
        exchange.getAttributes().put(Constants.CONTEXT, soulContext);
        return chain.execute(exchange);
    }

BodyParamPlugin,sofa插件参数解析

@Override
    // sofa参数解析
    public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
        // 获取请求
        final ServerHttpRequest request = exchange.getRequest();
        // 获取soulContext
        final SoulContext soulContext = exchange.getAttribute(Constants.CONTEXT);
        // soulContext不为空且是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)) {
                // json请求方式参数封装
                return body(exchange, serverRequest, chain);
            }
            if (MediaType.APPLICATION_FORM_URLENCODED.isCompatibleWith(mediaType)) {
                // 表单请求方式参数封装
                return formData(exchange, serverRequest, chain);
            }
            // get方式的直接解析封装
            return query(exchange, serverRequest, chain);
        }
        return chain.execute(exchange);
    }

SofaPlugin,调用sofa主要插件

@Override
    // sofa请求执行
    protected Mono<Void> doExecute(final ServerWebExchange exchange, final SoulPluginChain chain, final SelectorData selector, final RuleData rule) {
        // 获取sofa处理好的参数
        String body = exchange.getAttribute(Constants.SOFA_PARAMS);
        // 获取soulContext
        SoulContext soulContext = exchange.getAttribute(Constants.CONTEXT);
        assert soulContext != null;
        // 获取元数据信息
        MetaData metaData = exchange.getAttribute(Constants.META_DATA);
        // 检查元数据
        if (!checkMetaData(metaData)) {
            // 返回错误信息
            assert metaData != 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);
        }
        // 代理sofa执行
        final Mono<Object> result = sofaProxyService.genericInvoker(body, metaData, exchange);
        return result.then(chain.execute(exchange));
    }

SofaProxyService,sofa服务调用执行主要服务

    // sofa泛化调用
    public Mono<Object> genericInvoker(final String body, final MetaData metaData, final ServerWebExchange exchange) throws SoulException {
        // 根据元数据中的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 {
            // 根据body和参数类型构造请求参数
            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);
            }
        });
        // 执行sofa请求
        genericService.$genericInvoke(metaData.getMethodName(), pair.getLeft(), pair.getRight());
        return Mono.fromFuture(future.thenApply(ret -> {
            if (Objects.isNull(ret)) {
                ret = Constants.SOFA_RPC_RESULT_EMPTY;
            }

            GenericObject genericObject = (GenericObject) ret;
            exchange.getAttributes().put(Constants.SOFA_RPC_RESULT, genericObject.getFields());
            exchange.getAttributes().put(Constants.CLIENT_RESPONSE_RESULT_TYPE, ResultEnum.SUCCESS.getName());
            return ret;
        })).onErrorMap(SoulException::new);
    }

SofaResponsePlugin

    @Override
    // 封装返回结果
    public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
        return chain.execute(exchange).then(Mono.defer(() -> {
            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));
            return WebFluxResultUtils.result(exchange, success);
        }));
    }
上一篇下一篇

猜你喜欢

热点阅读