solution: Websocket Ethereum proxy

fix: #110

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

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

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<String> {
+        // 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<BlockchainOuterClass.NativeCallItem>, handler: AccessHandlerHttp.RequestHandler): Flux<NativeCall.CallResult> {
+        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()
+                }
+            }
+    }
+}

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<HttpServerRequest, HttpServerResponse, Publisher<Void>> {
+        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<ByteArray>,
+        handler: AccessHandlerHttp.RequestHandler
+    ): Flux<ByteBuf> {
+        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()) }
+    }
+}

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<String> {
-        // 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<ByteArray>,
-        handler: AccessHandlerHttp.RequestHandler
-    ): Flux<ByteBuf> {
-        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<HttpServerRequest, HttpServerResponse, Publisher<Void>> {
-        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 {

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<ByteArray, ProxyCall> {
         private val spaces = " \n\t".toByteArray()
     }
 
-    private val jsonExtractor: Function<Map<*, *>, RequestJson<Any>>
+    val jsonExtractor: (Map<*, *>) -> RequestJson<Any>
     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<ByteArray, ProxyCall> {
      * Convert payload to the proxy call details
      */
     override fun apply(data: ByteArray): ProxyCall {
-        val list: MutableList<Map<*, *>>
+        val list: List<Map<*, *>>
         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<Map<*, *>>(data)
-                list.add(json)
-            }
-            val context = ProxyCall(type)
-            // our internal ids for calls
-            var seq = 0
-            val batch = list.stream()
-                .map<RequestJson<Any>>(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<ByteArray, ProxyCall> {
             throw RpcException(RpcResponseError.CODE_INVALID_JSON, e.message)
         }
     }
+
+    fun extract(type: ProxyCall.RpcType, data: ByteArray): List<Map<*, *>> {
+        return if (ProxyCall.RpcType.BATCH == type) {
+            objectMapper.readerFor(MutableList::class.java).readValue(data)
+        } else {
+            val list = ArrayList<Map<*, *>>(1)
+            val json = objectMapper.readerFor(MutableMap::class.java).readValue<Map<*, *>>(data)
+            list.add(json)
+            list
+        }
+    }
+
+    fun convertMapToNativeCall(type: ProxyCall.RpcType, list: List<Map<*, *>>): ProxyCall {
+        return convertToNativeCall(type, list.map(jsonExtractor))
+    }
+
+    fun convertToNativeCall(type: ProxyCall.RpcType, list: List<RequestJson<Any>>): 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<RequestJson<Any>>): List<BlockchainOuterClass.NativeCallItem> {
+        // 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()
+            }
+    }
 }

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<WebsocketInbound, WebsocketOutbound, Publisher<Void>> {
+        return BiFunction { req, resp ->
+            val requests: Flux<RequestJson<Any>> = 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<RequestJson<Any>> {
+        // 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<RequestJson<Any>>, eventHandler: AccessHandlerHttp.RequestHandler): Flux<String> {
+        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<String, Any>().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<Any?>): Pair<String, Any?>? {
+        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
+    )
+}

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<out Any> {
+    open fun subscribe(chain: Chain, method: String, params: Any?): Flux<out Any> {
         val up = multistreamHolder.getUpstream(chain) ?: return Flux.error(SilentException.UnsupportedBlockchain(chain))
         return (up as EthereumMultistream)
             .getSubscribe()