StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

solution: upgrade to api 0.2

Browse Source
This commit is contained in:
Igor Artamonov
2019-07-03 23:05:24 -04:00
parent dab115c9b2
commit 56da82118e
8 changed files with 100 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -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}"

26
src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt
View File

@@ -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<BlockchainOuterClass.CallBlockchainReplyItem>) {
private val log = LoggerFactory.getLogger(BlockchainRpc::class.java)
override fun nativeCall(request: BlockchainOuterClass.NativeCallRequest, responseObserver: StreamObserver<BlockchainOuterClass.NativeCallReplyItem>) {
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<BlockchainOuterClass.TxStatus>) {
override fun streamTxStatus(request: BlockchainOuterClass.TxStatusRequest, responseObserver: StreamObserver<BlockchainOuterClass.TxStatus>) {
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<BlockchainOuterClass.AddressStatus>) {
override fun streamBalance(request: BlockchainOuterClass.BalanceRequest, responseObserver: StreamObserver<BlockchainOuterClass.AddressBalance>) {
trackAddress.add(request, responseObserver)
}
override fun getBalance(request: BlockchainOuterClass.BalanceRequest, responseObserver: StreamObserver<BlockchainOuterClass.AddressBalance>) {
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()
}
}
}

9
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
View File

@@ -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<BlockchainOuterClass.CallBlockchainReplyItem>) {
open fun nativeCall(request: BlockchainOuterClass.NativeCallRequest, responseObserver: StreamObserver<BlockchainOuterClass.NativeCallReplyItem>) {
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()

7
src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt
View File

@@ -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 {

41
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt
View File

@@ -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<Chain, ConcurrentLinkedQueue<TrackedAddress>>()
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<BlockchainOuterClass.AddressStatus>) {
fun initializeFor(request: BlockchainOuterClass.BalanceRequest, responseObserver: StreamObserver<BlockchainOuterClass.AddressBalance>): List<TrackedAddress> {
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<TrackedAddress>()
val observer = StreamSender<BlockchainOuterClass.AddressStatus>(responseObserver)
val observer = StreamSender<BlockchainOuterClass.AddressBalance>(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<BlockchainOuterClass.AddressStatus>): TrackedAddress {
fun send(request: BlockchainOuterClass.BalanceRequest, addresses: List<TrackedAddress>): Mono<Long> {
val chain = Chain.byId(request.asset.chainValue)
return verify(chain, addresses)
.map { updated -> notify(updated); 1 }
.sum()
}
fun add(request: BlockchainOuterClass.BalanceRequest, responseObserver: StreamObserver<BlockchainOuterClass.AddressBalance>) {
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<BlockchainOuterClass.AddressBalance>): 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<Wei>)
class TrackedAddress(val chain: Chain,
val stream: StreamSender<BlockchainOuterClass.AddressStatus>,
val stream: StreamSender<BlockchainOuterClass.AddressBalance>,
val address: Address,
val since: Instant = Instant.now(),
var lastPing: Instant = Instant.now(),

43
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt
View File

@@ -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<TrackedTx> {
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)
}
}

7
src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt
View File

@@ -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"

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt
View File

@@ -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