Multiple DBConfigs with custom conversions called only once

[hfhbd]

Hey,

I have 3 databases, each having its own DBConfig class, extending the AbstractR2dbcConfiguration.

From the docs: To add custom conversions override the getCustomConversions().

Unfortunally, there is only one global R2dbcCustomConversions Bean instead one for each configuration. This results into only one call of the overriden getCustomConversions() function.

Example:

@Configuration
@EnableR2dbcRepositories(ADBConfig.basePackage, databaseClientRef = ADBConfig.clientName)
class ADBConfig(private val properties: AProperties): AbstractR2dbcConfiguration() {

   companion object {
       const val basePackage = "org.example.a"
       const val clientName = "aDatabaseClient"
       const val factory = "aFactory"
   }

   @Bean(factory)
   override fun connectionFactory(): ConnectionFactory {
       return H2ConnectionFactory(
           H2ConnectionConfiguration.builder()
               .url(properties.datasource.jdbcURL)
               .username(properties.datasource.userName)
               .password(properties.datasource.password)
               .build()
       )
   }

   override fun customConverters() = listOf(AWritingConverter(), AReadConverter())

   @Bean(clientName)
   fun databaseClient(@Qualifier(factory) connectionFactory: ConnectionFactory): DatabaseClient = DatabaseClient.create(connectionFactory)
}

Alternative:

Don't override the getCustomConversions function, but instead the Bean directly with a given name.

@Bean(conversions)
    override fun r2dbcCustomConversions()=  R2dbcCustomConversions(storeConversions, listOf())

The @Bean R2dbcCustomConversions has now be called by @Qualifier in AbstractR2dbcConfiguration. The qualifier comes from a new attribute with name R2dbcCustomConversionsRefin @EnableR2dbcRepositories.

@Configuration
@EnableR2dbcRepositories(ADBConfig.basePackage, databaseClientRef = ADBConfig.clientName, R2dbcCustomConversionsRef = "aCustomConversions")
class ADBConfig(private val properties: AProperties): AbstractR2dbcConfiguration() {

[mp911de]

AbstractR2dbcConfiguration is intended for a simple, single database configuration. There's a pretty extensive chain of dependencies that depend on R2dbcCustomConversions (RelationalMappingContext <- ReactiveDataAccessStrategy <-DatabaseClient). If you require a customized configuration, please provide your own config class. AbstractR2dbcConfiguration isn't capable of providing more than a single configuration chain without heavy customization.

[hfhbd]

Hey @mp911de,
thanks for your fast answer! I used now a workaround to implement a MultiDBConfig later.
But now I noticed a next problem: How is the transaction handled with multi datasources? Currently, I use a TransactionManager Bean for each DB and create TransactionOperator for each transaction call.

@Table
data class Test(@Id val id: Int? = null, val name: String)

suspend fun findByOrSave(name: String, new: Test): Int =
    repo.findByUnique(name = name)
.switchIfEmpty(repo.save(new))
.transactional(TransactionalOperator.create(tm)).id!!

Unfortunately, the transaction/connection is shared and results in primary key violation. How can I enforce a new connection/not-shared transaction?

[mp911de]

There are multiple approaches. Each R2dbcTransactionManager is associated with a ConnectionFactory. If all your databases are of the same kind and you could reuse the same DatabaseClient, then AbstractRoutingConnectionFactory would be the way to go where you can route database calls based on contextual information.

If you're required to maintain multiple ConnectionFactory instances, then you also need to associate each of them with a R2dbcTransactionManager and TransactionalOperator. This pattern pretty much corresponds with what Spring provides for JDBC DataSource. You could also use an annotation-based programming model via @Transactional("txManager1")/@Transactional("txManager2") and so on.

Please note that .as(transactionalOperator::transactional) associates the transactional context with all upstream publishers. So in your case, findByUnique(…) and save(…) share the same transaction.

Please note that we use GitHub issues for feature requests and bug reports. We prefer Stack Overflow and Gitter for questions respective discussions.

As this ticket is neither a bug nor a feature request that we're going to implement, we're closing this ticket.

[hfhbd]

Hey, thanks for your patience.

Of course I did create a transaction manager and operator for each dbconfig, but this was not my problem.

Please note that .as(transactionalOperator::transactional) associates the transactional context with all upstream publishers. So in your case, findByUnique(…) and save(…) share the same transaction.

I was aware of this and expected this behavior, but in a parallel Flow using Kotlin Coroutines, the same transaction was send to all parallel flows instead to the single one, which results into transaction misbehavior.

Your comment helped me to check this parallelism and I switched to ParallelFlux, which does not share the transaction and results in the expected behavior.

So thank you.

[mp911de]

Thanks a lot for the detail. Parallelism wasn’t obvious from the previous comment. Glad it worked out for you.