Merge branch 'master' into eclair_bench

eclair-rpc-test/src/test/scala/org/bitcoins/eclair/rpc/EclairRpcClientTest.scala

@@ -554,6 +554,12 @@ class EclairRpcClientTest extends BitcoinSAsyncTest {
             _ = assert(channels.exists(_.state == ChannelState.NORMAL),
                        "Nodes did not have open channel!")
             preimage = PaymentPreimage.random
+            wsEventP = Promise[WebSocketEvent]
+            _ <- client.connectToWebSocket({ event =>
+              if (!wsEventP.isCompleted) {
+                wsEventP.success(event)
+              }
+            })
             invoice <- otherClient.createInvoice("foo", amt, preimage)
             paymentId <- client.sendToNode(otherClientNodeId,
                                            amt,
@@ -562,13 +568,19 @@ class EclairRpcClientTest extends BitcoinSAsyncTest {
                                            None,
                                            None,
                                            Some("ext_id"))
-            _ <- EclairRpcTestUtil.awaitUntilPaymentSucceeded(client, paymentId)
+            wsEvent <- wsEventP.future
             succeeded <- client.getSentInfo(invoice.lnTags.paymentHash.hash)
             _ <- client.close(channelId)
             bitcoind <- bitcoindRpcClientF
             address <- bitcoind.getNewAddress
             _ <- bitcoind.generateToAddress(6, address)
           } yield {
+            assert(wsEvent.isInstanceOf[WebSocketEvent.PaymentSent])
+            val paymentSent = wsEvent.asInstanceOf[WebSocketEvent.PaymentSent]
+            assert(paymentSent.parts.nonEmpty)
+            assert(paymentSent.id == paymentId)
+            assert(paymentSent.parts.head.amount == amt)
+            assert(paymentSent.parts.head.id == paymentId)
             assert(succeeded.nonEmpty)
 
             val succeededPayment = succeeded.head

eclair-rpc/src/main/scala/org/bitcoins/eclair/rpc/api/EclairApi.scala

@@ -262,4 +262,7 @@ trait EclairApi {
       externalId: Option[String]): Future[PaymentId]
 
   def usableBalances(): Future[Vector[UsableBalancesResult]]
+
+  /** Connects to the Eclair web socket end point and passes [[WebSocketEvent]]s to the given [[eventHandler]] */
+  def connectToWebSocket(eventHandler: WebSocketEvent => Unit): Future[Unit]
 }

eclair-rpc/src/main/scala/org/bitcoins/eclair/rpc/client/EclairRpcClient.scala

@@ -842,7 +842,9 @@ class EclairRpcClient(val instance: EclairInstance, binary: Option[File] = None)
     f
   }
 
-  def connectToWebSocket[T](f: WebSocketEvent => T): Future[Unit] = {
+  /** @inheritdoc */
+  override def connectToWebSocket(
+      eventHandler: WebSocketEvent => Unit): Future[Unit] = {
     val incoming: Sink[Message, Future[Done]] =
       Sink.foreach[Message] {
         case message: TextMessage.Strict =>
@@ -850,7 +852,8 @@ class EclairRpcClient(val instance: EclairInstance, binary: Option[File] = None)
           val validated: JsResult[WebSocketEvent] =
             parsed.validate[WebSocketEvent]
           val event = parseResult[WebSocketEvent](validated, parsed, "ws")
-          f(event)
+          eventHandler(event)
+        case _: Message => ()
       }
 
     val flow =
@@ -861,7 +864,7 @@ class EclairRpcClient(val instance: EclairInstance, binary: Option[File] = None)
     instance.authCredentials.bitcoinAuthOpt
     val request = WebSocketRequest(
       uri,
-      extraHeaders = Seq(
+      extraHeaders = Vector(
         Authorization(
           BasicHttpCredentials("", instance.authCredentials.password))))
     val (upgradeResponse, _) = Http().singleWebSocketRequest(request, flow)