Vertx

Module provides a repository implementation based on the Vertx reactive protocol.

Dependency

Java Kotlin

Dependency build.gradle:

annotationProcessor "ru.tinkoff.kora:annotation-processors"
implementation "ru.tinkoff.kora:database-vertx"

Module:

@KoraApp
public interface Application extends VertxDatabaseModule { }

Dependency build.gradle.kts:

ksp("ru.tinkoff.kora:symbol-processors")
implementation("ru.tinkoff.kora:database-vertx")

Module:

@KoraApp
interface Application : VertxDatabaseModule

It is also required to provide a driver implementation as a dependency of a version no higher than 4.3.8

In some cases, such as with a PostgreSQL database, it is also required to add dependency.

Configuration

Example of the complete configuration described in the VertxDatabaseConfig class (default or example values are specified):

Hocon YAML

db {
    connectionUri = "postgresql://localhost:5432/postgres" //(1)!
    username = "postgres" //(2)!
    password = "postgres" //(3)!
    poolName = "kora" //(4)!
    maxPoolSize = 10 //(5)!
    connectionTimeout = "10s" //(6)!
    acquireTimeout = "0s" //(7)!
    idleTimeout = "10m" //(8)!
    cachePreparedStatements = true //(9)!
    initializationFailTimeout = "0s" //(10)!
    readinessProbe = false //(11)!
    telemetry {
        logging {
            enabled = false //(12)!
        }
        metrics {
            enabled = true //(13)!
            slo = [ 1, 10, 50, 100, 200, 500, 1000, 2000, 5000, 10000, 20000, 30000, 60000, 90000 ] //(14)!
        }
        tracing {
            enabled = true //(15)!
        }
    }
}

1. URI connection URI (required)
2. User name for connection (required)
3. Password of the user to connect (required)
4. Database connection set name (required)
5. Maximum size of the database connection set
6. Maximum time to establish a connection
7. Maximum time to get a connection from a connection set (optional)
8. Maximum time for connection downtime
9. Whether to cache prepared requests
10. Maximum time to wait for connection initialization at service startup (optional)
11. Whether to enable probes.md#_2 for database connection
12. Enables module logging (default false)
13. Enables module metrics (default true)
14. Configures SLO for DistributionSummary metrics
15. Enables module tracing (default true)

db:
  connectionUri = "postgresql://localhost:5432/postgres" #(1)!
  username: "postgres" #(2)!
  password: "postgres" #(3)!
  poolName: "kora" #(4)!
  maxPoolSize: 10 #(5)!
  connectionTimeout: "10s" #(6)!
  acquireTimeout: "10s" #(7)!
  idleTimeout: "10m" #(8)!
  cachePreparedStatements: true #(9)!
  initializationFailTimeout: "0s" #(10)!
  readinessProbe: false #(11)!
  telemetry:
    logging:
      enabled: false #(12)!
    metrics:
      enabled: true #(13)!
      slo: [ 1, 10, 50, 100, 200, 500, 1000, 2000, 5000, 10000, 20000, 30000, 60000, 90000 ] #(14)!
    tracing:
      enabled: true #(15)!

1. URI connection URI (required)
2. User name for connection (required)
3. Password of the user to connect (required)
4. Database connection set name (required)
5. Maximum size of the database connection set
6. Maximum time to establish a connection
7. Maximum time to get a connection from a connection set (optional)
8. Maximum time for connection downtime
9. Whether to cache prepared requests
10. Maximum time to wait for connection initialization at service startup (optional)
11. Whether to enable probes.md#_2 for database connection
12. Enables module logging (default false)
13. Enables module metrics (default true)
14. Configures SLO for DistributionSummary metrics
15. Enables module tracing (default true)

You can also configure Netty transport.

Usage

Java Kotlin

@Repository
public interface EntityRepository extends VertxRepository { }

@Repository
interface EntityRepository : VertxRepository

Mapping

It is possible to override the conversion of different parts of entity and query parameters, Kora provides special interfaces for this.

Result

If you need to convert the result manually, it is suggested to use VertxRowSetMapper:

Java Kotlin

final class ResultMapper implements VertxRowSetMapper<List<UUID>> {

    @Override
    public List<UUID> apply(RowSet<Row> rows) {
        // mapping code
    }
}

@Repository
public interface EntityRepository extends VertxRepository {

    @Mapping(ResultMapper.class)
    @Query("SELECT id FROM entities")
    Mono<List<UUID>> getIds();
}

class ResultMapper : VertxRowSetMapper<List<UUID>> {
    override fun apply(rows: RowSet<Row>): List<UUID> {
        // mapping code
    }
}

@Repository
interface EntityRepository : VertxRepository {

    @Mapping(ResultMapper::class)
    @Query("SELECT id FROM entities")
    fun getIds(): Mono<List<UUID>>
}

Row

If you need to convert the string manually, it is suggested to use VertxRowMapper:

Java Kotlin

final class RowMapper implements VertxRowMapper<UUID> {

    @Override
    public UUID apply(Row row) {
        return UUID.fromString(rs.get(0, String.class));
    }
}

@Repository
public interface EntityRepository extends VertxRepository {

    @Mapping(RowMapper.class)
    @Query("SELECT id FROM entities")
    Flux<UUID> findAll();
}

class RowMapper : VertxRowMapper<UUID> {

    override fun apply(row: Row): UUID {
        return UUID.fromString(rs.get(0, String.class))
    }
}

@Repository
interface EntityRepository : VertxRepository {

    @Mapping(RowMapper::class)
    @Query("SELECT id FROM entities")
    fun findAll(): Flux<UUID>
}

Column

If you need to convert the column value manually, it is suggested to use the VertxResultColumnMapper:

Java Kotlin

public final class ColumnMapper implements VertxResultColumnMapper<UUID> {

    @Override
    public UUID apply(Row row, int index) {
        return UUID.fromString(row.get(String.class, index));
    }
}

@Table("entities")
public record Entity(@Mapping(ColumnMapper.class) @Id UUID id, String name) { }

@Repository
public interface EntityRepository extends VertxRepository {

    @Query("SELECT * FROM entities")
    Flux<Entity> findAll();
}

class ColumnMapper : VertxResultColumnMapper<UUID> {

    override fun apply(row: Row, index: Int): UUID {
        return UUID.fromString(row.get(String.class, index))
    }
}

@Table("entities")
data class Entity(
    @Id @Mapping(ColumnMapper::class) val id: UUID,
    val name: String
)

@Repository
interface EntityRepository : VertxRepository {

    @Query("SELECT * FROM entities")
    fun findAll(): Flux<Entity>
}

Parameter

If you want to convert the value of a query parameter manually, it is suggested to use VertxParameterColumnMapper:

Java Kotlin

public final class ParameterMapper implements VertxParameterColumnMapper<UUID> {

    @Override
    public Object apply(@Nullable UUID value) {
        return value.toString();
    }
}

@Repository
public interface EntityRepository extends VertxRepository {

    @Query("SELECT * FROM entities WHERE id = :id")
    Flux<Entity> findById(@Mapping(ParameterMapper.class) UUID id);
}

class ParameterMapper : VertxParameterColumnMapper<UUID?> {
    override fun apply(value: UUID?): Any {
        return value.toString()
    }
}

@Repository
interface EntityRepository : VertxRepository {

    @Query("SELECT * FROM entities WHERE id = :id")
    fun findById(@Mapping(ParameterMapper::class) id: UUID): Flux<Entity>
}

Transactions

In order to perform manual queries, Kora has an interface ru.tinkoff.kora.database.vertx.VertxConnectionFactory, which is provided in a method within the VertxRepository contract. All repository methods called within a transaction lambda will be executed in that transaction.

In order to perform queries transactionally, the inTx contract can be used:

Java Kotlin

@Component
public final class SomeService {

    private final EntityRepository repository;

    public SomeService(EntityRepository repository) {
        this.repository = repository;
    }

    public Mono<List<Entity>> saveAll(Entity one, Entity two) {
        return repository.getVertxConnectionFactory().inTx(connection -> {
            // do some work
            return repository.insert(one) //(1)!
                    .zipWith(repository.insert(two), //(2)!
                        (r1, r2) -> List.of(one, two));
        });
    }
}

1. will be executed within the transaction or rolled back if the entire lambda throws an exception
2. will be executed within the transaction or rolled back if the entire lambda throws an exception

@Component
class SomeService(private val repository: EntityRepository) {

    fun saveAll(
        one: Entity,
        two: Entity
    ): Mono<List<Entity>> {
        return repository.getVertxConnectionFactory.inTx {
            repository.insert(one).zipWith(repository.insert(two)) //(1)!
            { r1: UpdateCount, r2: UpdateCount -> listOf(one, two) }
        }
    }
}

1. will be executed within the transaction or will be rolled back if the entire lambda throws an exception

Connection

If some more complex logic is needed for the query, and @Query is not enough, you can use io.r2dbc.spi.Connection:

Java Kotlin

@Component
public final class SomeService {

    private final EntityRepository repository;

    public SomeService(EntityRepository repository) {
        this.repository = repository;
    }

    public Mono<List<Entity>> saveAll(Entity one, Entity two) {
        return repository.getVertxConnectionFactory().inTx(connection -> {
            // do some work
        });
    }
}

@Component
class SomeService(private val repository: EntityRepository) {

    fun saveAll(
        one: Entity,
        two: Entity
    ): Mono<List<Entity>> {
        return repository.getVertxConnectionFactory.inTx { connection ->
            // do some work
        }
    }
}

Signatures

Available signatures for repository methods out of the box:

Java Kotlin

The T refers to the type of the return value, either List<T>, either Void or UpdateCount.

• T myMethod()
• @Nullable T myMethod()
• Optional<T> myMethod()
• Mono<T> myMethod() Project Reactor (require dependency)
• Flux<T> myMethod() Project Reactor (require dependency)

By T we mean the type of the return value, either T?, either List<T>, either Unit or UpdateCount.

• myMethod(): T
• suspend myMethod(): T Kotlin Coroutine (require dependency as implementation)
• myMethod(): Flow<T> Kotlin Coroutine (require dependency as implementation)