solution: track Bitcoin transactions

src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt

@@ -31,7 +31,7 @@ import reactor.core.publisher.Mono
 class BlockchainRpc(
         @Autowired private val nativeCall: NativeCall,
         @Autowired private val streamHead: StreamHead,
-        @Autowired private val trackEthereumTx: TrackEthereumTx,
+        @Autowired private val trackTx: List<TrackTx>,
         @Autowired private val trackAddress: List<TrackAddress>,
         @Autowired private val describe: Describe,
         @Autowired private val subscribeStatus: SubscribeStatus
@@ -48,7 +48,11 @@ class BlockchainRpc(
     }
 
     override fun subscribeTxStatus(request: Mono<BlockchainOuterClass.TxStatusRequest>): Flux<BlockchainOuterClass.TxStatus> {
-        return trackEthereumTx.add(request)
+        return request.flatMapMany { request ->
+            val chain = Chain.byId(request.chainValue)
+            trackTx.find { it.isSupported(chain) }?.subscribe(request)
+                    ?: Flux.error(SilentException.UnsupportedBlockchain(chain))
+        }
     }
 
     override fun subscribeBalance(requestMono: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {

src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinTx.kt

@@ -0,0 +1,158 @@
+package io.emeraldpay.dshackle.rpc
+
+import com.google.protobuf.ByteString
+import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.api.proto.Common
+import io.emeraldpay.dshackle.BlockchainType
+import io.emeraldpay.dshackle.SilentException
+import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.Upstreams
+import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinApi
+import io.emeraldpay.dshackle.upstream.bitcoin.ExtractBlock
+import io.emeraldpay.grpc.Chain
+import org.slf4j.LoggerFactory
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.stereotype.Service
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import java.math.BigInteger
+import java.time.Duration
+import java.time.Instant
+import kotlin.math.max
+import kotlin.math.min
+
+@Service
+class TrackBitcoinTx(
+        @Autowired private val upstreams: Upstreams
+) : TrackTx {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(TrackBitcoinTx::class.java)
+    }
+
+    override fun isSupported(chain: Chain): Boolean {
+        return BlockchainType.fromBlockchain(chain) == BlockchainType.BITCOIN && upstreams.isAvailable(chain)
+    }
+
+    override fun subscribe(request: BlockchainOuterClass.TxStatusRequest): Flux<BlockchainOuterClass.TxStatus> {
+        val chain = Chain.byId(request.chainValue)
+        val upstream = upstreams.getUpstream(chain)?.castApi(BitcoinApi::class.java)
+                ?: return Flux.error(SilentException.UnsupportedBlockchain(chain))
+        val txid = request.txId
+        val confirmations = max(min(1, request.confirmationLimit), 12)
+        return upstream.getApi(Selector.empty).flatMapMany { api ->
+            subscribe(chain, api, upstream, txid)
+        }.takeUntil { tx ->
+            tx.confirmations >= confirmations
+        }.map(this::asProto)
+    }
+
+    fun subscribe(chain: Chain, api: BitcoinApi, upstream: Upstream<BitcoinApi>, txid: String): Flux<TxStatus> {
+        return loadExisting(api, txid)
+                .flatMapMany { status ->
+                    if (status.mined) {
+                        //Head almost always knows the current height, so it can continue with calculating confirmations
+                        //without publishing an empty TxStatus first
+                        continueWithMined(api, upstream, status)
+                    } else {
+                        loadMempool(api, txid)
+                                .flatMapMany { tx ->
+                                    val next = if (tx.found) {
+                                        untilMined(upstream, tx)
+                                    } else {
+                                        untilFound(chain, api, upstream, txid)
+                                    }
+                                    //fist provide the current status, then updates
+                                    Flux.concat(Mono.just(tx), next)
+                                }
+                    }
+                }
+    }
+
+    fun continueWithMined(api: BitcoinApi, upstream: Upstream<BitcoinApi>, status: TxStatus): Flux<TxStatus> {
+        return api.getBlock(status.blockHash!!)
+                .map { block ->
+                    TxStatus(status.txid, true, ExtractBlock.getHeight(block), true, status.blockHash, ExtractBlock.getTime(block), ExtractBlock.getDifficulty(block))
+                }.flatMapMany { tx ->
+                    withConfirmations(upstream, tx)
+                }
+    }
+
+    fun untilFound(chain: Chain, api: BitcoinApi, upstream: Upstream<BitcoinApi>, txid: String): Flux<TxStatus> {
+        return Flux.interval(Duration.ofSeconds(1))
+                .take(Duration.ofMinutes(10))
+                .flatMap { loadMempool(api, txid) }
+                .skipUntil { it.found }
+                .flatMap { subscribe(chain, api, upstream, txid) }
+                .doOnError { t ->
+                    log.error("Failed to wait until found", t)
+                }
+    }
+
+    fun untilMined(upstream: Upstream<BitcoinApi>, tx: TxStatus): Mono<TxStatus> {
+        return upstream.getHead().getFlux().flatMap {
+            upstream.getApi(Selector.empty).flatMap { api ->
+                loadExisting(api, tx.txid)
+            }.filter { it.mined }
+        }.single()
+    }
+
+    fun withConfirmations(upstream: Upstream<BitcoinApi>, tx: TxStatus): Flux<TxStatus> {
+        return upstream.getHead().getFlux().map {
+            tx.withHead(it.height)
+        }
+    }
+
+    fun loadExisting(api: BitcoinApi, txid: String): Mono<TxStatus> {
+        val mined = api.getTx(txid)
+        return mined.map {
+            val block = it["blockhash"] as String?
+            TxStatus(txid, found = true, mined = block != null, blockHash = block, height = ExtractBlock.getHeight(it))
+        }
+    }
+
+    fun loadMempool(api: BitcoinApi, txid: String): Mono<TxStatus> {
+        val mempool = api.getMempool()
+        return mempool.map {
+            if (it.contains(txid)) {
+                TxStatus(txid, found = true, mined = false)
+            } else {
+                TxStatus(txid, found = false, mined = false)
+            }
+        }
+    }
+
+    private fun asProto(tx: TxStatus): BlockchainOuterClass.TxStatus {
+        val data = BlockchainOuterClass.TxStatus.newBuilder()
+                .setTxId(tx.txid)
+                .setConfirmations(tx.confirmations.toInt())
+
+        data.broadcasted = tx.found
+        val isMined = tx.mined
+        data.mined = isMined
+        if (isMined) {
+            data.setBlock(
+                    Common.BlockInfo.newBuilder()
+                            .setBlockId(tx.blockHash!!.substring(2))
+                            .setTimestamp(tx.blockTime!!.toEpochMilli())
+                            .setWeight(ByteString.copyFrom(tx.blockTotalDifficulty!!.toByteArray()))
+                            .setHeight(tx.height!!)
+            )
+        }
+        return data.build()
+    }
+
+    class TxStatus(
+            val txid: String,
+            val found: Boolean = false,
+            val height: Long? = null,
+            val mined: Boolean = false,
+            val blockHash: String? = null,
+            val blockTime: Instant? = null,
+            val blockTotalDifficulty: BigInteger? = null,
+            val confirmations: Long = 0) {
+
+        fun withHead(headHeight: Long) = TxStatus(txid, found, height, mined, blockHash, blockTime, blockTotalDifficulty, headHeight - height!! + 1)
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt

@@ -58,7 +58,7 @@ import kotlin.math.min
 class TrackEthereumTx(
         @Autowired private val upstreams: Upstreams,
         @Autowired private val upstreamScheduler: Scheduler
-) {
+) : TrackTx {
 
     companion object {
         private val ZERO_BLOCK = BlockHash.from("0x0000000000000000000000000000000000000000000000000000000000000000")
@@ -75,6 +75,10 @@ class TrackEthereumTx(
 
     val notFound = ConcurrentLinkedQueue<TrackedTx>()
 
+    override fun isSupported(chain: Chain): Boolean {
+        return BlockchainType.fromBlockchain(chain) == BlockchainType.ETHEREUM && upstreams.isAvailable(chain)
+    }
+
     @PostConstruct
     fun init() {
         upstreams.observeChains().subscribe { chain ->
@@ -165,7 +169,7 @@ class TrackEthereumTx(
 
     fun streamAllUpdates(tx: TxDetails): Flux<BlockchainOuterClass.TxStatus> {
         val current = checkForUpdate(tx)
-                .doOnNext (this::onFirstUpdate)
+                .doOnNext(this::onFirstUpdate)
                 .map { Notification(it, asProto(it)) }
         val updates = Flux.from(tx.bus)
 
@@ -175,12 +179,9 @@ class TrackEthereumTx(
                 .map { it.proto }
     }
 
-    fun add(requestMono: Mono<BlockchainOuterClass.TxStatusRequest>): Flux<BlockchainOuterClass.TxStatus> {
+    override fun subscribe(request: BlockchainOuterClass.TxStatusRequest): Flux<BlockchainOuterClass.TxStatus> {
-        return requestMono.map { request ->
+        val tx = prepareTracking(request)
-            prepareTracking(request)
+        return streamAllUpdates(tx)
-        }.flatMapMany { tx ->
-            streamAllUpdates(tx)
-        }
     }
 
     private fun verifyAll(chain: Chain) {

src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt

@@ -0,0 +1,10 @@
+package io.emeraldpay.dshackle.rpc
+
+import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.grpc.Chain
+import reactor.core.publisher.Flux
+
+interface TrackTx {
+    fun isSupported(chain: Chain): Boolean
+    fun subscribe(request: BlockchainOuterClass.TxStatusRequest): Flux<BlockchainOuterClass.TxStatus>
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt

@@ -49,4 +49,5 @@ abstract class AbstractHead : Head {
     fun getCurrent(): BlockContainer? {
         return head.get()
     }
+
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt

@@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.upstream
 
 import io.emeraldpay.dshackle.data.BlockContainer
 import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
 
 interface Head {
     fun getFlux(): Flux<BlockContainer>

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinApi.kt

@@ -41,4 +41,16 @@ open class BitcoinApi(
                     }
                 }
     }
+
+    open fun getBlock(hash: String): Mono<Map<String, Any>> {
+        return executeAndResult(0, "getblock", listOf(hash), Map::class.java) as Mono<Map<String, Any>>
+    }
+
+    open fun getTx(txid: String): Mono<Map<String, Any>> {
+        return executeAndResult(0, "getrawtransaction", listOf(txid, true), Map::class.java) as Mono<Map<String, Any>>
+    }
+
+    open fun getMempool(): Mono<List<String>> {
+        return executeAndResult(0, "getrawmempool", emptyList(), List::class.java) as Mono<List<String>>
+    }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/ExtractBlock.kt

@@ -15,26 +15,42 @@ class ExtractBlock(
 
     companion object {
         private val log = LoggerFactory.getLogger(ExtractBlock::class.java)
+
+        @JvmStatic
+        fun getHeight(data: Map<String, Any>): Long? {
+            val height = data["height"] as Number? ?: return null
+            return height.toLong()
+        }
+
+        @JvmStatic
+        fun getTime(data: Map<String, Any>): Instant? {
+            val time = data["time"] as Number? ?: return null
+            return Instant.ofEpochMilli(time.toLong() * 1000)
+        }
+
+        @JvmStatic
+        fun getDifficulty(data: Map<String, Any>): BigInteger? {
+            val chainwork = data["chainwork"] as String? ?: return null
+            return BigInteger(1, Hex.decodeHex(chainwork))
+        }
     }
 
     fun extract(json: ByteArray): BlockContainer {
         val data = objectMapper.readValue(json, Map::class.java) as Map<String, Any>
 
-        val height = data["height"] as Number? ?: throw IllegalArgumentException("Block JSON has no height")
-        val time = data["time"] as Number? ?: throw IllegalArgumentException("Block JSON has no time")
         val hash = data["hash"] as String? ?: throw IllegalArgumentException("Block JSON has no hash")
-        val chainwork = data["chainwork"] as String? ?: throw IllegalArgumentException("Block JSON has no chainwork")
         val transactions = (data["tx"] as List<String>?)?.map(TxId.Companion::from) ?: emptyList()
 
         return BlockContainer(
-                height.toLong(),
+                getHeight(data) ?: throw IllegalArgumentException("Block JSON has no height"),
                 BlockId.from(hash),
-                BigInteger(1, Hex.decodeHex(chainwork)),
+                getDifficulty(data) ?: throw IllegalArgumentException("Block JSON has no chainwork"),
-                Instant.ofEpochMilli(time.toLong() * 1000),
+                getTime(data) ?: throw IllegalArgumentException("Block JSON has no time"),
                 false,
                 json,
                 transactions
         )
     }
 
+
 }

src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinTxSpec.groovy

@@ -0,0 +1,257 @@
+package io.emeraldpay.dshackle.rpc
+
+import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.data.BlockId
+import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.Upstreams
+import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinApi
+import io.emeraldpay.grpc.Chain
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.test.StepVerifier
+import spock.lang.Specification
+
+import java.time.Duration
+import java.time.Instant
+
+class TrackBitcoinTxSpec extends Specification {
+
+    def "loadMempool() returns not found when not found"() {
+        setup:
+        TrackBitcoinTx track = new TrackBitcoinTx(Stub(Upstreams))
+        BitcoinApi api = Mock(BitcoinApi) {
+            1 * getMempool() >> Mono.just([
+                    "69cd44d7c641db82e69824523c7ac0c5c1e5628f025474529cf5ffe64527efc9",
+                    "d296c6d47335a7f283574b06f1d6303b30ac75631e081ab128346a549ad93350"
+            ])
+        }
+        when:
+        def act = track.loadMempool(api, "65ce58db064bd105b14dc76a0bce0df14653cf5263d22d17e78864cf272ee367")
+
+        then:
+        StepVerifier.create(act)
+                .expectNextMatches {
+                    it.found == false && it.mined == false && it.blockHash == null
+                }
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+
+    def "loadMempool() returns ok when found"() {
+        setup:
+        TrackBitcoinTx track = new TrackBitcoinTx(Stub(Upstreams))
+        BitcoinApi api = Mock(BitcoinApi) {
+            1 * getMempool() >> Mono.just([
+                    "69cd44d7c641db82e69824523c7ac0c5c1e5628f025474529cf5ffe64527efc9",
+                    "d296c6d47335a7f283574b06f1d6303b30ac75631e081ab128346a549ad93350"
+            ])
+        }
+        when:
+        def act = track.loadMempool(api, "69cd44d7c641db82e69824523c7ac0c5c1e5628f025474529cf5ffe64527efc9")
+
+        then:
+        StepVerifier.create(act)
+                .expectNextMatches {
+                    it.found == true && it.mined == false && it.blockHash == null
+                }
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+
+    def "loadExiting() returns not found if not mined"() {
+        setup:
+        TrackBitcoinTx track = new TrackBitcoinTx(Stub(Upstreams))
+        def txid = "69cd44d7c641db82e69824523c7ac0c5c1e5628f025474529cf5ffe64527efc9"
+        BitcoinApi api = Mock(BitcoinApi) {
+            1 * getTx(txid) >> Mono.just([
+                    txid: txid
+            ])
+        }
+        when:
+        def act = track.loadExisting(api, txid)
+
+        then:
+        StepVerifier.create(act)
+                .expectNextMatches {
+                    it.found == true && it.mined == false && it.blockHash == null
+                }
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+
+    def "loadExiting() returns block if mined"() {
+        setup:
+        TrackBitcoinTx track = new TrackBitcoinTx(Stub(Upstreams))
+        def txid = "69cd44d7c641db82e69824523c7ac0c5c1e5628f025474529cf5ffe64527efc9"
+        BitcoinApi api = Mock(BitcoinApi) {
+            1 * getTx(txid) >> Mono.just([
+                    txid     : txid,
+                    blockhash: "0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f",
+                    height   : 100
+            ])
+        }
+        when:
+        def act = track.loadExisting(api, txid)
+
+        then:
+        StepVerifier.create(act)
+                .expectNextMatches {
+                    it.found == true && it.mined == true &&
+                            it.blockHash == "0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f" &&
+                            it.height == 100
+                }
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+
+    def "Goes with confirmations"() {
+        setup:
+        TrackBitcoinTx track = new TrackBitcoinTx(Stub(Upstreams))
+        def txid = "69cd44d7c641db82e69824523c7ac0c5c1e5628f025474529cf5ffe64527efc9"
+        // start with the current block
+        def next = Flux.fromIterable([10, 12, 13, 14, 15]).map { h ->
+            new BlockContainer(h.longValue(), BlockId.from("0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f"), BigInteger.ONE, Instant.now(), false, null, [])
+        }
+        Head head = Mock(Head) {
+            1 * getFlux() >> next
+        }
+        Upstream upstream = Mock(Upstream) {
+            1 * getHead() >> head
+        }
+        def status = new TrackBitcoinTx.TxStatus(
+                txid, true, 10, true, "0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f", Instant.now(), BigInteger.ONE, 0
+        )
+        when:
+        def act = track.withConfirmations(upstream, status)
+
+        then:
+        StepVerifier.create(act)
+                .expectNextMatches { it.confirmations == 1 }
+                .expectNextMatches { it.confirmations == 3 }
+                .expectNextMatches { it.confirmations == 4 }
+                .expectNextMatches { it.confirmations == 5 }
+                .expectNextMatches { it.confirmations == 6 }
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+
+    def "Wait until mined"() {
+        setup:
+        TrackBitcoinTx track = new TrackBitcoinTx(Stub(Upstreams))
+        def txid = "69cd44d7c641db82e69824523c7ac0c5c1e5628f025474529cf5ffe64527efc9"
+        // start with the current block
+        def next = Flux.fromIterable([10, 12, 13]).map { h ->
+            new BlockContainer(h.longValue(), BlockId.from("0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f"), BigInteger.ONE, Instant.now(), false, null, [])
+        }
+        Head head = Mock(Head) {
+            1 * getFlux() >> next
+        }
+        BitcoinApi api = Mock(BitcoinApi) {
+            3 * getTx(txid) >>> [
+                    Mono.just([
+                            txid: txid
+                    ]),
+                    Mono.just([
+                            txid: txid
+                    ]),
+                    Mono.just([
+                            txid     : txid,
+                            blockhash: "0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f",
+                            height   : 100
+                    ])
+            ]
+        }
+        Upstream upstream = Mock(Upstream) {
+            1 * getHead() >> head
+            _ * getApi(_) >> Mono.just(api)
+        }
+        def status = new TrackBitcoinTx.TxStatus(
+                txid, false, null, false, null, null, null, 0
+        )
+        when:
+        def act = track.untilMined(upstream, status)
+
+        then:
+        StepVerifier.create(act)
+                .expectNextMatches { it.mined && it.height == 100 && it.blockHash == "0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f" }
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+
+    def "Check mempool until found"() {
+        setup:
+        TrackBitcoinTx track = new TrackBitcoinTx(Stub(Upstreams))
+        def txid = "69cd44d7c641db82e69824523c7ac0c5c1e5628f025474529cf5ffe64527efc9"
+        BitcoinApi api = Mock(BitcoinApi) {
+            4 * getMempool() >>> [
+                    Mono.just([]),
+                    Mono.just(["4523c7ac0c5c1e5628f025474529c69cd44d7c641db82e6982f5ffe64527efc9"]),
+                    Mono.just(["4523c7ac0c5c1e5628f025474529c69cd44d7c641db82e6982f5ffe64527efc9", txid]),
+                    Mono.just(["4523c7ac0c5c1e5628f025474529c69cd44d7c641db82e6982f5ffe64527efc9", txid]) //second call when started over
+            ]
+            1 * getTx(txid) >> Mono.just([
+                    txid: txid
+            ])
+        }
+        Head head = Mock(Head) {
+            _ * getFlux() >> Flux.empty()
+        }
+        Upstream upstream = Mock(Upstream) {
+            _ * getApi(_) >> Mono.just(api)
+            _ * getHead() >> head
+        }
+
+        when:
+        def steps = StepVerifier.withVirtualTime {
+            track.untilFound(Chain.BITCOIN, api, upstream, txid).take(1)
+        }
+
+        then:
+        steps
+                .expectSubscription()
+                .expectNoEvent(Duration.ofSeconds(3))
+                .expectNextMatches { it.found && !it.mined }
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+
+    def "Subscribe to an existing tx"() {
+        setup:
+        TrackBitcoinTx track = new TrackBitcoinTx(Stub(Upstreams))
+        def txid = "69cd44d7c641db82e69824523c7ac0c5c1e5628f025474529cf5ffe64527efc9"
+        BitcoinApi api = Mock(BitcoinApi) {
+            _ * getTx(txid) >> Mono.just([
+                    txid     : txid,
+                    blockhash: "0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f",
+                    height   : 1
+            ])
+            _ * getBlock("0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f") >> Mono.just([
+                    height   : 1,
+                    chainwork: "01",
+                    time     : 10000
+            ])
+        }
+        def next = Flux.fromIterable([10, 11, 12]).map { h ->
+            new BlockContainer(h.longValue(), BlockId.from("0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f"), BigInteger.ONE, Instant.now(), false, null, [])
+        }
+        Head head = Mock(Head) {
+            _ * getFlux() >> next
+        }
+        Upstream upstream = Mock(Upstream) {
+            _ * getApi(_) >> Mono.just(api)
+            _ * getHead() >> head
+        }
+
+        when:
+        def act = track.subscribe(Chain.BITCOIN, api, upstream, txid)
+
+        then:
+        StepVerifier.create(act)
+                .expectNextMatches { it.found && it.mined && it.confirmations == 10 && it.blockHash == "0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f" }
+                .expectNextMatches { it.found && it.mined && it.confirmations == 11 }
+                .expectNextMatches { it.found && it.mined && it.confirmations == 12 }
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy

@@ -103,7 +103,7 @@ class TrackEthereumTxSpec extends Specification {
         upstreamMock.nextBlock(BlockContainer.from(blockHeadJson, TestingCommons.objectMapper()))
 
         when:
-        def flux = trackTx.add(Mono.just(req))
+        def flux = trackTx.subscribe(req)
         then:
         StepVerifier.create(flux)
                 .expectNext(exp1)
@@ -134,7 +134,7 @@ class TrackEthereumTxSpec extends Specification {
 
         when:
         def act = StepVerifier.withVirtualTime {
-            return trackTx.add(Mono.just(req))
+            return trackTx.subscribe(req)
         }
         then:
         act
@@ -192,7 +192,7 @@ class TrackEthereumTxSpec extends Specification {
 
         when:
         def act = StepVerifier.withVirtualTime {
-            return trackTx.add(Mono.just(req))
+            return trackTx.subscribe(req)
         }
         then:
         act
@@ -301,7 +301,7 @@ class TrackEthereumTxSpec extends Specification {
         }
 
         when:
-        def flux = trackTx.add(Mono.just(req))
+        def flux = trackTx.subscribe(req)
         then:
         StepVerifier.create(flux)
                 .expectNext(exp1.build())

src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/ExtractBlockSpec.groovy

@@ -21,4 +21,70 @@ class ExtractBlockSpec extends Specification {
         act.transactions.size() == 1487
         act.json == json
     }
+
+    def "Shouldn't extract time from empty"() {
+        when:
+        def act = ExtractBlock.getTime([:])
+        then:
+        act == null
+    }
+
+    def "Shouldn't extract time from null"() {
+        when:
+        def act = ExtractBlock.getTime([time: null])
+        then:
+        act == null
+    }
+
+    def "Extract correct time"() {
+        expect:
+        ExtractBlock.getTime(block).toString() == time
+
+        where:
+        time                   | block
+        "2020-04-18T00:58:46Z" | [time: 1587171526]
+        "1970-01-01T00:00:00Z" | [time: 0]
+    }
+
+    def "Shouldn't extract height from empty"() {
+        when:
+        def act = ExtractBlock.getHeight([:])
+        then:
+        act == null
+    }
+
+    def "Shouldn't extract height from null"() {
+        when:
+        def act = ExtractBlock.getHeight([height: null])
+        then:
+        act == null
+    }
+
+    def "Should extract height"() {
+        when:
+        def act = ExtractBlock.getHeight([height: 123456])
+        then:
+        act == 123456L
+    }
+
+    def "Shouldn't extract difficulty from empty"() {
+        when:
+        def act = ExtractBlock.getDifficulty([:])
+        then:
+        act == null
+    }
+
+    def "Shouldn't extract difficulty from null"() {
+        when:
+        def act = ExtractBlock.getDifficulty([chainwork: null])
+        then:
+        act == null
+    }
+
+    def "Should extract difficulty"() {
+        when:
+        def act = ExtractBlock.getDifficulty([chainwork: "00000000000000000000000000000000000000000ea25fe034fa07aed9e338eb"])
+        then:
+        act.toString(16) == "ea25fe034fa07aed9e338eb"
+    }
 }