Add zeromq publisher for unconfirmed transactions and new blocks

build.sbt

@@ -62,6 +62,8 @@ libraryDependencies ++= Seq(
   "com.typesafe.akka" %% "akka-parsing" % akkaHttpVersion,
   "com.typesafe.akka" %% "akka-stream" % akkaVersion,
   "org.bitlet" % "weupnp" % "0.1.4",
+
+  "org.zeromq" % "jeromq" % "0.5.3",
 
   // api dependencies 
   "io.circe" %% "circe-core" % circeVersion,

src/main/scala/org/ergoplatform/ErgoApp.scala

@@ -12,7 +12,7 @@ import org.ergoplatform.mining.ErgoMiner.StartMining
 import org.ergoplatform.network.{ErgoNodeViewSynchronizer, ErgoSyncTracker}
 import org.ergoplatform.nodeView.history.ErgoSyncInfoMessageSpec
 import org.ergoplatform.nodeView.history.extra.ExtraIndexer
-import org.ergoplatform.nodeView.{ErgoNodeViewRef, ErgoReadersHolderRef}
+import org.ergoplatform.nodeView.{ErgoEventPublisherRef, ErgoNodeViewRef, ErgoReadersHolderRef}
 import org.ergoplatform.settings.{Args, ErgoSettings, NetworkType}
 import scorex.core.api.http._
 import scorex.core.app.ScorexContext
@@ -98,6 +98,8 @@ class ErgoApp(args: Args) extends ScorexLogging {
 
   private val readersHolderRef: ActorRef = ErgoReadersHolderRef(nodeViewHolderRef)
 
+  ErgoEventPublisherRef()
+
   // Create an instance of ErgoMiner actor if "mining = true" in config
   private val minerRefOpt: Option[ActorRef] =
     if (ergoSettings.nodeSettings.mining) {

src/main/scala/org/ergoplatform/nodeView/ErgoEventPublisher.scala

@@ -0,0 +1,42 @@
+package org.ergoplatform.nodeView
+
+import akka.actor.{Actor, ActorRef, ActorRefFactory, Props}
+import org.ergoplatform.modifiers.history.header.Header
+import org.zeromq.SocketType
+import org.zeromq.ZMQ
+import org.zeromq.ZContext
+import org.ergoplatform.modifiers.mempool.UnconfirmedTransaction
+import org.ergoplatform.network.ErgoNodeViewSynchronizer.ReceivableMessages.{FullBlockApplied, SuccessfulTransaction}
+import scorex.util.ScorexLogging
+
+class ErgoEventPublisher(socket: ZMQ.Socket) extends Actor with ScorexLogging {
+
+  override def preStart(): Unit = {
+    context.system.eventStream.subscribe(self, classOf[SuccessfulTransaction])
+    context.system.eventStream.subscribe(self, classOf[FullBlockApplied])
+  }
+
+  @SuppressWarnings(Array("IsInstanceOf"))
+  override def receive: Receive = {
+    case SuccessfulTransaction(transaction: UnconfirmedTransaction) =>
+      socket.send(s"utx${transaction.id}")
+
+    case FullBlockApplied(header: Header) =>
+      socket.send(s"blk${header.id}${header.height.toString}")
+
+    case a: Any => log.warn(s"ErgoEventPublisher got improper input: $a")
+  }
+}
+
+object ErgoEventPublisherRef {
+
+  def apply()
+           (implicit context: ActorRefFactory): ActorRef = {
+    val zContext: ZContext = new ZContext()
+    val socket = zContext.createSocket(SocketType.PUB)
+    socket.bind("tcp://127.0.0.1:5555")
+    val props = Props(new ErgoEventPublisher(socket))
+    context.actorOf(props)
+  }
+
+}