diff --git a/build.gradle b/build.gradle index 27f3a45b..b6708b56 100644 --- a/build.gradle +++ b/build.gradle @@ -44,7 +44,7 @@ configurations { } dependencies { - compile "io.emeraldpay:emerald-grpc:0.1-SNAPSHOT" + compile "io.emeraldpay:emerald-grpc:0.2" compile "io.grpc:grpc-protobuf:${grpcVersion}" compile "io.grpc:grpc-stub:${grpcVersion}" diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt index e670aed7..16abd45f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt @@ -6,6 +6,7 @@ import io.emeraldpay.api.proto.Common import io.emeraldpay.grpc.Chain import io.grpc.stub.StreamObserver import io.infinitape.etherjar.domain.TransactionId +import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service import java.time.Instant @@ -18,7 +19,9 @@ class BlockchainRpc( @Autowired private val trackAddress: TrackAddress ): BlockchainGrpc.BlockchainImplBase() { - override fun nativeCall(request: BlockchainOuterClass.CallBlockchainRequest, responseObserver: StreamObserver) { + private val log = LoggerFactory.getLogger(BlockchainRpc::class.java) + + override fun nativeCall(request: BlockchainOuterClass.NativeCallRequest, responseObserver: StreamObserver) { nativeCall.nativeCall(request, responseObserver) } @@ -26,18 +29,31 @@ class BlockchainRpc( streamHead.add(Chain.byId(request.type.number), responseObserver) } - override fun trackTx(request: BlockchainOuterClass.TrackTxRequest, responseObserver: StreamObserver) { + override fun streamTxStatus(request: BlockchainOuterClass.TxStatusRequest, responseObserver: StreamObserver) { val tx = TrackTx.TrackedTx( Chain.byId(request.chainValue), StreamSender(responseObserver), Instant.now(), - TransactionId.from(request.txid), - Math.min(Math.max(1, request.confirmations), 100) + TransactionId.from(request.txId), + Math.min(Math.max(1, request.confirmationLimit), 100) ) trackTx.add(tx) } - override fun trackAddress(request: BlockchainOuterClass.TrackAddressRequest, responseObserver: StreamObserver) { + override fun streamBalance(request: BlockchainOuterClass.BalanceRequest, responseObserver: StreamObserver) { trackAddress.add(request, responseObserver) } + + override fun getBalance(request: BlockchainOuterClass.BalanceRequest, responseObserver: StreamObserver) { + val addresses = trackAddress.initializeFor(request, responseObserver) + trackAddress.send(request, addresses) + .doOnError { t -> + log.error("Failed to process balance", t) + responseObserver.onError(Exception("Internal error")) + } + .subscribe { + responseObserver.onCompleted() + } + + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index a5bd308c..dddd5fbe 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -6,7 +6,6 @@ import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.grpc.Chain import io.grpc.stub.StreamObserver -import io.infinitape.etherjar.rpc.json.ResponseJson import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service @@ -23,9 +22,7 @@ class NativeCall( private val log = LoggerFactory.getLogger(NativeCall::class.java) - - - open fun nativeCall(request: BlockchainOuterClass.CallBlockchainRequest, responseObserver: StreamObserver) { + open fun nativeCall(request: BlockchainOuterClass.NativeCallRequest, responseObserver: StreamObserver) { val chain= Chain.byId(request.chain.number) if (chain == Chain.UNSPECIFIED) { throw Exception("Invalid chain id: ${request.chain.number}") @@ -49,7 +46,7 @@ class NativeCall( } } .map { - BlockchainOuterClass.CallBlockchainReplyItem.newBuilder() + BlockchainOuterClass.NativeCallReplyItem.newBuilder() .setSucceed(true) .setId(it.id) .setPayload(ByteString.copyFrom(it.payload)) @@ -62,7 +59,7 @@ class NativeCall( log.error("Lost context for a native call", it) 0 } - return@onErrorResume BlockchainOuterClass.CallBlockchainReplyItem.newBuilder() + return@onErrorResume BlockchainOuterClass.NativeCallReplyItem.newBuilder() .setSucceed(false) .setId(id) .build() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt index 60f3f6a3..9d5f2320 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt @@ -1,5 +1,6 @@ package io.emeraldpay.dshackle.rpc +import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.grpc.Chain @@ -26,7 +27,7 @@ class StreamHead( @PostConstruct fun init() { - listOf(Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC, Chain.MORDEN).forEach { chain -> + listOf(Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC, Chain.TESTNET_MORDEN, Chain.TESTNET_KOVAN).forEach { chain -> if (upstreams.ethereumUpstream(chain)?.head != null) { clients[chain] = ConcurrentLinkedQueue() subscribe(chain) @@ -77,7 +78,9 @@ class StreamHead( val data = BlockchainOuterClass.ChainHead.newBuilder() .setChainValue(chain.id) .setHeight(block.number) - .setHash(block.hash.toHex()) + .setTimestamp(block.timestamp.time) + .setWeight(ByteString.copyFrom(block.totalDifficulty.toByteArray())) + .setBlockId(block.hash.toHex().substring(2)) .build() var sent: Boolean = false try { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt index cdefc743..48fda2c0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt @@ -1,6 +1,5 @@ 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 @@ -17,11 +16,13 @@ import org.springframework.stereotype.Service import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.core.publisher.toFlux +import reactor.math.sum 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.* import java.util.concurrent.ConcurrentLinkedQueue import java.util.concurrent.Future import javax.annotation.PostConstruct @@ -33,7 +34,7 @@ class TrackAddress( private val clients = HashMap>() - private val allChains = listOf(Chain.MORDEN, Chain.ETHEREUM_CLASSIC, Chain.ETHEREUM) + private val allChains = listOf(Chain.TESTNET_MORDEN, Chain.ETHEREUM_CLASSIC, Chain.ETHEREUM, Chain.TESTNET_KOVAN) @PostConstruct fun init() { @@ -59,18 +60,18 @@ class TrackAddress( } } - fun add(request: BlockchainOuterClass.TrackAddressRequest, responseObserver: StreamObserver) { + fun initializeFor(request: BlockchainOuterClass.BalanceRequest, responseObserver: StreamObserver): List { val chain = Chain.byId(request.asset.chainValue) if (!allChains.contains(chain)) { responseObserver.onError(Exception("Unsupported chain ${request.asset.chainValue}")) - return + return Collections.emptyList() } if (request.asset.code?.toLowerCase() != "ether") { responseObserver.onError(Exception("Unsupported asset ${request.asset.code}")) - return + return Collections.emptyList() } val new = java.util.ArrayList() - val observer = StreamSender(responseObserver) + 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) { @@ -78,11 +79,27 @@ class TrackAddress( new.add(forAddress(address, chain, observer)) } } - verify(chain, new).subscribe() - clients[chain]?.addAll(new) + return new } - private fun forAddress(address: Common.SingleAddress, chain: Chain, observer: StreamSender): TrackedAddress { + fun send(request: BlockchainOuterClass.BalanceRequest, addresses: List): Mono { + val chain = Chain.byId(request.asset.chainValue) + return verify(chain, addresses) + .map { updated -> notify(updated); 1 } + .sum() + } + + fun add(request: BlockchainOuterClass.BalanceRequest, responseObserver: StreamObserver) { + val chain = Chain.byId(request.asset.chainValue) + val new = initializeFor(request, responseObserver) + send(request, new) + .doFinally { + clients[chain]?.addAll(new) + } + .subscribe() + } + + private fun forAddress(address: Common.SingleAddress, chain: Chain, observer: StreamSender): TrackedAddress { val addressParsed = Address.from(address.address) return TrackedAddress( chain, @@ -130,8 +147,8 @@ class TrackAddress( private fun notify(address: TrackedAddress): Boolean { val sent = address.stream.send( - BlockchainOuterClass.AddressStatus.newBuilder() - .setBalance(ByteString.copyFrom(address.balance!!.amount!!.toByteArray())) + BlockchainOuterClass.AddressBalance.newBuilder() + .setBalance(address.balance!!.amount!!.toString(10)) .setAsset(Common.Asset.newBuilder() .setChainValue(address.chain.id) .setCode("ETHER") @@ -149,7 +166,7 @@ class TrackAddress( class Update(val addr: TrackedAddress, val value: Future) class TrackedAddress(val chain: Chain, - val stream: StreamSender, + val stream: StreamSender, val address: Address, val since: Instant = Instant.now(), var lastPing: Instant = Instant.now(), diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt index 04dfb41f..aeb93dda 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt @@ -17,6 +17,7 @@ import reactor.core.publisher.Mono import reactor.core.publisher.toFlux import reactor.kotlin.core.publisher.switchIfEmpty import java.lang.Exception +import java.math.BigInteger import java.time.Duration import java.time.Instant import java.util.concurrent.ConcurrentLinkedQueue @@ -34,7 +35,7 @@ class TrackTx( @PostConstruct fun init() { - listOf(Chain.MORDEN, Chain.ETHEREUM_CLASSIC, Chain.ETHEREUM).forEach { chain -> + listOf(Chain.TESTNET_MORDEN, Chain.ETHEREUM_CLASSIC, Chain.ETHEREUM, Chain.TESTNET_KOVAN).forEach { chain -> clients[chain] = ConcurrentLinkedQueue() upstreams.ethereumUpstream(chain)?.head?.let { head -> head.getFlux().subscribe { verifyAll(chain) } @@ -61,6 +62,27 @@ class TrackTx( } } + private fun loadWeight(tx: TrackedTx): Mono { + val batch = Batch() + val execution = Mono + .fromCompletionStage(batch.add(Commands.eth().getBlock(tx.status.blockHash))) + .map { block -> + if (block != null && block.number != null && block.totalDifficulty != null) { + tx.withStatus( + blockTotalDifficulty = block.totalDifficulty, + blockTime = block.timestamp.toInstant() + ) + } else { + tx.withStatus( + mined = false + ) + } + } + val upstream = upstreams.ethereumUpstream(tx.chain)!! + upstream.api.execute(batch) + return execution + } + private fun verify(tx: TrackedTx): Boolean { val found = tx.status.found val mined = tx.status.mined @@ -80,10 +102,9 @@ class TrackTx( ) return@flatMap upstream.head.getHead().map { head -> tx.withStatus( - confirmation = head.number - tx.status.height!! + 1, - blockTime = head.timestamp.toInstant() + confirmation = head.number - tx.status.height!! + 1 ) - } + }.flatMap(this::loadWeight) } else { tx.withStatus( found = true, @@ -110,8 +131,7 @@ class TrackTx( private fun notify(tx: TrackedTx): Boolean { val client = tx.stream val data = BlockchainOuterClass.TxStatus.newBuilder() - .setChainValue(tx.chain.id) - .setTxid(tx.txid.toHex()) + .setTxId(tx.txid.toHex()) .setConfirmations(tx.status.confirmation.toInt()) .setMined(tx.status.mined) .setBroadcasted(tx.status.found) @@ -119,7 +139,9 @@ class TrackTx( if (tx.status.mined) { data.setBlock( Common.BlockInfo.newBuilder() - .setHash(ByteString.copyFrom(tx.status.blockHash!!.bytes)) + .setBlockId(tx.status.blockHash!!.toHex().substring(2)) + .setTimestamp(tx.status.blockTime!!.toEpochMilli()) + .setWeight(ByteString.copyFrom(tx.status.blockTotalDifficulty!!.toByteArray())) .setHeight(tx.status.height!!) .setTimestamp(tx.status.blockTime!!.toEpochMilli()) ) @@ -151,8 +173,9 @@ class TrackTx( mined: Boolean = this.status.mined, blockHash: BlockHash? = this.status.blockHash, blockTime: Instant? = this.status.blockTime, + blockTotalDifficulty: BigInteger? = this.status.blockTotalDifficulty, confirmation: Long = this.status.confirmation): TrackedTx { - this.status = this.status.copy(found, height, mined, blockHash, blockTime, confirmation) + this.status = this.status.copy(found, height, mined, blockHash, blockTime, blockTotalDifficulty, confirmation) return this } @@ -167,14 +190,16 @@ class TrackTx( var mined: Boolean = false, var blockHash: BlockHash? = null, var blockTime: Instant? = null, + var blockTotalDifficulty: BigInteger? = null, var confirmation: Long = 0) { fun copy(found: Boolean = this.found, height: Long? = this.height, mined: Boolean = this.mined, blockHash: BlockHash? = this.blockHash, blockTime: Instant? = this.blockTime, + blockTotalDifficulty: BigInteger? = this.blockTotalDifficulty, confirmation: Long = this.confirmation) - = TxStatus(found, height, mined, blockHash, blockTime, confirmation) + = TxStatus(found, height, mined, blockHash, blockTime, blockTotalDifficulty, confirmation) } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt index 3de76481..5914347f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt @@ -123,9 +123,12 @@ class EthereumApi( if (Chain.ETHEREUM_CLASSIC == chain) { return "1" } - if (Chain.MORDEN == chain) { + if (Chain.TESTNET_MORDEN == chain) { return "2" } + if (Chain.TESTNET_KOVAN == chain) { + return "42" + } throw RpcException(-32602, "Invalid chain") } if ("net_peerCount" == method) { @@ -135,7 +138,7 @@ class EthereumApi( return true } if ("web3_clientVersion" == method) { - return "EmeraldDshackle/v0.1" + return "EmeraldDshackle/v0.2" } if ("eth_protocolVersion" == method) { return "0x3f" diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt index 9f81b0a0..0499c86b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt @@ -30,7 +30,8 @@ class Upstreams( "ethereum-classic" to Chain.ETHEREUM_CLASSIC, "eth" to Chain.ETHEREUM, "etc" to Chain.ETHEREUM_CLASSIC, - "morden" to Chain.MORDEN + "morden" to Chain.TESTNET_MORDEN, + "kovan" to Chain.TESTNET_KOVAN ) @PostConstruct