Set metadata on events by persisence id query (#814)

akka · Sep 11, 2020 · 64ff267 · 64ff267
1 parent 5be6ee0
commit 64ff267
Show file tree

Hide file tree

Showing 3 changed files with 31 additions and 3 deletions.
diff --git a/core/src/main/scala/akka/persistence/cassandra/query/scaladsl/CassandraReadJournal.scala b/core/src/main/scala/akka/persistence/cassandra/query/scaladsl/CassandraReadJournal.scala
@@ -663,7 +663,13 @@ class CassandraReadJournal protected (
 
  private def toEventEnvelope(persistentRepr: PersistentRepr, offset: Offset): immutable.Iterable[EventEnvelope] =
  adaptFromJournal(persistentRepr).map { payload =>
- EventEnvelope(offset, persistentRepr.persistenceId, persistentRepr.sequenceNr, payload, timestampFrom(offset))
+ EventEnvelope(
+ offset,
+ persistentRepr.persistenceId,
+ persistentRepr.sequenceNr,
+ payload,
+ timestampFrom(offset),
+ persistentRepr.metadata)
  }
 
  private def offsetToInternalOffset(offset: Offset): (UUID, Boolean) =

diff --git a/core/src/test/scala/akka/persistence/cassandra/query/DirectWriting.scala b/core/src/test/scala/akka/persistence/cassandra/query/DirectWriting.scala
@@ -30,7 +30,7 @@ trait DirectWriting extends BeforeAndAfterAll {
 
  private lazy val writeStatements: CassandraJournalStatements = new CassandraJournalStatements(settings)
 
- private lazy val preparedWriteMessage = cluster.prepare(writeStatements.writeMessage(withMeta = false))
+ private lazy val preparedWriteMessage = cluster.prepare(writeStatements.writeMessage(withMeta = true))
 
  private lazy val preparedDeleteMessage = cluster.prepare(writeStatements.deleteMessage)
 
@@ -42,7 +42,7 @@ trait DirectWriting extends BeforeAndAfterAll {
  val now = Uuids.unixTimestamp(nowUuid)
  val serManifest = Serializers.manifestFor(serializer, persistent)
 
- val bs = preparedWriteMessage
+ var bs = preparedWriteMessage
  .bind()
  .setString("persistence_id", persistent.persistenceId)
  .setLong("partition_nr", partitionNr)
@@ -53,6 +53,20 @@ trait DirectWriting extends BeforeAndAfterAll {
  .setString("ser_manifest", serManifest)
  .setString("event_manifest", persistent.manifest)
  .setByteBuffer("event", serialized)
+
+ bs = persistent.metadata match {
+ case Some(meta) =>
+ val metaPayload = meta.asInstanceOf[AnyRef]
+ val metaSerializer = serialization.findSerializerFor(metaPayload)
+ val metaSerialized = ByteBuffer.wrap(serialization.serialize(metaPayload).get)
+ val metaSerializedManifest = Serializers.manifestFor(metaSerializer, metaPayload)
+ bs.setString("meta_ser_manifest", metaSerializedManifest)
+ .setInt("meta_ser_id", metaSerializer.identifier)
+ .setByteBuffer("meta", metaSerialized)
+ case _ =>
+ bs
+ }
+
  cluster.execute(bs)
  system.log.debug("Directly wrote payload [{}] for entity [{}]", persistent.payload, persistent.persistenceId)
  }

diff --git a/core/src/test/scala/akka/persistence/cassandra/query/EventsByPersistenceIdSpec.scala b/core/src/test/scala/akka/persistence/cassandra/query/EventsByPersistenceIdSpec.scala
@@ -284,6 +284,14 @@ class EventsByPersistenceIdSpec extends CassandraSpec(EventsByPersistenceIdSpec.
  }
  }
  }
+
+ "return metadata" in {
+ val meta = "cats"
+ val pr1 = PersistentRepr("e1", 1L, "with-meta", "").withMetadata(meta)
+ writeTestEvent(pr1)
+ val src = queries.currentEventsByPersistenceId("with-meta", 0L, Long.MaxValue)
+ src.map(_.eventMetadata).runWith(TestSink.probe[Any]).request(2).expectNext(Some(meta)).expectComplete()
+ }
  }
 
  "Cassandra live query EventsByPersistenceId" must {