ptrdom commented on code in PR #534: URL: https://github.com/apache/pekko-persistence-r2dbc/pull/534#discussion_r4401311234
########## 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: Implementing `Wrapped` on `AdaptedConnection` alone would only add a second `unwrap()` call: the pool wraps the adapted connection, so `unwrap()` on the pooled connection still returns the adapter. And `Wrapped.unwrap(Class)` in r2dbc-spi 1.0 only recurses while the unwrapped object is `Wrapped`, which the MariaDB driver connection isn't. So instead the adapter now wraps the pool rather than the driver connections in it, and `AdaptedConnection.unwrap()` returns what the pooled connection unwraps to. A single `unwrap()` returns the `MariadbConnection`, as for the pooled connections of the other databases, and `close()` still returns the connection to the pool. The adapted connection factory is `Wrapped` too, unwrapping to the pool, which `ConnectionFactoryProvider` disposes at shutdown. `MariaDBPooledConnectionSpec` checks the unwrapping against MariaDB (with the adapter inside the pool it unwraps to `AdaptedConnection`), and `MariaDBConnectionFactoryProviderSpec` checks that the p ool is disposed. Done in 9e03012. ########## 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: Moved after the list of values. Done in 9e03012. ########## 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: Added a sentence for MySQL, linking to the connection configuration, as there's no MySQL section in Getting Started yet. Done in 9e03012. -- 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]
