pjfanning commented on code in PR #510:
URL:
https://github.com/apache/pekko-persistence-r2dbc/pull/510#discussion_r4376113346
##########
core/src/test/scala/org/apache/pekko/persistence/r2dbc/query/EventsBySliceBacktrackingSpec.scala:
##########
@@ -186,6 +201,86 @@ class EventsBySliceBacktrackingSpec
env9.sequenceNr shouldBe 4L
}
+ "mark FilteredPayload as filtered in backtracking results" in {
+ val entityType = nextEntityType()
+ val pid = nextPid(entityType)
+ val slice = query.sliceForPersistenceId(pid)
+ val sinkProbe = TestSink[EventEnvelope[String]]()(system.classicSystem)
+
+ // don't let behind-current-time be a reason for not finding events
+ val startTime = InstantFactory.now().minusSeconds(10 * 60)
+
+ val filteredSerializer =
SerializationExtension(system).findSerializerFor(FilteredPayload)
+ writeSerializedEvent(slice, pid, 1L, startTime,
filteredSerializer.identifier,
+ Serializers.manifestFor(filteredSerializer, FilteredPayload),
filteredSerializer.toBinary(FilteredPayload))
+ writeEvent(slice, pid, 2L, startTime.plusMillis(1), "e-2")
+
+ val result: TestSubscriber.Probe[EventEnvelope[String]] =
+ query
+ .eventsBySlices[String](entityType, slice, slice, NoOffset)
+ .runWith(sinkProbe)
+ .request(100)
+
+ val env1 = result.expectNext()
+ env1.sequenceNr shouldBe 1L
+ env1.source shouldBe EnvelopeOrigin.SourceQuery
+ env1.filtered shouldBe true
+ env1.eventOption shouldBe None
+
+ val env2 = result.expectNext()
+ env2.sequenceNr shouldBe 2L
+ env2.filtered shouldBe false
+ env2.eventOption shouldBe Some("e-2")
+
+ val env3 = result.expectNext()
+ env3.sequenceNr shouldBe 1L
+ env3.source shouldBe EnvelopeOrigin.SourceBacktracking
+ env3.filtered shouldBe true
+
+ val env4 = result.expectNext()
+ env4.sequenceNr shouldBe 2L
+ env4.source shouldBe EnvelopeOrigin.SourceBacktracking
+ env4.filtered shouldBe false
+
+ result.cancel()
+ }
+
+ "only mark events with the FilteredPayload manifest as filtered" in {
+ // In Pekko 2.0.0-M5 SerializedEvent shares the FilteredPayload
serializer, with manifest "S".
+ // The manifest must be checked, not only the serializer id.
+ val entityType = nextEntityType()
+ val pid = nextPid(entityType)
+ val slice = query.sliceForPersistenceId(pid)
+ val sinkProbe = TestSink[EventEnvelope[Any]]()(system.classicSystem)
+
+ // don't let behind-current-time be a reason for not finding events
+ val startTime = InstantFactory.now().minusSeconds(10 * 60)
+
+ val filteredSerializer =
SerializationExtension(system).findSerializerFor(FilteredPayload)
+ writeSerializedEvent(slice, pid, 1L, startTime,
filteredSerializer.identifier, "S", Array.emptyByteArray)
Review Comment:
Good catch, thanks. Fixed in a641df0: the test now writes a
`MessageFormats.PersistentPayload` wrapping a serialized `"e-1"`, as you
suggested, so it no longer relies on M4 ignoring the manifest.
##########
core/src/test/scala/org/apache/pekko/persistence/r2dbc/query/EventsBySliceSpec.scala:
##########
@@ -238,6 +239,29 @@ class EventsBySliceSpec
}
}
+ "mark FilteredPayload as filtered with no payload when reading it" in
new Setup {
+ persister ! PersistWithAck(FilteredPayload, probe.ref)
Review Comment:
`EventsBySliceSpec` sets `journal.publish-events = off`, so its Live variant
only ever reads from the database and can't exercise PubSub. The PubSub fix is
in #511, with a test in `EventsBySlicePubSubSpec` (which has `publish-events =
on`) that fails before that fix. #511 also handles `Tagged(FilteredPayload,
tags)`, since the journal publishes the raw `PersistentRepr`.
--
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]
