From 2aaee675a6e292d074f775f4f794695882340cd3 Mon Sep 17 00:00:00 2001
From: Damien Bailly <damien.bailly@gmail.com>
Date: Thu, 12 Jan 2017 13:42:35 +0000
Subject: [PATCH] Wraps CombinedReadEventAdapter into NoopWriteEventAdapter

When multiple event-adapters are configured they are combined into a single instance of `CombinedReadEventAdapter`.

However `CombinedReadEventAdpater` is not just a `ReadEventAdapter` but a full `EventAdapter`
which throws an `IllegalStateException("CombinedReadEventAdapter must not be used when writing (creating manifests) events!")`
when its `toJournal` method is called.

The `CombinedReadEventAdapter` is not wrapped into `NoopWriteEventAdapter` (because it's not a `ReadEventAdapter`)
and therefore it doesn't avoid the `toJournal` calls.

This is especially problematic when multiple `ReadEventAdapter` are combined together as illustrated in the new testcase.
---
 .../persistence/journal/EventAdapters.scala    |  2 +-
 .../journal/InmemEventAdaptersSpec.scala       | 18 ++++++++++++++++++
 2 files changed, 19 insertions(+), 1 deletion(-)

diff --git a/akka-persistence/src/main/scala/akka/persistence/journal/EventAdapters.scala b/akka-persistence/src/main/scala/akka/persistence/journal/EventAdapters.scala
index bf6fd3fd61..d31253fbcb 100644
--- a/akka-persistence/src/main/scala/akka/persistence/journal/EventAdapters.scala
+++ b/akka-persistence/src/main/scala/akka/persistence/journal/EventAdapters.scala
@@ -92,7 +92,7 @@ private[akka] object EventAdapters {
     val bindings: immutable.Seq[ClassHandler] = {
       val bs = for ((k: FQN, as: BoundAdapters) ← adapterBindings)
         yield if (as.size == 1) (system.dynamicAccess.getClassFor[Any](k).get, handlers(as.head))
-      else (system.dynamicAccess.getClassFor[Any](k).get, CombinedReadEventAdapter(as.map(handlers)))
+      else (system.dynamicAccess.getClassFor[Any](k).get, NoopWriteEventAdapter(CombinedReadEventAdapter(as.map(handlers))))
 
       sort(bs)
     }
diff --git a/akka-persistence/src/test/scala/akka/persistence/journal/InmemEventAdaptersSpec.scala b/akka-persistence/src/test/scala/akka/persistence/journal/InmemEventAdaptersSpec.scala
index b2a0180dca..6c9bd03c68 100644
--- a/akka-persistence/src/test/scala/akka/persistence/journal/InmemEventAdaptersSpec.scala
+++ b/akka-persistence/src/test/scala/akka/persistence/journal/InmemEventAdaptersSpec.scala
@@ -29,6 +29,7 @@ class InmemEventAdaptersSpec extends AkkaSpec {
       |      precise  = ${classOf[PreciseAdapter].getCanonicalName}
       |      reader  = ${classOf[ReaderAdapter].getCanonicalName}
       |      writer  = ${classOf[WriterAdapter].getCanonicalName}
+      |      another-reader = ${classOf[AnotherReaderAdapter].getCanonicalName}
       |    }
       |    event-adapter-bindings = {
       |      "${classOf[EventMarkerInterface].getCanonicalName}" = marker
@@ -36,6 +37,7 @@ class InmemEventAdaptersSpec extends AkkaSpec {
       |      "akka.persistence.journal.PreciseAdapterEvent" = precise
       |      "akka.persistence.journal.ReadMeEvent" = reader
       |      "akka.persistence.journal.WriteMeEvent" = writer
+      |      "akka.persistence.journal.ReadMeTwiceEvent" = [reader, another-reader]
       |    }
       |  }
       |}
@@ -99,6 +101,17 @@ class InmemEventAdaptersSpec extends AkkaSpec {
       val w: EventAdapter = adapters.get(classOf[WriteMeEvent])
       w.fromJournal(w.toJournal(WriteMeEvent()), "").events.head.toString should ===("to-WriteMeEvent()")
     }
+
+    "allow combining only the read-side (CombinedReadEventAdapter)" in {
+      val adapters = EventAdapters(extendedActorSystem, inmemConfig)
+
+      // combined-read-side only adapter
+      val r: EventAdapter = adapters.get(classOf[ReadMeTwiceEvent])
+      r.fromJournal(r.toJournal(ReadMeTwiceEvent()), "").events.map(_.toString) shouldBe Seq(
+        "from-ReadMeTwiceEvent()",
+        "again-ReadMeTwiceEvent()"
+      )
+    }
   }
 
 }
@@ -117,10 +130,15 @@ class PreciseAdapter extends BaseTestAdapter {
 }
 
 case class ReadMeEvent()
+case class ReadMeTwiceEvent()
 class ReaderAdapter extends ReadEventAdapter {
   override def fromJournal(event: Any, manifest: String): EventSeq =
     EventSeq("from-" + event)
 }
+class AnotherReaderAdapter extends ReadEventAdapter {
+  override def fromJournal(event: Any, manifest: String): EventSeq =
+    EventSeq("again-" + event)
+}
 
 case class WriteMeEvent()
 class WriterAdapter extends WriteEventAdapter {