From a41230a00ab60b952fd89d64abf4f554feba1409 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Wed, 22 Apr 2020 22:54:27 -0400 Subject: [PATCH] solution: track Bitcoin transactions --- .../emeraldpay/dshackle/rpc/BlockchainRpc.kt | 8 +- .../emeraldpay/dshackle/rpc/TrackBitcoinTx.kt | 158 +++++++++++ .../dshackle/rpc/TrackEthereumTx.kt | 17 +- .../io/emeraldpay/dshackle/rpc/TrackTx.kt | 10 + .../dshackle/upstream/AbstractHead.kt | 1 + .../io/emeraldpay/dshackle/upstream/Head.kt | 1 + .../dshackle/upstream/bitcoin/BitcoinApi.kt | 12 + .../dshackle/upstream/bitcoin/ExtractBlock.kt | 28 +- .../dshackle/rpc/TrackBitcoinTxSpec.groovy | 257 ++++++++++++++++++ .../dshackle/rpc/TrackEthereumTxSpec.groovy | 8 +- .../upstream/bitcoin/ExtractBlockSpec.groovy | 66 +++++ 11 files changed, 546 insertions(+), 20 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinTx.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinTxSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt index 0e34507f..acc48916 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt +++ b/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, @Autowired private val trackAddress: List, @Autowired private val describe: Describe, @Autowired private val subscribeStatus: SubscribeStatus @@ -48,7 +48,11 @@ class BlockchainRpc( } override fun subscribeTxStatus(request: Mono): Flux { - 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): Flux { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinTx.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinTx.kt new file mode 100644 index 00000000..4fb191df --- /dev/null +++ b/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 { + 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, txid: String): Flux { + 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, status: TxStatus): Flux { + 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, txid: String): Flux { + 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, tx: TxStatus): Mono { + return upstream.getHead().getFlux().flatMap { + upstream.getApi(Selector.empty).flatMap { api -> + loadExisting(api, tx.txid) + }.filter { it.mined } + }.single() + } + + fun withConfirmations(upstream: Upstream, tx: TxStatus): Flux { + return upstream.getHead().getFlux().map { + tx.withHead(it.height) + } + } + + fun loadExisting(api: BitcoinApi, txid: String): Mono { + 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 { + 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) + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt index c15f0048..651b0e9c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt +++ b/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() + 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 { 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): Flux { - return requestMono.map { request -> - prepareTracking(request) - }.flatMapMany { tx -> - streamAllUpdates(tx) - } + override fun subscribe(request: BlockchainOuterClass.TxStatusRequest): Flux { + val tx = prepareTracking(request) + return streamAllUpdates(tx) } private fun verifyAll(chain: Chain) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt new file mode 100644 index 00000000..40bcb0fc --- /dev/null +++ b/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 +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt index b2eb7ec0..e1795162 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt @@ -49,4 +49,5 @@ abstract class AbstractHead : Head { fun getCurrent(): BlockContainer? { return head.get() } + } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt index 142d0d6b..f5546b18 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt +++ b/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 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinApi.kt index beb8dcab..ae4d22f2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinApi.kt @@ -41,4 +41,16 @@ open class BitcoinApi( } } } + + open fun getBlock(hash: String): Mono> { + return executeAndResult(0, "getblock", listOf(hash), Map::class.java) as Mono> + } + + open fun getTx(txid: String): Mono> { + return executeAndResult(0, "getrawtransaction", listOf(txid, true), Map::class.java) as Mono> + } + + open fun getMempool(): Mono> { + return executeAndResult(0, "getrawmempool", emptyList(), List::class.java) as Mono> + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/ExtractBlock.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/ExtractBlock.kt index 25e94a01..330aa9ec 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/ExtractBlock.kt +++ b/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): Long? { + val height = data["height"] as Number? ?: return null + return height.toLong() + } + + @JvmStatic + fun getTime(data: Map): Instant? { + val time = data["time"] as Number? ?: return null + return Instant.ofEpochMilli(time.toLong() * 1000) + } + + @JvmStatic + fun getDifficulty(data: Map): 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 - 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?)?.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)), - Instant.ofEpochMilli(time.toLong() * 1000), + getDifficulty(data) ?: throw IllegalArgumentException("Block JSON has no chainwork"), + getTime(data) ?: throw IllegalArgumentException("Block JSON has no time"), false, json, transactions ) } + } \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinTxSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinTxSpec.groovy new file mode 100644 index 00000000..77a21f6b --- /dev/null +++ b/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)) + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy index c12b618c..414ce232 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy +++ b/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()) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/ExtractBlockSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/ExtractBlockSpec.groovy index 7acd5a18..2e5b0dd5 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/ExtractBlockSpec.groovy +++ b/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" + } }