From bdd25da3873ffbc0dfa26c72266ef71ba06a9841 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Tue, 26 Oct 2021 23:03:40 -0400 Subject: [PATCH] solution: Websocket Ethereum proxy fix: #110 --- .../io/emeraldpay/dshackle/ProxyStarter.kt | 4 +- .../emeraldpay/dshackle/config/ProxyConfig.kt | 1 + .../emeraldpay/dshackle/proxy/BaseHandler.kt | 86 ++++++++++ .../emeraldpay/dshackle/proxy/HttpHandler.kt | 86 ++++++++++ .../emeraldpay/dshackle/proxy/ProxyServer.kt | 104 ++---------- .../emeraldpay/dshackle/proxy/ReadRpcJson.kt | 70 ++++---- .../dshackle/proxy/WebsocketHandler.kt | 155 ++++++++++++++++++ .../dshackle/rpc/NativeSubscribe.kt | 4 +- .../dshackle/proxy/BaseHandlerSpec.groovy | 105 ++++++++++++ ...rverSpec.groovy => HttpHandlerSpec.groovy} | 145 ++++++++-------- .../dshackle/proxy/ReadRpcJsonSpec.groovy | 8 + .../proxy/WebsocketHandlerSpec.groovy | 117 +++++++++++++ 12 files changed, 681 insertions(+), 204 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/proxy/HttpHandler.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/proxy/BaseHandlerSpec.groovy rename src/test/groovy/io/emeraldpay/dshackle/proxy/{ProxyServerSpec.groovy => HttpHandlerSpec.groovy} (67%) create mode 100644 src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/ProxyStarter.kt b/src/main/kotlin/io/emeraldpay/dshackle/ProxyStarter.kt index f1f248c8..8d05044f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/ProxyStarter.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/ProxyStarter.kt @@ -23,6 +23,7 @@ import io.emeraldpay.dshackle.proxy.ProxyServer import io.emeraldpay.dshackle.proxy.ReadRpcJson import io.emeraldpay.dshackle.proxy.WriteRpcJson import io.emeraldpay.dshackle.rpc.NativeCall +import io.emeraldpay.dshackle.rpc.NativeSubscribe import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service @@ -37,6 +38,7 @@ class ProxyStarter( @Autowired private val readRpcJson: ReadRpcJson, @Autowired private val writeRpcJson: WriteRpcJson, @Autowired private val nativeCall: NativeCall, + @Autowired private val nativeSubscribe: NativeSubscribe, @Autowired private val tlsSetup: TlsSetup, @Autowired private val accessHandlerHttp: AccessHandlerHttp, // depend on Monitoring, declared here just to ensure it's properly initialized before the Proxy @@ -54,7 +56,7 @@ class ProxyStarter( log.debug("Proxy server is not configured") return } - val server = ProxyServer(config, readRpcJson, writeRpcJson, nativeCall, tlsSetup, accessHandlerHttp.factory) + val server = ProxyServer(config, readRpcJson, writeRpcJson, nativeCall, nativeSubscribe, tlsSetup, accessHandlerHttp.factory) server.start() } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfig.kt index 2c53e5cb..fc01f00d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfig.kt @@ -28,6 +28,7 @@ open class ProxyConfig { } var enabled: Boolean = true + var websocketEnabled: Boolean = true /** * Host to bind server. Default: 127.0.0.1 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt new file mode 100644 index 00000000..8ca2fdbd --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt @@ -0,0 +1,86 @@ +/** + * 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.proxy + +import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.api.proto.Common +import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp +import io.emeraldpay.dshackle.rpc.NativeCall +import io.emeraldpay.grpc.Chain +import org.reactivestreams.Publisher +import org.slf4j.LoggerFactory +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import java.util.concurrent.TimeUnit + +abstract class BaseHandler( + private val writeRpcJson: WriteRpcJson, + private val nativeCall: NativeCall, + private val requestMetrics: ProxyServer.RequestMetricsFactory, +) { + + companion object { + private val log = LoggerFactory.getLogger(BaseHandler::class.java) + } + + fun execute(chain: Chain, call: ProxyCall, handler: AccessHandlerHttp.RequestHandler): Publisher { + // return empty response for empty request + if (call.items.isEmpty()) { + return if (call.type == ProxyCall.RpcType.BATCH) { + Mono.just("[]") + } else { + Mono.just("") + } + } + val jsons = execute(chain, call.items, handler) + .transform(writeRpcJson.toJsons(call)) + return if (call.type == ProxyCall.RpcType.SINGLE) { + jsons.next() + } else { + jsons.transform(writeRpcJson.asArray()) + } + } + + fun execute(chain: Chain, items: List, handler: AccessHandlerHttp.RequestHandler): Flux { + val startTime = System.currentTimeMillis() + // during the execution we know only ID of the call, so we use it to find the origin call and associated metrics + val metricById = { id: Int -> + items.find { it.id == id }?.let { item -> + requestMetrics.get(chain, item.method) + } + } + val request = BlockchainOuterClass.NativeCallRequest.newBuilder() + .setChain(Common.ChainRef.forNumber(chain.id)) + .addAllItems(items) + .build() + handler.onRequest(request) + return nativeCall + .nativeCallResult(Mono.just(request)) + .doOnNext { + metricById(it.id)?.let { metrics -> + metrics.requestMetric.increment() + metrics.callMetric.record(System.currentTimeMillis() - startTime, TimeUnit.MILLISECONDS) + } + handler.onResponse(it) + } + .doOnError { + // when error happened the whole flux is stopped and no result is produced, so we should mark all the requests as failed + items.forEach { item -> + requestMetrics.get(chain, item.method).errorMetric.increment() + } + } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/HttpHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/HttpHandler.kt new file mode 100644 index 00000000..6ebe5177 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/HttpHandler.kt @@ -0,0 +1,86 @@ +/** + * 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.proxy + +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.config.ProxyConfig +import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp +import io.emeraldpay.dshackle.rpc.NativeCall +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.etherjar.rpc.RpcException +import io.emeraldpay.grpc.Chain +import io.netty.buffer.ByteBuf +import io.netty.buffer.Unpooled +import org.reactivestreams.Publisher +import org.slf4j.LoggerFactory +import org.springframework.http.HttpHeaders +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.netty.http.server.HttpServerRequest +import reactor.netty.http.server.HttpServerResponse +import java.util.function.BiFunction + +/** + * Responds to HTTP requests made to the Ethereum Proxy Server + */ +class HttpHandler( + private val readRpcJson: ReadRpcJson, + writeRpcJson: WriteRpcJson, + nativeCall: NativeCall, + private val accessHandler: AccessHandlerHttp.HandlerFactory, + requestMetrics: ProxyServer.RequestMetricsFactory, +) : BaseHandler(writeRpcJson, nativeCall, requestMetrics) { + + companion object { + private val log = LoggerFactory.getLogger(HttpHandler::class.java) + } + + fun proxy(routeConfig: ProxyConfig.Route): BiFunction> { + return BiFunction { req, resp -> + // handle access events + val eventHandler = accessHandler.create(req, routeConfig.blockchain) + val request = req.receive() + .aggregate() + .asByteArray() + val results = processRequest(routeConfig.blockchain, request, eventHandler) + // make sure that the access log handler is closed at the end, so it can render the logs + .doFinally { eventHandler.close() } + resp.addHeader(HttpHeaders.CONTENT_TYPE, "application/json") + .send(results) + } + } + + fun processRequest( + chain: Chain, + request: Mono, + handler: AccessHandlerHttp.RequestHandler + ): Flux { + return request + .map(readRpcJson) + .flatMapMany { call -> + execute(chain, call, handler) + } + .onErrorResume(RpcException::class.java) { err -> + val id = err.details?.let { + if (it is JsonRpcResponse.Id) it else JsonRpcResponse.NumberId(-1) + } ?: JsonRpcResponse.NumberId(-1) + + val json = JsonRpcResponse.error(err.code, err.rpcMessage, id) + Mono.just(Global.objectMapper.writeValueAsString(json)) + } + .map { Unpooled.wrappedBuffer(it.toByteArray()) } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt index 6e5ab4f9..a16c40a6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt @@ -16,36 +16,23 @@ */ package io.emeraldpay.dshackle.proxy -import io.emeraldpay.api.proto.BlockchainOuterClass -import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.TlsSetup import io.emeraldpay.dshackle.config.ProxyConfig import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp import io.emeraldpay.dshackle.rpc.NativeCall -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse -import io.emeraldpay.etherjar.rpc.RpcException +import io.emeraldpay.dshackle.rpc.NativeSubscribe import io.emeraldpay.grpc.Chain import io.micrometer.core.instrument.Counter import io.micrometer.core.instrument.Metrics import io.micrometer.core.instrument.Timer -import io.netty.buffer.ByteBuf -import io.netty.buffer.Unpooled import io.netty.channel.ChannelHandler import io.netty.channel.ChannelHandlerContext -import org.reactivestreams.Publisher import org.slf4j.LoggerFactory -import org.springframework.http.HttpHeaders -import reactor.core.publisher.Flux -import reactor.core.publisher.Mono import reactor.netty.http.server.HttpServer -import reactor.netty.http.server.HttpServerRequest -import reactor.netty.http.server.HttpServerResponse import reactor.netty.http.server.HttpServerRoutes import java.util.EnumMap -import java.util.concurrent.TimeUnit import java.util.concurrent.locks.ReentrantReadWriteLock -import java.util.function.BiFunction import kotlin.concurrent.read import kotlin.concurrent.write @@ -57,6 +44,7 @@ class ProxyServer( private val readRpcJson: ReadRpcJson, private val writeRpcJson: WriteRpcJson, private val nativeCall: NativeCall, + private val nativeSubscribe: NativeSubscribe, private val tlsSetup: TlsSetup, private val accessHandler: AccessHandlerHttp.HandlerFactory ) { @@ -92,6 +80,11 @@ class ProxyServer( StandardRequestMetrics() } + private val httpHandler = HttpHandler(readRpcJson, writeRpcJson, nativeCall, accessHandler, requestMetrics) + private val wsHandler: WebsocketHandler? = if (config.websocketEnabled) { + WebsocketHandler(readRpcJson, writeRpcJson, nativeCall, nativeSubscribe, requestMetrics) + } else null + fun start() { if (!config.enabled) { log.debug("Proxy server is not enabled") @@ -116,88 +109,11 @@ class ProxyServer( fun setupRoutes(routes: HttpServerRoutes) { config.routes.forEach { routeConfig -> - routes.post("/" + routeConfig.id, proxy(routeConfig)) - } - } - - fun execute(chain: Chain, call: ProxyCall, handler: AccessHandlerHttp.RequestHandler): Publisher { - // return empty response for empty request - if (call.items.isEmpty()) { - return if (call.type == ProxyCall.RpcType.BATCH) { - Mono.just("[]") - } else { - Mono.just("") + routes.post("/" + routeConfig.id, httpHandler.proxy(routeConfig)) + if (config.websocketEnabled && wsHandler != null) { + routes.ws("/" + routeConfig.id, wsHandler.proxy(routeConfig)) } } - val startTime = System.currentTimeMillis() - // during the execution we know only ID of the call, we use it to find the origin call and associated metrics - val metricById = { id: Int -> - call.items.find { it.id == id }?.let { item -> - requestMetrics.get(chain, item.method) - } - } - val request = BlockchainOuterClass.NativeCallRequest.newBuilder() - .setChain(Common.ChainRef.forNumber(chain.id)) - .addAllItems(call.items) - .build() - handler.onRequest(request) - val jsons = nativeCall - .nativeCallResult(Mono.just(request)) - .doOnNext { - metricById(it.id)?.requestMetric?.increment() - } - .doOnNext { - handler.onResponse(it) - metricById(it.id)?.callMetric?.record(System.currentTimeMillis() - startTime, TimeUnit.MILLISECONDS) - } - .doOnError { - // when error happened the whole flux is stopped and no result is produced, so we should mark all the requests as failed - call.items.forEach { item -> - requestMetrics.get(chain, item.method).errorMetric.increment() - } - } - .transform(writeRpcJson.toJsons(call)) - return if (call.type == ProxyCall.RpcType.SINGLE) { - jsons.next() - } else { - jsons.transform(writeRpcJson.asArray()) - } - } - - fun processRequest( - chain: Chain, - request: Mono, - handler: AccessHandlerHttp.RequestHandler - ): Flux { - return request - .map(readRpcJson) - .flatMapMany { call -> - execute(chain, call, handler) - } - .onErrorResume(RpcException::class.java) { err -> - val id = err.details?.let { - if (it is JsonRpcResponse.Id) it else JsonRpcResponse.NumberId(-1) - } ?: JsonRpcResponse.NumberId(-1) - - val json = JsonRpcResponse.error(err.code, err.rpcMessage, id) - Mono.just(Global.objectMapper.writeValueAsString(json)) - } - .map { Unpooled.wrappedBuffer(it.toByteArray()) } - } - - fun proxy(routeConfig: ProxyConfig.Route): BiFunction> { - return BiFunction { req, resp -> - // handle access events - val eventHandler = accessHandler.create(req, routeConfig.blockchain) - val request = req.receive() - .aggregate() - .asByteArray() - val results = processRequest(routeConfig.blockchain, request, eventHandler) - // make sure that the access log handler is closed at the end, so it can render the logs - .doFinally { eventHandler.close() } - resp.addHeader(HttpHeaders.CONTENT_TYPE, "application/json") - .send(results) - } } interface RequestMetricsFactory { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt index a5901f4c..feb785e9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt @@ -28,7 +28,6 @@ import org.slf4j.LoggerFactory import org.springframework.stereotype.Service import java.io.IOException import java.util.function.Function -import java.util.stream.Collectors /** * Reader for JSON RPC request @@ -41,11 +40,11 @@ open class ReadRpcJson : Function { private val spaces = " \n\t".toByteArray() } - private val jsonExtractor: Function, RequestJson> + val jsonExtractor: (Map<*, *>) -> RequestJson private val objectMapper: ObjectMapper = Global.objectMapper init { - jsonExtractor = Function { json -> + jsonExtractor = { json -> if (json["id"] == null) { throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "ID is not set") } @@ -126,33 +125,11 @@ open class ReadRpcJson : Function { * Convert payload to the proxy call details */ override fun apply(data: ByteArray): ProxyCall { - val list: MutableList> + val list: List> try { val type = getType(data) - if (ProxyCall.RpcType.BATCH == type) { - list = objectMapper.readerFor(MutableList::class.java).readValue(data) - } else { - list = ArrayList(1) - val json = objectMapper.readerFor(MutableMap::class.java).readValue>(data) - list.add(json) - } - val context = ProxyCall(type) - // our internal ids for calls - var seq = 0 - val batch = list.stream() - .map>(jsonExtractor) - .map { json -> - val id = seq++ - context.ids[id] = json.id - BlockchainOuterClass.NativeCallItem.newBuilder() - .setId(id) - .setMethod(json.method) - .setPayload(ByteString.copyFrom(objectMapper.writeValueAsBytes(json.params))) - .build() - } - .collect(Collectors.toList()) - context.items.addAll(batch) - return context + list = extract(type, data) + return convertMapToNativeCall(type, list) } catch (e: RpcException) { throw e } catch (e: Exception) { @@ -160,4 +137,41 @@ open class ReadRpcJson : Function { throw RpcException(RpcResponseError.CODE_INVALID_JSON, e.message) } } + + fun extract(type: ProxyCall.RpcType, data: ByteArray): List> { + return if (ProxyCall.RpcType.BATCH == type) { + objectMapper.readerFor(MutableList::class.java).readValue(data) + } else { + val list = ArrayList>(1) + val json = objectMapper.readerFor(MutableMap::class.java).readValue>(data) + list.add(json) + list + } + } + + fun convertMapToNativeCall(type: ProxyCall.RpcType, list: List>): ProxyCall { + return convertToNativeCall(type, list.map(jsonExtractor)) + } + + fun convertToNativeCall(type: ProxyCall.RpcType, list: List>): ProxyCall { + val context = ProxyCall(type) + val batch = convertToNativeCall(0, context, list) + context.items.addAll(batch) + return context + } + + fun convertToNativeCall(seqStart: Int, context: ProxyCall, items: List>): List { + // internal ids for calls + var seq = seqStart + return items + .map { json -> + val id = seq++ + context.ids[id] = json.id + BlockchainOuterClass.NativeCallItem.newBuilder() + .setId(id) + .setMethod(json.method) + .setPayload(ByteString.copyFrom(objectMapper.writeValueAsBytes(json.params))) + .build() + } + } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt new file mode 100644 index 00000000..e36007d2 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt @@ -0,0 +1,155 @@ +/** + * 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.proxy + +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.config.ProxyConfig +import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp +import io.emeraldpay.dshackle.rpc.NativeCall +import io.emeraldpay.dshackle.rpc.NativeSubscribe +import io.emeraldpay.etherjar.rpc.json.RequestJson +import io.emeraldpay.etherjar.rpc.json.ResponseJson +import io.emeraldpay.grpc.Chain +import io.netty.buffer.ByteBufInputStream +import io.netty.buffer.Unpooled +import org.apache.commons.lang3.StringUtils +import org.reactivestreams.Publisher +import org.slf4j.LoggerFactory +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.netty.http.websocket.WebsocketInbound +import reactor.netty.http.websocket.WebsocketOutbound +import java.util.concurrent.atomic.AtomicLong +import java.util.function.BiFunction + +/** + * Responds to Websocket requests made to the Ethereum Proxy Server + */ +class WebsocketHandler( + private val readRpcJson: ReadRpcJson, + writeRpcJson: WriteRpcJson, + nativeCall: NativeCall, + private val nativeSubscribe: NativeSubscribe, + requestMetrics: ProxyServer.RequestMetricsFactory, +) : BaseHandler(writeRpcJson, nativeCall, requestMetrics) { + + companion object { + private val log = LoggerFactory.getLogger(WebsocketHandler::class.java) + } + + private val subscriptionId = AtomicLong(0) + + fun nextSubscriptionId(): String { + val n = subscriptionId.incrementAndGet() + return StringUtils.leftPad(n.toString(16), 16, "0") + } + + fun proxy(routeConfig: ProxyConfig.Route): BiFunction> { + return BiFunction { req, resp -> + val requests: Flux> = req.aggregateFrames() + .receiveFrames() + .map { ByteBufInputStream(it.content()).readAllBytes() } + .flatMap(this@WebsocketHandler::parseRequest) + + val eventHandler: AccessHandlerHttp.RequestHandler = AccessHandlerHttp.NoOpHandler() + val responses = respond(routeConfig.blockchain, requests, eventHandler) + .map { Unpooled.wrappedBuffer(it.toByteArray()) } + + resp.send(responses) + .then() + } + } + + fun parseRequest(data: ByteArray): Mono> { + // try to parse JSON call. If received an invalid value just silently ignore it, that's what other Ethereum servers do + try { + val type = readRpcJson.getType(data) + // WS is not supposed to have batches, so ignore them too + if (type != ProxyCall.RpcType.SINGLE) { + return Mono.empty() + } + val items = readRpcJson.extract(type, data) + if (items.isEmpty()) { + //empty should never happen for a SINGLE type of request, but anyway, just return nothing + return Mono.empty() + } + return Mono + .just(items.first()) + .map(readRpcJson.jsonExtractor) + .onErrorResume { Mono.empty() } + } catch (t: Throwable) { + return Mono.empty() + } + } + + fun respond(blockchain: Chain, requests: Flux>, eventHandler: AccessHandlerHttp.RequestHandler): Flux { + return requests.flatMap { call -> + val method = call.method + if (method == "eth_subscribe") { + val methodParams = splitMethodParams(call.params) + if (methodParams != null) { + val subscriptionId = nextSubscriptionId() + // first need to respond with ID of the subscription, and the following responses would have it in "subscription" param + val start = ResponseJson().also { + it.id = call.id + it.result = subscriptionId + } + // produce actual responses + val responses = nativeSubscribe + .subscribe(blockchain, methodParams.first, methodParams.second) + .map { event -> + WsSubscriptionResponse(params = WsSubscriptionData(event, subscriptionId)) + } + Flux.concat(Mono.just(start), responses) + .map { Global.objectMapper.writeValueAsString(it) } + } else { + Mono.empty() + } + } else { + val proxyCall = readRpcJson.convertToNativeCall(ProxyCall.RpcType.SINGLE, listOf(call)) + execute(blockchain, proxyCall, eventHandler) + } + } + } + + fun splitMethodParams(params: List): Pair? { + if (params.isEmpty()) { + return null + } + if (params.size == 1) { + return Pair(params.first().toString(), null) + } + if (params.size == 2) { + return Pair(params.first().toString(), params[1]) + } + return null + } + + // classes only to render WebSocket subscription response. + // the difference with standard JSON RPC responses that it + // (1) it doesn't have id on the top level, but rather as part of params, + // and (2) it has the `method` field + data class WsSubscriptionResponse( + val jsonrpc: String = "2.0", + val method: String = "eth_subscription", + val params: WsSubscriptionData, + ) + + data class WsSubscriptionData( + val result: Any?, + val subscription: String + ) +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt index fd8ffbe2..63640e3a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt @@ -33,7 +33,7 @@ import reactor.core.publisher.Flux import reactor.core.publisher.Mono @Service -class NativeSubscribe( +open class NativeSubscribe( @Autowired private val multistreamHolder: MultistreamHolder ) { @@ -81,7 +81,7 @@ class NativeSubscribe( } } - fun subscribe(chain: Chain, method: String, params: Any?): Flux { + open fun subscribe(chain: Chain, method: String, params: Any?): Flux { val up = multistreamHolder.getUpstream(chain) ?: return Flux.error(SilentException.UnsupportedBlockchain(chain)) return (up as EthereumMultistream) .getSubscribe() diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/BaseHandlerSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/BaseHandlerSpec.groovy new file mode 100644 index 00000000..607c1bab --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/BaseHandlerSpec.groovy @@ -0,0 +1,105 @@ +/** + * 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.proxy + +import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp +import io.emeraldpay.dshackle.rpc.NativeCall +import io.emeraldpay.grpc.Chain +import org.jetbrains.annotations.NotNull +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import spock.lang.Specification + +import java.time.Duration + +class BaseHandlerSpec extends Specification { + + def requestHandler = new AccessHandlerHttp.NoOpHandler() + + def "Return empty for empty single call"() { + setup: + def handler = new BaseHandlerImpl(new WriteRpcJson(), Stub(NativeCall), Stub(ProxyServer.RequestMetricsFactory)) + when: + def act = Mono.from(handler.execute(Chain.ETHEREUM, new ProxyCall(ProxyCall.RpcType.SINGLE), requestHandler)) + .block(Duration.ofSeconds(1)) + then: + act == "" + } + + def "Return empty array for empty batch call"() { + setup: + def handler = new BaseHandlerImpl(new WriteRpcJson(), Stub(NativeCall), Stub(ProxyServer.RequestMetricsFactory)) + when: + def act = Mono.from(handler.execute(Chain.ETHEREUM, new ProxyCall(ProxyCall.RpcType.BATCH), requestHandler)) + .block(Duration.ofSeconds(1)) + then: + act == "[]" + } + + def "Execute single call"() { + setup: + def nativeCall = Mock(NativeCall) + def handler = new BaseHandlerImpl(new WriteRpcJson(), nativeCall, Stub(ProxyServer.RequestMetricsFactory)) + + def request = BlockchainOuterClass.NativeCallItem.newBuilder() + .setMethod("eth_test") + .setId(0) + .build() + def call = new ProxyCall(ProxyCall.RpcType.SINGLE) + call.items.add(request) + call.ids[0] = 5 + def response = new NativeCall.CallResult(0, '{"foo": 1}'.bytes, null) + when: + def act = Flux.from(handler.execute(Chain.ETHEREUM, call, requestHandler)) + .collectList() + .block(Duration.ofSeconds(1)) + .join("") + then: + act == '{"jsonrpc":"2.0","id":5,"result":{"foo": 1}}' + 1 * nativeCall.nativeCallResult(_) >> Flux.fromIterable([response]) + } + + def "Execute batch call with one item"() { + setup: + def nativeCall = Mock(NativeCall) + def handler = new BaseHandlerImpl(new WriteRpcJson(), nativeCall, Stub(ProxyServer.RequestMetricsFactory)) + + def request = BlockchainOuterClass.NativeCallItem.newBuilder() + .setMethod("eth_test") + .setId(0) + .build() + def call = new ProxyCall(ProxyCall.RpcType.BATCH) + call.items.add(request) + call.ids[0] = 5 + def response = new NativeCall.CallResult(0, '{"foo": 1}'.bytes, null) + when: + def act = Flux.from(handler.execute(Chain.ETHEREUM, call, requestHandler)) + .collectList() + .block(Duration.ofSeconds(1)) + .join("") + then: + act == '[{"jsonrpc":"2.0","id":5,"result":{"foo": 1}}]' + 1 * nativeCall.nativeCallResult(_) >> Flux.fromIterable([response]) + } + + class BaseHandlerImpl extends BaseHandler { + + BaseHandlerImpl(@NotNull WriteRpcJson writeRpcJson, @NotNull NativeCall nativeCall, @NotNull ProxyServer.RequestMetricsFactory requestMetrics) { + super(writeRpcJson, nativeCall, requestMetrics) + } + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/ProxyServerSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/HttpHandlerSpec.groovy similarity index 67% rename from src/test/groovy/io/emeraldpay/dshackle/proxy/ProxyServerSpec.groovy rename to src/test/groovy/io/emeraldpay/dshackle/proxy/HttpHandlerSpec.groovy index 8bd5d75b..59b1d211 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/proxy/ProxyServerSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/HttpHandlerSpec.groovy @@ -1,6 +1,5 @@ /** - * Copyright (c) 2020 ETCDEV GmbH - * Copyright (c) 2020 EmeraldPay, Inc + * 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. @@ -19,11 +18,8 @@ package io.emeraldpay.dshackle.proxy import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common -import io.emeraldpay.dshackle.TlsSetup -import io.emeraldpay.dshackle.config.ProxyConfig import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp import io.emeraldpay.dshackle.rpc.NativeCall -import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.etherjar.rpc.RpcException import io.emeraldpay.grpc.Chain @@ -35,64 +31,7 @@ import spock.lang.Specification import java.time.Duration import java.util.function.Function -class ProxyServerSpec extends Specification { - - def "Uses NativeCall"() { - setup: - NativeCall nativeCall = Mock(NativeCall) - def predefined = { a -> Flux.just("hello") } as Function - - WriteRpcJson writeRpcJson = Mock { - 1 * toJsons(_) >> predefined - } - - ProxyServer server = new ProxyServer( - new ProxyConfig(), - new ReadRpcJson(), - writeRpcJson, - nativeCall, - new TlsSetup(TestingCommons.fileResolver()), - new AccessHandlerHttp.NoOpFactory() - ) - - def call = new ProxyCall(ProxyCall.RpcType.SINGLE) - call.ids[1] = 1 - call.items.add( - BlockchainOuterClass.NativeCallItem.newBuilder() - .setMethod("eth_hello") - .build() - ) - when: - def act = server.execute(Chain.ETHEREUM, call, new AccessHandlerHttp.NoOpHandler()) - - then: - 1 * nativeCall.nativeCallResult(_) >> Flux.just(new NativeCall.CallResult(1, "".bytes, null)) - StepVerifier.create(act) - .expectNext("hello") - .expectComplete() - .verify(Duration.ofSeconds(1)) - } - - def "Return error on invalid request"() { - setup: - ReadRpcJson read = Mock(ReadRpcJson) { - 1 * apply(_) >> { throw new RpcException(-32123, "test", new JsonRpcResponse.NumberId(4)) } - } - def server = new ProxyServer( - Stub(ProxyConfig), - read, - Stub(WriteRpcJson), Stub(NativeCall), Stub(TlsSetup), - new AccessHandlerHttp.NoOpFactory() - ) - when: - def act = server.processRequest(Chain.ETHEREUM, Mono.just("".bytes), new AccessHandlerHttp.NoOpHandler()) - .map { new String(it.array()) } - then: - StepVerifier.create(act) - .expectNext('{"jsonrpc":"2.0","id":4,"error":{"code":-32123,"message":"test"}}') - .expectComplete() - .verify(Duration.ofSeconds(1)) - } +class HttpHandlerSpec extends Specification { def "Calls access log handler"() { setup: @@ -107,30 +46,78 @@ class ProxyServerSpec extends Specification { .addItems(reqItem) .build() - - ReadRpcJson read = Mock(ReadRpcJson) { - 1 * apply(_) >> new ProxyCall(ProxyCall.RpcType.SINGLE).tap { it.items.add(reqItem) } - } NativeCall nativeCall = Mock(NativeCall) { 1 * nativeCallResult(_) >> Flux.fromIterable([respItem]) } - def handler = Mock(AccessHandlerHttp.RequestHandler.class) - - def server = new ProxyServer( - Stub(ProxyConfig), - read, - new WriteRpcJson(), - nativeCall, - Stub(TlsSetup), - new AccessHandlerHttp.NoOpFactory() + def accessHandler = Mock(AccessHandlerHttp.RequestHandler) + def accessHandlerFactory = Mock(AccessHandlerHttp.HandlerFactory) { + _ * it.create(_,) >> accessHandler + } + def handler = new HttpHandler( + new ReadRpcJson(), new WriteRpcJson(), + nativeCall, accessHandlerFactory, Stub(ProxyServer.RequestMetricsFactory) ) when: - server.processRequest(Chain.ETHEREUM, Mono.just("".bytes), handler) + handler.execute(Chain.ETHEREUM, [reqItem], accessHandler) .blockLast() then: - 1 * handler.onRequest(req) - 1 * handler.onResponse(respItem) + 1 * accessHandler.onRequest(req) + 1 * accessHandler.onResponse(respItem) + } + + def "Return error on invalid request"() { + setup: + ReadRpcJson read = Mock(ReadRpcJson) { + 1 * apply(_) >> { throw new RpcException(-32123, "test", new JsonRpcResponse.NumberId(4)) } + } + + def handler = new HttpHandler( + read, new WriteRpcJson(), + Stub(NativeCall), Stub(AccessHandlerHttp.HandlerFactory), Stub(ProxyServer.RequestMetricsFactory) + ) + when: + + def act = handler.processRequest(Chain.ETHEREUM, Mono.just("".bytes), new AccessHandlerHttp.NoOpHandler()) + .map { new String(it.array()) } + then: + StepVerifier.create(act) + .expectNext('{"jsonrpc":"2.0","id":4,"error":{"code":-32123,"message":"test"}}') + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "Uses NativeCall"() { + setup: + NativeCall nativeCall = Mock(NativeCall) + def predefined = { a -> Flux.just("hello") } as Function + + WriteRpcJson writeRpcJson = Mock { + 1 * toJsons(_) >> predefined + } + + + def handler = new HttpHandler( + new ReadRpcJson(), writeRpcJson, + nativeCall, Stub(AccessHandlerHttp.HandlerFactory), Stub(ProxyServer.RequestMetricsFactory) + ) + + def call = new ProxyCall(ProxyCall.RpcType.SINGLE) + call.ids[1] = 1 + call.items.add( + BlockchainOuterClass.NativeCallItem.newBuilder() + .setMethod("eth_hello") + .build() + ) + when: + def act = handler.execute(Chain.ETHEREUM, call, new AccessHandlerHttp.NoOpHandler()) + + then: + 1 * nativeCall.nativeCallResult(_) >> Flux.just(new NativeCall.CallResult(1, "".bytes, null)) + StepVerifier.create(act) + .expectNext("hello") + .expectComplete() + .verify(Duration.ofSeconds(1)) } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/ReadRpcJsonSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/ReadRpcJsonSpec.groovy index 53b31983..3cab7db4 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/proxy/ReadRpcJsonSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/ReadRpcJsonSpec.groovy @@ -224,4 +224,12 @@ class ReadRpcJsonSpec extends Specification { t.rpcMessage.toLowerCase() == "params must be an array" t.details == new JsonRpcResponse.NumberId(2) } + + def "Error if json is broken"() { + when: + reader.apply('{"id":2, "method":"net_peerCount", "params"'.bytes) + then: + def t = thrown(RpcException) + t.code == -32700 + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy new file mode 100644 index 00000000..35027283 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy @@ -0,0 +1,117 @@ +/** + * 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.proxy + +import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp +import io.emeraldpay.dshackle.rpc.NativeCall +import io.emeraldpay.dshackle.rpc.NativeSubscribe +import io.emeraldpay.etherjar.rpc.json.RequestJson +import io.emeraldpay.grpc.Chain +import reactor.core.publisher.Flux +import spock.lang.Specification + +import java.time.Duration + +class WebsocketHandlerSpec extends Specification { + + def requestHandler = new AccessHandlerHttp.NoOpHandler() + + def "Parse standard RPC request"() { + setup: + def handler = new WebsocketHandler( + new ReadRpcJson(), Stub(WriteRpcJson), Stub(NativeCall), Stub(NativeSubscribe), Stub(ProxyServer.RequestMetricsFactory) + ) + when: + def act = handler.parseRequest('{"id": 5, "jsonrpc": "2.0", "method": "eth_getBlockByNumber", "params": ["0x100001", false]}'.bytes) + .block(Duration.ofSeconds(1)) + + then: + act.id == 5 + act.method == "eth_getBlockByNumber" + act.params == ["0x100001", false] + } + + def "Parse to empty an invalid request"() { + setup: + def handler = new WebsocketHandler( + new ReadRpcJson(), Stub(WriteRpcJson), Stub(NativeCall), Stub(NativeSubscribe), Stub(ProxyServer.RequestMetricsFactory) + ) + when: + def act = handler.parseRequest('hello world'.bytes) + .block(Duration.ofSeconds(1)) + + then: + act == null + } + + def "Parse to empty a batch request"() { + setup: + def req1 = '{"id": 5, "jsonrpc": "2.0", "method": "eth_getBlockByNumber", "params": ["0x100001", false]}' + def handler = new WebsocketHandler( + new ReadRpcJson(), Stub(WriteRpcJson), Stub(NativeCall), Stub(NativeSubscribe), Stub(ProxyServer.RequestMetricsFactory) + ) + when: + def act = handler.parseRequest("[$req1]".bytes) + .block(Duration.ofSeconds(1)) + + then: + act == null + } + + def "Respond to a single call"() { + setup: + def response = new NativeCall.CallResult(0, '{"foo": 1}'.bytes, null) + + def nativeCall = Mock(NativeCall) { + 1 * it.nativeCallResult(_) >> Flux.fromIterable([response]) + } + def handler = new WebsocketHandler( + new ReadRpcJson(), new WriteRpcJson(), nativeCall, Stub(NativeSubscribe), Stub(ProxyServer.RequestMetricsFactory) + ) + + def request = new RequestJson("foo_test", [], 2) + when: + def act = handler.respond(Chain.ETHEREUM, Flux.just(request), requestHandler) + .single() + .block(Duration.ofSeconds(1)) + then: + act == '{"jsonrpc":"2.0","id":2,"result":{"foo": 1}}' + } + + def "Respond to a subscription call"() { + setup: + def response1 = [foo: 1] + def response2 = [foo: 2] + + def nativeSubscribe = Mock(NativeSubscribe) { + 1 * it.subscribe(Chain.ETHEREUM, "foo_test", null) >> Flux.fromIterable([response1, response2]) + } + def handler = new WebsocketHandler( + new ReadRpcJson(), new WriteRpcJson(), Stub(NativeCall), nativeSubscribe, Stub(ProxyServer.RequestMetricsFactory) + ) + + def request = new RequestJson("eth_subscribe", ["foo_test"], 2) + when: + def act = handler.respond(Chain.ETHEREUM, Flux.just(request), requestHandler) + .collectList() + .block(Duration.ofSeconds(1)) + then: + act[0] == '{"jsonrpc":"2.0","id":2,"result":"0000000000000001"}' + act[1] == '{"jsonrpc":"2.0","method":"eth_subscription","params":{"result":{"foo":1},"subscription":"0000000000000001"}}' + act[2] == '{"jsonrpc":"2.0","method":"eth_subscription","params":{"result":{"foo":2},"subscription":"0000000000000001"}}' + act.size() == 3 + } +}