@@ -5,6 +5,7 @@ import io.emeraldpay.api.proto.Common
|
|||||||
import io.emeraldpay.dshackle.BlockchainType
|
import io.emeraldpay.dshackle.BlockchainType
|
||||||
import io.emeraldpay.dshackle.SilentException
|
import io.emeraldpay.dshackle.SilentException
|
||||||
import io.emeraldpay.dshackle.config.TokensConfig
|
import io.emeraldpay.dshackle.config.TokensConfig
|
||||||
|
import io.emeraldpay.dshackle.upstream.Head
|
||||||
import io.emeraldpay.dshackle.upstream.MultistreamHolder
|
import io.emeraldpay.dshackle.upstream.MultistreamHolder
|
||||||
import io.emeraldpay.dshackle.upstream.Selector
|
import io.emeraldpay.dshackle.upstream.Selector
|
||||||
import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
|
import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
|
||||||
@@ -14,6 +15,7 @@ import io.emeraldpay.grpc.Chain
|
|||||||
import io.infinitape.etherjar.domain.Address
|
import io.infinitape.etherjar.domain.Address
|
||||||
import io.infinitape.etherjar.erc20.ERC20Token
|
import io.infinitape.etherjar.erc20.ERC20Token
|
||||||
import io.infinitape.etherjar.hex.Hex32
|
import io.infinitape.etherjar.hex.Hex32
|
||||||
|
import io.infinitape.etherjar.hex.HexQuantity
|
||||||
import org.slf4j.LoggerFactory
|
import org.slf4j.LoggerFactory
|
||||||
import org.springframework.beans.factory.annotation.Autowired
|
import org.springframework.beans.factory.annotation.Autowired
|
||||||
import org.springframework.stereotype.Service
|
import org.springframework.stereotype.Service
|
||||||
@@ -84,10 +86,11 @@ class TrackERC20Address(
|
|||||||
}
|
}
|
||||||
|
|
||||||
fun getBalance(addr: TrackedAddress): Mono<BigInteger> {
|
fun getBalance(addr: TrackedAddress): Mono<BigInteger> {
|
||||||
return getUpstream(addr.chain)
|
val upstream = getUpstream(addr.chain)
|
||||||
|
return upstream
|
||||||
.getDirectApi(Selector.empty)
|
.getDirectApi(Selector.empty)
|
||||||
.flatMap { api ->
|
.flatMap { api ->
|
||||||
api.read(prepareEthCall(addr.token, addr.address))
|
api.read(prepareEthCall(addr.token, addr.address, upstream.getHead()))
|
||||||
.flatMap(JsonRpcResponse::requireStringResult)
|
.flatMap(JsonRpcResponse::requireStringResult)
|
||||||
.map {
|
.map {
|
||||||
Hex32.from(it).asQuantity().value
|
Hex32.from(it).asQuantity().value
|
||||||
@@ -95,11 +98,12 @@ class TrackERC20Address(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fun prepareEthCall(token: ERC20Token, target: Address): JsonRpcRequest {
|
fun prepareEthCall(token: ERC20Token, target: Address, head: Head): JsonRpcRequest {
|
||||||
val call = token
|
val call = token
|
||||||
.readBalanceOf(target)
|
.readBalanceOf(target)
|
||||||
.toJson()
|
.toJson()
|
||||||
return JsonRpcRequest("eth_call", listOf(call, "latest"))
|
val height = head.getCurrentHeight()?.let { HexQuantity.from(it).toHex() } ?: "latest"
|
||||||
|
return JsonRpcRequest("eth_call", listOf(call, height))
|
||||||
}
|
}
|
||||||
|
|
||||||
fun getUpstream(chain: Chain): EthereumMultistream {
|
fun getUpstream(chain: Chain): EthereumMultistream {
|
||||||
|
|||||||
@@ -89,7 +89,8 @@ class EthereumDirectReader(
|
|||||||
}
|
}
|
||||||
balanceReader = object : Reader<Address, Wei> {
|
balanceReader = object : Reader<Address, Wei> {
|
||||||
override fun read(key: Address): Mono<Wei> {
|
override fun read(key: Address): Mono<Wei> {
|
||||||
val request = JsonRpcRequest("eth_getBalance", listOf(key.toHex(), "latest"))
|
val height = up.getHead().getCurrentHeight()?.let { HexQuantity.from(it).toHex() } ?: "latest"
|
||||||
|
val request = JsonRpcRequest("eth_getBalance", listOf(key.toHex(), height))
|
||||||
return readWithQuorum(request)
|
return readWithQuorum(request)
|
||||||
.timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Balance not read $key")))
|
.timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Balance not read $key")))
|
||||||
.map {
|
.map {
|
||||||
|
|||||||
@@ -130,6 +130,7 @@ open class EthereumReader(
|
|||||||
}
|
}
|
||||||
|
|
||||||
fun balance(): Reader<Address, Wei> {
|
fun balance(): Reader<Address, Wei> {
|
||||||
|
//TODO include height as part of cache?
|
||||||
return CompoundReader(
|
return CompoundReader(
|
||||||
balanceCache, directReader.balanceReader
|
balanceCache, directReader.balanceReader
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -101,7 +101,7 @@ class TrackEthereumAddressSpec extends Specification {
|
|||||||
TrackEthereumAddress trackAddress = new TrackEthereumAddress(upstreams)
|
TrackEthereumAddress trackAddress = new TrackEthereumAddress(upstreams)
|
||||||
|
|
||||||
apiMock.answerOnce("eth_getBalance", ["0xe2c8fa8120d813cd0b5e6add120295bf20cfa09f", "latest"], "0x499602D2")
|
apiMock.answerOnce("eth_getBalance", ["0xe2c8fa8120d813cd0b5e6add120295bf20cfa09f", "latest"], "0x499602D2")
|
||||||
apiMock.answerOnce("eth_getBalance", ["0xe2c8fa8120d813cd0b5e6add120295bf20cfa09f", "latest"], "0xff98")
|
apiMock.answerOnce("eth_getBalance", ["0xe2c8fa8120d813cd0b5e6add120295bf20cfa09f", "0x1"], "0xff98")
|
||||||
when:
|
when:
|
||||||
def flux = trackAddress.subscribe(req)
|
def flux = trackAddress.subscribe(req)
|
||||||
then:
|
then:
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ import io.emeraldpay.dshackle.quorum.QuorumReaderFactory
|
|||||||
import io.emeraldpay.dshackle.quorum.QuorumRpcReader
|
import io.emeraldpay.dshackle.quorum.QuorumRpcReader
|
||||||
import io.emeraldpay.dshackle.reader.Reader
|
import io.emeraldpay.dshackle.reader.Reader
|
||||||
import io.emeraldpay.dshackle.upstream.ApiSource
|
import io.emeraldpay.dshackle.upstream.ApiSource
|
||||||
|
import io.emeraldpay.dshackle.upstream.Head
|
||||||
import io.emeraldpay.dshackle.upstream.Multistream
|
import io.emeraldpay.dshackle.upstream.Multistream
|
||||||
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
|
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
|
||||||
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
|
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
|
||||||
@@ -142,10 +143,13 @@ class EthereumDirectReaderSpec extends Specification {
|
|||||||
.verify(Duration.ofSeconds(1))
|
.verify(Duration.ofSeconds(1))
|
||||||
}
|
}
|
||||||
|
|
||||||
def "Reads balance"() {
|
def "Reads balance - height is unknown"() {
|
||||||
setup:
|
setup:
|
||||||
def up = Mock(Multistream) {
|
def up = Mock(Multistream) {
|
||||||
1 * getApiSource(_) >> Stub(ApiSource)
|
1 * getApiSource(_) >> Stub(ApiSource)
|
||||||
|
1 * getHead() >> Mock(Head) {
|
||||||
|
1 * getCurrentHeight() >> null
|
||||||
|
}
|
||||||
}
|
}
|
||||||
def calls = Mock(Factory) {
|
def calls = Mock(Factory) {
|
||||||
1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM)
|
1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM)
|
||||||
@@ -171,4 +175,36 @@ class EthereumDirectReaderSpec extends Specification {
|
|||||||
.verify(Duration.ofSeconds(1))
|
.verify(Duration.ofSeconds(1))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
def "Reads balance - height is known"() {
|
||||||
|
setup:
|
||||||
|
def up = Mock(Multistream) {
|
||||||
|
1 * getApiSource(_) >> Stub(ApiSource)
|
||||||
|
1 * getHead() >> Mock(Head) {
|
||||||
|
1 * getCurrentHeight() >> 11_061_691
|
||||||
|
}
|
||||||
|
}
|
||||||
|
def calls = Mock(Factory) {
|
||||||
|
1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM)
|
||||||
|
}
|
||||||
|
EthereumDirectReader reader = new EthereumDirectReader(
|
||||||
|
up, Caches.default(), new CurrentBlockCache(), calls
|
||||||
|
)
|
||||||
|
reader.quorumReaderFactory = Mock(QuorumReaderFactory) {
|
||||||
|
1 * create(_, _) >> Mock(Reader) {
|
||||||
|
1 * read(new JsonRpcRequest("eth_getBalance", [address1, "0xa8c9bb"])) >> Mono.just(
|
||||||
|
new QuorumRpcReader.Result(
|
||||||
|
Global.objectMapper.writeValueAsBytes("0x100"), 1
|
||||||
|
)
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
when:
|
||||||
|
def act = reader.balanceReader.read(Address.from(address1))
|
||||||
|
then:
|
||||||
|
StepVerifier.create(act)
|
||||||
|
.expectNext(Wei.from("0x100"))
|
||||||
|
.expectComplete()
|
||||||
|
.verify(Duration.ofSeconds(1))
|
||||||
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -197,8 +197,10 @@ class EthereumReaderSpec extends Specification {
|
|||||||
def "Caches balance until block mined"() {
|
def "Caches balance until block mined"() {
|
||||||
setup:
|
setup:
|
||||||
def api = TestingCommons.api()
|
def api = TestingCommons.api()
|
||||||
|
// no height
|
||||||
api.answerOnce("eth_getBalance", ["0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c", "latest"], "0x10")
|
api.answerOnce("eth_getBalance", ["0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c", "latest"], "0x10")
|
||||||
api.answerOnce("eth_getBalance", ["0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c", "latest"], "0xff")
|
// height 101 + 1 => 102 => 0x66
|
||||||
|
api.answerOnce("eth_getBalance", ["0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c", "0x66"], "0xff")
|
||||||
EthereumUpstreamMock upstream = new EthereumUpstreamMock(Chain.ETHEREUM, api)
|
EthereumUpstreamMock upstream = new EthereumUpstreamMock(Chain.ETHEREUM, api)
|
||||||
def upstreams = TestingCommons.multistream(upstream)
|
def upstreams = TestingCommons.multistream(upstream)
|
||||||
def reader = new EthereumReader(upstreams, Caches.default(), calls)
|
def reader = new EthereumReader(upstreams, Caches.default(), calls)
|
||||||
|
|||||||
Reference in New Issue
Block a user