configure akka for remoting, add payment listener to channels

eclair-node/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>fr.acinq.eclair</groupId>
         <artifactId>eclair_2.11</artifactId>
-        <version>0.2-SNAPSHOT</version>
+        <version>0.2-REMOTE-SNAPSHOT</version>
     </parent>
 
     <artifactId>eclair-node_2.11</artifactId>
@@ -123,7 +123,8 @@
                 <activeByDefault>true</activeByDefault>
             </activation>
             <properties>
-                <bitcoind.url>https://bitcoin.org/bin/bitcoin-core-0.14.0/bitcoin-0.14.0-x86_64-linux-gnu.tar.gz</bitcoind.url>
+                <bitcoind.url>https://bitcoin.org/bin/bitcoin-core-0.14.0/bitcoin-0.14.0-x86_64-linux-gnu.tar.gz
+                </bitcoind.url>
                 <bitcoind.md5>c811c157d4d618f7d7f4b9f24834551c</bitcoind.md5>
                 <bitcoind.sha1>3ab7e537bd00bf35e6a78fca108d0d886f8289c1</bitcoind.sha1>
             </properties>
@@ -150,6 +151,11 @@
             <artifactId>akka-actor_${scala.version.short}</artifactId>
             <version>${akka.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.typesafe.akka</groupId>
+            <artifactId>akka-remote_${scala.version.short}</artifactId>
+            <version>${akka.version}</version>
+        </dependency>
         <dependency>
             <groupId>com.typesafe.akka</groupId>
             <artifactId>akka-slf4j_${scala.version.short}</artifactId>

eclair-node/src/main/resources/application.conf

@@ -10,10 +10,10 @@ eclair {
   }
   bitcoind {
     host = "localhost"
-    rpcport = 18332
+    rpcport = 18334
     rpcuser = "foo"
     rpcpassword = "bar"
-    zmq = "tcp://127.0.0.1:29000"
+    zmq = "tcp://127.0.0.1:30000"
   }
 
   node-alias = "eclair"
@@ -45,6 +45,7 @@ eclair {
 
   payment-handler = "local"
 }
+
 akka {
   loggers = ["akka.event.slf4j.Slf4jLogger"]
   loglevel = "DEBUG"
@@ -54,6 +55,15 @@ akka {
       # enable DEBUG logging of all LoggingFSMs for events, transitions and timers
       fsm = on
     }
+    provider = remote
+  }
+
+  remote {
+    enabled-transports = ["akka.remote.netty.tcp"]
+    netty.tcp {
+      hostname = "127.0.0.1"
+      port = 2550
+    }
   }
 
   http {

eclair-node/src/main/scala/fr/acinq/eclair/channel/Channel.scala

@@ -1,6 +1,6 @@
 package fr.acinq.eclair.channel
 
-import akka.actor.{ActorRef, FSM, LoggingFSM, OneForOneStrategy, Props, SupervisorStrategy}
+import akka.actor.{ActorRef, FSM, LoggingFSM, OneForOneStrategy, Props, SupervisorStrategy, Terminated}
 import fr.acinq.bitcoin.Crypto.{PrivateKey, PublicKey}
 import fr.acinq.bitcoin._
 import fr.acinq.eclair._
@@ -29,6 +29,7 @@ object Channel {
 class Channel(val nodeParams: NodeParams, remoteNodeId: PublicKey, blockchain: ActorRef, router: ActorRef, relayer: ActorRef)(implicit ec: ExecutionContext = ExecutionContext.Implicits.global) extends LoggingFSM[State, Data] {
 
   val forwarder = context.actorOf(Props(new Forwarder(nodeParams)), "forwarder")
+  var listener: Option[ActorRef] = None
 
   /*
           8888888 888b    888 8888888 88888888888
@@ -425,7 +426,9 @@ class Channel(val nodeParams: NodeParams, remoteNodeId: PublicKey, blockchain: A
 
     case Event(add: UpdateAddHtlc, d@DATA_NORMAL(commitments, _)) =>
       Try(Commitments.receiveAdd(commitments, add)) match {
-        case Success(commitments1) => goto(stateName) using d.copy(commitments = commitments1)
+        case Success(commitments1) =>
+          listener.map(l => l ! add)
+          goto(stateName) using d.copy(commitments = commitments1)
         case Failure(cause) => handleLocalError(cause, d)
       }
 
@@ -440,6 +443,7 @@ class Channel(val nodeParams: NodeParams, remoteNodeId: PublicKey, blockchain: A
     case Event(fulfill@UpdateFulfillHtlc(_, id, r), d@DATA_NORMAL(commitments, _)) =>
       Try(Commitments.receiveFulfill(d.commitments, fulfill)) match {
         case Success(Right(commitments1)) =>
+          listener.map(l => l ! fulfill)
           relayer ! ForwardFulfill(fulfill)
           goto(stateName) using d.copy(commitments = commitments1)
         case Success(Left(_)) => goto(stateName)
@@ -988,6 +992,24 @@ class Channel(val nodeParams: NodeParams, remoteNodeId: PublicKey, blockchain: A
 
     // we only care about this event in NORMAL and SHUTDOWN state, and we never unregister to the event stream
     case Event(CurrentBlockCount(_), _) => stay
+
+    case Event('ping, _) =>
+      sender ! 'pong
+      stay
+
+    case Event('setListener, _) =>
+      listener = Some(sender)
+      context.watch(sender)
+      log.info(s"adding listener $sender")
+      sender ! 'ok
+      stay()
+
+    case Event(Terminated(actor), _) if listener == Some(actor) =>
+      // this can be the connection or the listener, either way it is a cause of death
+      log.info(s"lost listener $actor")
+      listener = None
+      stay()
+
   }
 
   onTransition {

pom.xml

@@ -4,7 +4,7 @@
 
     <groupId>fr.acinq.eclair</groupId>
     <artifactId>eclair_2.11</artifactId>
-    <version>0.2-SNAPSHOT</version>
+    <version>0.2-REMOTE-SNAPSHOT</version>
     <packaging>pom</packaging>
 
     <modules>