Tagged pubsub (#258)

* Unwrap tagged events in pubsub
* ignore metals
* move unwrapping of Tagged payloads into PubSub

Author: leviramsey

.gitignore

@@ -15,3 +15,4 @@ target/
 ext-lib-src/
 .classpath_nb
 .bsp
+metals.sbt

core/src/main/scala/akka/persistence/r2dbc/internal/PubSub.scala

@@ -17,6 +17,7 @@ import akka.actor.typed.pubsub.Topic
 import akka.annotation.InternalApi
 import akka.persistence.Persistence
 import akka.persistence.PersistentRepr
+import akka.persistence.journal.Tagged
 import akka.persistence.query.TimestampOffset
 import akka.persistence.query.typed.EventEnvelope
 import akka.persistence.typed.PersistenceId
@@ -55,11 +56,20 @@ import akka.persistence.typed.PersistenceId
     val slice = persistenceExt.sliceForPersistenceId(pid)
 
     val offset = TimestampOffset(timestamp, timestamp, Map(pid -> pr.sequenceNr))
+    val payload =
+      pr.payload match {
+        case Tagged(payload, _) =>
+          // eventsByTag not implemented (see issue #82), but events can still be tagged, so we unwrap this tagged event.
+          payload
+
+        case other => other
+      }
+
     val envelope = new EventEnvelope(
       offset,
       pid,
       pr.sequenceNr,
-      Option(pr.payload),
+      Option(payload),
       timestamp.toEpochMilli,
       pr.metadata,
       entityType,

core/src/main/scala/akka/persistence/r2dbc/journal/R2dbcJournal.scala

@@ -183,18 +183,20 @@ private[r2dbc] final class R2dbcJournal(config: Config, cfgPath: String) extends
     writeAndPublishResult.map(_ => Nil)(ExecutionContexts.parasitic)
   }
 
-  private def publish(messages: immutable.Seq[AtomicWrite], dbTimestamp: Future[Instant]): Future[Done] = {
-    if (pubSub.isDefined) {
-      dbTimestamp.map { timestamp =>
-        pubSub.foreach { p =>
-          messages.iterator.flatMap(_.payload.iterator).foreach(pr => p.publish(pr, timestamp))
+  private def publish(messages: immutable.Seq[AtomicWrite], dbTimestamp: Future[Instant]): Future[Done] =
+    pubSub match {
+      case Some(ps) =>
+        dbTimestamp.map { timestamp =>
+          messages.iterator
+            .flatMap(_.payload.iterator)
+            .foreach(pr => ps.publish(pr, timestamp))
+
+          Done
         }
-        Done
-      }
-    } else {
-      dbTimestamp.map(_ => Done)(ExecutionContexts.parasitic)
+
+      case None =>
+        dbTimestamp.map(_ => Done)(ExecutionContexts.parasitic)
     }
-  }
 
   private def logEventsByTagsNotImplemented(): Unit = {
     if (!eventsByTagNotImplementedLogged) {