solution: working rpc to get bitcoin address balance

2020-04-19 20:47:16 -04:00
parent ba996a8d2b
commit 4f0b32f705
14 changed files with 5269 additions and 67 deletions
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt
@@ -19,6 +19,7 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
 import io.emeraldpay.api.proto.ReactorBlockchainGrpc
 import io.emeraldpay.dshackle.BlockchainType
+import io.emeraldpay.dshackle.SilentException
 import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
@@ -31,7 +32,7 @@ class BlockchainRpc(
        @Autowired private val nativeCall: NativeCall,
        @Autowired private val streamHead: StreamHead,
        @Autowired private val trackEthereumTx: TrackEthereumTx,
-        @Autowired private val trackEthereumAddress: TrackEthereumAddress,
+        @Autowired private val trackAddress: List<TrackAddress>,
        @Autowired private val describe: Describe,
        @Autowired private val subscribeStatus: SubscribeStatus
 ): ReactorBlockchainGrpc.BlockchainImplBase() {
@@ -50,12 +51,20 @@ class BlockchainRpc(
        return trackEthereumTx.add(request)
    }

-    override fun subscribeBalance(request: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {
-        return trackEthereumAddress.subscribe(request)
+    override fun subscribeBalance(requestMono: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {
+        return requestMono.flatMapMany { request ->
+            val chain = Chain.byId(request.asset.chainValue)
+            trackAddress.find { it.isSupported(chain) }?.subscribe(request)
+                    ?: Flux.error(SilentException.UnsupportedBlockchain(chain))
+        }
    }

-    override fun getBalance(request: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {
-        return trackEthereumAddress.getBalance(request)
+    override fun getBalance(requestMono: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {
+        return requestMono.flatMapMany { request ->
+            val chain = Chain.byId(request.asset.chainValue)
+            trackAddress.find { it.isSupported(chain) }?.getBalance(request)
+                    ?: Flux.error(SilentException.UnsupportedBlockchain(chain))
+        }
    }

    override fun describe(request: Mono<BlockchainOuterClass.DescribeRequest>): Mono<BlockchainOuterClass.DescribeResponse> {
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt
@@ -0,0 +1,17 @@
+package io.emeraldpay.dshackle.rpc
+
+import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.grpc.Chain
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+
+/**
+ * Base interface to tracking balance on a single blockchain
+ */
+interface TrackAddress {
+
+    fun isSupported(chain: Chain): Boolean
+    fun getBalance(requestMono: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance>
+    fun subscribe(requestMono: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance>
+
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt
@@ -0,0 +1,120 @@
+package io.emeraldpay.dshackle.rpc
+
+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.Upstreams
+import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinApi
+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.util.function.Tuples
+import java.math.BigDecimal
+import java.math.BigInteger
+import java.util.*
+
+@Service
+class TrackBitcoinAddress(
+        @Autowired private val upstreams: Upstreams
+) : TrackAddress {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(TrackBitcoinAddress::class.java)
+    }
+
+    override fun isSupported(chain: Chain): Boolean {
+        return BlockchainType.fromBlockchain(chain) == BlockchainType.BITCOIN && upstreams.isAvailable(chain)
+    }
+
+    override fun getBalance(req: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> {
+        if (!req.hasAddress()) {
+            return Flux.error(SilentException("Address not provided"))
+        }
+        val chain = Chain.byId(req.asset.chainValue)
+        val upstream = upstreams.getUpstream(chain)?.castApi(BitcoinApi::class.java)
+                ?: return Flux.error(SilentException.UnsupportedBlockchain(req.asset.chainValue))
+        val addressesAll = when {
+            req.address.hasAddressSingle() -> {
+                listOf(req.address.addressSingle.address)
+            }
+            req.address.hasAddressMulti() -> {
+                req.address.addressMulti.addressesList
+                        .map { addr -> addr.address }
+            }
+            else -> {
+                return Flux.error(SilentException("Unsupported address"))
+            }
+        }
+        val result = upstream.getApi(Selector.empty).flatMapMany { api ->
+            val addresses = addressesAll.sorted()
+            val results = api.executeAndResult(0, "listunspent", emptyList(), List::class.java)
+                    .flatMapMany { unspents ->
+                        val result = getTotal(chain, addresses, unspents)
+                        Flux.fromIterable(result)
+                    }
+            results.map { addr ->
+                buildResponse(addr)
+            }
+        }
+        return result
+    }
+
+    fun getTotal(chain: Chain, addresses: List<String>, unspents: List<*>): List<AddressBalance> {
+        return unspents.asSequence()
+                .filterIsInstance<Map<String, Any>>()
+                .filter { unspent ->
+                    unspent.containsKey("address")
+                            && Collections.binarySearch(addresses, unspent["address"] as String) >= 0
+                            && unspent.containsKey("amount")
+                }
+                .map {
+                    Tuples.of(it["address"] as String, it["amount"] as Number)
+                }
+                .map {
+                    AddressBalance(chain, it.t1,
+                            //use toString because toDecimal makes rounding
+                            BigDecimal(it.t2.toString()).multiply(BigDecimal.TEN.pow(8)).toBigInteger()
+                    )
+                }
+                .plus(
+                        //add default ZERO value
+                        addresses.map {
+                            AddressBalance(chain, it, BigInteger.ZERO)
+                        }
+                )
+                .groupBy {
+                    it.address
+                }
+                .map {
+                    it.value.reduceRight { x, acc ->
+                        x.plus(acc)
+                    }
+                }.toList()
+    }
+
+
+    override fun subscribe(request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> {
+        return Flux.error(SilentException("Not Implemented"))
+    }
+
+    private fun buildResponse(address: AddressBalance): BlockchainOuterClass.AddressBalance {
+        return BlockchainOuterClass.AddressBalance.newBuilder()
+                .setBalance(address.balance.toString(10))
+                .setAsset(Common.Asset.newBuilder()
+                        .setChainValue(address.chain.id)
+                        .setCode("BTC"))
+                .setAddress(Common.SingleAddress.newBuilder().setAddress(address.address))
+                .build()
+    }
+
+    open class AddressBalance(val chain: Chain, val address: String, var balance: BigInteger = BigInteger.ZERO) {
+        open fun withBalance(balance: BigInteger) = AddressBalance(chain, address, balance)
+
+        open fun plus(other: AddressBalance) = AddressBalance(chain, address, balance + other.balance)
+
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt
@@ -51,7 +51,7 @@ import javax.annotation.PostConstruct
 class TrackEthereumAddress(
        @Autowired private val upstreams: Upstreams,
        @Autowired private val upstreamScheduler: Scheduler
-) {
+) : TrackAddress {

    private val log = LoggerFactory.getLogger(TrackEthereumAddress::class.java)
    private val clients = HashMap<Chain, ConcurrentLinkedQueue<TrackedAddress>>()
@@ -83,6 +83,10 @@ class TrackEthereumAddress(
        }
    }

+    override fun isSupported(chain: Chain): Boolean {
+        return BlockchainType.fromBlockchain(chain) == BlockchainType.ETHEREUM && upstreams.isAvailable(chain)
+    }
+
    private fun startTracking(client: TrackedAddress) {
        clients[client.chain]?.add(client) ?: log.warn("Chain ${client.chain} is not available for tracking")
    }
@@ -99,9 +103,6 @@ class TrackEthereumAddress(

    private fun initializeSimple(request: BlockchainOuterClass.BalanceRequest): Flux<SimpleAddress> {
        val chain = Chain.byId(request.asset.chainValue)
-        if (BlockchainType.fromBlockchain(chain) != BlockchainType.ETHEREUM) {
-            return Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue))
-        }
        if (!upstreams.isAvailable(chain)) {
            return Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue))
        }
@@ -128,43 +129,35 @@ class TrackEthereumAddress(
                }
    }

-    fun subscribe(requestMono: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {
-        return requestMono.flatMapMany { request ->
-            val chain = Chain.byId(request.asset.chainValue)
-            if (BlockchainType.fromBlockchain(chain) != BlockchainType.ETHEREUM) {
-                return@flatMapMany Flux.error<BlockchainOuterClass.AddressBalance>(SilentException.UnsupportedBlockchain(request.asset.chainValue))
-            }
-            val bus = TopicProcessor.create<BlockchainOuterClass.AddressBalance>()
-            initializeSubscription(request, bus)
-                    .flatMap { tracked ->
-                        val current = getBalance(tracked).map {
-                            tracked.withBalance(it)
-                        }.doOnNext {
-                            startTracking(it)
-                        }.map {
-                            buildResponse(it)
-                        }
-                        Flux.merge(current, bus).doFinally { stopTracking(tracked) }
+    override fun subscribe(request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> {
+        val bus = TopicProcessor.create<BlockchainOuterClass.AddressBalance>()
+        return initializeSubscription(request, bus)
+                .flatMap { tracked ->
+                    val current = getBalance(tracked).map {
+                        tracked.withBalance(it)
+                    }.doOnNext {
+                        startTracking(it)
+                    }.map {
+                        buildResponse(it)
                    }
-                    .doOnError { t ->
-                        if (t is SilentException) {
-                            if (t is SilentException.UnsupportedBlockchain) {
-                                log.warn("Unsupported blockchain: ${t.blockchainId}")
-                            }
-                            log.debug("Failed to process subscription", t)
-                        } else {
-                            log.warn("Failed to process subscription", t)
+                    Flux.merge(current, bus).doFinally { stopTracking(tracked) }
+                }
+                .doOnError { t ->
+                    if (t is SilentException) {
+                        if (t is SilentException.UnsupportedBlockchain) {
+                            log.warn("Unsupported blockchain: ${t.blockchainId}")
                        }
+                        log.debug("Failed to process subscription", t)
+                    } else {
+                        log.warn("Failed to process subscription", t)
                    }
-        }
+                }
    }

-    fun getBalance(requestMono: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {
-        return requestMono.flatMapMany { request ->
-            initializeSimple(request)
-                    .flatMap { a -> getBalance(a).map { a.withBalance(it) } }
-                    .map { buildResponse(it) }
-        }
+    override fun getBalance(request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> {
+        return initializeSimple(request)
+                .flatMap { a -> getBalance(a).map { a.withBalance(it) } }
+                .map { buildResponse(it) }
    }

    private fun simpleAddress(address: Common.SingleAddress, chain: Chain): SimpleAddress {
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt
@@ -33,5 +33,6 @@ interface Upstream<out T : UpstreamApi> {
    fun getMethods(): CallMethods
    fun getId(): String

-    fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, upstreamType: Class<TA>): T
+    fun <A : UpstreamApi> castApi(apiType: Class<A>): Upstream<A>
+    fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, apiType: Class<TA>): T
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinChainUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinChainUpstreams.kt
@@ -5,7 +5,6 @@ import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.upstream.*
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumHeadLagObserver
 import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
@@ -66,14 +65,18 @@ class BitcoinChainUpstreams(
        return upstreams.flatMap { it.getLabels() }
    }

-    override fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, upstreamType: Class<TA>): T {
+    override fun <A : UpstreamApi> castApi(apiType: Class<A>): Upstream<A> {
+        if (!apiType.isAssignableFrom(BitcoinApi::class.java)) {
+            throw ClassCastException("Cannot cast ${EthereumApi::class.java} to $apiType")
+        }
+        return this as Upstream<A>
+    }
+
+    override fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, apiType: Class<TA>): T {
        if (!selfType.isAssignableFrom(this.javaClass)) {
            throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
        }
-        if (!upstreamType.isAssignableFrom(BitcoinApi::class.java)) {
-            throw ClassCastException("Cannot cast ${EthereumApi::class.java} to $upstreamType")
-        }
-        return this as T
+        return castApi(apiType) as T
    }

 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt
@@ -3,7 +3,6 @@ package io.emeraldpay.dshackle.upstream.bitcoin
 import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.upstream.*
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator
 import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
@@ -44,14 +43,18 @@ class BitcoinUpstream(
        return listOf(UpstreamsConfig.Labels())
    }

-    override fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, upstreamType: Class<TA>): T {
+    override fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, apiType: Class<TA>): T {
        if (!selfType.isAssignableFrom(this.javaClass)) {
            throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
        }
-        if (!upstreamType.isAssignableFrom(BitcoinApi::class.java)) {
-            throw ClassCastException("Cannot cast ${BitcoinApi::class.java} to $upstreamType")
+        return castApi(apiType) as T
+    }
+
+    override fun <A : UpstreamApi> castApi(apiType: Class<A>): Upstream<A> {
+        if (!apiType.isAssignableFrom(BitcoinApi::class.java)) {
+            throw ClassCastException("Cannot cast ${BitcoinApi::class.java} to $apiType")
        }
-        return this as T
+        return this as Upstream<A>
    }

    override fun isRunning(): Boolean {
@@ -88,4 +91,5 @@ class BitcoinUpstream(
        validatorSubscription?.dispose()
    }

+
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainUpstreams.kt
@@ -86,14 +86,18 @@ class EthereumChainUpstreams(
    }

    @SuppressWarnings("unchecked")
-    override fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, upstreamType: Class<TA>): T {
+    override fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, apiType: Class<TA>): T {
        if (!selfType.isAssignableFrom(this.javaClass)) {
            throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
        }
-        if (!upstreamType.isAssignableFrom(EthereumApi::class.java)) {
-            throw ClassCastException("Cannot cast ${EthereumApi::class.java} to $upstreamType")
+        return castApi(apiType) as T
+    }
+
+    override fun <A : UpstreamApi> castApi(apiType: Class<A>): Upstream<A> {
+        if (!apiType.isAssignableFrom(EthereumApi::class.java)) {
+            throw ClassCastException("Cannot cast ${EthereumApi::class.java} to $apiType")
        }
-        return this as T
+        return this as Upstream<A>
    }

 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt
@@ -120,14 +120,18 @@ open class EthereumUpstream(
    }

    @Suppress("unchecked")
-    override fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, upstreamType: Class<TA>): T {
+    override fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, apiType: Class<TA>): T {
        if (!selfType.isAssignableFrom(this.javaClass)) {
            throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
        }
-        if (!upstreamType.isAssignableFrom(EthereumApi::class.java)) {
-            throw ClassCastException("Cannot cast ${EthereumApi::class.java} to $upstreamType")
+        return castApi(apiType) as T
+    }
+
+    override fun <A : UpstreamApi> castApi(apiType: Class<A>): Upstream<A> {
+        if (!apiType.isAssignableFrom(EthereumApi::class.java)) {
+            throw ClassCastException("Cannot cast ${EthereumApi::class.java} to $apiType")
        }
-        return this as T
+        return this as Upstream<A>
    }

 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
@@ -216,13 +216,17 @@ open class EthereumGrpcUpstream(
    }

    @SuppressWarnings("unchecked")
-    override fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, upstreamType: Class<TA>): T {
+    override fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, apiType: Class<TA>): T {
        if (!selfType.isAssignableFrom(this.javaClass)) {
            throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
        }
-        if (!upstreamType.isAssignableFrom(EthereumApi::class.java)) {
-            throw ClassCastException("Cannot cast ${EthereumApi::class.java} to $upstreamType")
+        return castApi(apiType) as T
+    }
+
+    override fun <A : UpstreamApi> castApi(apiType: Class<A>): Upstream<A> {
+        if (!apiType.isAssignableFrom(EthereumApi::class.java)) {
+            throw ClassCastException("Cannot cast ${EthereumApi::class.java} to $apiType")
        }
-        return this as T
+        return this as Upstream<A>
    }
 }