feat: Typed envelope version of the eventsByPersistenceId queries #31892
Conversation
| * Copyright (C) 2015-2023 Lightbend Inc. <https://www.lightbend.com> | ||
| */ | ||
|
|
||
| package akka.persistence.query.scaladsl |
There was a problem hiding this comment.
We can place these in package akka.persistence.query.typed.scaladsl, where we have this EventEnvelope and the bySlices
There was a problem hiding this comment.
Done 2e8c7c1
Also noticed I had missed adding javadsl.
| } | ||
| val envelope = | ||
| new EventEnvelope[Event]( | ||
| offset = Sequence(pr.sequenceNr), |
There was a problem hiding this comment.
can we use TimestampOffset here so that it corresponds to the offset used for bySlices?
There was a problem hiding this comment.
In the testkit currentEventsBySlices we use Sequence, so then I should change that as well I guess?
There was a problem hiding this comment.
I see, maybe leave it as Sequence then, at least until we see that as a problem.
There was a problem hiding this comment.
Already went ahead and did it in 2e8c7c1
Do you think anyone may be using the slicing testkit stuff and expect a problem?
There was a problem hiding this comment.
I think that is fine to change this. Shouldn't be a big deal. The seqNr is also in the envelope if that is needed for some reason.
| storage, | ||
| persistence.sliceForPersistenceId)) | ||
| .map(env => | ||
| EventEnvelope(env.offset, env.persistenceId, env.sequenceNr, env.event, env.timestamp, env.eventMetadata)) |
There was a problem hiding this comment.
If we change the EventsByPersistenceIdStage to emit TimestampOffset we should convert that back to Sequence here for compatibility.
| */ | ||
| @InternalApi | ||
| private[akka] def timestampOffsetFor(pr: PersistentRepr) = { | ||
| val timestamp = Instant.ofEpochMilli(pr.timestamp) // Note: we don't really have microsecond granularity here |
There was a problem hiding this comment.
A good thing is that we always use an increasing and unique timestamp in the testkit, see akka.persistence.testkit.internal.CurrentTime
| @InternalApi | ||
| private[akka] def timestampOffsetFor(pr: PersistentRepr) = { | ||
| val timestamp = Instant.ofEpochMilli(pr.timestamp) // Note: we don't really have microsecond granularity here | ||
| val readTimestamp = Instant.now().truncatedTo(ChronoUnit.MICROS) |
There was a problem hiding this comment.
Shall we use CurrentTime.now() for the readTimestamp also, or maybe better to use Instant.now() but make sure it is greater than or equal to the pr.timestamp so that it doesn't "go backwards" (read can't happen before write)
No description provided.