|
|
|
|
@@ -122,6 +122,8 @@ public class RequestInput extends RPCInput implements IInput{
|
|
|
|
|
protected void doRequestMapping(InputConfig aConfig, InputContext inputContext) {
|
|
|
|
|
RequestInputConfig config = (RequestInputConfig) aConfig;
|
|
|
|
|
|
|
|
|
|
Map<String, Object> params = new HashMap<>();
|
|
|
|
|
synchronized (inputContext.getStepContext()) {
|
|
|
|
|
// 把请求信息放入stepContext
|
|
|
|
|
Map<String, Object> group = new HashMap<>();
|
|
|
|
|
group.put("request", request);
|
|
|
|
|
@@ -131,15 +133,16 @@ public class RequestInput extends RPCInput implements IInput{
|
|
|
|
|
HttpMethod method = HttpMethod.valueOf(config.getMethod().toUpperCase());
|
|
|
|
|
request.put("method", method);
|
|
|
|
|
|
|
|
|
|
Map<String, Object> params = new HashMap<>();
|
|
|
|
|
params.putAll(MapUtil.toHashMap(config.getQueryParams()));
|
|
|
|
|
request.put("params", params);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
ONode ctxNode = null;
|
|
|
|
|
// 数据转换
|
|
|
|
|
if (inputContext != null && inputContext.getStepContext() != null) {
|
|
|
|
|
StepContext<String, Object> stepContext = inputContext.getStepContext();
|
|
|
|
|
ctxNode = PathMapping.toONode(stepContext);
|
|
|
|
|
synchronized (stepContext) {
|
|
|
|
|
Map<String, Object> dataMapping = this.getConfig().getDataMapping();
|
|
|
|
|
if (dataMapping != null) {
|
|
|
|
|
Map<String, Object> requestMapping = (Map<String, Object>) dataMapping.get("request");
|
|
|
|
|
@@ -199,6 +202,7 @@ public class RequestInput extends RPCInput implements IInput{
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (config.isNewVersion()) {
|
|
|
|
|
String host = config.getServiceName();
|
|
|
|
|
@@ -223,13 +227,17 @@ public class RequestInput extends RPCInput implements IInput{
|
|
|
|
|
UriComponents uriComponents = UriComponentsBuilder.fromUriString(sb.toString())
|
|
|
|
|
.queryParams(MapUtil.toMultiValueMap(params)).build();
|
|
|
|
|
|
|
|
|
|
synchronized (inputContext.getStepContext()) {
|
|
|
|
|
request.put("url", uriComponents.toUriString());
|
|
|
|
|
}
|
|
|
|
|
} else {
|
|
|
|
|
UriComponents uriComponents = UriComponentsBuilder.fromUriString(config.getBaseUrl() + setPathVariable(ctxNode, config.getPath()))
|
|
|
|
|
.queryParams(MapUtil.toMultiValueMap(params)).build();
|
|
|
|
|
synchronized (inputContext.getStepContext()) {
|
|
|
|
|
request.put("url", uriComponents.toUriString());
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
private String setPathVariable(ONode ctxNode, String path) {
|
|
|
|
|
if (ctxNode == null || StringUtils.isBlank(path)) {
|
|
|
|
|
@@ -277,14 +285,16 @@ public class RequestInput extends RPCInput implements IInput{
|
|
|
|
|
ct = CONTENT_TYPE_JSON;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
synchronized (inputContext.getStepContext()) {
|
|
|
|
|
response.put("body", this.parseBody(ct, (String)responseBody));
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// 数据转换
|
|
|
|
|
if (inputContext != null && inputContext.getStepContext() != null) {
|
|
|
|
|
StepContext<String, Object> stepContext = inputContext.getStepContext();
|
|
|
|
|
if (!CollectionUtils.isEmpty(responseMapping)) {
|
|
|
|
|
ONode ctxNode = PathMapping.toONode(stepContext);
|
|
|
|
|
|
|
|
|
|
synchronized (stepContext) {
|
|
|
|
|
// headers
|
|
|
|
|
Map<String, Object> fixedHeaders = MapUtil.upperCaseKey((Map<String, Object>) responseMapping.get("fixedHeaders"));
|
|
|
|
|
Map<String, Object> headerMapping = MapUtil.upperCaseKey((Map<String, Object>) responseMapping.get("headers"));
|
|
|
|
|
@@ -332,6 +342,7 @@ public class RequestInput extends RPCInput implements IInput{
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@Override
|
|
|
|
|
protected Mono<RPCResponse> getClientSpecFromContext(InputConfig aConfig, InputContext inputContext) {
|
|
|
|
|
@@ -385,12 +396,16 @@ public class RequestInput extends RPCInput implements IInput{
|
|
|
|
|
|
|
|
|
|
headers.remove(CommonConstants.HEADER_CONTENT_LENGTH);
|
|
|
|
|
headers.add(systemConfig.fizzTraceIdHeader(), inputContext.getStepContext().getTraceId());
|
|
|
|
|
synchronized (inputContext.getStepContext()) {
|
|
|
|
|
request.put("headers", MapUtil.headerToHashMap(headers));
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
Object body = null;
|
|
|
|
|
if (CONTENT_TYPE_XML.equals(reqContentType) || CONTENT_TYPE_TEXT_XML.equals(reqContentType)) {
|
|
|
|
|
// convert JSON to XML if it is XML content type
|
|
|
|
|
synchronized (inputContext.getStepContext()) {
|
|
|
|
|
request.put("jsonBody", request.get("body"));
|
|
|
|
|
}
|
|
|
|
|
String jsonStr = null;
|
|
|
|
|
if (TypeUtils.isBasicType(request.get("body"))) {
|
|
|
|
|
jsonStr = request.get("body").toString();
|
|
|
|
|
@@ -404,7 +419,9 @@ public class RequestInput extends RPCInput implements IInput{
|
|
|
|
|
} else {
|
|
|
|
|
body = jsonStr;
|
|
|
|
|
}
|
|
|
|
|
synchronized (inputContext.getStepContext()) {
|
|
|
|
|
request.put("body", body);
|
|
|
|
|
}
|
|
|
|
|
LOGGER.info("body={}", body);
|
|
|
|
|
LOGGER.info("headers={}", JSON.toJSONString(headers));
|
|
|
|
|
} else if (CONTENT_TYPE_MULTIPART_FORM_DATA.equals(reqContentType)) {
|
|
|
|
|
@@ -460,6 +477,7 @@ public class RequestInput extends RPCInput implements IInput{
|
|
|
|
|
headers.put("ELAPSEDTIME", elapsedMillis + "ms");
|
|
|
|
|
|
|
|
|
|
RequestRPCResponse reqCr = (RequestRPCResponse) cr;
|
|
|
|
|
synchronized (inputContext.getStepContext()) {
|
|
|
|
|
if (reqCr.getStatusCode() != null) {
|
|
|
|
|
this.response.put("httpStatus", reqCr.getStatusCode().value());
|
|
|
|
|
}
|
|
|
|
|
@@ -468,6 +486,7 @@ public class RequestInput extends RPCInput implements IInput{
|
|
|
|
|
inputContext.getStepContext().addElapsedTime(prefix + request.get("url"),
|
|
|
|
|
elapsedMillis);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
protected Mono<Object> bodyToMono(ClientResponse cr){
|
|
|
|
|
return cr.bodyToMono(String.class);
|
|
|
|
|
}
|
|
|
|
|
@@ -475,9 +494,11 @@ public class RequestInput extends RPCInput implements IInput{
|
|
|
|
|
protected void doOnBodyError(Throwable ex, long elapsedMillis) {
|
|
|
|
|
LogService.setBizId(inputContext.getStepContext().getTraceId());
|
|
|
|
|
LOGGER.warn("failed to call {}", request.get("url"), ex);
|
|
|
|
|
synchronized (inputContext.getStepContext()) {
|
|
|
|
|
inputContext.getStepContext().addElapsedTime(
|
|
|
|
|
stepResponse.getStepName() + "-" + "调用接口 failed " + request.get("url"), elapsedMillis);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Parse response body according to content-type header
|
|
|
|
|
public Object parseBody(String contentType, String responseBody) {
|
|
|
|
|
|