solution: esplora upstream connection for bitcoin

2020-09-14 23:08:13 -04:00
parent 46d90ce932
commit 56318d800e
28 changed files with 962 additions and 136 deletions
--- a/src/main/kotlin/io/emeraldpay/dshackle/Global.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/Global.kt
@@ -19,9 +19,11 @@ import com.fasterxml.jackson.core.Version
 import com.fasterxml.jackson.databind.DeserializationFeature
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.databind.module.SimpleModule
+import io.emeraldpay.dshackle.upstream.bitcoin.data.EsploraUnspent
+import io.emeraldpay.dshackle.upstream.bitcoin.data.EsploraUnspentDeserializer
+import io.emeraldpay.dshackle.upstream.bitcoin.data.RpcUnspent
+import io.emeraldpay.dshackle.upstream.bitcoin.data.RpcUnspentDeserializer
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
-import io.infinitape.etherjar.rpc.json.TransactionReceiptJson
-import io.infinitape.etherjar.rpc.json.TransactionReceiptJsonDeserializer
 import java.text.SimpleDateFormat
 import java.util.*

@@ -36,6 +38,9 @@ class Global {
            val module = SimpleModule("EmeraldDshackle", Version(1, 0, 0, null, null, null))
            module.addSerializer(JsonRpcResponse::class.java, JsonRpcResponse.ResponseJsonSerializer())

+            module.addDeserializer(EsploraUnspent::class.java, EsploraUnspentDeserializer())
+            module.addDeserializer(RpcUnspent::class.java, RpcUnspentDeserializer())
+
            val objectMapper = ObjectMapper()
            objectMapper.registerModule(module)
            objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
--- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt
@@ -107,6 +107,7 @@ class UpstreamsConfig {
    }

    class BitcoinConnection : RpcConnection() {
+        var esplora: HttpEndpoint? = null
    }

    class HttpEndpoint(val url: URI) {
--- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt
@@ -132,6 +132,14 @@ class UpstreamsConfigReader(
                            http.tls = authConfigReader.readClientTls(node)
                        }
                    }
+                    getMapping(connConfigNode, "esplora")?.let { node ->
+                        getValueAsString(node, "url")?.let { url ->
+                            val http = UpstreamsConfig.HttpEndpoint(URI(url))
+                            http.basicAuth = authConfigReader.readClientBasicAuth(node)
+                            http.tls = authConfigReader.readClientTls(node)
+                            connection.esplora = http
+                        }
+                    }
                } else {
                    log.error("Upstream at #0 has invalid configuration")
                }
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt
@@ -21,15 +21,16 @@ import io.emeraldpay.dshackle.BlockchainType
 import io.emeraldpay.dshackle.SilentException
 import io.emeraldpay.dshackle.upstream.MultistreamHolder
 import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream
+import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent
 import io.emeraldpay.grpc.Chain
+import org.bitcoinj.params.MainNetParams
+import org.bitcoinj.params.TestNet3Params
 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 reactor.core.publisher.Mono
 import java.math.BigInteger
-import java.util.*
 import kotlin.collections.HashMap

@Service
@@ -64,10 +65,18 @@ class TrackBitcoinAddress(
    }

    fun requestBalances(chain: Chain, api: BitcoinMultistream, addresses: List<String>): Flux<AddressBalance> {
-        return api.getReader().listUnspent()
-                .flatMapMany { unspents ->
-                    val result = getTotal(chain, addresses, unspents)
-                    Flux.fromIterable(result)
+        return Flux.fromIterable(addresses)
+                .map { Address(chain, it) }
+                .flatMap { address ->
+                    api.getReader()
+                            .listUnspent(address.bitcoinAddress)
+                            .map { unspents ->
+                                getTotal(address, unspents)
+                            }
+                            .onErrorResume { t ->
+                                log.error("Failed to get unspent", t)
+                                Mono.empty()
+                            }
                }
    }

@@ -83,37 +92,13 @@ class TrackBitcoinAddress(
                .map(this@TrackBitcoinAddress::buildResponse)
    }

-    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()
+    fun getTotal(address: Address, unspents: List<SimpleUnspent>): AddressBalance {
+        val total = if (unspents.isEmpty()) {
+            0L
+        } else {
+            unspents.map { it.value }.reduce(Long::plus)
+        }
+        return AddressBalance(address, BigInteger.valueOf(total))
    }


@@ -158,5 +143,14 @@ class TrackBitcoinAddress(
        fun plus(other: AddressBalance) = AddressBalance(address, balance + other.balance)
    }

-    data class Address(val chain: Chain, val address: String)
+    class Address(val chain: Chain, val address: String) {
+        val network = if (chain == Chain.BITCOIN) {
+            MainNetParams()
+        } else {
+            TestNet3Params()
+        }
+        val bitcoinAddress = org.bitcoinj.core.Address.fromString(
+                network, address
+        )
+    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
@@ -23,6 +23,7 @@ import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder
 import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream
+import io.emeraldpay.dshackle.upstream.bitcoin.EsploraClient
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream
@@ -142,12 +143,21 @@ open class ConfiguredUpstreams(
            return
        }

+        val esplora = conn.esplora?.let { endpoint ->
+            val tls = endpoint.tls?.let { tls ->
+                tls.ca?.let { ca ->
+                    fileResolver.resolve(ca).readBytes()
+                }
+            }
+            EsploraClient(endpoint.url, endpoint.basicAuth, tls)
+        }
+
        val methods = buildMethods(config, chain)
        val upstream = BitcoinRpcUpstream(config.id
                ?: "bitcoin-${seq.getAndIncrement()}", chain, directApi,
                options, config.role,
                QuorumForLabels.QuorumItem(1, config.labels),
-                methods)
+                methods, esplora)

        upstream.start()
        currentUpstreams.update(UpstreamChange(chain, upstream, UpstreamChange.ChangeType.ADDED))
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Capability.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Capability.kt
@@ -0,0 +1,6 @@
+package io.emeraldpay.dshackle.upstream
+
+enum class Capability {
+    RPC,
+    BALANCE
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
@@ -78,14 +78,14 @@ abstract class Multistream(
     */
    fun addUpstream(upstream: Upstream) {
        upstreams.add(upstream)
-        setHead(updateHead())
        onUpstreamsUpdated()
+        setHead(updateHead())
    }

    fun removeUpstream(id: String) {
        if (upstreams.removeIf { it.getId() == id }) {
-            setHead(updateHead())
            onUpstreamsUpdated()
+            setHead(updateHead())
        }
    }

@@ -103,7 +103,7 @@ abstract class Multistream(
    /**
     * Finds an API that executed directly on a remote.
     */
-    fun getDirectApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
+    open fun getDirectApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
        val apis = getApiSource(matcher)
        apis.request(1)
        return Mono.from(apis)
@@ -120,7 +120,7 @@ abstract class Multistream(
        throw NotImplementedError("Immediate direct API is not implemented for Aggregated Upstream")
    }

-    fun onUpstreamsUpdated() {
+    open fun onUpstreamsUpdated() {
        reconfigLock.withLock {
            getAll().map { it.getMethods() }.let {
                //TODO made list of uniq instances, and then if only one, just use it directly
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
@@ -15,7 +15,6 @@
 */
 package io.emeraldpay.dshackle.upstream.bitcoin

-import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.reader.EmptyReader
@@ -38,8 +37,9 @@ open class BitcoinMultistream(
        private val log = LoggerFactory.getLogger(BitcoinMultistream::class.java)
    }

-    private var head: Head? = null
-    private var reader = BitcoinReader(this, EmptyHead())
+    private var head: Head = EmptyHead()
+    private var esplora = upstreams.find { it.esploraClient != null }?.esploraClient
+    private var reader = BitcoinReader(this, head, esplora)

    override fun init() {
        if (upstreams.size > 0) {
@@ -49,7 +49,7 @@ open class BitcoinMultistream(
    }

    override fun updateHead(): Head {
-        head?.let {
+        head.let {
            if (it is Lifecycle) {
                it.stop()
            }
@@ -81,13 +81,19 @@ open class BitcoinMultistream(
        return reader
    }

+    override fun onUpstreamsUpdated() {
+        super.onUpstreamsUpdated()
+        esplora = upstreams.find { it.esploraClient != null }?.esploraClient
+        reader = BitcoinReader(this, this.head, esplora)
+    }
+
    override fun setHead(head: Head) {
        this.head = head
-        reader = BitcoinReader(this, head)
+        reader = BitcoinReader(this, head, esplora)
    }

    override fun getHead(): Head {
-        return head!!
+        return head
    }

    override fun getLabels(): Collection<UpstreamsConfig.Labels> {
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt
@@ -19,8 +19,10 @@ import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import org.bitcoinj.core.Address
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.publisher.Mono
@@ -28,7 +30,8 @@ import reactor.kotlin.core.publisher.cast

 open class BitcoinReader(
        private val upstreams: BitcoinMultistream,
-        head: Head
+        head: Head,
+        esploraClient: EsploraClient?
 ) : Lifecycle {

    companion object {
@@ -38,6 +41,12 @@ open class BitcoinReader(
    private val objectMapper: ObjectMapper = Global.objectMapper
    private val mempool = CachingMempoolData(upstreams, head)

+    private val unspentReader: UnspentReader = if (esploraClient != null) {
+        EsploraUnspentReader(esploraClient, head)
+    } else {
+        RpcUnspentReader(upstreams)
+    }
+
    open fun getMempool(): CachingMempoolData {
        return mempool
    }
@@ -50,8 +59,8 @@ open class BitcoinReader(
        return castedRead(JsonRpcRequest("getrawtransaction", listOf(txid, true)), Map::class.java).cast()
    }

-    open fun listUnspent(): Mono<List<String>> {
-        return castedRead(JsonRpcRequest("listunspent", emptyList()), List::class.java).cast()
+    open fun listUnspent(address: Address): Mono<List<SimpleUnspent>> {
+        return unspentReader.read(address)
    }

    override fun isRunning(): Boolean {
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt
@@ -34,8 +34,9 @@ open class BitcoinRpcUpstream(
        options: UpstreamsConfig.Options,
        role: UpstreamsConfig.UpstreamRole,
        node: QuorumForLabels.QuorumItem,
-        callMethods: CallMethods
-) : BitcoinUpstream(id, chain, options, role, callMethods, node), Lifecycle {
+        callMethods: CallMethods,
+        esploraClient: EsploraClient? = null
+) : BitcoinUpstream(id, chain, options, role, callMethods, node, esploraClient), Lifecycle {

    companion object {
        private val log = LoggerFactory.getLogger(BitcoinRpcUpstream::class.java)
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt
@@ -29,7 +29,8 @@ abstract class BitcoinUpstream(
        options: UpstreamsConfig.Options,
        role: UpstreamsConfig.UpstreamRole,
        callMethods: CallMethods,
-        node: QuorumForLabels.QuorumItem
+        node: QuorumForLabels.QuorumItem,
+        val esploraClient: EsploraClient? = null
 ) : DefaultUpstream(id, options, role, callMethods, node) {

    constructor(id: String,
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/EsploraClient.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/EsploraClient.kt
@@ -0,0 +1,104 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.config.AuthConfig
+import io.emeraldpay.dshackle.upstream.bitcoin.data.EsploraUnspent
+import io.netty.handler.codec.http.HttpHeaderNames
+import io.netty.handler.codec.http.HttpHeaders
+import io.netty.handler.ssl.SslContextBuilder
+import org.bitcoinj.core.Address
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
+import reactor.netty.http.client.HttpClient
+import java.io.ByteArrayInputStream
+import java.net.URI
+import java.security.KeyStore
+import java.security.cert.CertificateFactory
+import java.security.cert.X509Certificate
+import java.util.*
+import java.util.function.Consumer
+
+class EsploraClient(
+        private val url: URI,
+        basicAuth: AuthConfig.ClientBasicAuth? = null,
+        tlsCAAuth: ByteArray? = null
+) {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(EsploraClient::class.java)
+    }
+
+    private val httpClient: HttpClient
+
+    init {
+        var build = HttpClient.create()
+
+        build = build.headers { h ->
+            h.add(HttpHeaderNames.CONTENT_TYPE, "application/json")
+        }
+
+        basicAuth?.let { auth ->
+            val authString: String = auth.username + ":" + auth.password
+            val authBase64 = Base64.getEncoder().encodeToString(authString.toByteArray())
+            val encodedAuth = "Basic $authBase64"
+            val headers = Consumer { h: HttpHeaders -> h.add(HttpHeaderNames.AUTHORIZATION, encodedAuth) }
+            build = build.headers(headers)
+        }
+
+        tlsCAAuth?.let { auth ->
+            val cf = CertificateFactory.getInstance("X.509")
+            val cert = cf.generateCertificate(ByteArrayInputStream(auth)) as X509Certificate
+            val ks = KeyStore.getInstance(KeyStore.getDefaultType())
+            ks.load(null, "".toCharArray())
+            ks.setCertificateEntry("server", cert)
+            val sslContext = SslContextBuilder.forClient().trustManager(cert).build()
+
+            build.secure { spec ->
+                spec.sslContext(sslContext)
+            }
+        }
+
+        this.httpClient = build
+    }
+
+    fun getUtxo(address: Address): Mono<List<EsploraUnspent>> {
+        val response = httpClient
+                .get()
+                .uri("$url/address/$address/utxo")
+
+        val json = response
+                .response { header, bytes ->
+                    if (header.status().code() != 200) {
+                        Mono.error(EsploraException("HTTP Code: ${header.status().code()}"))
+                    } else {
+                        bytes.aggregate().asByteArray()
+                    }
+                }
+                .single()
+                .map { bytes -> String(bytes) }
+
+        return json.map {
+            val parsed = Global.objectMapper.readerFor(EsploraUnspent::class.java)
+                    .readValues<EsploraUnspent>(it);
+            parsed.readAll()
+        }
+    }
+
+    class EsploraException(msg: String) : Exception(msg)
+
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/EsploraUnspentReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/EsploraUnspentReader.kt
@@ -0,0 +1,53 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.bitcoin.data.EsploraUnspent
+import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent
+import org.bitcoinj.core.Address
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
+import java.util.function.Function
+
+class EsploraUnspentReader(
+        private val esploraClient: EsploraClient,
+        private val head: Head
+) : UnspentReader {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(EsploraUnspentReader::class.java)
+    }
+
+    private val convert: (T: EsploraUnspent) -> SimpleUnspent = { base ->
+        SimpleUnspent(
+                base.txid,
+                base.vout,
+                base.value,
+                head.getCurrentHeight()?.let { base.height - it } ?: 0
+        )
+    }
+
+    private val convertAll: Function<List<EsploraUnspent>, List<SimpleUnspent>> = Function { base ->
+        base.map(convert)
+    }
+
+    override fun read(key: Address): Mono<List<SimpleUnspent>> {
+        return esploraClient.getUtxo(key)
+                .map(convertAll)
+    }
+
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RpcUnspentReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RpcUnspentReader.kt
@@ -0,0 +1,61 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.bitcoin.data.RpcUnspent
+import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import org.bitcoinj.core.Address
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
+
+class RpcUnspentReader(
+        private val upstreams: BitcoinMultistream
+) : UnspentReader {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(RpcUnspentReader::class.java)
+    }
+
+    private val convert: (T: RpcUnspent) -> SimpleUnspent = { base ->
+        SimpleUnspent(
+                base.txid,
+                base.vout,
+                base.amount,
+                base.confirmations
+        )
+    }
+
+    override fun read(key: Address): Mono<List<SimpleUnspent>> {
+        val address = key.toString()
+        return upstreams.getDirectApi(Selector.empty).flatMap { api ->
+            api.read(JsonRpcRequest("listunspent", emptyList()))
+                    .flatMap(JsonRpcResponse::requireResult)
+                    .map {
+                        Global.objectMapper.readerFor(RpcUnspent::class.java).readValues<RpcUnspent>(it).readAll()
+                    }
+                    .map {
+                        it.filter {
+                            it.address == address
+                        }.map(convert)
+                    }
+        }
+    }
+
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/UnspentReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/UnspentReader.kt
@@ -0,0 +1,23 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent
+import org.bitcoinj.core.Address
+
+interface UnspentReader : Reader<Address, List<SimpleUnspent>> {
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/EsploraUnspent.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/EsploraUnspent.kt
@@ -0,0 +1,26 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.bitcoin.data
+
+import java.time.Instant
+
+data class EsploraUnspent(
+        val txid: String,
+        val vout: Int,
+        val value: Long,
+        val timestamp: Instant,
+        val height: Long
+)
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/EsploraUnspentDeserializer.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/EsploraUnspentDeserializer.kt
@@ -0,0 +1,36 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.bitcoin.data
+
+import com.fasterxml.jackson.core.JsonParser
+import com.fasterxml.jackson.databind.DeserializationContext
+import com.fasterxml.jackson.databind.JsonDeserializer
+import com.fasterxml.jackson.databind.JsonNode
+import java.time.Instant
+
+class EsploraUnspentDeserializer : JsonDeserializer<EsploraUnspent>() {
+    override fun deserialize(jp: JsonParser, ctxt: DeserializationContext): EsploraUnspent {
+        val node: JsonNode = jp.readValueAsTree()
+        val status = node.get("status")
+        return EsploraUnspent(
+                node.get("txid").asText(),
+                node.get("vout").asInt(),
+                node.get("value").asLong(),
+                Instant.ofEpochSecond(status.get("block_time").asLong()),
+                status.get("block_height").asLong()
+        )
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/RpcUnspent.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/RpcUnspent.kt
@@ -0,0 +1,24 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.bitcoin.data
+
+data class RpcUnspent(
+        val txid: String,
+        val vout: Int,
+        val address: String,
+        val amount: Long,
+        val confirmations: Long
+)
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/RpcUnspentDeserializer.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/RpcUnspentDeserializer.kt
@@ -0,0 +1,37 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.bitcoin.data
+
+import com.fasterxml.jackson.core.JsonParser
+import com.fasterxml.jackson.databind.DeserializationContext
+import com.fasterxml.jackson.databind.JsonDeserializer
+import com.fasterxml.jackson.databind.JsonNode
+import java.math.BigDecimal
+
+class RpcUnspentDeserializer : JsonDeserializer<RpcUnspent>() {
+
+    override fun deserialize(jp: JsonParser, ctxt: DeserializationContext): RpcUnspent {
+        val node: JsonNode = jp.readValueAsTree()
+        return RpcUnspent(
+                node.get("txid").asText(),
+                node.get("vout").asInt(),
+                node.get("address").asText(),
+                BigDecimal(node.get("amount").asText()).multiply(BigDecimal.TEN.pow(8)).longValueExact(),
+                node.get("confirmations").asLong()
+        )
+    }
+
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/SimpleUnspent.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/SimpleUnspent.kt
@@ -0,0 +1,23 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.bitcoin.data
+
+data class SimpleUnspent(
+        val txid: String,
+        val vout: Int,
+        val value: Long,
+        val confirmations: Long
+)