problem: doesn't return ERC20 balance

src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt

@@ -21,5 +21,9 @@ import org.reactivestreams.Publisher
 interface ApiSource : Publisher<Upstream> {
 
     fun resolve()
+
+    /**
+     * Must be called before actual use, it spins off control flow of the API Source
+     */
     fun request(tries: Int)
 }

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<Upstream>
     private val standardWithFallback: List<Upstream>
 
+    private var started = false
     private val control = Sinks.many().unicast().onBackpressureBuffer<Boolean>()
 
     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)

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<BigInteger> {
+        apis.request(1)
+        return Flux.from(apis)
             .flatMap {
                 getBalance(it.cast(EthereumUpstream::class.java), token, address)
             }
+            .doOnNext {
+                apis.resolve()
+            }
             .next()
     }
 

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
+    }
 }