-
Notifications
You must be signed in to change notification settings - Fork 18
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Separate sql query definition+binding from the actual locic+error han…
…dling to reuse the latter one while adding new dialects
- Loading branch information
1 parent
7b15676
commit c194e4f
Showing
18 changed files
with
1,571 additions
and
1,466 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
66 changes: 66 additions & 0 deletions
66
core/src/main/scala/akka/persistence/r2dbc/internal/TimestampCodec.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,66 @@ | ||
| /* | ||
| * Copyright (C) 2022 - 2023 Lightbend Inc. <https://www.lightbend.com> | ||
| */ | ||
|
|
||
| package akka.persistence.r2dbc.internal | ||
|
|
||
| import java.nio.charset.StandardCharsets.UTF_8 | ||
| import akka.annotation.InternalApi | ||
| import io.r2dbc.postgresql.codec.Json | ||
| import io.r2dbc.spi.Row | ||
| import io.r2dbc.spi.Statement | ||
|
|
||
| import java.time.{ Instant, LocalDateTime } | ||
| import java.util.TimeZone | ||
|
|
||
| /** | ||
| * INTERNAL API | ||
| */ | ||
| @InternalApi private[akka] sealed trait TimestampCodec { | ||
|
|
||
| def encode(timestamp: Instant): Any | ||
| def decode(row: Row, name: String): Instant | ||
|
|
||
| protected def instantNow() = InstantFactory.now() | ||
|
|
||
| def now[T](): T | ||
| } | ||
|
|
||
| /** | ||
| * INTERNAL API | ||
| */ | ||
| @InternalApi private[akka] object TimestampCodec { | ||
| case object PostgresTimestampCodec extends TimestampCodec { | ||
| override def decode(row: Row, name: String): Instant = row.get(name, classOf[Instant]) | ||
|
|
||
| override def encode(timestamp: Instant): Any = timestamp | ||
|
|
||
| override def now[T](): T = instantNow().asInstanceOf[T] | ||
| } | ||
|
|
||
| case object SqlServerCodec extends TimestampCodec { | ||
|
|
||
| // should this come from config? | ||
| private val zone = TimeZone.getTimeZone("UTC").toZoneId | ||
|
|
||
| override def decode(row: Row, name: String): Instant = { | ||
| row | ||
| .get(name, classOf[LocalDateTime]) | ||
| .atZone(zone) | ||
| .toInstant | ||
| } | ||
|
|
||
| override def encode(timestamp: Instant): Any = LocalDateTime.ofInstant(timestamp, zone) | ||
|
|
||
| override def now[T](): T = LocalDateTime.ofInstant(instantNow(), zone).asInstanceOf[T] | ||
| } | ||
|
|
||
|
|
||
| implicit class RichStatement[T](val statement: Statement)(implicit codec: TimestampCodec) extends AnyRef { | ||
| def bindTimestamp(name: String, timestamp: Instant): Statement = statement.bind(name, codec.encode(timestamp)) | ||
| def bindTimestamp(index: Int, timestamp: Instant): Statement = statement.bind(index, codec.encode(timestamp)) | ||
| } | ||
| implicit class RichRow[T](val row: Row)(implicit codec: TimestampCodec) extends AnyRef { | ||
| def getTimestamp(rowName: String = "db_timestamp"): Instant = codec.decode(row, rowName) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
32 changes: 32 additions & 0 deletions
32
core/src/main/scala/akka/persistence/r2dbc/internal/h2/sql/H2DurableStateSql.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| /* | ||
| * Copyright (C) 2022 - 2023 Lightbend Inc. <https://www.lightbend.com> | ||
| */ | ||
|
|
||
| package akka.persistence.r2dbc.internal.h2.sql | ||
|
|
||
| import akka.persistence.r2dbc.R2dbcSettings | ||
| import akka.persistence.r2dbc.internal.DurableStateDao.SerializedStateRow | ||
| import akka.persistence.r2dbc.internal.PayloadCodec.RichStatement | ||
| import akka.persistence.r2dbc.internal.Sql.Interpolation | ||
| import akka.persistence.r2dbc.internal.TimestampCodec.{ RichStatement => TimestampRichStatement } | ||
| import akka.persistence.r2dbc.internal.postgres.PostgresDurableStateDao.EvaluatedAdditionalColumnBindings | ||
| import akka.persistence.r2dbc.internal.postgres.sql.PostgresDurableStateSql | ||
| import akka.persistence.r2dbc.internal.{ PayloadCodec, TimestampCodec } | ||
| import akka.persistence.r2dbc.state.scaladsl.AdditionalColumn | ||
| import io.r2dbc.spi.Statement | ||
|
|
||
| import java.lang | ||
| import java.time.Instant | ||
| import scala.collection.immutable | ||
| import scala.concurrent.duration.{ Duration, FiniteDuration } | ||
|
|
||
| class H2DurableStateSql(settings: R2dbcSettings)(implicit | ||
| statePayloadCodec: PayloadCodec, | ||
| timestampCodec: TimestampCodec) | ||
| extends PostgresDurableStateSql(settings) { | ||
| protected override def behindCurrentTimeIntervalConditionFor(behindCurrentTime: FiniteDuration): String = | ||
| if (behindCurrentTime > Duration.Zero) | ||
| s"AND db_timestamp < CURRENT_TIMESTAMP - interval '${behindCurrentTime.toMillis.toDouble / 1000}' second" | ||
| else "" | ||
|
|
||
| } |
Oops, something went wrong.