diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt index 5f8c0443..c9f747fe 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt @@ -21,5 +21,9 @@ import org.reactivestreams.Publisher interface ApiSource : Publisher { fun resolve() + + /** + * Must be called before actual use, it spins off control flow of the API Source + */ fun request(tries: Int) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt index 4a3519e6..51603269 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt @@ -35,6 +35,7 @@ import kotlin.math.min import kotlin.math.pow import kotlin.math.roundToLong import kotlin.random.Random +import org.slf4j.LoggerFactory class FilteredApis( val chain: Chain, @@ -49,6 +50,8 @@ class FilteredApis( ) : ApiSource { companion object { + private val log = LoggerFactory.getLogger(FilteredApis::class.java) + private const val DEFAULT_DELAY_STEP = 100 private const val MAX_WAIT_MILLIS = 5000L @@ -86,6 +89,7 @@ class FilteredApis( private val secondaryUpstreams: List private val standardWithFallback: List + private var started = false private val control = Sinks.many().unicast().onBackpressureBuffer() init { @@ -176,6 +180,12 @@ class FilteredApis( result.filter { up -> up.isAvailable() && matcher.matches(up) } .zipWith(control.asFlux()) .map { it.t1 } + .doOnSubscribe { + if (!started) { + // in addition to subscription the FilteredAPI should use request() method to prepare the control flow + log.warn("API Source subscribed before preparing a request") + } + } .subscribe(subscriber) } @@ -184,6 +194,7 @@ class FilteredApis( } override fun request(tries: Int) { + started = true // TODO check the buffer size before submitting repeat(tries) { control.tryEmitNext(true) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/ERC20Balance.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/ERC20Balance.kt index 496bfc93..c5250488 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/ERC20Balance.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/ERC20Balance.kt @@ -15,6 +15,7 @@ */ package io.emeraldpay.dshackle.upstream.ethereum +import io.emeraldpay.dshackle.upstream.ApiSource import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest @@ -41,10 +42,18 @@ open class ERC20Balance { return upstreams // use only up-to-date upstreams .getApiSource(Selector.HeightMatcher(upstreams.getHead().getCurrentHeight() ?: 0)) - .let { Flux.from(it) } + .let { getBalance(it, token, address) } + } + + open fun getBalance(apis: ApiSource, token: ERC20Token, address: Address): Mono { + apis.request(1) + return Flux.from(apis) .flatMap { getBalance(it.cast(EthereumUpstream::class.java), token, address) } + .doOnNext { + apis.resolve() + } .next() } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/ERC20BalanceSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/ERC20BalanceSpec.groovy index 27ecdf7d..d1f6cfa1 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/ERC20BalanceSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/ERC20BalanceSpec.groovy @@ -17,6 +17,9 @@ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.dshackle.test.EthereumUpstreamMock import io.emeraldpay.dshackle.test.ReaderMock +import io.emeraldpay.dshackle.upstream.ApiSource +import io.emeraldpay.dshackle.upstream.FilteredApis +import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.etherjar.domain.Address @@ -55,4 +58,34 @@ class ERC20BalanceSpec extends Specification { then: act.toLong() == 0x1f28d72868 } + + def "Gets balance from api source"() { + setup: + ReaderMock api = new ReaderMock() + .with( + new JsonRpcRequest("eth_call", [ + new TransactionCallJson().tap { json -> + json.setTo(Address.from("0x54EedeAC495271d0F6B175474E89094C44Da98b9")) + json.setData(HexData.from("0x70a0823100000000000000000000000016c15c65ad00b6dfbcc2cb8a7b6c2d0103a3883b")) + }, + "latest" + ]), + JsonRpcResponse.ok('"0x0000000000000000000000000000000000000000000000000000001f28d72868"') + ) + + EthereumUpstream upstream = new EthereumUpstreamMock(Chain.ETHEREUM, api) + ERC20Token token = new ERC20Token(Address.from("0x54EedeAC495271d0F6B175474E89094C44Da98b9")) + ERC20Balance query = new ERC20Balance() + + ApiSource apiSource = new FilteredApis( + Chain.ETHEREUM, [upstream], Selector.empty + ) + + when: + def act = query.getBalance(apiSource, token, Address.from("0x16c15c65ad00b6dfbcc2cb8a7b6c2d0103a3883b")) + .block(Duration.ofSeconds(1)) + + then: + act.toLong() == 0x1f28d72868 + } }