Delta value from Datomic txReport using Datomisca

queryTxReport.scala

import datomisca._
import scala.concurrent._
import scala.concurrent.duration.Duration
import datomisca.executioncontext.ExecutionContextHelper._

object TxReportQuery {

  object PersonSchema {

    object ns {
      val person = new Namespace("person") {
        val character = Namespace("person.character")
      }
    }

    val name = Attribute(ns.person / "name", SchemaType.string, Cardinality.one).withDoc("The name of a person")
    val email = Attribute(ns.person / "email", SchemaType.string, Cardinality.one).withDoc("The email address of a person")

    val txData = Seq(name, email)
  }

  def main(args: Array[String]) {

    val uri = "datomic:mem:transaction_reporting"

    Datomic.createDatabase(uri)

    implicit val conn = Datomic.connect(uri)

    Await.result(Datomic.transact(PersonSchema.txData), Duration("2 seconds"))

    val rs = conn.txReportQueue.stream

    val johnId = DId(Partition.USER)
    val john = SchemaEntity.add(johnId)(Props() +
      (PersonSchema.name -> "John") +
      (PersonSchema.email -> "[email protected]"))

    val fid = Datomic.transact(john) map { tx => tx.resolve(johnId) }
    val realJohnId = Await.result(fid, Duration("2 seconds"))

    val updateJohn = SchemaEntity.add(FinalId(realJohnId))(Props() +
      (PersonSchema.email -> "[email protected]"))
    Await.result(Datomic.transact(updateJohn), Duration("2 seconds"))

    val queryByReport = Query("""
         [:find ?name ?v ?added ?prev
             :in $db $tx 
             :where
              [$tx ?e ?emailId ?v _ ?added]
              [$db ?emailId :db/ident :person/email  ]
              [$db ?e :person/email ?prev ]
              [$db ?e :person/name ?name]]
""")

    val emailId = conn.database.entid(PersonSchema.ns.person / "email")

    val qe = rs.tail.head.get
    println(Datomic.q(queryByReport, qe.dbBefore, DColl(qe.txData)))

    val qe2 = rs.tail.tail.head.get
    Datomic.q(queryByReport, qe2.dbBefore, DColl(qe2.txData)).
      collect { case (DString(name), DString(after), DBoolean(true), DString(before)) => (name, before, after) }

    conn.removeTxReportQueue

    conn.release

    Datomic.shutdown(true)

    defaultExecutorService.shutdownNow()
  }
}