mirrors/bitcoin-s
1
0
Fork 0
mirror of https://github.com/bitcoin-s/bitcoin-s.git synced 2025-02-22 14:33:06 +01:00
Code Issues Projects Releases Packages Wiki Activity

Add guard for calling NodeCallbackStreamManager.stop() (#4523)

Browse source 
* Add guard for calling NodeCallbackStreamManager.stop()

* scalafmt
This commit is contained in:
Chris Stewart 2022-07-20 10:16:26 -05:00 • committed by GitHub
parent c210052640
commit f5cca7e5e1
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 17 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
node/src/main/scala/org/bitcoins/node/callback/NodeCallbackStreamManager.scala
View file

@ -5,6 +5,7 @@ import akka.actor.ActorSystem
import akka.stream.BoundedSourceQueue
import akka.stream.scaladsl.{Keep, Sink, Source}
import grizzled.slf4j.{Logger, Logging}
import monix.execution.atomic.AtomicBoolean
import org.bitcoins.core.api.CallbackHandler
import org.bitcoins.core.gcs.GolombFilter
import org.bitcoins.core.protocol.blockchain.{Block, BlockHeader, MerkleBlock}
@ -112,15 +113,25 @@ case class NodeCallbackStreamManager(callbacks: NodeCallbacks)(implicit
override def start(): Future[Unit] = Future.unit
private val isStopped: AtomicBoolean = AtomicBoolean(false)
/** Completes all streams and waits until they are fully drained */
override def stop(): Future[Unit] = {
val start = System.currentTimeMillis()
//complete all queues
filterQueue.complete()
txQueue.complete()
headerQueue.complete()
merkleBlockQueue.complete()
blockQueue.complete()
//can't complete a stream twice
if (!isStopped.get()) {
//complete all queues
filterQueue.complete()
txQueue.complete()
headerQueue.complete()
merkleBlockQueue.complete()
blockQueue.complete()
isStopped.set(true)
} else {
logger.warn(
s"Already stopped all queues associated with this NodeCallBackStreamManager")
}
for {
_ <- filterSinkCompleteF