Add abillity to clear callbacks (#4512)

Fix missing clearCallbacks()

Add some tests

Remove uncessary code from test

chain-test/src/test/scala/org/bitcoins/chain/config/ChainAppConfigTest.scala

@@ -1,14 +1,15 @@
 package org.bitcoins.chain.config
 
 import java.nio.file.Files
-
 import com.typesafe.config.ConfigFactory
+import org.bitcoins.chain.{ChainCallbacks, OnBlockHeaderConnected}
 import org.bitcoins.core.config.{MainNet, RegTest, TestNet3}
+import org.bitcoins.core.protocol.blockchain.BlockHeader
 import org.bitcoins.testkit.chain.ChainUnitTest
 import org.bitcoins.testkit.util.FileUtil
 import org.scalatest.FutureOutcome
 
-import scala.concurrent.Await
+import scala.concurrent.{Await, Future}
 
 class ChainAppConfigTest extends ChainUnitTest {
   val tempDir = Files.createTempDirectory("bitcoin-s")
@@ -79,6 +80,25 @@ class ChainAppConfigTest extends ChainUnitTest {
     assert(appConfig.network == TestNet3)
   }
 
+  it must "add a callback and then remove the callback" in { _ =>
+    val tempDir = Files.createTempDirectory("bitcoin-s")
+    val appConfig = ChainAppConfig(baseDatadir = tempDir, Vector.empty)
+    assert(appConfig.isCallbackEmpty)
+
+    val dummyCallback: OnBlockHeaderConnected = {
+      case _: Vector[(Int, BlockHeader)] =>
+        Future.unit
+    }
+    val printlnCallback = ChainCallbacks.onBlockHeaderConnected(dummyCallback)
+
+    appConfig.addCallbacks(printlnCallback)
+    assert(!appConfig.isCallbackEmpty)
+
+    //clear the callback
+    appConfig.clearCallbacks()
+    assert(appConfig.isCallbackEmpty)
+  }
+
   override def afterAll(): Unit = {
     FileUtil.deleteTmpDir(chainAppConfig.baseDatadir)
     val stopF = for {

chain/src/main/scala/org/bitcoins/chain/ChainCallbacks.scala

@@ -1,12 +1,13 @@
 package org.bitcoins.chain
 
 import grizzled.slf4j.Logger
+import org.bitcoins.core.api.callback.{CallbackFactory, ModuleCallbacks}
 import org.bitcoins.core.api.{Callback, CallbackHandler}
 import org.bitcoins.core.protocol.blockchain.BlockHeader
 
 import scala.concurrent.{ExecutionContext, Future}
 
-trait ChainCallbacks {
+trait ChainCallbacks extends ModuleCallbacks[ChainCallbacks] {
 
   def onBlockHeaderConnected: CallbackHandler[
     Vector[(Int, BlockHeader)],
@@ -14,7 +15,7 @@ trait ChainCallbacks {
 
   def onSyncFlagChanged: CallbackHandler[Boolean, OnSyncFlagChanged]
 
-  def +(other: ChainCallbacks): ChainCallbacks
+  override def +(other: ChainCallbacks): ChainCallbacks
 
   def executeOnBlockHeaderConnectedCallbacks(
       logger: Logger,
@@ -45,7 +46,7 @@ trait OnBlockHeaderConnected extends Callback[Vector[(Int, BlockHeader)]]
 
 trait OnSyncFlagChanged extends Callback[Boolean]
 
-object ChainCallbacks {
+object ChainCallbacks extends CallbackFactory[ChainCallbacks] {
 
   private case class ChainCallbacksImpl(
       onBlockHeaderConnected: CallbackHandler[
@@ -67,7 +68,7 @@ object ChainCallbacks {
   def onOnSyncFlagChanged(f: OnSyncFlagChanged): ChainCallbacks =
     ChainCallbacks(onSyncFlagChanged = Vector(f))
 
-  lazy val empty: ChainCallbacks =
+  override val empty: ChainCallbacks =
     ChainCallbacks(onBlockHeaderConnected = Vector.empty)
 
   def apply(

chain/src/main/scala/org/bitcoins/chain/config/ChainAppConfig.scala

@@ -8,7 +8,6 @@ import org.bitcoins.chain.pow.Pow
 import org.bitcoins.commons.config.AppConfigFactory
 import org.bitcoins.core.api.CallbackConfig
 import org.bitcoins.core.api.chain.db.BlockHeaderDbHelper
-import org.bitcoins.core.util.Mutable
 import org.bitcoins.db._
 
 import java.nio.file.Path
@@ -35,13 +34,7 @@ case class ChainAppConfig(baseDatadir: Path, configOverrides: Vector[Config])(
 
   override lazy val appConfig: ChainAppConfig = this
 
-  private val callbacks = new Mutable(ChainCallbacks.empty)
-
-  override def callBacks: ChainCallbacks = callbacks.atomicGet
-
-  override def addCallbacks(newCallbacks: ChainCallbacks): ChainCallbacks = {
-    callbacks.atomicUpdate(newCallbacks)(_ + _)
-  }
+  override lazy val callbackFactory: ChainCallbacks.type = ChainCallbacks
 
   /** Checks whether or not the chain project is initialized by
     * trying to read the genesis block header from our block
@@ -102,6 +95,7 @@ case class ChainAppConfig(baseDatadir: Path, configOverrides: Vector[Config])(
 
   override def stop(): Future[Unit] = {
     val _ = stopHikariLogger()
+    clearCallbacks()
     super.stop()
   }
 

core/src/main/scala/org/bitcoins/core/api/CallbackConfig.scala

@@ -1,7 +1,29 @@
 package org.bitcoins.core.api
 
-trait CallbackConfig[T] {
-  def addCallbacks(newCallbacks: T): T
+import org.bitcoins.core.api.callback.{CallbackFactory, ModuleCallbacks}
+import org.bitcoins.core.util.Mutable
 
-  def callBacks: T
+trait CallbackConfig[T <: ModuleCallbacks[T]] {
+
+  private[this] val atomicCallbacks: Mutable[T] = new Mutable(
+    callbackFactory.empty)
+
+  def isCallbackEmpty: Boolean =
+    atomicCallbacks.atomicGet == callbackFactory.empty
+
+  def addCallbacks(newCallbacks: T): T = {
+    atomicCallbacks.atomicUpdate(newCallbacks) { case (t1, t2) =>
+      t1.+(t2)
+    }
+  }
+
+  def callBacks: T = atomicCallbacks.atomicGet
+
+  /** Clears all callbacks */
+  def clearCallbacks(): Unit = {
+    atomicCallbacks.atomicSet(callbackFactory.empty)
+    ()
+  }
+
+  def callbackFactory: CallbackFactory[T]
 }

core/src/main/scala/org/bitcoins/core/api/callback/CallbackFactory.scala

@@ -0,0 +1,5 @@
+package org.bitcoins.core.api.callback
+
+trait CallbackFactory[T] {
+  def empty: T
+}

core/src/main/scala/org/bitcoins/core/api/callback/ModuleCallbacks.scala

@@ -0,0 +1,6 @@
+package org.bitcoins.core.api.callback
+
+/** A data structure to represent all callbacks for a specific module */
+trait ModuleCallbacks[T <: ModuleCallbacks[T]] {
+  def +(other: T): T
+}

node/src/main/scala/org/bitcoins/node/NodeCallbacks.scala

@@ -1,6 +1,7 @@
 package org.bitcoins.node
 
 import grizzled.slf4j.Logger
+import org.bitcoins.core.api.callback.{CallbackFactory, ModuleCallbacks}
 import org.bitcoins.core.api.{Callback, Callback2, CallbackHandler}
 import org.bitcoins.core.gcs.GolombFilter
 import org.bitcoins.core.protocol.blockchain.{Block, BlockHeader, MerkleBlock}
@@ -13,7 +14,7 @@ import scala.concurrent.{ExecutionContext, Future}
   * The appropriate callback is executed whenever the node receives
   * a `getdata` message matching it.
   */
-trait NodeCallbacks {
+trait NodeCallbacks extends ModuleCallbacks[NodeCallbacks] {
 
   def onCompactFiltersReceived: CallbackHandler[
     Vector[(DoubleSha256Digest, GolombFilter)],
@@ -31,7 +32,7 @@ trait NodeCallbacks {
     Vector[BlockHeader],
     OnBlockHeadersReceived]
 
-  def +(other: NodeCallbacks): NodeCallbacks
+  override def +(other: NodeCallbacks): NodeCallbacks
 
   def executeOnTxReceivedCallbacks(logger: Logger, tx: Transaction)(implicit
       ec: ExecutionContext): Future[Unit] = {
@@ -103,7 +104,7 @@ trait OnCompactFiltersReceived
 /** Callback for handling a received block header */
 trait OnBlockHeadersReceived extends Callback[Vector[BlockHeader]]
 
-object NodeCallbacks {
+object NodeCallbacks extends CallbackFactory[NodeCallbacks] {
 
   // Use Impl pattern here to enforce the correct names on the CallbackHandlers
   private case class NodeCallbacksImpl(
@@ -154,7 +155,7 @@ object NodeCallbacks {
     NodeCallbacks(onBlockHeadersReceived = Vector(f))
 
   /** Empty callbacks that does nothing with the received data */
-  val empty: NodeCallbacks =
+  override val empty: NodeCallbacks =
     NodeCallbacks(Vector.empty,
                   Vector.empty,
                   Vector.empty,

node/src/main/scala/org/bitcoins/node/config/NodeAppConfig.scala

@@ -13,7 +13,7 @@ import org.bitcoins.chain.models.{
 import org.bitcoins.core.api.CallbackConfig
 import org.bitcoins.core.api.node.NodeType
 import org.bitcoins.core.config.{MainNet, RegTest, SigNet, TestNet3}
-import org.bitcoins.core.util.{Mutable, TimeUtil}
+import org.bitcoins.core.util.TimeUtil
 import org.bitcoins.db.{DbAppConfig, JdbcProfileComponent}
 import org.bitcoins.node._
 import org.bitcoins.node.db.NodeDbManagement
@@ -50,13 +50,7 @@ case class NodeAppConfig(baseDatadir: Path, configOverrides: Vector[Config])(
 
   override def appConfig: NodeAppConfig = this
 
-  private val callbacks = new Mutable(NodeCallbacks.empty)
-
-  override def callBacks: NodeCallbacks = callbacks.atomicGet
-
-  override def addCallbacks(newCallbacks: NodeCallbacks): NodeCallbacks = {
-    callbacks.atomicUpdate(newCallbacks)(_ + _)
-  }
+  override lazy val callbackFactory: NodeCallbacks.type = NodeCallbacks
 
   /** Ensures correct tables and other required information is in
     * place for our node.
@@ -96,6 +90,7 @@ case class NodeAppConfig(baseDatadir: Path, configOverrides: Vector[Config])(
   }
 
   override def stop(): Future[Unit] = {
+    clearCallbacks()
     val _ = stopHikariLogger()
     super.stop()
   }

tor/src/main/scala/org/bitcoins/tor/TorCallbacks.scala

@@ -1,13 +1,14 @@
 package org.bitcoins.tor
 
 import grizzled.slf4j.Logging
+import org.bitcoins.core.api.callback.{CallbackFactory, ModuleCallbacks}
 import org.bitcoins.core.api.{Callback, CallbackHandler}
 
 import scala.concurrent.{ExecutionContext, Future}
 
 trait OnTorStarted extends Callback[Unit]
 
-trait TorCallbacks extends Logging {
+trait TorCallbacks extends ModuleCallbacks[TorCallbacks] with Logging {
   def onTorStarted: CallbackHandler[Unit, OnTorStarted]
 
   def executeOnTorStarted()(implicit ec: ExecutionContext): Future[Unit] = {
@@ -20,7 +21,7 @@ trait TorCallbacks extends Logging {
   def +(other: TorCallbacks): TorCallbacks
 }
 
-object TorCallbacks {
+object TorCallbacks extends CallbackFactory[TorCallbacks] {
 
   private case class TorCallbacksImpl(
       onTorStarted: CallbackHandler[Unit, OnTorStarted]
@@ -31,7 +32,7 @@ object TorCallbacks {
     }
   }
 
-  val empty = apply(Vector.empty)
+  override val empty: TorCallbacks = apply(Vector.empty)
 
   def apply(onTorStarted: OnTorStarted): TorCallbacks = {
     apply(Vector(onTorStarted))

tor/src/main/scala/org/bitcoins/tor/config/TorAppConfig.scala

@@ -4,7 +4,7 @@ import com.typesafe.config.Config
 import org.bitcoins.asyncutil.AsyncUtil
 import org.bitcoins.commons.config.{AppConfig, AppConfigFactory, ConfigOps}
 import org.bitcoins.core.api.CallbackConfig
-import org.bitcoins.core.util.{Mutable, NetworkUtil}
+import org.bitcoins.core.util.NetworkUtil
 import org.bitcoins.tor.TorProtocolHandler.{Password, SafeCookie}
 import org.bitcoins.tor.client.TorClient
 import org.bitcoins.tor.{Socks5ProxyParams, TorCallbacks, TorParams}
@@ -27,13 +27,6 @@ case class TorAppConfig(
     extends AppConfig
     with CallbackConfig[TorCallbacks] {
 
-  private val callbacks = new Mutable(TorCallbacks.empty)
-
-  override def callBacks: TorCallbacks = callbacks.atomicGet
-
-  override def addCallbacks(newCallbacks: TorCallbacks): TorCallbacks = {
-    callbacks.atomicUpdate(newCallbacks)(_ + _)
-  }
   override protected[bitcoins] def moduleName: String = TorAppConfig.moduleName
   override protected[bitcoins] type ConfigType = TorAppConfig
 
@@ -41,6 +34,7 @@ case class TorAppConfig(
       configs: Vector[Config]): TorAppConfig =
     TorAppConfig(baseDatadir, subModuleNameOpt, configs)
 
+  override lazy val callbackFactory: TorCallbacks.type = TorCallbacks
   private val isStarted: AtomicBoolean = new AtomicBoolean(false)
 
   lazy val torDir: Path = baseDatadir.resolve("tor")
@@ -163,6 +157,7 @@ case class TorAppConfig(
   }
 
   override def stop(): Future[Unit] = {
+    clearCallbacks()
     if (torProvided) {
       Future.unit
     } else {
@@ -182,7 +177,7 @@ case class TorAppConfig(
     AsyncUtil
       .retryUntilSatisfied(checkIfLogExists, 1.second, 60)
       //execute started callbacks
-      .flatMap(_ => callbacks.atomicGet.executeOnTorStarted())
+      .flatMap(_ => callBacks.executeOnTorStarted())
       .recover { case _: AsyncUtil.RpcRetryException =>
         throw new RuntimeException(
           s"Could not start tor, please try again in a few minutes")

wallet/src/main/scala/org/bitcoins/wallet/WalletCallbacks.scala

@@ -1,6 +1,7 @@
 package org.bitcoins.wallet
 
 import grizzled.slf4j.Logger
+import org.bitcoins.core.api.callback.{CallbackFactory, ModuleCallbacks}
 import org.bitcoins.core.api.wallet.db.SpendingInfoDb
 import org.bitcoins.core.api.{Callback, CallbackHandler}
 import org.bitcoins.core.protocol.BitcoinAddress
@@ -13,7 +14,7 @@ import scala.concurrent.{ExecutionContext, Future}
   * The appropriate callback is executed whenever the wallet finishes,
   * the corresponding function.
   */
-trait WalletCallbacks {
+trait WalletCallbacks extends ModuleCallbacks[WalletCallbacks] {
 
   def onTransactionProcessed: CallbackHandler[
     Transaction,
@@ -107,7 +108,7 @@ trait OnBlockProcessed extends Callback[Block]
 /** Triggered when a rescan is */
 trait OnRescanComplete extends Callback[Unit]
 
-object WalletCallbacks {
+object WalletCallbacks extends CallbackFactory[WalletCallbacks] {
 
   private case class WalletCallbacksImpl(
       onTransactionProcessed: CallbackHandler[
@@ -159,7 +160,7 @@ object WalletCallbacks {
   }
 
   /** Empty callbacks that does nothing with the received data */
-  val empty: WalletCallbacks =
+  override val empty: WalletCallbacks =
     apply(Vector.empty,
           Vector.empty,
           Vector.empty,

wallet/src/main/scala/org/bitcoins/wallet/config/WalletAppConfig.scala

@@ -8,7 +8,6 @@ import org.bitcoins.core.api.chain.ChainQueryApi
 import org.bitcoins.core.api.feeprovider.FeeRateApi
 import org.bitcoins.core.api.node.NodeApi
 import org.bitcoins.core.hd._
-import org.bitcoins.core.util.Mutable
 import org.bitcoins.core.wallet.fee.SatoshisPerVirtualByte
 import org.bitcoins.core.wallet.keymanagement._
 import org.bitcoins.crypto.AesPassword
@@ -70,13 +69,7 @@ case class WalletAppConfig(baseDatadir: Path, configOverrides: Vector[Config])(
     Executors.newFixedThreadPool(Runtime.getRuntime.availableProcessors(),
                                  rescanThreadFactory)
 
-  private val callbacks = new Mutable(WalletCallbacks.empty)
-
-  override def callBacks: WalletCallbacks = callbacks.atomicGet
-
-  override def addCallbacks(newCallbacks: WalletCallbacks): WalletCallbacks = {
-    callbacks.atomicUpdate(newCallbacks)(_ + _)
-  }
+  override lazy val callbackFactory: WalletCallbacks.type = WalletCallbacks
 
   lazy val kmConf: KeyManagerAppConfig =
     KeyManagerAppConfig(baseDatadir, configOverrides)
@@ -217,6 +210,7 @@ case class WalletAppConfig(baseDatadir: Path, configOverrides: Vector[Config])(
       stopHikariLogger()
     }
 
+    clearCallbacks()
     stopRebroadcastTxsScheduler()
     //this eagerly shuts down all scheduled tasks on the scheduler
     //in the future, we should actually cancel all things that are scheduled