pjfanning commented on code in PR #534: URL: https://github.com/apache/pekko-persistence-r2dbc/pull/534#discussion_r4398923761
########## core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/ConnectionAdapter.scala: ########## @@ -0,0 +1,209 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.pekko.persistence.r2dbc.internal + +import java.time.Duration +import java.time.Instant +import java.time.LocalDateTime +import java.time.ZoneOffset +import java.util.function.BiFunction +import java.util.function.Predicate + +import org.apache.pekko +import pekko.annotation.InternalApi +import io.r2dbc.spi.Batch +import io.r2dbc.spi.Connection +import io.r2dbc.spi.ConnectionFactory +import io.r2dbc.spi.ConnectionFactoryMetadata +import io.r2dbc.spi.ConnectionMetadata +import io.r2dbc.spi.IsolationLevel +import io.r2dbc.spi.Readable +import io.r2dbc.spi.Result +import io.r2dbc.spi.Row +import io.r2dbc.spi.RowMetadata +import io.r2dbc.spi.Statement +import io.r2dbc.spi.TransactionDefinition +import io.r2dbc.spi.ValidationDepth +import org.reactivestreams.Publisher +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono + +/** + * INTERNAL API + * + * Adapts an R2DBC driver to the types used by the plugin, for drivers that don't support all of them. Wraps the + * connections of a `ConnectionFactory` and converts the values bound to statements and read from rows. Subclasses + * override the conversions they need, the defaults pass the values through unchanged. + */ +@InternalApi +private[r2dbc] abstract class ConnectionAdapter { + + def bindValue(value: Any): Any = value + + def bindNullType(tpe: Class[?]): Class[?] = tpe + + def get[T](row: Row, index: Int, tpe: Class[T]): T = row.get(index, tpe) + + def get[T](row: Row, name: String, tpe: Class[T]): T = row.get(name, tpe) + + final def wrap(connectionFactory: ConnectionFactory): ConnectionFactory = + new ConnectionAdapter.AdaptedConnectionFactory(connectionFactory, this) +} + +/** + * INTERNAL API + */ +@InternalApi +private[r2dbc] object ConnectionAdapter { + + /** + * For drivers without `java.time.Instant` support. Instants are bound and read as `LocalDateTime` in UTC, so the + * database session must use the UTC time zone. + */ + trait InstantAsUtcLocalDateTime extends ConnectionAdapter { + + override def bindValue(value: Any): Any = + value match { + case instant: Instant => LocalDateTime.ofInstant(instant, ZoneOffset.UTC) + case other => super.bindValue(other) + } + + override def bindNullType(tpe: Class[?]): Class[?] = + if (tpe == classOf[Instant]) classOf[LocalDateTime] + else super.bindNullType(tpe) + + override def get[T](row: Row, index: Int, tpe: Class[T]): T = + if (tpe == classOf[Instant]) toInstant(super.get(row, index, classOf[LocalDateTime])).asInstanceOf[T] + else super.get(row, index, tpe) + + override def get[T](row: Row, name: String, tpe: Class[T]): T = + if (tpe == classOf[Instant]) toInstant(super.get(row, name, classOf[LocalDateTime])).asInstanceOf[T] + else super.get(row, name, tpe) + + private def toInstant(localDateTime: LocalDateTime): Instant = + if (localDateTime == null) null else localDateTime.toInstant(ZoneOffset.UTC) + } + + private final class AdaptedConnectionFactory(underlying: ConnectionFactory, adapter: ConnectionAdapter) + extends ConnectionFactory { + override def create(): Publisher[? <: Connection] = + Mono.from[Connection](underlying.create()).map[Connection]((connection: Connection) => + new AdaptedConnection(connection, adapter)) + + override def getMetadata: ConnectionFactoryMetadata = underlying.getMetadata + } + + private final class AdaptedConnection(underlying: Connection, adapter: ConnectionAdapter) extends Connection { Review Comment: This is the first wrapper the plugin puts around connections. For the other dialects, `R2dbcSession` hands out r2dbc-pool's `PooledConnection`, which implements `io.r2dbc.spi.Wrapped`, so `unwrap()` returns the native driver connection. With MariaDB, `unwrap()` returns this `AdaptedConnection`, one level short of the driver connection. Could `AdaptedConnection` implement `Wrapped[Connection]` and return `underlying`, so `unwrap()` behaves the same as for the other databases? (`R2dbcExecutor.closeAfterTimeout` only unwraps once and still closes the physical connection, since `close()` delegates, so that path isn't affected either way.) ########## core/src/main/resources/reference.conf: ########## @@ -280,6 +280,7 @@ pekko.persistence.r2dbc { ssl { enabled = off # See PostgresqlConnectionFactoryProvider.SSL_MODE + # The MariaDB driver supports require, verify-ca and verify-full, see the MariaDB section of the documentation Review Comment: Nit: this line sits between `# See PostgresqlConnectionFactoryProvider.SSL_MODE` and `# Possible values:`, so it reads as if the list that follows is the MariaDB one. Could it move after the list of values? ########## docs/src/main/paradox/overview.md: ########## @@ -5,6 +5,8 @@ The Pekko Persistence R2DBC plugin allows for using SQL database with R2DBC as a Currently, the R2DBC plugin has support for [PostgreSQL](https://www.postgresql.org) and [Yugabyte](https://www.yugabyte.com). It is specifically designed to work well for distributed SQL databases. +[MariaDB](https://mariadb.org) 10.11 or later is supported, see @ref:[MariaDB](getting-started.md#mariadb). Review Comment: Nit: the overview now mentions MariaDB but still not MySQL, so it reads as if MariaDB is supported and MySQL isn't. Could MySQL be mentioned here too (it can be a separate change if you prefer)? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
