Blocking code in reactive package

[bekirtaskin]

In EventuateCommonReactiveJdbcOperations class, There is a block call in columnToJson method.
`public String columnToJson(EventuateSchema eventuateSchema, String column) {

BiFunction<String, List<Object>, List<Map<String, Object>>> selectCallback =
        (sql, params) -> reactiveJdbcStatementExecutor
                .query(sql, params.toArray())
                .collectList()
                .block(Duration.ofMillis(blockingTimeoutForRetrievingMetadata));

return eventuateSqlDialect.castToJson("?",
        eventuateSchema, "message", column, selectCallback);

}`

Blockhoud would not let it run.

I had to change the code a bit to get rid of blocking code part.

`import static io.eventuate.common.jdbc.EventuateJdbcOperationsUtils.MESSAGE_AUTO_GENERATED_ID_COLUMN;

import java.util.HashMap;
import java.util.Map;

import io.eventuate.common.id.IdGenerator;
import io.eventuate.common.jdbc.EventuateJdbcOperationsUtils;
import io.eventuate.common.jdbc.EventuateSchema;
import io.eventuate.common.jdbc.sqldialect.EventuateSqlDialect;
import io.eventuate.common.json.mapper.JSonMapper;
import io.eventuate.common.reactive.jdbc.EventuateCommonReactiveJdbcOperations;
import io.eventuate.common.reactive.jdbc.EventuateReactiveJdbcStatementExecutor;
import reactor.core.publisher.Mono;

public class EventuateCommonReactiveOperations extends EventuateCommonReactiveJdbcOperations {

private static final String COLUMN_TYPE_SQL = "select data_type from information_schema.columns where table_schema = ? and table_name = ? and column_name = ?";
private static final String MESSAGE_TABLE_NAME = "message";

private EventuateJdbcOperationsUtils eventuateJdbcOperationsUtils;
private EventuateReactiveJdbcStatementExecutor reactiveJdbcStatementExecutor;
private EventuateSqlDialect eventuateSqlDialect;

public EventuateCommonReactiveOperations(
    EventuateJdbcOperationsUtils eventuateJdbcOperationsUtils,
    EventuateReactiveJdbcStatementExecutor reactiveJdbcStatementExecutor,
    EventuateSqlDialect eventuateSqlDialect,
    int blockingTimeoutForRetrievingMetadata) {
super(eventuateJdbcOperationsUtils, reactiveJdbcStatementExecutor, eventuateSqlDialect, blockingTimeoutForRetrievingMetadata);
this.eventuateJdbcOperationsUtils = eventuateJdbcOperationsUtils;
this.reactiveJdbcStatementExecutor = reactiveJdbcStatementExecutor;
this.eventuateSqlDialect = eventuateSqlDialect;
}

@Override
public Mono<String> insertIntoMessageTable(IdGenerator idGenerator,
    String payload,
    String destination,
    Map<String, String> headers,
    EventuateSchema eventuateSchema) {

return insertIntoMessageTableImpl(idGenerator, payload, destination, headers, eventuateSchema, false);
}

@Override
public Mono<String> insertPublishedMessageIntoMessageTable(IdGenerator idGenerator,
    String payload,
    String destination,
    Map<String, String> headers,
    EventuateSchema eventuateSchema) {

return insertIntoMessageTableImpl(idGenerator, payload, destination, headers, eventuateSchema, true);
}

private Mono<String> insertIntoMessageTableImpl(
    IdGenerator idGenerator,
    String payload,
    String destination,
    Map<String, String> headers,
    EventuateSchema eventuateSchema,
    boolean published) {
if (idGenerator.databaseIdRequired()) {
    return insertIntoMessageTableDatabaseIdImpl(idGenerator,
	    payload, destination, headers, published, eventuateSchema);
} else {
    return insertIntoMessageTableApplicationIdImpl(idGenerator,
	    payload, destination, headers, published, eventuateSchema);
}
}

private Mono<String> insertIntoMessageTableApplicationIdImpl(IdGenerator idGenerator,
    String payload,
    String destination,
    Map<String, String> headers,
    boolean published,
    EventuateSchema eventuateSchema) {
headers = new HashMap<>(headers);
String messageId = idGenerator.genId(null).asString();
headers.put("ID", messageId);
String serializedHeaders = JSonMapper.toJson(headers);
return this
	.createInsertQueryIntoMessageTableApplicationIdSql(eventuateSchema)
	.flatMap(sql -> reactiveJdbcStatementExecutor.update(
		sql,
		messageId,
		destination,
		serializedHeaders,
		payload,
		eventuateJdbcOperationsUtils.booleanToInt(published)))
	.map(rowsUpdated -> messageId);
}

private Mono<String> insertIntoMessageTableDatabaseIdImpl(IdGenerator idGenerator,
    String payload,
    String destination,
    Map<String, String> headers,
    boolean published,
    EventuateSchema eventuateSchema) {
String serializedHeaders = JSonMapper.toJson(headers);
return this
	.createInsertQueryIntoMessageTableDbIdSql(eventuateSchema)
	.flatMap(sql -> reactiveJdbcStatementExecutor.insertAndReturnId(
		sql,
		MESSAGE_AUTO_GENERATED_ID_COLUMN,
		destination,
		serializedHeaders,
		payload,
		eventuateJdbcOperationsUtils.booleanToInt(published)))
	.map(id -> idGenerator.genId(id).asString());
}

public Mono<String> createInsertQueryIntoMessageTableApplicationIdSql(final EventuateSchema eventuateSchema) {
return Mono.zip(
	this.getHeadersColumnType(eventuateSchema),
	this.getPayloadColumnType(eventuateSchema),
	(headersJson, payloadJson) -> {
	    String sql = "insert into %s(id, destination, headers, payload, creation_time, published) values(?, ?, %s, %s, %s, ?)";
	    return String.format(sql,
		    eventuateSchema.qualifyTable(MESSAGE_TABLE_NAME),
		    headersJson,
		    payloadJson,
		    eventuateSqlDialect.getCurrentTimeInMillisecondsExpression());
	});
}

public Mono<String> createInsertQueryIntoMessageTableDbIdSql(final EventuateSchema eventuateSchema) {
return Mono.zip(
	this.getHeadersColumnType(eventuateSchema),
	this.getPayloadColumnType(eventuateSchema),
	(headersJson, payloadJson) -> {
	    String sql = "insert into %s(id, destination, headers, payload, creation_time, published) values('', ?, %s, %s, %s, ?)";
	    return String.format(sql,
		    eventuateSchema.qualifyTable(MESSAGE_TABLE_NAME),
		    headersJson,
		    payloadJson,
		    eventuateSqlDialect.getCurrentTimeInMillisecondsExpression());
	});
}

private Mono<String> getHeadersColumnType(final EventuateSchema eventuateSchema) {
return reactiveJdbcStatementExecutor
	.query(COLUMN_TYPE_SQL, eventuateSchema.getEventuateDatabaseSchema(), MESSAGE_TABLE_NAME, "headers")
	.last()
	.map(r -> r.get("data_type"))
	.map(columnType -> String.format("?::%s", columnType))
	.defaultIfEmpty("?");
}

private Mono<String> getPayloadColumnType(final EventuateSchema eventuateSchema) {
return reactiveJdbcStatementExecutor
	.query(COLUMN_TYPE_SQL, eventuateSchema.getEventuateDatabaseSchema(), MESSAGE_TABLE_NAME, "payload")
	.last()
	.map(r -> r.get("data_type"))
	.map(columnType -> String.format("?::%s", columnType))
	.defaultIfEmpty("?");
}`