From be40fce6c5c142d870f164d73deb5ce746ff4edd Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Mon, 10 Jun 2019 23:03:29 -0400 Subject: [PATCH] solution: naive trackAddress implementation --- .../kotlin/io/emeraldpay/dshackle/Config.kt | 5 - .../emeraldpay/dshackle/rpc/BlockchainRpc.kt | 7 +- .../emeraldpay/dshackle/rpc/TrackAddress.kt | 158 ++++++++++++++++++ 3 files changed, 164 insertions(+), 6 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt diff --git a/src/main/kotlin/io/emeraldpay/dshackle/Config.kt b/src/main/kotlin/io/emeraldpay/dshackle/Config.kt index 41d73b83..54fd15c2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/Config.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/Config.kt @@ -29,9 +29,4 @@ open class Config { return objectMapper } - // Temporally hack to let Spring Boot know it has something active (i.e shouldn't shutdown, as non-web server) - @Scheduled(fixedRate = 60000) - fun readCurrentTime() { - } - } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt index 1d2f7428..e670aed7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt @@ -14,7 +14,8 @@ import java.time.Instant class BlockchainRpc( @Autowired private val nativeCall: NativeCall, @Autowired private val streamHead: StreamHead, - @Autowired private val trackTx: TrackTx + @Autowired private val trackTx: TrackTx, + @Autowired private val trackAddress: TrackAddress ): BlockchainGrpc.BlockchainImplBase() { override fun nativeCall(request: BlockchainOuterClass.CallBlockchainRequest, responseObserver: StreamObserver) { @@ -35,4 +36,8 @@ class BlockchainRpc( ) trackTx.add(tx) } + + override fun trackAddress(request: BlockchainOuterClass.TrackAddressRequest, responseObserver: StreamObserver) { + trackAddress.add(request, responseObserver) + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt new file mode 100644 index 00000000..cdefc743 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt @@ -0,0 +1,158 @@ +package io.emeraldpay.dshackle.rpc + +import com.google.protobuf.ByteString +import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.api.proto.Common +import io.emeraldpay.dshackle.upstream.Upstreams +import io.emeraldpay.grpc.Chain +import io.grpc.stub.StreamObserver +import io.infinitape.etherjar.domain.Address +import io.infinitape.etherjar.domain.Wei +import io.infinitape.etherjar.rpc.Batch +import io.infinitape.etherjar.rpc.Commands +import io.infinitape.etherjar.rpc.json.BlockTag +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.scheduling.annotation.Scheduled +import org.springframework.stereotype.Service +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.core.publisher.toFlux +import reactor.util.function.Tuple2 +import reactor.util.function.Tuples +import java.lang.Exception +import java.time.Duration +import java.time.Instant +import java.util.concurrent.ConcurrentLinkedQueue +import java.util.concurrent.Future +import javax.annotation.PostConstruct + +@Service +class TrackAddress( + @Autowired private val upstreams: Upstreams +) { + + private val clients = HashMap>() + + private val allChains = listOf(Chain.MORDEN, Chain.ETHEREUM_CLASSIC, Chain.ETHEREUM) + + @PostConstruct + fun init() { + allChains.forEach { chain -> + clients[chain] = ConcurrentLinkedQueue() + upstreams.ethereumUpstream(chain)?.head?.let { head -> + head.getFlux().subscribe { verifyAll(chain) } + } + } + } + + @Scheduled(fixedDelay = 120_000) + fun pingOld() { + val period = Duration.ofMinutes(15) + allChains.forEach { chain -> + clients[chain]?.let { clients -> + clients.toFlux().filter { + it.lastPing < Instant.now().minus(period) + }.subscribe { + notify(it) + } + } + } + } + + fun add(request: BlockchainOuterClass.TrackAddressRequest, responseObserver: StreamObserver) { + val chain = Chain.byId(request.asset.chainValue) + if (!allChains.contains(chain)) { + responseObserver.onError(Exception("Unsupported chain ${request.asset.chainValue}")) + return + } + if (request.asset.code?.toLowerCase() != "ether") { + responseObserver.onError(Exception("Unsupported asset ${request.asset.code}")) + return + } + val new = java.util.ArrayList() + val observer = StreamSender(responseObserver) + if (request.address.addrTypeCase == Common.AnyAddress.AddrTypeCase.ADDRESS_SINGLE) { + new.add(forAddress(request.address.addressSingle, chain, observer)) + } else if (request.address.addrTypeCase == Common.AnyAddress.AddrTypeCase.ADDRESS_MULTI) { + request.address.addressMulti.addressesList.forEach { address -> + new.add(forAddress(address, chain, observer)) + } + } + verify(chain, new).subscribe() + clients[chain]?.addAll(new) + } + + private fun forAddress(address: Common.SingleAddress, chain: Chain, observer: StreamSender): TrackedAddress { + val addressParsed = Address.from(address.address) + return TrackedAddress( + chain, + observer, + addressParsed + ) + } + + private fun verifyAll(chain: Chain) { + clients[chain]?.let { all -> + all.toFlux() + .buffer(20) + .map { group -> + verify(chain, group).subscribe { updated -> notify(updated) } + } + .subscribe() + } + } + + private fun verify(chain: Chain, group: List): Flux { + val up = upstreams.ethereumUpstream(chain)!! + return group.toFlux() + .reduce>>(Tuples.of(Batch(), ArrayList())) { batch, a -> + val f = batch.t1.add(Commands.eth().getBalance(a.address, BlockTag.LATEST)); + batch.t2.add(Update(a, f)) + batch + } + .flatMap { + Mono.fromCompletionStage(up.api.execute(it.t1)) + .thenReturn(it.t2) + } + .flatMapMany { + it.toFlux() + } + .filter { + it.addr.balance == null || it.addr.balance != it.value.get() + } + .doOnNext { + it.addr.balance = it.value.get() + } + .map { + it.addr + } + } + + private fun notify(address: TrackedAddress): Boolean { + val sent = address.stream.send( + BlockchainOuterClass.AddressStatus.newBuilder() + .setBalance(ByteString.copyFrom(address.balance!!.amount!!.toByteArray())) + .setAsset(Common.Asset.newBuilder() + .setChainValue(address.chain.id) + .setCode("ETHER") + ) + .setAddress(Common.SingleAddress.newBuilder().setAddress(address.address.toHex())) + .build() + ) + if (!sent) { + clients[address.chain]?.remove(address) + } + address.lastPing = Instant.now() + return sent + } + + class Update(val addr: TrackedAddress, val value: Future) + + class TrackedAddress(val chain: Chain, + val stream: StreamSender, + val address: Address, + val since: Instant = Instant.now(), + var lastPing: Instant = Instant.now(), + var balance: Wei? = null + ) +} \ No newline at end of file