-
Notifications
You must be signed in to change notification settings - Fork 36
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
fix: A few smaller follow ups for the edge replication (#1094)
* Java types in Java DSL * Aligned Java function types (throwing factories vs non-throwing where user calls) * Wrong type of replication impl in Java API fixed * Initial consumer filter for edge replication * Pass initial consumer filters to producer for RES * Include internal replication settings when converting back and forth to Java API EventProducerPushDestination * JavaDSL test for edge replication * Fixes and refactoring of SourceProviderAdapters --------- Co-authored-by: Patrik Nordwall <[email protected]>
- Loading branch information
1 parent
d410abe
commit 9985eb9
Showing
21 changed files
with
913 additions
and
204 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
4 changes: 4 additions & 0 deletions
4
...ojection-core/src/main/mima-filters/1.5.1-M1.backwards.excludes/edge-replication.excludes
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,4 @@ | ||
| # internal/renamed | ||
| ProblemFilters.exclude[MissingClassProblem]("akka.projection.internal.ScalaBySlicesSourceProviderAdapter") | ||
| # (was internal stable but only because it was historically used by akka-persistence-r2dbc) | ||
| ProblemFilters.exclude[MissingClassProblem]("akka.projection.internal.SourceProviderAdapter") |
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
99 changes: 99 additions & 0 deletions
99
...ction-core/src/main/scala/akka/projection/internal/ScalaToJavaSourceProviderAdapter.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,99 @@ | ||
| /* | ||
| * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com> | ||
| */ | ||
|
|
||
| package akka.projection.internal | ||
|
|
||
| import akka.NotUsed | ||
| import akka.annotation.InternalApi | ||
| import akka.dispatch.ExecutionContexts | ||
| import akka.persistence.query.typed.EventEnvelope | ||
| import akka.persistence.query.typed.javadsl.EventTimestampQuery | ||
| import akka.persistence.query.typed.javadsl.LoadEventQuery | ||
| import akka.persistence.query.typed.scaladsl.{ EventTimestampQuery => ScalaEventTimestampQuery } | ||
| import akka.persistence.query.typed.scaladsl.{ LoadEventQuery => ScalaLoadEventQuery } | ||
| import akka.projection.BySlicesSourceProvider | ||
| import akka.projection.javadsl | ||
| import akka.projection.scaladsl | ||
| import akka.stream.javadsl.{ Source => JSource } | ||
|
|
||
| import java.time.Instant | ||
| import java.util.Optional | ||
| import java.util.concurrent.CompletionStage | ||
| import java.util.function.Supplier | ||
| import scala.compat.java8.FutureConverters._ | ||
| import scala.compat.java8.OptionConverters._ | ||
|
|
||
| /** | ||
| * INTERNAL API: Adapter from scaladsl.SourceProvider with BySlicesSourceProvider to javadsl.SourceProvider with BySlicesSourceProvider | ||
| */ | ||
| @InternalApi private[projection] object ScalaToJavaBySlicesSourceProviderAdapter { | ||
| def apply[Offset, Envelope]( | ||
| delegate: scaladsl.SourceProvider[Offset, Envelope] | ||
| with BySlicesSourceProvider): javadsl.SourceProvider[Offset, Envelope] = | ||
| delegate match { | ||
| case adapted: JavaToScalaBySliceSourceProviderAdapter[_, _] => | ||
| // just unwrap rather than wrapping further | ||
| adapted.delegate | ||
| case delegate: CanTriggerReplay => new ScalaToJavaBySlicesSourceProviderAdapterWithCanTriggerReplay(delegate) | ||
| case _ => new ScalaToJavaBySlicesSourceProviderAdapter(delegate) | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * INTERNAL API: Adapter from scaladsl.SourceProvider with BySlicesSourceProvider to javadsl.SourceProvider with BySlicesSourceProvider | ||
| */ | ||
| @InternalApi | ||
| private[projection] sealed class ScalaToJavaBySlicesSourceProviderAdapter[Offset, Envelope] private[internal] ( | ||
| val delegate: scaladsl.SourceProvider[Offset, Envelope] with BySlicesSourceProvider) | ||
| extends javadsl.SourceProvider[Offset, Envelope] | ||
| with BySlicesSourceProvider | ||
| with EventTimestampQuery | ||
| with LoadEventQuery { | ||
| override def source( | ||
| offset: Supplier[CompletionStage[Optional[Offset]]]): CompletionStage[JSource[Envelope, NotUsed]] = | ||
| delegate | ||
| .source(() => offset.get().toScala.map(_.asScala)(ExecutionContexts.parasitic)) | ||
| .map(_.asJava)(ExecutionContexts.parasitic) | ||
| .toJava | ||
|
|
||
| override def extractOffset(envelope: Envelope): Offset = delegate.extractOffset(envelope) | ||
|
|
||
| override def extractCreationTime(envelope: Envelope): Long = delegate.extractCreationTime(envelope) | ||
|
|
||
| def minSlice: Int = delegate.minSlice | ||
|
|
||
| def maxSlice: Int = delegate.maxSlice | ||
|
|
||
| override def timestampOf(persistenceId: String, sequenceNr: Long): CompletionStage[Optional[Instant]] = | ||
| delegate match { | ||
| case etq: ScalaEventTimestampQuery => | ||
| etq.timestampOf(persistenceId, sequenceNr).map(_.asJava)(ExecutionContexts.parasitic).toJava | ||
| case _ => | ||
| throw new IllegalStateException( | ||
| s"timestampOf was called but delegate of type [${delegate.getClass}] does not implement akka.persistence.query.typed.scaladsl.EventTimestampQuery") | ||
| } | ||
|
|
||
| override def loadEnvelope[Event](persistenceId: String, sequenceNr: Long): CompletionStage[EventEnvelope[Event]] = | ||
| delegate match { | ||
| case etq: ScalaLoadEventQuery => | ||
| etq.loadEnvelope[Event](persistenceId, sequenceNr).toJava | ||
| case _ => | ||
| throw new IllegalStateException( | ||
| s"loadEnvelope was called but delegate of type [${delegate.getClass}] does not implement akka.persistence.query.typed.scaladsl.LoadEventQuery") | ||
| } | ||
|
|
||
| } | ||
|
|
||
| /** | ||
| * INTERNAL API: Adapter from scaladsl.SourceProvider with BySlicesSourceProvider to javadsl.SourceProvider with BySlicesSourceProvider | ||
| */ | ||
| @InternalApi | ||
| private[projection] final class ScalaToJavaBySlicesSourceProviderAdapterWithCanTriggerReplay[Offset, Envelope] private[internal] ( | ||
| delegate: scaladsl.SourceProvider[Offset, Envelope] with BySlicesSourceProvider with CanTriggerReplay) | ||
| extends ScalaToJavaBySlicesSourceProviderAdapter[Offset, Envelope](delegate) | ||
| with CanTriggerReplay { | ||
|
|
||
| override private[akka] def triggerReplay(persistenceId: String, fromSeqNr: Long): Unit = | ||
| delegate.triggerReplay(persistenceId, fromSeqNr) | ||
| } |
66 changes: 0 additions & 66 deletions
66
akka-projection-core/src/main/scala/akka/projection/internal/SourceProviderAdapter.scala
This file was deleted.
Oops, something went wrong.
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
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
Oops, something went wrong.