From 1598797767cf97f876bd8df613260375ba8878bb Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Tue, 12 Jul 2022 20:56:17 -0400 Subject: [PATCH 1/4] problem: unclear how grpc upstream labels works --- .../io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt index a118d956..ba06b973 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt @@ -224,7 +224,11 @@ class UpstreamsConfigReader( upNode: MappingNode, upstream: UpstreamsConfig.Upstream ) { + // Dshackle gRPC connection dispatches requests to different upstreams, which may + // be on different blockchains, and each may have different set of labels. + // So the labels and chains assigned to the gRPC connection make no sense. if (hasAny(upNode, "labels")) { + // Actual labels from underlying upstreams are handled by GrpcUpstreamStatus log.warn("Labels should be not applied to gRPC upstream") } if (hasAny(upNode, "chain")) { From afebfd7363c26f662051f557f2450d59202063e0 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Wed, 20 Jul 2022 18:35:39 -0400 Subject: [PATCH 2/4] problem: not always clear why an upstream is not available --- .../io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt index c55fcb5a..480d6f8c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt @@ -91,7 +91,7 @@ class GrpcUpstreams( client.describe(BlockchainOuterClass.DescribeRequest.newBuilder().build()) }.onErrorContinue { t, _ -> if (ExceptionUtils.indexOfType(t, ConnectException::class.java) >= 0) { - log.warn("gRPC upstream $host:$port is unavailable") + log.warn("gRPC upstream $host:$port is unavailable. (${t.javaClass}: ${t.message})") known.values.forEach { it.setStatus(UpstreamAvailability.UNAVAILABLE) } From 7f67171ad6174d8b769900bbac14d43f8af85513 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Thu, 21 Jul 2022 22:07:12 -0400 Subject: [PATCH 3/4] problem: doesn't request Bitcoin balancer from a remote Grpc node --- .../io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt | 11 +++++------ .../dshackle/startup/ConfiguredUpstreams.kt | 2 +- 2 files changed, 6 insertions(+), 7 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt index 31490ee9..1a535c45 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt @@ -57,7 +57,7 @@ class TrackBitcoinAddress( /** * Keep tracking of the current state of local upstreams. True for a chain that has an upstream with balance data. */ - private val balanceAvailable: MutableMap = ConcurrentHashMap() + private val localBalanceAvailable: MutableMap = ConcurrentHashMap() /** * Criteria for a remote grpc upstream that can provide a balance @@ -72,7 +72,7 @@ class TrackBitcoinAddress( multistreamHolder.observeChains().subscribe { chain -> multistreamHolder.getUpstream(chain)?.let { mup -> val available = mup.getAll().any { up -> - !up.isGrpc() && (up.getOptions().providesBalance ?: false) + !up.isGrpc() && up.getCapabilities().contains(Capability.BALANCE) } setBalanceAvailability(chain, available) } @@ -80,14 +80,14 @@ class TrackBitcoinAddress( } fun setBalanceAvailability(chain: Chain, enabled: Boolean) { - balanceAvailable[chain] = enabled + localBalanceAvailable[chain] = enabled } /** * @return true if the current instance has data sources to provide the balance */ fun isBalanceAvailable(chain: Chain): Boolean { - return balanceAvailable[chain] ?: false + return localBalanceAvailable[chain] ?: false } fun allAddresses(api: BitcoinMultistream, request: BlockchainOuterClass.BalanceRequest): Flux { @@ -179,8 +179,7 @@ class TrackBitcoinAddress( } .timeout(Defaults.timeoutInternal, Mono.empty()) .switchIfEmpty( - Mono.just(0) - .doOnNext { + Mono.fromCallable { log.warn("No upstream providing balance for ${api.chain}") } .then(Mono.error(SilentException.DataUnavailable("BALANCE"))) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 48e4f189..264294be 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -257,7 +257,7 @@ open class ConfiguredUpstreams( log.info("Using ALL CHAINS (gRPC) upstream, at ${endpoint.host}:${endpoint.port}") ds.start() .doOnNext { - log.info("Chain ${it.chain} has ${it.type} through gRPC at ${endpoint.host}:${endpoint.port}") + log.info("Chain ${it.chain} ${it.type} through gRPC at ${endpoint.host}:${endpoint.port}. With caps: ${it.upstream.getCapabilities()}") } .subscribe(currentUpstreams::update) } From d65a21753504718e1833757502f6ce025a90095c Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Thu, 21 Jul 2022 22:09:37 -0400 Subject: [PATCH 4/4] problem: Bitcoin listunspent is not working when the upstream do not provide it, though have other sources of unspent data solution: wrapper for listunspent requests that uses Dshackle mechanism to gather data --- .../dshackle/rpc/TrackBitcoinAddress.kt | 4 +- .../upstream/bitcoin/BitcoinMultistream.kt | 31 +++++++----- .../upstream/bitcoin/BitcoinReader.kt | 3 ++ .../upstream/bitcoin/EsploraUnspentReader.kt | 1 - .../upstream/bitcoin/LocalCallRouter.kt | 39 +++++++++++++++ .../upstream/bitcoin/RemoteUnspentReader.kt | 48 +++++++++++++++++++ .../upstream/bitcoin/RpcUnspentReader.kt | 13 +++-- .../upstream/bitcoin/data/RpcUnspent.kt | 1 - .../bitcoin/data/RpcUnspentDeserializer.kt | 1 - .../upstream/bitcoin/data/SimpleUnspent.kt | 1 - .../upstream/grpc/BitcoinGrpcUpstream.kt | 4 ++ .../upstream/grpc/EthereumGrpcUpstream.kt | 4 ++ .../dshackle/upstream/grpc/GrpcUpstream.kt | 6 ++- .../rpc/TrackBitcoinAddressSpec.groovy | 12 ++--- .../upstream/bitcoin/BitcoinReaderSpec.groovy | 3 +- .../bitcoin/RpcUnspentReaderSpec.groovy | 14 ++---- 16 files changed, 144 insertions(+), 41 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RemoteUnspentReader.kt diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt index 1a535c45..0536d655 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt @@ -180,8 +180,8 @@ class TrackBitcoinAddress( .timeout(Defaults.timeoutInternal, Mono.empty()) .switchIfEmpty( Mono.fromCallable { - log.warn("No upstream providing balance for ${api.chain}") - } + log.warn("No upstream providing balance for ${api.chain}") + } .then(Mono.error(SilentException.DataUnavailable("BALANCE"))) ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt index c7740f1e..45f95684 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt @@ -26,7 +26,7 @@ import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.RequestPostprocessor import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Upstream -import io.emeraldpay.dshackle.upstream.ethereum.LocalCallRouter +import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.Chain @@ -37,28 +37,34 @@ import reactor.core.publisher.Mono @Suppress("UNCHECKED_CAST") open class BitcoinMultistream( chain: Chain, - val upstreams: MutableList, + private val sourceUpstreams: MutableList, caches: Caches -) : Multistream(chain, upstreams as MutableList, caches, RequestPostprocessor.Empty()), Lifecycle { +) : Multistream(chain, sourceUpstreams as MutableList, caches, RequestPostprocessor.Empty()), Lifecycle { companion object { private val log = LoggerFactory.getLogger(BitcoinMultistream::class.java) } private var head: Head = EmptyHead() - private var esplora = upstreams.find { it.esploraClient != null }?.esploraClient + private var esplora = sourceUpstreams.find { it.esploraClient != null }?.esploraClient private var reader = BitcoinReader(this, head, esplora) private var addressActiveCheck: AddressActiveCheck? = null private var xpubAddresses: XpubAddresses? = null private val feeEstimation = BitcoinFees(this, reader, 6) + private var callRouter: LocalCallRouter = LocalCallRouter(DefaultBitcoinMethods(), reader) override fun init() { - if (upstreams.size > 0) { + if (sourceUpstreams.size > 0) { head = updateHead() } super.init() } + open val upstreams: List + get() { + return sourceUpstreams + } + override fun getFeeEstimation(): ChainFees { return feeEstimation } @@ -75,8 +81,8 @@ open class BitcoinMultistream( } lagObserver?.stop() lagObserver = null - val head = if (upstreams.size == 1) { - val upstream = upstreams.first() + val head = if (sourceUpstreams.size == 1) { + val upstream = sourceUpstreams.first() upstream.setLag(0) upstream.getHead().apply { if (this is Lifecycle) { @@ -84,10 +90,10 @@ open class BitcoinMultistream( } } } else { - val newHead = MergedHead(upstreams.map { it.getHead() }).apply { + val newHead = MergedHead(sourceUpstreams.map { it.getHead() }).apply { this.start() } - val lagObserver = BitcoinHeadLagObserver(newHead, upstreams) + val lagObserver = BitcoinHeadLagObserver(newHead, sourceUpstreams) this.lagObserver = lagObserver lagObserver.start() newHead @@ -97,7 +103,7 @@ open class BitcoinMultistream( } override fun getRoutedApi(matcher: Selector.Matcher): Mono> { - return Mono.just(LocalCallRouter(getMethods())) + return Mono.just(callRouter) } open fun getReader(): BitcoinReader { @@ -106,10 +112,11 @@ open class BitcoinMultistream( override fun onUpstreamsUpdated() { super.onUpstreamsUpdated() - esplora = upstreams.find { it.esploraClient != null }?.esploraClient + esplora = sourceUpstreams.find { it.esploraClient != null }?.esploraClient reader = BitcoinReader(this, this.head, esplora) addressActiveCheck = esplora?.let { AddressActiveCheck(it) } xpubAddresses = addressActiveCheck?.let { XpubAddresses(it) } + callRouter = LocalCallRouter(getMethods(), reader) } override fun setHead(head: Head) { @@ -122,7 +129,7 @@ open class BitcoinMultistream( } override fun getLabels(): Collection { - return upstreams.flatMap { it.getLabels() } + return sourceUpstreams.flatMap { it.getLabels() } } @Suppress("UNCHECKED_CAST") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt index d72148f6..f9f0496c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt @@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.upstream.bitcoin import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.upstream.Capability import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent @@ -43,6 +44,8 @@ open class BitcoinReader( private val unspentReader: UnspentReader = if (esploraClient != null) { EsploraUnspentReader(esploraClient, head) + } else if (upstreams.upstreams.any { it.isGrpc() && it.getCapabilities().contains(Capability.BALANCE) }) { + RemoteUnspentReader(upstreams) } else { RpcUnspentReader(upstreams) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/EsploraUnspentReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/EsploraUnspentReader.kt index 2b2e2e11..5c5317a8 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/EsploraUnspentReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/EsploraUnspentReader.kt @@ -37,7 +37,6 @@ class EsploraUnspentReader( base.txid, base.vout, base.value, - head.getCurrentHeight()?.let { base.height - it } ?: 0 ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/LocalCallRouter.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/LocalCallRouter.kt index 5f77cfca..f1abee27 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/LocalCallRouter.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/LocalCallRouter.kt @@ -15,12 +15,17 @@ */ package io.emeraldpay.dshackle.upstream.bitcoin +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.SilentException import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.bitcoin.data.RpcUnspent +import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.etherjar.rpc.RpcException import io.emeraldpay.etherjar.rpc.RpcResponseError +import org.bitcoinj.core.Address import org.slf4j.LoggerFactory import reactor.core.publisher.Mono @@ -33,6 +38,7 @@ import reactor.core.publisher.Mono */ class LocalCallRouter( private val methods: CallMethods, + private val reader: BitcoinReader, ) : Reader { companion object { @@ -47,6 +53,39 @@ class LocalCallRouter( if (!methods.isCallable(key.method)) { return Mono.error(RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Unsupported method")) } + if (key.method == "listunspent") { + return processUnspentRequest(key) + } return Mono.empty() } + + /** + * + */ + fun processUnspentRequest(key: JsonRpcRequest): Mono { + if (key.params.size < 3) { + return Mono.error(SilentException("Invalid call to unspent. Address is missing")) + } + val addresses = key.params[2] + if (addresses is List<*> && addresses.size > 0) { + val address = addresses[0].toString().let { Address.fromString(null, it) } + return reader.listUnspent(address).map { + val rpc = it.map(convertUnspent(address)) + val json = Global.objectMapper.writeValueAsBytes(rpc) + JsonRpcResponse.ok(json, JsonRpcResponse.NumberId(key.id)) + } + } + return Mono.error(SilentException("Invalid call to unspent")) + } + + fun convertUnspent(address: Address): (SimpleUnspent) -> RpcUnspent { + return { base -> + RpcUnspent( + base.txid, + base.vout, + address.toString(), + base.value, + ) + } + } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RemoteUnspentReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RemoteUnspentReader.kt new file mode 100644 index 00000000..60d20209 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RemoteUnspentReader.kt @@ -0,0 +1,48 @@ +package io.emeraldpay.dshackle.upstream.bitcoin + +import io.emeraldpay.api.proto.BlockchainOuterClass.BalanceRequest +import io.emeraldpay.dshackle.upstream.Capability +import io.emeraldpay.dshackle.upstream.Selector +import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent +import io.emeraldpay.dshackle.upstream.grpc.BitcoinGrpcUpstream +import org.bitcoinj.core.Address +import org.slf4j.LoggerFactory +import reactor.core.publisher.Mono + +class RemoteUnspentReader( + val upstreams: BitcoinMultistream +) : UnspentReader { + + companion object { + private val log = LoggerFactory.getLogger(RemoteUnspentReader::class.java) + } + + private val selector = Selector.LocalAndMatcher( + Selector.GrpcMatcher(), + Selector.CapabilityMatcher(Capability.BALANCE) + ) + + override fun read(key: Address): Mono> { + val apis = upstreams.getApiSource(selector) + apis.request(1) + return Mono.from(apis) + .map { up -> + up.cast(BitcoinGrpcUpstream::class.java).remote + } + .flatMapMany { + val request = BalanceRequest.newBuilder() + .build() + it.getBalance(request) + } + .map { resp -> + resp.utxoList.map { utxo -> + SimpleUnspent( + utxo.txId, + utxo.index.toInt(), + utxo.balance.toLong(), + ) + } + } + .reduce(List::plus) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RpcUnspentReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RpcUnspentReader.kt index 773109bc..adb382e8 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RpcUnspentReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RpcUnspentReader.kt @@ -16,6 +16,8 @@ package io.emeraldpay.dshackle.upstream.bitcoin import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.SilentException +import io.emeraldpay.dshackle.upstream.Capability import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.bitcoin.data.RpcUnspent import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent @@ -38,14 +40,17 @@ class RpcUnspentReader( base.txid, base.vout, base.amount, - base.confirmations ) } + private val selector = Selector.CapabilityMatcher(Capability.BALANCE) + override fun read(key: Address): Mono> { + // docs: https://developer.bitcoin.org/reference/rpc/listunspent.html + // val address = key.toString() - return upstreams.getDirectApi(Selector.empty).flatMap { api -> - api.read(JsonRpcRequest("listunspent", emptyList())) + return upstreams.getDirectApi(selector).flatMap { api -> + api.read(JsonRpcRequest("listunspent", listOf(1, 9999999, listOf(address)))) .flatMap(JsonRpcResponse::requireResult) .map { Global.objectMapper.readerFor(RpcUnspent::class.java).readValues(it).readAll() @@ -55,6 +60,6 @@ class RpcUnspentReader( it.address == address }.map(convert) } - } + }.switchIfEmpty(Mono.error(SilentException.DataUnavailable("BALANCE"))) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/RpcUnspent.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/RpcUnspent.kt index ee3be9c2..d5c91fa6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/RpcUnspent.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/RpcUnspent.kt @@ -20,5 +20,4 @@ data class RpcUnspent( val vout: Int, val address: String, val amount: Long, - val confirmations: Long ) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/RpcUnspentDeserializer.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/RpcUnspentDeserializer.kt index 85d1d308..19a830a5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/RpcUnspentDeserializer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/RpcUnspentDeserializer.kt @@ -30,7 +30,6 @@ class RpcUnspentDeserializer : JsonDeserializer() { node.get("vout").asInt(), node.get("address").asText(), BigDecimal(node.get("amount").asText()).multiply(BigDecimal.TEN.pow(8)).longValueExact(), - node.get("confirmations").asLong() ) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/SimpleUnspent.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/SimpleUnspent.kt index 2c6bb471..0c032b71 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/SimpleUnspent.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/SimpleUnspent.kt @@ -19,5 +19,4 @@ data class SimpleUnspent( val txid: String, val vout: Int, val value: Long, - val confirmations: Long ) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt index 542688d3..07789163 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt @@ -99,6 +99,10 @@ class BitcoinGrpcUpstream( var timeout = Defaults.timeout private var capabilities: Set = emptySet() + override fun getBlockchainApi(): ReactorBlockchainGrpc.ReactorBlockchainStub { + return remote + } + override fun getHead(): Head { return grpcHead } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt index 3951d981..4bf85743 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt @@ -103,6 +103,10 @@ open class EthereumGrpcUpstream( private val defaultReader: Reader = client.forSelector(Selector.empty) var timeout = Defaults.timeout + override fun getBlockchainApi(): ReactorBlockchainGrpc.ReactorBlockchainStub { + return remote + } + override fun start() { } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt index 166252cb..5b001af4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt @@ -16,12 +16,16 @@ package io.emeraldpay.dshackle.upstream.grpc import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.api.proto.ReactorBlockchainGrpc +import io.emeraldpay.dshackle.upstream.Upstream -interface GrpcUpstream { +interface GrpcUpstream : Upstream { /** * Update the configuration of the upstream with the new data. * Called on the first creation, and each time a new state received from upstream */ fun update(conf: BlockchainOuterClass.DescribeChain) + + fun getBlockchainApi(): ReactorBlockchainGrpc.ReactorBlockchainStub } diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy index 59a9008d..49cd8024 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy @@ -51,7 +51,7 @@ class TrackBitcoinAddressSpec extends Specification { def "Correct sum for single"() { setup: def unspents = [ - new SimpleUnspent("f14b222e652c58d11435fa9172ddea000c6f5e20e6b715eb940fc28d1c4adeef", 0, 100L, 123L) + new SimpleUnspent("f14b222e652c58d11435fa9172ddea000c6f5e20e6b715eb940fc28d1c4adeef", 0, 100L) ] TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder)) def address = new TrackBitcoinAddress.Address( @@ -68,8 +68,8 @@ class TrackBitcoinAddressSpec extends Specification { def "Correct sum for few"() { setup: def unspents = [ - new SimpleUnspent("f14b222e652c58d11435fa9172ddea000c6f5e20e6b715eb940fc28d1c4adeef", 0, 100L, 123L), - new SimpleUnspent("17d1c4adf14b222e652c58d11435fa9ee2ddea000c6f5e20e6b715eb940fc28f", 0, 123L, 123L), + new SimpleUnspent("f14b222e652c58d11435fa9172ddea000c6f5e20e6b715eb940fc28d1c4adeef", 0, 100L), + new SimpleUnspent("17d1c4adf14b222e652c58d11435fa9ee2ddea000c6f5e20e6b715eb940fc28f", 0, 123L), ] TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder)) def address = new TrackBitcoinAddress.Address( @@ -120,8 +120,8 @@ class TrackBitcoinAddressSpec extends Specification { def "Correct sum for few with utxo"() { setup: def unspents = [ - new SimpleUnspent("f14b222e652c58d11435fa9172ddea000c6f5e20e6b715eb940fc28d1c4adeef", 0, 100L, 123L), - new SimpleUnspent("17d1c4adf14b222e652c58d11435fa9ee2ddea000c6f5e20e6b715eb940fc28f", 0, 123L, 123L), + new SimpleUnspent("f14b222e652c58d11435fa9172ddea000c6f5e20e6b715eb940fc28d1c4adeef", 0, 100L), + new SimpleUnspent("17d1c4adf14b222e652c58d11435fa9ee2ddea000c6f5e20e6b715eb940fc28f", 0, 123L), ] TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder)) def address = new TrackBitcoinAddress.Address( @@ -282,7 +282,7 @@ class TrackBitcoinAddressSpec extends Specification { 2 * listUnspent(_) >>> [ Mono.just([]), Mono.just([ - new SimpleUnspent("f14b222e652c58d11435fa9172ddea000c6f5e20e6b715eb940fc28d1c4adeef", 0, 1230000L, 123L) + new SimpleUnspent("f14b222e652c58d11435fa9172ddea000c6f5e20e6b715eb940fc28d1c4adeef", 0, 1230000L) ]) ] } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReaderSpec.groovy index 5e65cd5f..777277eb 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReaderSpec.groovy @@ -34,7 +34,8 @@ class BitcoinReaderSpec extends Specification { api.answerOnce("getblockhash", [100000], "000000000003ba27aa200b1cecaad478d2b00432346c3f1f3986da1afd33e506") api.answerOnce("getblock", ["000000000003ba27aa200b1cecaad478d2b00432346c3f1f3986da1afd33e506"], block) def ups = Mock(BitcoinMultistream) { - _ * it.getDirectApi(_) >> Mono.just(api) + _ * getDirectApi(_) >> Mono.just(api) + _ * upstreams >> [] } def reader = new BitcoinReader(ups, Stub(Head), null) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/RpcUnspentReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/RpcUnspentReaderSpec.groovy index f65bc832..e47df718 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/RpcUnspentReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/RpcUnspentReaderSpec.groovy @@ -29,7 +29,7 @@ class RpcUnspentReaderSpec extends Specification { setup: def json = this.class.getClassLoader().getResourceAsStream("bitcoin/unspent-one-addr.json").bytes def rpcReader = Mock(Reader) { - 1 * read(new JsonRpcRequest("listunspent", [])) >> Mono.just(JsonRpcResponse.ok(json)) + 1 * read(new JsonRpcRequest("listunspent", [1, 9999999, ["1K7xkspJg7DDKNwzXgoRSDCUxiFsRegsSK"]])) >> Mono.just(JsonRpcResponse.ok(json)) } def upstreams = Mock(BitcoinMultistream) { 1 * getDirectApi(_) >> Mono.just(rpcReader) @@ -45,19 +45,16 @@ class RpcUnspentReaderSpec extends Specification { with(act[0]) { txid == "e0f946c8f971b25cdffa64eed71d886019e437c0bf6a1b280584c0be5d1b5409" vout == 29 - confirmations == 2010 value == 1230030 } with(act[1]) { txid == "66e1e4d14ed6f454d2fda036f35cba423274ecdf5d46deb93f172c412a0f650d" vout == 83 - confirmations == 4963 value == 756339 } with(act[35]) { txid == "f14b222e652c58d11435fa9172ddea000c6f5e20e6b715eb940fc28d1c4adeef" vout == 58 - confirmations == 2139 value == 1105047 } } @@ -66,7 +63,7 @@ class RpcUnspentReaderSpec extends Specification { setup: def json = this.class.getClassLoader().getResourceAsStream("bitcoin/unspent-two-addr.json").bytes def rpcReader = Mock(Reader) { - 1 * read(new JsonRpcRequest("listunspent", [])) >> Mono.just(JsonRpcResponse.ok(json)) + 1 * read(new JsonRpcRequest("listunspent", [1, 9999999, ["35hK24tcLEWcgNA4JxpvbkNkoAcDGqQPsP"]])) >> Mono.just(JsonRpcResponse.ok(json)) } def upstreams = Mock(BitcoinMultistream) { 1 * getDirectApi(_) >> Mono.just(rpcReader) @@ -83,19 +80,16 @@ class RpcUnspentReaderSpec extends Specification { txid == "8ad0d954a01eeb4f2c62d58d291699af847f9c8df43b775c27ffe8a5f76eba00" vout == 1 value == 216465 - confirmations == 2583 } with(act[11]) { txid == "777671a46b30b068052a73387e036bc8515cd3ba6adf9be4c70dfc0699f67c09" vout == 0 - confirmations == 13705 value == 307906 } // cat src/test/resources/bitcoin/unspent-two-addr.json | jq '[.[] | select(.address == "35hK24tcLEWcgNA4JxpvbkNkoAcDGqQPsP")] | .[211]' with(act[211]) { txid == "f20727393b0a586a3062a615fb71f43ec21c24258c3c6ec546fee5cbc1fa2ba7" vout == 0 - confirmations == 21890 value == 50000000000 } } @@ -104,7 +98,7 @@ class RpcUnspentReaderSpec extends Specification { setup: def json = this.class.getClassLoader().getResourceAsStream("bitcoin/unspent-two-addr.json").bytes def rpcReader = Mock(Reader) { - 1 * read(new JsonRpcRequest("listunspent", [])) >> Mono.just(JsonRpcResponse.ok(json)) + 1 * read(_) >> Mono.just(JsonRpcResponse.ok(json)) } def upstreams = Mock(BitcoinMultistream) { 1 * getDirectApi(_) >> Mono.just(rpcReader) @@ -122,13 +116,11 @@ class RpcUnspentReaderSpec extends Specification { txid == "e0f946c8f971b25cdffa64eed71d886019e437c0bf6a1b280584c0be5d1b5409" vout == 29 value == 1230030 - confirmations == 2030 } // cat src/test/resources/bitcoin/unspent-two-addr.json | jq '[.[] | select(.address == "1K7xkspJg7DDKNwzXgoRSDCUxiFsRegsSK")] | .[35]' with(act[35]) { txid == "f14b222e652c58d11435fa9172ddea000c6f5e20e6b715eb940fc28d1c4adeef" vout == 58 - confirmations == 2159 value == 1105047 } }