From 49bfa163ef3416c234eaf6d386f651b0c72ce8e1 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Mon, 4 Oct 2021 20:58:25 -0400 Subject: [PATCH] solution: base implementation for NativeSubscribe --- proto/blockchain.proto | 49 +++++++--- .../monitoring/accesslog/AccessHandlerGrpc.kt | 13 +++ .../dshackle/monitoring/accesslog/Events.kt | 20 ++++ .../monitoring/accesslog/EventsBuilder.kt | 30 ++++++ .../emeraldpay/dshackle/rpc/BlockchainRpc.kt | 22 +++++ .../dshackle/rpc/NativeSubscribe.kt | 98 +++++++++++++++++++ .../upstream/ethereum/EthereumMultistream.kt | 4 + .../upstream/ethereum/EthereumSubscribe.kt | 15 +++ .../upstream/ethereum/EthereumWsFactory.kt | 13 ++- .../dshackle/rpc/NativeSubscribeSpec.groovy | 92 +++++++++++++++++ 10 files changed, 338 insertions(+), 18 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy diff --git a/proto/blockchain.proto b/proto/blockchain.proto index 691906b8..7343d653 100644 --- a/proto/blockchain.proto +++ b/proto/blockchain.proto @@ -4,23 +4,17 @@ option java_package = "io.emeraldpay.api.proto"; import "common.proto"; service Blockchain { - rpc SubscribeHead (Chain) returns (stream ChainHead) { - } - rpc SubscribeBalance (BalanceRequest) returns (stream AddressBalance) { - } - rpc SubscribeTxStatus (TxStatusRequest) returns (stream TxStatus) { - } + rpc SubscribeHead (Chain) returns (stream ChainHead) {} + rpc SubscribeBalance (BalanceRequest) returns (stream AddressBalance) {} + rpc SubscribeTxStatus (TxStatusRequest) returns (stream TxStatus) {} - rpc GetBalance (BalanceRequest) returns (stream AddressBalance) { - } + rpc GetBalance (BalanceRequest) returns (stream AddressBalance) {} - rpc NativeCall (NativeCallRequest) returns (stream NativeCallReplyItem) { - } + rpc NativeCall (NativeCallRequest) returns (stream NativeCallReplyItem) {} + rpc NativeSubscribe (NativeSubscribeRequest) returns (stream NativeSubscribeReplyItem) {} - rpc Describe (DescribeRequest) returns (DescribeResponse) { - } - rpc SubscribeStatus (StatusRequest) returns (stream ChainStatus) { - } + rpc Describe (DescribeRequest) returns (DescribeResponse) {} + rpc SubscribeStatus (StatusRequest) returns (stream ChainStatus) {} } message NativeCallRequest { @@ -44,6 +38,16 @@ message NativeCallReplyItem { string errorMessage = 4; } +message NativeSubscribeRequest { + ChainRef chain = 1; + string method = 2; + bytes payload = 3; +} + +message NativeSubscribeReplyItem { + bytes payload = 1; +} + message ChainHead { ChainRef chain = 1; uint64 height = 2; @@ -70,12 +74,22 @@ message TxStatus { message BalanceRequest { Asset asset = 1; AnyAddress address = 2; + bool include_utxo = 3; } message AddressBalance { Asset asset = 1; SingleAddress address = 2; string balance = 3; + bool confirmed = 4; + repeated Utxo utxo = 5; +} + +message Utxo { + string tx_id = 1; + uint64 index = 2; + string balance = 3; + bool spent = 4; } message DescribeRequest { @@ -91,6 +105,7 @@ message DescribeChain { repeated NodeDetails nodes = 3; repeated string supportedMethods = 4; repeated string excludedMethods = 5; + repeated Capabilities capabilities = 6; } message StatusRequest { @@ -117,6 +132,12 @@ message NodeDetails { repeated Label labels = 2; } +enum Capabilities { + CAP_NONE = 0; + CAP_CALLS = 1; + CAP_BALANCE = 2; +} + message Label { string name = 1; string value = 2; diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandlerGrpc.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandlerGrpc.kt index 2e5d0377..ea5e0c16 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandlerGrpc.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandlerGrpc.kt @@ -40,6 +40,7 @@ class AccessHandlerGrpc( "SubscribeTxStatus" -> processSubscribeTxStatus(call, headers, next) "GetBalance" -> processSubscribeBalance(call, headers, next, false) "NativeCall" -> processNativeCall(call, headers, next) + "NativeSubscribe" -> processNativeSubscribe(call, headers, next) "Describe" -> processDescribe(call, headers, next) "SubscribeStatus" -> processStatus(call, headers, next) else -> { @@ -110,6 +111,18 @@ class AccessHandlerGrpc( ) } + @Suppress("UNCHECKED_CAST") + private fun processNativeSubscribe( + call: ServerCall, + headers: Metadata, + next: ServerCallHandler + ): ServerCall.Listener { + return process(call, headers, next, + EventsBuilder.NativeSubscribe() as EventsBuilder.RequestReply<*, ReqT, RespT> + ) + } + + @Suppress("UNCHECKED_CAST") private fun processDescribe( call: ServerCall, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt index 39045728..427d1d9e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt @@ -102,6 +102,16 @@ class Events { val nativeCall: NativeCallItemDetails ) : ChainBase(blockchain, "NativeCall", id, channel) + @JsonInclude(JsonInclude.Include.NON_NULL) + class NativeSubscribe( + blockchain: Chain, id: UUID, channel: Channel, + + // info about the initial request, that may include several native calls + val request: StreamRequestDetails, + val payloadSizeBytes: Long, + val nativeSubscribe: NativeSubscribeItemDetails + ) : ChainBase(blockchain, "NativeSubscribe", id, channel) + @JsonInclude(JsonInclude.Include.NON_NULL) class Describe( id: UUID, @@ -139,6 +149,16 @@ class Events { val ts: Instant = Instant.now() ) + data class NativeSubscribeItemDetails( + val method: String, + val payloadSizeBytes: Long + ) + + data class NativeSubscribeReplyDetails( + val replySizeBytes: Long, + val ts: Instant = Instant.now() + ) + data class BalanceRequest( val asset: String, val addressType: String diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt index 25ad3a8f..b0e6d882 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt @@ -293,6 +293,36 @@ class EventsBuilder { } } + class NativeSubscribe : + Base(), + RequestReply { + var item: Events.NativeSubscribeItemDetails? = null + val replies = HashMap() + + override fun getT(): NativeSubscribe { + return this + } + + override fun onRequest(msg: BlockchainOuterClass.NativeSubscribeRequest) { + withChain(msg.chain.number) + this.item = Events.NativeSubscribeItemDetails( + msg.method, + msg.payload.size().toLong() + ) + } + + override fun onReply(msg: BlockchainOuterClass.NativeSubscribeReplyItem): Events.NativeSubscribe { + return Events.NativeSubscribe( + request = requestDetails, + blockchain = chain, + nativeSubscribe = item!!, + payloadSizeBytes = msg.payload?.size()?.toLong() ?: 0L, + id = UUID.randomUUID(), + channel = Events.Channel.GRPC + ) + } + } + class Describe : Base(), RequestReply { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt index ea122fd2..0d9f0ff7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt @@ -37,6 +37,7 @@ import java.util.concurrent.TimeUnit @Service @DependsOn("monitoringSetup") class BlockchainRpc( @Autowired private val nativeCall: NativeCall, + @Autowired private val nativeSubscribe: NativeSubscribe, @Autowired private val streamHead: StreamHead, @Autowired private val trackTx: List, @Autowired private val trackAddress: List, @@ -73,6 +74,19 @@ class BlockchainRpc( }.doOnError { errorMetric.increment() } } + override fun nativeSubscribe(request: Mono): Flux { + var metrics: RequestMetrics? = null + return nativeSubscribe.nativeSubscribe( + request + .doOnNext { + metrics = chainMetrics.get(it.chain) + metrics!!.nativeSubscribeMetric.increment() + } + ).doOnNext { + metrics?.nativeSubscribeRespMetric?.increment() + }.doOnError { errorMetric.increment() } + } + override fun subscribeHead(request: Mono): Flux { return streamHead.add( request @@ -169,6 +183,14 @@ class BlockchainRpc( .tag("chain", chain.chainCode) .publishPercentileHistogram() .register(Metrics.globalRegistry) + val nativeSubscribeMetric = Counter.builder("request.grpc.request") + .tag("type", "nativeSubscribe") + .tag("chain", chain.chainCode) + .register(Metrics.globalRegistry) + val nativeSubscribeRespMetric = Counter.builder("request.grpc.response") + .tag("type", "nativeSubscribe") + .tag("chain", chain.chainCode) + .register(Metrics.globalRegistry) val subscribeHeadMetric = Counter.builder("request.grpc.request") .tag("type", "subscribeHead") .tag("chain", chain.chainCode) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt new file mode 100644 index 00000000..fa98b663 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt @@ -0,0 +1,98 @@ +/** + * Copyright (c) 2021 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.rpc + +import com.google.protobuf.ByteString +import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.SilentException +import io.emeraldpay.dshackle.upstream.MultistreamHolder +import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream +import io.emeraldpay.grpc.BlockchainType +import io.emeraldpay.grpc.Chain +import io.grpc.Status +import io.grpc.StatusException +import org.reactivestreams.Publisher +import org.slf4j.LoggerFactory +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.stereotype.Service +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono + +@Service +class NativeSubscribe( + @Autowired private val multistreamHolder: MultistreamHolder +) { + + companion object { + private val log = LoggerFactory.getLogger(NativeSubscribe::class.java) + } + + private val objectMapper = Global.objectMapper + + fun nativeSubscribe(request: Mono): Flux { + return request + .flatMapMany(this@NativeSubscribe::start) + .map(this@NativeSubscribe::convertToProto) + .onErrorMap(this@NativeSubscribe::convertToStatus) + } + + fun start(it: BlockchainOuterClass.NativeSubscribeRequest): Publisher { + val chain = Chain.byId(it.chainValue) + if (BlockchainType.from(chain) != BlockchainType.ETHEREUM) { + return Mono.error(UnsupportedOperationException("Native subscribe is not supported for ${chain.chainCode}")) + } + val method = it.method + val params: List<*> = it.payload?.let { payload -> + if (payload.size() > 0) { + listOf(objectMapper.readValue(payload.newInput(), Map::class.java)) + } else { + emptyList() + } + } ?: emptyList() + return subscribe(chain, method, params) + } + + fun convertToStatus(t: Throwable) = when (t) { + is SilentException.UnsupportedBlockchain -> StatusException( + Status.UNAVAILABLE.withDescription("BLOCKCHAIN UNAVAILABLE: ${t.blockchainId}") + ) + is UnsupportedOperationException -> StatusException( + Status.UNIMPLEMENTED.withDescription(t.message) + ) + else -> { + log.warn("Unhandled error", t) + StatusException( + Status.INTERNAL.withDescription(t.message) + ) + } + } + + fun subscribe(chain: Chain, method: String, params: List<*>): Flux { + val up = multistreamHolder.getUpstream(chain) ?: return Flux.error(SilentException.UnsupportedBlockchain(chain)) + return (up as EthereumMultistream) + .getSubscribe() + .subscribe(method, params) + } + + fun convertToProto(value: Any): BlockchainOuterClass.NativeSubscribeReplyItem { + val result = objectMapper.writeValueAsBytes(value) + return BlockchainOuterClass.NativeSubscribeReplyItem.newBuilder() + .setPayload(ByteString.copyFrom(result)) + .build() + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt index 963aac5c..34849d69 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt @@ -41,6 +41,7 @@ open class EthereumMultistream( private var head: Head? = null private val reader: EthereumReader = EthereumReader(this, this.caches, getMethodsFactory()) + private val subscribe = EthereumSubscribe() init { this.init() @@ -122,4 +123,7 @@ open class EthereumMultistream( return Mono.just(NativeCallRouter(reader, getMethods(), getHead())) } + open fun getSubscribe(): EthereumSubscribe { + return subscribe + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt new file mode 100644 index 00000000..b3f1c0a9 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt @@ -0,0 +1,15 @@ +package io.emeraldpay.dshackle.upstream.ethereum + +import org.slf4j.LoggerFactory +import reactor.core.publisher.Flux + +open class EthereumSubscribe { + + companion object { + private val log = LoggerFactory.getLogger(EthereumSubscribe::class.java) + } + + open fun subscribe(method: String, params: List<*>): Flux { + return Flux.error(UnsupportedOperationException("Method $method is not supported")) + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt index 48324ffc..96752883 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt @@ -24,10 +24,8 @@ import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse -import io.emeraldpay.dshackle.upstream.rpcclient.ResponseWSParser -import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics +import io.emeraldpay.dshackle.upstream.rpcclient.* +import io.emeraldpay.etherjar.rpc.RpcResponseError import io.emeraldpay.etherjar.rpc.json.BlockJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import io.netty.buffer.ByteBuf @@ -365,6 +363,7 @@ class EthereumWsFactory( Flux.from(rpcReceive.asFlux()) .doOnSubscribe { sendRpc(request) } .filter { resp -> resp.id.asNumber() == expectedId } + .take(Defaults.timeout) .take(1) .singleOrEmpty() .doOnNext { @@ -374,6 +373,12 @@ class EthereumWsFactory( rpcMetrics?.errors?.increment() } .map { it.copyWithId(JsonRpcResponse.Id.from(originalId)) } + .defaultIfEmpty( + JsonRpcResponse(null, + JsonRpcError(RpcResponseError.CODE_INTERNAL_ERROR, "Response not received from WebSocket"), + JsonRpcResponse.Id.from(originalId) + ) + ) } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy new file mode 100644 index 00000000..91a04b33 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy @@ -0,0 +1,92 @@ +/** + * Copyright (c) 2021 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.rpc + +import com.google.protobuf.ByteString +import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.dshackle.test.MultistreamHolderMock +import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream +import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscribe +import io.emeraldpay.grpc.Chain +import reactor.core.publisher.Flux +import reactor.test.StepVerifier +import spock.lang.Specification + +import java.time.Duration + +class NativeSubscribeSpec extends Specification { + + def "Call with empty params when not provided"() { + setup: + def subscribe = Mock(EthereumSubscribe) { + 1 * it.subscribe("newHeads", []) >> Flux.just("{}") + } + def up = Mock(EthereumMultistream) { + 1 * it.getSubscribe() >> subscribe + } + + def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM, up)) + def call = BlockchainOuterClass.NativeSubscribeRequest.newBuilder() + .setChainValue(Chain.ETHEREUM.id) + .setMethod("newHeads") + .build() + when: + def act = nativeSubscribe.start(call) + + then: + StepVerifier.create(act) + .expectNext("{}") + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "Call with params when provided"() { + setup: + def subscribe = Mock(EthereumSubscribe) { + 1 * it.subscribe("newHeads", { params -> + println("params: $params") + def ok = params.size() == 1 && + params[0] instanceof Map && + params[0]["address"] == "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2" && + params[0]["topics"] instanceof List && + params[0]["topics"][0] == "0x7fcf532c15f0a6db0bd6d0e038bea71d30d808c7d98cb3bf7268a95bf5081b65" + println("ok: $ok") + ok + }) >> Flux.just("{}") + } + def up = Mock(EthereumMultistream) { + 1 * it.getSubscribe() >> subscribe + } + + def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM, up)) + def call = BlockchainOuterClass.NativeSubscribeRequest.newBuilder() + .setChainValue(Chain.ETHEREUM.id) + .setMethod("newHeads") + .setPayload(ByteString.copyFromUtf8( + '{"address": "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2", ' + + '"topics": ["0x7fcf532c15f0a6db0bd6d0e038bea71d30d808c7d98cb3bf7268a95bf5081b65"]}' + )) + .build() + when: + def act = nativeSubscribe.start(call) + + then: + StepVerifier.create(act) + .expectNext("{}") + .expectComplete() + .verify(Duration.ofSeconds(1)) + } +}