DLC connection checks and notifications (#4720)

* DLC connection checks and notifications

* asynchronous connection checks

app-commons/src/main/scala/org/bitcoins/commons/jsonmodels/ws/WsModels.scala

@@ -9,8 +9,11 @@ import org.bitcoins.core.protocol.dlc.models.DLCStatus
 import org.bitcoins.core.protocol.transaction.Transaction
 import org.bitcoins.core.wallet.fee.FeeUnit
 import org.bitcoins.crypto.{Sha256Digest, StringFactory}
+import ujson.Value
 
-/** The event type being sent over the websocket. An example is [[WalletWsType.BlockProcessed]] */
+import java.net.InetSocketAddress
+
+/** The event type being sent over the websocket. An example is [[WalletWsType.NewAddress]] */
 sealed trait WsType
 
 object WsType extends StringFactory[WsType] {
@@ -27,6 +30,7 @@ object WsType extends StringFactory[WsType] {
 sealed trait WalletWsType extends WsType
 sealed trait ChainWsType extends WsType
 sealed trait TorWsType extends WsType
+sealed trait DLCNodeWsType extends WsType
 
 object WalletWsType extends StringFactory[WalletWsType] {
   case object TxProcessed extends WalletWsType
@@ -92,6 +96,25 @@ object TorWsType extends StringFactory[TorWsType] {
   }
 }
 
+object DLCNodeWsType extends StringFactory[DLCNodeWsType] {
+  case object DLCConnectionInitiated extends DLCNodeWsType
+  case object DLCConnectionEstablished extends DLCNodeWsType
+  case object DLCConnectionFailed extends DLCNodeWsType
+
+  private val all = Vector(DLCConnectionInitiated,
+                           DLCConnectionEstablished,
+                           DLCConnectionFailed)
+
+  override def fromStringOpt(string: String): Option[DLCNodeWsType] = {
+    all.find(_.toString.toLowerCase() == string.toLowerCase)
+  }
+
+  override def fromString(string: String): DLCNodeWsType = {
+    fromStringOpt(string)
+      .getOrElse(sys.error(s"Cannot find chain ws type for string=$string"))
+  }
+}
+
 /** A notification that we send over the websocket.
   * The type of the notification is indicated by [[WsType]].
   * An example is [[org.bitcoins.commons.jsonmodels.ws.WalletNotification.NewAddressNotification]]
@@ -115,6 +138,10 @@ sealed trait TorNotification[T] extends WsNotification[T] {
   override def `type`: TorWsType
 }
 
+sealed trait DLCNodeNotification[T] extends WsNotification[T] {
+  override def `type`: DLCNodeWsType
+}
+
 object WalletNotification {
 
   case class NewAddressNotification(payload: BitcoinAddress)
@@ -231,3 +258,30 @@ object TorNotification {
     }
   }
 }
+
+object DLCNodeNotification {
+
+  case class DLCNodeConnectionInitiated(payload: InetSocketAddress)
+      extends DLCNodeNotification[InetSocketAddress] {
+    override def `type`: DLCNodeWsType = DLCNodeWsType.DLCConnectionInitiated
+
+    override def json: Value = upickle.default.writeJs(this)(
+      WsPicklers.dlcNodeConnectionInitiatedPickler)
+  }
+
+  case class DLCNodeConnectionEstablished(payload: InetSocketAddress)
+      extends DLCNodeNotification[InetSocketAddress] {
+    override def `type`: DLCNodeWsType = DLCNodeWsType.DLCConnectionEstablished
+
+    override def json: Value = upickle.default.writeJs(this)(
+      WsPicklers.dlcNodeConnectionEstablishedPickler)
+  }
+
+  case class DLCNodeConnectionFailed(payload: InetSocketAddress)
+      extends DLCNodeNotification[InetSocketAddress] {
+    override def `type`: DLCNodeWsType = DLCNodeWsType.DLCConnectionFailed
+
+    override def json: Value =
+      upickle.default.writeJs(this)(WsPicklers.dlcNodeConnectionFailedPickler)
+  }
+}

app-commons/src/main/scala/org/bitcoins/commons/rpc/Command.scala

@@ -1599,6 +1599,31 @@ object DLCContactRemove {
   }
 }
 
+case class DLCCheckConnection(address: InetSocketAddress)
+    extends CommandRpc
+    with AppServerCliCommand
+    with ServerJsonModels
+
+object DLCCheckConnection {
+
+  def fromJsArr(arr: ujson.Arr): Try[DLCCheckConnection] = {
+    arr.arr.toList match {
+      case addressJs :: Nil =>
+        Try {
+          val address = {
+            val uri = new URI(s"tcp://${addressJs.str}")
+            InetSocketAddress.createUnresolved(uri.getHost, uri.getPort)
+          }
+          DLCCheckConnection(address)
+        }
+      case other =>
+        val exn = new IllegalArgumentException(
+          s"Bad number or arguments to checkconnection, got=${other.length} expected=1")
+        Failure(exn)
+    }
+  }
+}
+
 case class LoadWallet(
     walletNameOpt: Option[String],
     passwordOpt: Option[AesPassword],

app-commons/src/main/scala/org/bitcoins/commons/serializers/WsPicklers.scala

@@ -4,6 +4,11 @@ import org.bitcoins.commons.jsonmodels.ws.ChainNotification.{
   BlockProcessedNotification,
   SyncFlagChangedNotification
 }
+import org.bitcoins.commons.jsonmodels.ws.DLCNodeNotification.{
+  DLCNodeConnectionEstablished,
+  DLCNodeConnectionFailed,
+  DLCNodeConnectionInitiated
+}
 import org.bitcoins.commons.jsonmodels.ws.WalletNotification.{
   DLCOfferAddNotification,
   DLCOfferRemoveNotification,
@@ -18,14 +23,20 @@ import org.bitcoins.commons.jsonmodels.ws.WalletNotification.{
 import org.bitcoins.commons.jsonmodels.ws.{
   ChainNotification,
   ChainWsType,
+  DLCNodeNotification,
+  DLCNodeWsType,
   TorNotification,
   TorWsType,
   WalletNotification,
   WalletWsType
 }
+import org.bitcoins.core.config.DLC
 import org.bitcoins.core.serializers.PicklerKeys
+import org.bitcoins.core.util.NetworkUtil
 import upickle.default._
 
+import java.net.InetSocketAddress
+
 object WsPicklers {
 
   implicit val chainWsTypePickler: ReadWriter[ChainWsType] = {
@@ -43,6 +54,11 @@ object WsPicklers {
       .bimap(_.toString.toLowerCase, str => TorWsType.fromString(str.str))
   }
 
+  implicit val dlcNodeWsTypePickler: ReadWriter[DLCNodeWsType] = {
+    readwriter[ujson.Str]
+      .bimap(_.toString.toLowerCase, str => DLCNodeWsType.fromString(str.str))
+  }
+
   private def writeChainNotification(
       notification: ChainNotification[_]): ujson.Obj = {
     val payloadJson: ujson.Value = notification match {
@@ -164,6 +180,46 @@ object WsPicklers {
     }
   }
 
+  private def writeDLCNodeNotification(
+      notification: DLCNodeNotification[_]): ujson.Obj = {
+    def addr2str(address: InetSocketAddress) =
+      address.getHostName + ":" + address.getPort
+    val payloadJson: ujson.Value = notification match {
+      case DLCNodeConnectionInitiated(address) =>
+        upickle.default.writeJs(addr2str(address))
+      case DLCNodeConnectionEstablished(address) =>
+        upickle.default.writeJs(addr2str(address))
+      case DLCNodeConnectionFailed(address) =>
+        upickle.default.writeJs(addr2str(address))
+    }
+    val notificationObj = ujson.Obj(
+      PicklerKeys.typeKey -> writeJs(notification.`type`),
+      PicklerKeys.payloadKey -> payloadJson
+    )
+    notificationObj
+  }
+
+  private def readDLCNodeNotification(
+      obj: ujson.Obj): DLCNodeNotification[_] = {
+    val typeObj = read[DLCNodeWsType](obj(PicklerKeys.typeKey))
+    val payloadObj = obj(PicklerKeys.payloadKey)
+
+    typeObj match {
+      case DLCNodeWsType.DLCConnectionInitiated =>
+        val address: InetSocketAddress =
+          NetworkUtil.parseInetSocketAddress(payloadObj.str, DLC.DefaultPort)
+        DLCNodeConnectionInitiated(address)
+      case DLCNodeWsType.DLCConnectionEstablished =>
+        val address: InetSocketAddress =
+          NetworkUtil.parseInetSocketAddress(payloadObj.str, DLC.DefaultPort)
+        DLCNodeConnectionEstablished(address)
+      case DLCNodeWsType.DLCConnectionFailed =>
+        val address: InetSocketAddress =
+          NetworkUtil.parseInetSocketAddress(payloadObj.str, DLC.DefaultPort)
+        DLCNodeConnectionFailed(address)
+    }
+  }
+
   implicit val newAddressPickler: ReadWriter[NewAddressNotification] = {
     readwriter[ujson.Obj].bimap(
       writeWalletNotification(_),
@@ -251,4 +307,25 @@ object WsPicklers {
         .asInstanceOf[TorNotification.TorStartedNotification.type])
   }
 
+  implicit val dlcNodeConnectionInitiatedPickler: ReadWriter[
+    DLCNodeConnectionInitiated] = {
+    readwriter[ujson.Obj].bimap(
+      writeDLCNodeNotification(_),
+      readDLCNodeNotification(_).asInstanceOf[DLCNodeConnectionInitiated])
+  }
+
+  implicit val dlcNodeConnectionFailedPickler: ReadWriter[
+    DLCNodeConnectionFailed] = {
+    readwriter[ujson.Obj].bimap(
+      writeDLCNodeNotification(_),
+      readDLCNodeNotification(_).asInstanceOf[DLCNodeConnectionFailed])
+  }
+
+  implicit val dlcNodeConnectionEstablishedPickler: ReadWriter[
+    DLCNodeConnectionEstablished] = {
+    readwriter[ujson.Obj].bimap(
+      writeDLCNodeNotification(_),
+      readDLCNodeNotification(_).asInstanceOf[DLCNodeConnectionEstablished])
+  }
+
 }

app/server/src/main/scala/org/bitcoins/server/BitcoinSServerMain.scala

@@ -298,6 +298,9 @@ class BitcoinSServerMain(override val serverArgParser: ServerArgParser)(implicit
     val torCallbacks = WebsocketUtil.buildTorCallbacks(wsQueue)
     torConf.addCallbacks(torCallbacks)
 
+    val dlcNodeCallbacks = WebsocketUtil.buildDLCNodeCallbacks(wsQueue)
+    dlcNodeConf.addCallbacks(dlcNodeCallbacks)
+
     ()
   }
 
@@ -459,6 +462,8 @@ class BitcoinSServerMain(override val serverArgParser: ServerArgParser)(implicit
         }
         dlcWalletCallbacks = WebsocketUtil.buildDLCWalletCallbacks(wsQueue)
         _ = dlcConfig.addCallbacks(dlcWalletCallbacks)
+        dlcNodeCallbacks = WebsocketUtil.buildDLCNodeCallbacks(wsQueue)
+        _ = dlcNodeConf.addCallbacks(dlcNodeCallbacks)
         _ <- startedTorConfigF
       } yield {
         logger.info(s"Done starting Main!")

app/server/src/main/scala/org/bitcoins/server/DLCRoutes.scala

@@ -7,19 +7,14 @@ import org.bitcoins.commons.rpc._
 import org.bitcoins.commons.serializers.Picklers
 import org.bitcoins.core.api.dlc.node.DLCNodeApi
 import org.bitcoins.core.api.dlc.wallet.db.IncomingDLCOfferDb
-import org.bitcoins.core.protocol.dlc.models.{
-  EnumSingleOracleInfo,
-  NumericSingleOracleInfo,
-  SingleContractInfo
-}
-import org.bitcoins.core.protocol.tlv.{
-  EnumEventDescriptorV0TLV,
-  NumericEventDescriptorTLV
-}
+import org.bitcoins.core.protocol.dlc.models.{EnumSingleOracleInfo, NumericSingleOracleInfo, SingleContractInfo}
+import org.bitcoins.core.protocol.tlv.{EnumEventDescriptorV0TLV, NumericEventDescriptorTLV}
 import org.bitcoins.server.routes._
 import ujson._
 import upickle.default._
 
+import scala.concurrent.Future
+
 case class DLCRoutes(dlcNode: DLCNodeApi)(implicit system: ActorSystem)
     extends ServerRoute {
 
@@ -186,5 +181,13 @@ case class DLCRoutes(dlcNode: DLCNodeApi)(implicit system: ActorSystem)
           }
       }
 
+    case ServerCommand("checkconnection", arr) =>
+      withValidServerCommand(DLCCheckConnection.fromJsArr(arr)) { addr =>
+        complete {
+          Future(dlcNode.checkPeerConnection(addr.address)).map { _ =>
+            Server.httpSuccess("initiated")
+          }
+        }
+      }
   }
 }

app/server/src/main/scala/org/bitcoins/server/util/WebsocketUtil.scala

@@ -12,6 +12,7 @@ import org.bitcoins.commons.jsonmodels.bitcoind.GetBlockHeaderResult
 import org.bitcoins.commons.jsonmodels.ws.TorNotification.TorStartedNotification
 import org.bitcoins.commons.jsonmodels.ws.{
   ChainNotification,
+  DLCNodeNotification,
   WalletNotification,
   WalletWsType,
   WsNotification
@@ -23,6 +24,12 @@ import org.bitcoins.core.protocol.dlc.models.DLCStatus
 import org.bitcoins.core.protocol.transaction.Transaction
 import org.bitcoins.core.util.FutureUtil
 import org.bitcoins.crypto.{DoubleSha256DigestBE, Sha256Digest}
+import org.bitcoins.dlc.node.{
+  DLCNodeCallbacks,
+  OnPeerConnectionEstablished,
+  OnPeerConnectionFailed,
+  OnPeerConnectionInitiated
+}
 import org.bitcoins.dlc.wallet.{
   DLCWalletCallbacks,
   OnDLCOfferAdd,
@@ -209,4 +216,35 @@ object WebsocketUtil extends Logging {
     onDLCStateChange(onStateChange) + onDLCOfferAdd(
       onOfferAdd) + onDLCOfferRemove(onOfferRemove)
   }
+
+  def buildDLCNodeCallbacks(
+      walletQueue: SourceQueueWithComplete[WsNotification[_]])(implicit
+      ec: ExecutionContext): DLCNodeCallbacks = {
+
+    val onConnectionInitiated: OnPeerConnectionInitiated = { payload =>
+      val notification =
+        DLCNodeNotification.DLCNodeConnectionInitiated(payload)
+      val offerF = walletQueue.offer(notification)
+      offerF.map(_ => ())
+    }
+
+    val onConnectionEstablished: OnPeerConnectionEstablished = { payload =>
+      val notification =
+        DLCNodeNotification.DLCNodeConnectionEstablished(payload)
+      val offerF = walletQueue.offer(notification)
+      offerF.map(_ => ())
+    }
+
+    val onConnectionFailed: OnPeerConnectionFailed = { payload =>
+      val notification = DLCNodeNotification.DLCNodeConnectionFailed(payload)
+      val offerF = walletQueue.offer(notification)
+      offerF.map(_ => ())
+    }
+
+    import DLCNodeCallbacks._
+
+    onPeerConnectionInitiated(
+      onConnectionInitiated) + onPeerConnectionEstablished(
+      onConnectionEstablished) + onPeerConnectionFailed(onConnectionFailed)
+  }
 }

core/src/main/scala/org/bitcoins/core/api/dlc/node/DLCNodeApi.scala

@@ -31,5 +31,7 @@ trait DLCNodeApi extends StartStopAsync[Unit] {
       message: String,
       tempContractId: Sha256Digest): Future[Sha256Digest]
 
+  def checkPeerConnection(peerAddress: InetSocketAddress): Future[Unit]
+
   def getHostAddress: Future[InetSocketAddress]
 }

dlc-node-test/src/test/scala/org/bitcoins/dlc/node/DLCNodeTest.scala

@@ -2,12 +2,15 @@ package org.bitcoins.dlc.node
 
 import org.bitcoins.core.number.UInt32
 import org.bitcoins.core.protocol.dlc.models.DLCState
+import org.bitcoins.core.util.NetworkUtil
 import org.bitcoins.core.wallet.fee.SatoshisPerVirtualByte
 import org.bitcoins.testkit.async.TestAsyncUtil
 import org.bitcoins.testkit.dlc.BitcoinSDLCNodeTest
 import org.bitcoins.testkit.wallet.DLCWalletUtil._
 import org.scalatest.FutureOutcome
 
+import java.net.InetSocketAddress
+import scala.concurrent.{Future, Promise}
 import scala.concurrent.duration.DurationInt
 
 class DLCNodeTest extends BitcoinSDLCNodeTest {
@@ -17,6 +20,64 @@ class DLCNodeTest extends BitcoinSDLCNodeTest {
     witTwoFundedDLCNodes(test)
   }
 
+  it must "check a connection" in { nodes: (DLCNode, DLCNode) =>
+    val nodeA = nodes._1
+    val nodeB = nodes._2
+
+    val configA = nodeA.config
+
+    val errorP = Promise[String]()
+    val failure =
+      DLCNodeCallbacks.onPeerConnectionFailed(new OnPeerConnectionFailed {
+        override def apply(param: InetSocketAddress): Future[Unit] = {
+          errorP.success("err")
+          Future.unit
+        }
+      })
+    configA.addCallbacks(failure)
+
+    val okP = Promise[String]()
+    val established = DLCNodeCallbacks.onPeerConnectionEstablished(
+      new OnPeerConnectionEstablished {
+        override def apply(param: InetSocketAddress): Future[Unit] = {
+          okP.success("ok")
+          Future.unit
+        }
+      })
+    configA.addCallbacks(established)
+
+    val initiatedP = Promise[String]()
+    val init =
+      DLCNodeCallbacks.onPeerConnectionInitiated(new OnPeerConnectionInitiated {
+        override def apply(param: InetSocketAddress): Future[Unit] = {
+          initiatedP.success("init")
+          Future.unit
+        }
+      })
+    configA.addCallbacks(init)
+
+    for {
+      (addrB, _) <- nodeB.serverBindF
+      _ = assert(!initiatedP.isCompleted)
+      _ = assert(!errorP.isCompleted)
+      _ = assert(!okP.isCompleted)
+      // don't wait for the future completion here, we want to test OnPeerConnectionInitiated
+      _ = nodeA.checkPeerConnection(addrB)
+      initiated <- initiatedP.future
+      _ = assert(initiated == "init")
+      ok <- okP.future
+      _ = assert(ok == "ok")
+      _ = assert(!errorP.isCompleted)
+      invalidAddr = InetSocketAddress.createUnresolved(addrB.getHostString,
+                                                       NetworkUtil.randomPort())
+      _ <- recoverToSucceededIf[java.net.ConnectException](
+        nodeA.checkPeerConnection(invalidAddr))
+      error <- errorP.future
+    } yield {
+      assert(error == "err")
+    }
+  }
+
   it must "setup a DLC" in { nodes: (DLCNode, DLCNode) =>
     val nodeA = nodes._1
     val nodeB = nodes._2

dlc-node/src/main/scala/org/bitcoins/dlc/node/DLCClient.scala

@@ -51,6 +51,7 @@ class DLCClient(
     case c @ Tcp.CommandFailed(cmd: Tcp.Connect) =>
       val ex = c.cause.getOrElse(new IOException("Unknown Error"))
       log.error(s"Cannot connect to ${cmd.remoteAddress} ", ex)
+      connectedAddress.foreach(_.failure(ex))
       throw ex
     case Tcp.Connected(peerOrProxyAddress, _) =>
       val connection = sender()

dlc-node/src/main/scala/org/bitcoins/dlc/node/DLCNode.scala

@@ -13,6 +13,7 @@ import org.bitcoins.dlc.node.peer.Peer
 
 import java.net.InetSocketAddress
 import scala.concurrent._
+import scala.util.{Failure, Success}
 
 case class DLCNode(wallet: DLCWalletApi)(implicit
     system: ActorSystem,
@@ -111,15 +112,35 @@ case class DLCNode(wallet: DLCWalletApi)(implicit
     } yield res
   }
 
+  override def checkPeerConnection(
+      peerAddress: InetSocketAddress): Future[Unit] = {
+    for {
+      handler <- connectToPeer(peerAddress)
+    } yield {
+      handler ! DLCConnectionHandler.CloseConnection
+    }
+  }
+
   private def connectToPeer(
       peerAddress: InetSocketAddress): Future[ActorRef] = {
+    config.callBacks.executeOnPeerConnectionInitiated(peerAddress)
     val peer =
       Peer(socket = peerAddress, socks5ProxyParams = config.socks5ProxyParams)
 
     val handlerP = Promise[ActorRef]()
-    for {
+
+    val f = for {
       _ <- DLCClient.connect(peer, wallet, Some(handlerP))
       handler <- handlerP.future
     } yield handler
+
+    f.onComplete {
+      case Success(_) =>
+        config.callBacks.executeOnPeerConnectionEstablished(peerAddress)
+      case Failure(_) =>
+        config.callBacks.executeOnPeerConnectionFailed(peerAddress)
+    }
+
+    f
   }
 }

dlc-node/src/main/scala/org/bitcoins/dlc/node/DLCNodeCallbacks.scala

@@ -0,0 +1,127 @@
+package org.bitcoins.dlc.node
+
+import grizzled.slf4j.Logging
+import org.bitcoins.core.api.callback.{CallbackFactory, ModuleCallbacks}
+import org.bitcoins.core.api.{Callback, CallbackHandler}
+
+import java.net.InetSocketAddress
+import scala.concurrent.{ExecutionContext, Future}
+
+/** Callbacks for responding to events in the DLC node. */
+trait DLCNodeCallbacks extends ModuleCallbacks[DLCNodeCallbacks] with Logging {
+
+  def onPeerConnectionInitiated: CallbackHandler[
+    InetSocketAddress,
+    OnPeerConnectionInitiated]
+
+  def onPeerConnectionEstablished: CallbackHandler[
+    InetSocketAddress,
+    OnPeerConnectionEstablished]
+
+  def onPeerConnectionFailed: CallbackHandler[
+    InetSocketAddress,
+    OnPeerConnectionFailed]
+
+  override def +(other: DLCNodeCallbacks): DLCNodeCallbacks
+
+  def executeOnPeerConnectionInitiated(peerAddress: InetSocketAddress)(implicit
+      ec: ExecutionContext): Future[Unit] = {
+    onPeerConnectionInitiated.execute(
+      peerAddress,
+      (err: Throwable) =>
+        logger.error(
+          s"${onPeerConnectionInitiated.name} Callback failed with error: ",
+          err))
+  }
+
+  def executeOnPeerConnectionEstablished(peerAddress: InetSocketAddress)(
+      implicit ec: ExecutionContext): Future[Unit] = {
+    onPeerConnectionEstablished.execute(
+      peerAddress,
+      (err: Throwable) =>
+        logger.error(
+          s"${onPeerConnectionEstablished.name} Callback failed with error: ",
+          err))
+  }
+
+  def executeOnPeerConnectionFailed(peerAddress: InetSocketAddress)(implicit
+      ec: ExecutionContext): Future[Unit] = {
+    onPeerConnectionFailed.execute(
+      peerAddress,
+      (err: Throwable) =>
+        logger.error(
+          s"${onPeerConnectionFailed.name} Callback failed with error: ",
+          err))
+  }
+}
+
+trait OnPeerConnectionInitiated extends Callback[InetSocketAddress]
+
+trait OnPeerConnectionEstablished extends Callback[InetSocketAddress]
+
+trait OnPeerConnectionFailed extends Callback[InetSocketAddress]
+
+object DLCNodeCallbacks extends CallbackFactory[DLCNodeCallbacks] {
+
+  // Use Impl pattern here to enforce the correct names on the CallbackHandlers
+  private case class DLCNodeCallbacksImpl(
+      onPeerConnectionInitiated: CallbackHandler[
+        InetSocketAddress,
+        OnPeerConnectionInitiated],
+      onPeerConnectionEstablished: CallbackHandler[
+        InetSocketAddress,
+        OnPeerConnectionEstablished],
+      onPeerConnectionFailed: CallbackHandler[
+        InetSocketAddress,
+        OnPeerConnectionFailed])
+      extends DLCNodeCallbacks {
+
+    override def +(other: DLCNodeCallbacks): DLCNodeCallbacks =
+      copy(
+        onPeerConnectionInitiated =
+          onPeerConnectionInitiated ++ other.onPeerConnectionInitiated,
+        onPeerConnectionEstablished =
+          onPeerConnectionEstablished ++ other.onPeerConnectionEstablished,
+        onPeerConnectionFailed =
+          onPeerConnectionFailed ++ other.onPeerConnectionFailed
+      )
+  }
+
+  def onPeerConnectionInitiated(
+      f: OnPeerConnectionInitiated): DLCNodeCallbacks =
+    DLCNodeCallbacks(onPeerConnectionInitiated = Vector(f))
+
+  def onPeerConnectionEstablished(
+      f: OnPeerConnectionEstablished): DLCNodeCallbacks =
+    DLCNodeCallbacks(onPeerConnectionEstablished = Vector(f))
+
+  def onPeerConnectionFailed(f: OnPeerConnectionFailed): DLCNodeCallbacks =
+    DLCNodeCallbacks(onPeerConnectionFailed = Vector(f))
+
+  /** Empty callbacks that does nothing with the received data */
+  override val empty: DLCNodeCallbacks =
+    DLCNodeCallbacks(Vector.empty, Vector.empty, Vector.empty)
+
+  def apply(
+      onPeerConnectionInitiated: Vector[OnPeerConnectionInitiated] =
+        Vector.empty,
+      onPeerConnectionEstablished: Vector[OnPeerConnectionEstablished] =
+        Vector.empty,
+      onPeerConnectionFailed: Vector[OnPeerConnectionFailed] =
+        Vector.empty): DLCNodeCallbacks = {
+    DLCNodeCallbacksImpl(
+      onPeerConnectionInitiated =
+        CallbackHandler[InetSocketAddress, OnPeerConnectionInitiated](
+          "onPeerConnectionInitiated",
+          onPeerConnectionInitiated),
+      onPeerConnectionEstablished =
+        CallbackHandler[InetSocketAddress, OnPeerConnectionEstablished](
+          "onPeerConnectionEstablished",
+          onPeerConnectionEstablished),
+      onPeerConnectionFailed =
+        CallbackHandler[InetSocketAddress, OnPeerConnectionFailed](
+          "onPeerConnectionFailed",
+          onPeerConnectionFailed)
+    )
+  }
+}

dlc-node/src/main/scala/org/bitcoins/dlc/node/config/DLCNodeAppConfig.scala

@@ -3,9 +3,10 @@ package org.bitcoins.dlc.node.config
 import akka.actor.ActorSystem
 import com.typesafe.config.Config
 import org.bitcoins.commons.config.{AppConfig, AppConfigFactory}
+import org.bitcoins.core.api.CallbackConfig
 import org.bitcoins.core.api.dlc.wallet.DLCWalletApi
 import org.bitcoins.core.util.{FutureUtil, NetworkUtil}
-import org.bitcoins.dlc.node.DLCNode
+import org.bitcoins.dlc.node.{DLCNode, DLCNodeCallbacks}
 import org.bitcoins.tor.config.TorAppConfig
 import org.bitcoins.tor.{Socks5ProxyParams, TorParams}
 
@@ -20,7 +21,8 @@ import scala.concurrent._
   */
 case class DLCNodeAppConfig(baseDatadir: Path, configOverrides: Vector[Config])(
     implicit ec: ExecutionContext)
-    extends AppConfig {
+    extends AppConfig
+    with CallbackConfig[DLCNodeCallbacks] {
 
   override protected[bitcoins] def moduleName: String =
     DLCNodeAppConfig.moduleName
@@ -37,6 +39,8 @@ case class DLCNodeAppConfig(baseDatadir: Path, configOverrides: Vector[Config])(
 
   override def stop(): Future[Unit] = Future.unit
 
+  override lazy val callbackFactory: DLCNodeCallbacks.type = DLCNodeCallbacks
+
   lazy val torConf: TorAppConfig =
     TorAppConfig(baseDatadir, Some(moduleName), configOverrides)