Migrate ZmqWatcher to akka-typed (#1759)

And re-work watch types and bitcoin events.
The type architecture is cleaner and more future-proof.
More tests have also been added.

Author: t-bast

eclair-core/src/main/scala/fr/acinq/eclair/Logs.scala

@@ -20,7 +20,7 @@ import akka.event.DiagnosticLoggingAdapter
 import akka.io.Tcp
 import fr.acinq.bitcoin.ByteVector32
 import fr.acinq.bitcoin.Crypto.PublicKey
-import fr.acinq.eclair.blockchain.ValidateResult
+import fr.acinq.eclair.blockchain.bitcoind.ZmqWatcher.ValidateResult
 import fr.acinq.eclair.channel.{LocalChannelDown, LocalChannelUpdate}
 import fr.acinq.eclair.crypto.TransportHandler.HandshakeCompleted
 import fr.acinq.eclair.io.Peer.PeerRoutingMessage

eclair-core/src/main/scala/fr/acinq/eclair/Setup.scala

@@ -17,7 +17,10 @@
 package fr.acinq.eclair
 
 import akka.Done
-import akka.actor.{ActorContext, ActorRef, ActorSystem, Props, SupervisorStrategy}
+import akka.actor.typed
+import akka.actor.typed.scaladsl.Behaviors
+import akka.actor.typed.scaladsl.adapter.ClassicActorSystemOps
+import akka.actor.{ActorRef, ActorSystem, Props, SupervisorStrategy}
 import akka.pattern.after
 import akka.util.Timeout
 import com.softwaremill.sttp.okhttp.OkHttpFutureBackend
@@ -28,7 +31,7 @@ import fr.acinq.eclair.blockchain.bitcoind.rpc.{BasicBitcoinJsonRPCClient, Batch
 import fr.acinq.eclair.blockchain.bitcoind.zmq.ZMQActor
 import fr.acinq.eclair.blockchain.bitcoind.{BitcoinCoreWallet, ZmqWatcher}
 import fr.acinq.eclair.blockchain.fee._
-import fr.acinq.eclair.channel.{Channel, Register, TxPublisher}
+import fr.acinq.eclair.channel.{Channel, Register}
 import fr.acinq.eclair.crypto.keymanager.{LocalChannelKeyManager, LocalNodeKeyManager}
 import fr.acinq.eclair.db.Databases.FileBackup
 import fr.acinq.eclair.db.{Databases, DbEventHandler, FileBackupHandler}
@@ -234,7 +237,7 @@ class Setup(datadir: File,
       watcher = {
         system.actorOf(SimpleSupervisor.props(Props(new ZMQActor(config.getString("bitcoind.zmqblock"), Some(zmqBlockConnected))), "zmqblock", SupervisorStrategy.Restart))
         system.actorOf(SimpleSupervisor.props(Props(new ZMQActor(config.getString("bitcoind.zmqtx"), Some(zmqTxConnected))), "zmqtx", SupervisorStrategy.Restart))
-        system.actorOf(SimpleSupervisor.props(ZmqWatcher.props(nodeParams.chainHash, blockCount, extendedBitcoinClient), "watcher", SupervisorStrategy.Resume))
+        system.spawn(Behaviors.supervise(ZmqWatcher(nodeParams.chainHash, blockCount, extendedBitcoinClient)).onFailure(typed.SupervisorStrategy.resume), "watcher")
       }
 
       router = system.actorOf(SimpleSupervisor.props(Router.props(nodeParams, watcher, Some(routerInitialized)), "router", SupervisorStrategy.Resume))
@@ -347,7 +350,7 @@ object Setup {
 
 case class Kit(nodeParams: NodeParams,
                system: ActorSystem,
-               watcher: ActorRef,
+               watcher: typed.ActorRef[ZmqWatcher.Command],
                paymentHandler: ActorRef,
                register: ActorRef,
                relayer: ActorRef,