[Question] How to get RequestBody when defined plugins in mysql mode

[ColdFragrance]

Question

In ShenYu-2.5.0, I defined a plugin as follow

protected Mono doExecute(final ServerWebExchange exchange, final ShenyuPluginChain chain, final SelectorData selector, final RuleData rule) {
Flux<DataBuffer> fluxMap = exchange.getRequest().getBody();
    fluxMap.subscribe(dataBuffer -> {
    	InputStream inputStream = dataBuffer.asInputStream();
        try {
			byte[] bytes = new byte[0];
			bytes = new byte[inputStream.available()];
			inputStream.read(bytes);
			String body = new String(bytes);
	        log.info("请求body信息 >> {}", body);
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			if(null != inputStream) {
				try {
					inputStream.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
    });   
    return chain.execute(exchange);
}

And, request command is

curl -X POST http://localhost:9077/flow/one?n=999 -H "Content-Type: application/json" -H "X-Access-Token: eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyTmFtZSI6ImFkbWluIiwiZXhwIjoxNjc4MzI0ODAyfQ.9QjlqXGYpXFXR70w3RFCFJ2dcwbrnrY2gTHyiVyF5io" -d "{"a":"1","b":"2","c":"3"}"

When I use oracle model via changing spring.profiles.active to 'oracle' in application.yaml, it is works.But use mysql model,it cannot enter fluxMap.subscribe(dataBuffer -> {}) fragment.
My Questions:
1.How to get RequestBody?
2.What's the difference between oracle and mysql?

Addtionally,Shenyu-Admin is works when mysql model,but not works when oracle model,such as

org.springframework.dao.DuplicateKeyException: ### Error updating database. Cause: java.sql.SQLIntegrityConstraintViolationException: ORA-00001: 违反唯一约束条件 (WORKFLOW_USER.SYS_C0026987) ### The error may exist in class path resource [mappers/operation-record-log-sqlmap.xml] ### The error may involve org.apache.shenyu.admin.mapper.OperationRecordLogMapper.insert-Inline ### The error occurred while setting parameters ### SQL: insert into operation_record_log( id, color, context, operator, operation_time, operation_type) values ( ?, ?, ?, ?, ?, ? ) ### Cause: java.sql.SQLIntegrityConstraintViolationException: ORA-00001: 违反唯一约束条件 (WORKFLOW_USER.SYS_C0026987) ; ORA-00001: 违反唯一约束条件 (WORKFLOW_USER.SYS_C0026987) ; nested exception is java.sql.SQLIntegrityConstraintViolationException: ORA-00001: 违反唯一约束条件 (WORKFLOW_USER.SYS_C0026987)

[moremind]

this is not related to database

[moremind]

you can get request body like this:

shenyu/shenyu-plugin/shenyu-plugin-sign/src/main/java/org/apache/shenyu/plugin/sign/SignPlugin.java

Line 72 in d764f9d

protected Mono<Void> doExecute(final ServerWebExchange exchange, final ShenyuPluginChain chain, final SelectorData selectorData, final RuleData rule) {

you get request body, must put it to exchange.

[li-keguo]

You should provide detailed procedures and logs

[ColdFragrance]

Well, I want to get RequestBody by Synchronized way so that I can modify it, but fluxMap.subscribe(dataBuffer -> {}) is Asynchronized way.Can you provide a detailed Synchronized way that help me?

[ColdFragrance]

My procedure as follow
`
@slf4j
public class MyRequestPlugin extends AbstractShenyuPlugin {

private final List<HttpMessageReader<?>> messageReaders;

@Autowired
private RedisTemplate redisTemplate;

public MyRequestPlugin(final List<HttpMessageReader<?>> messageReaders) {
    this.messageReaders = messageReaders;
}
private static final List<HttpMessageReader<?>> MESSAGE_READERS = HandlerStrategies.builder().build().messageReaders();

@Override
protected Mono<Void> doExecute(final ServerWebExchange exchange, final ShenyuPluginChain chain, final SelectorData selector, final RuleData rule) {
    FlowRuleHandler ruleHandle = MyRequestPluginDataHandler.CACHED_HANDLE.get().obtainHandle(CacheKeyUtils.INST.getKey(rule));
    if (Objects.isNull(ruleHandle)) {
        log.error("\r\nCryptor request rule configuration is null :{}", rule.getId());
    }

    Map<String, String> headMap = exchange.getRequest().getHeaders().toSingleValueMap();
    log.info("请求Head信息 >> {}", headMap);
    Mono<MultiValueMap<String, String>> formData = exchange.getFormData();
    //获取form-data中的内容
    formData.subscribe(map -> {
        Map<String, String> formMap = map.toSingleValueMap();
        log.info("form-data信息 >> {}", formMap);
    });
    Map<String, String> queryParams = exchange.getRequest().getQueryParams().toSingleValueMap();
    log.info("请求queryParams信息 >> {}", queryParams);
    URI uri = exchange.getRequest().getURI();
    log.info("请求URI >> {}", uri);
    log.info("messageReaders >> {}", messageReaders);
    String iniId = headMap.get("Ini-Id");
    
	// How to get RequestBody by Synchronized way?
	ServerRequest serverRequest = ServerRequest.create(exchange, messageReaders);
	System.out.println(serverRequest);
	Builder b = ServerRequest.from(serverRequest);
	Builder bb = b.body(exchange.getRequest().getBody());
	System.out.println(b);
	System.out.println(bb);
	        
    return chain.execute(exchange);
}

@Override
public int getOrder() {
    return PluginEnum.REQUEST.getCode();
}

@Override
public String named() {
    return PluginEnum.REQUEST.getName();
}

}
`

[moremind]

reactor of spring-webflux is asynchronous,but you can get request body as follows:

ServerRequest serverRequest = ServerRequest.create(exchange, MESSAGE_READERS);
            Mono<String> mono = serverRequest.bodyToMono(String.class)
                    .switchIfEmpty(Mono.defer(() -> Mono.just("")))
                    .flatMap(originalBody -> signBody(originalBody, exchange));

            BodyInserter<Mono<String>, ReactiveHttpOutputMessage> bodyInserter = BodyInserters.fromPublisher(mono, String.class);
            CachedBodyOutputMessage outputMessage = ResponseUtils.newCachedBodyOutputMessage(exchange);
            return bodyInserter.insert(outputMessage, new BodyInserterContext())
                    .then(Mono.defer(() -> {
                        ServerHttpRequestDecorator decorator = new SignRequestDecorator(exchange, outputMessage);
                        return chain.execute(exchange.mutate().request(decorator).build());
                    })).onErrorResume((Function<Throwable, Mono<Void>>) throwable -> ResponseUtils.release(outputMessage, throwable));

[ColdFragrance]

But flatMap(originalBody -> signBody(originalBody, exchange)) is ASYNC.Is it works？