solution: base implementation for NativeSubscribe

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 SubscribeHead (Chain) returns (stream ChainHead) {}
-    }
+    rpc SubscribeBalance (BalanceRequest) returns (stream AddressBalance) {}
-    rpc SubscribeBalance (BalanceRequest) returns (stream AddressBalance) {
+    rpc SubscribeTxStatus (TxStatusRequest) returns (stream TxStatus) {}
-    }
-    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 Describe (DescribeRequest) returns (DescribeResponse) {}
-    }
+    rpc SubscribeStatus (StatusRequest) returns (stream ChainStatus) {}
-    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;

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 <ReqT : Any, RespT : Any> processNativeSubscribe(
+            call: ServerCall<ReqT, RespT>,
+            headers: Metadata,
+            next: ServerCallHandler<ReqT, RespT>
+    ): ServerCall.Listener<ReqT> {
+        return process(call, headers, next,
+                EventsBuilder.NativeSubscribe() as EventsBuilder.RequestReply<*, ReqT, RespT>
+        )
+    }
+
+
     @Suppress("UNCHECKED_CAST")
     private fun <ReqT : Any, RespT : Any> processDescribe(
             call: ServerCall<ReqT, RespT>,

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

src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt

@@ -293,6 +293,36 @@ class EventsBuilder {
         }
     }
 
+    class NativeSubscribe :
+            Base<NativeSubscribe>(),
+            RequestReply<Events.NativeSubscribe, BlockchainOuterClass.NativeSubscribeRequest, BlockchainOuterClass.NativeSubscribeReplyItem> {
+        var item: Events.NativeSubscribeItemDetails? = null
+        val replies = HashMap<Int, Events.NativeSubscribeReplyDetails>()
+
+        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<Describe>(),
             RequestReply<Events.Describe, BlockchainOuterClass.DescribeRequest, BlockchainOuterClass.DescribeResponse> {

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<TrackTx>,
         @Autowired private val trackAddress: List<TrackAddress>,
@@ -73,6 +74,19 @@ class BlockchainRpc(
         }.doOnError { errorMetric.increment() }
     }
 
+    override fun nativeSubscribe(request: Mono<BlockchainOuterClass.NativeSubscribeRequest>): Flux<BlockchainOuterClass.NativeSubscribeReplyItem> {
+        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<Common.Chain>): Flux<BlockchainOuterClass.ChainHead> {
         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)

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<BlockchainOuterClass.NativeSubscribeRequest>): Flux<BlockchainOuterClass.NativeSubscribeReplyItem> {
+        return request
+                .flatMapMany(this@NativeSubscribe::start)
+                .map(this@NativeSubscribe::convertToProto)
+                .onErrorMap(this@NativeSubscribe::convertToStatus)
+    }
+
+    fun start(it: BlockchainOuterClass.NativeSubscribeRequest): Publisher<Any> {
+        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<Any>()
+            }
+        } ?: emptyList<Any>()
+        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<Any> {
+        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()
+    }
+
+}

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
+    }
 }

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<Any> {
+        return Flux.error(UnsupportedOperationException("Method $method is not supported"))
+    }
+}

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.*
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.emeraldpay.etherjar.rpc.RpcResponseError
-import io.emeraldpay.dshackle.upstream.rpcclient.ResponseWSParser
-import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics
 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)
+                                        )
+                                )
                     }
         }
 

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))
+    }
+}