Merge pull request #123 from emeraldpay/feat/websocket-proxy

README.adoc

@@ -24,7 +24,7 @@ It automatically verifies their availability and the current status of the netwo
 
 Provides:
 
-- Standard Bitcoin and Ethereum JSON RPC API
+- Standard Bitcoin and Ethereum JSON RPC API over HTTP and WebSocket
 - Enhanced gRPC-based API, with upstream selection, async execution, etc
 - **Secure** TLS with optional client authentication
 - Blockchain-aware edge **caching**, in memory and Redis
@@ -107,12 +107,12 @@ Which sets the following:
 
 - gRPC access through 0.0.0.0:2449
 ** TLS security is disabled (_please don't use in production!_)
-- JSON RPC access through 0.0.0.0:8545
+- JSON RPC access through 0.0.0.0:8545 (both HTTP and WebsScket)
 ** proxy requests to Ethereum and Kovan upstreams
 ** request path for Ethereum Mainnet is `/eth`, `/kovan` for Kovan Testnet, and `/btc` for bitcoin
 ** i.e. call Ethereum Mainnet by `POST http://127.0.0.0:8545/eth` with JSON RPC payload
 - two upstreams, one for Ethereum Mainnet and another for Kovan Testnet (both upstreams are configured to use Infura endpoint)
-- for Ethereum Mainnet it connects using JSON RPC and Websockets connections,
+- for Ethereum Mainnet it connects using JSON RPC and WebSocket connections,
 - for Bitcoin Mainet only JSON RPC is used
 - `${INFURA_USER}` will be provided through environment variable
 
@@ -146,7 +146,7 @@ Tools such as https://github.com/fullstorydev/grpcurl[gRPCurl] can automatically
 
 Alternatively you can connect to port 8545 with traditional JSON RPC requests
 
-==== Access using JSON RPC
+==== Access using JSON RPC over HTTP
 
 Dshackle implements standard JSON RPC interface, providing additional caching layer, upstream readiness/liveness checks, retry and other features for building Fault Tolerant services.
 
@@ -165,6 +165,24 @@ curl --request POST \
 {"jsonrpc":"2.0","id":1,"result":"0x72fa5e0181"}
 ----
 
+==== Access using JSON RPC over WebSocket
+
+Or the same Proxy URL can be accessed through WebSocket
+
+[source,bash]
+----
+websocat ws://localhost:8545/eth
+----
+
+Then make RPC calls or subscriptions:
+
+----
+> | {"jsonrpc":"2.0", "id": 1, "method": "eth_subscribe", "params": ["newHeads"]}
+
+< | {"jsonrpc":"2.0","id":1,"result":"1f8"}
+< | {"jsonrpc":"2.0","method":"eth_subscription","params":{"result":{....},"subscription":"1f8"}}
+----
+
 ==== Access using gRPC
 
 NOTE: It's not necessary to use gRPC, as Dshackle can provide standard JSON RPC proxy, but Dshackle gRPC interface improves performance and provides additional features.

docs/03-server-config.adoc

@@ -62,6 +62,10 @@ a| `upstreams`
 
 === Enabling JSON RPC proxy
 
+In addition to the gRPC protocol, Dshackle provides access compatible with Bitcoin and Ethereum JSON RPC.
+The same server can be accessible as an HTTP JSON RPC and WebSocket JSON RPC.
+For Ethereum, besides the standard RPC calls, it provides RPC subscriptions with `eth_subscribe` method.
+
 .Example proxy:
 [source,yaml]
 ----
@@ -82,7 +86,9 @@ cluster:
 With that configuration Dshackle starts a JSON RPC proxy:
 
 - JSON RPC server is listening on `0.0.0.0:8080`
-- `http://0.0.0.0:8080/eth` provides access to Ethereum API routed to an available upstream
+- `http://0.0.0.0:8080/eth` (and `ws://0.0.0.0:8080/eth`) provides access to Ethereum API routed to an available upstream
+
+NOTE: Same URL should be used to access both HTTP RPC and WebSocket RPC
 
 .Full configuration:
 [source,yaml]

docs/reference-configuration.adoc

@@ -50,6 +50,7 @@ cache:
 proxy:
   host: 0.0.0.0
   port: 8080
+  websocket: true
   tls:
     enabled: true
     server:
@@ -373,20 +374,26 @@ proxy:
 
 | `host`
 | `127.0.0.0`
-| Host to bind gRPC server
+| Host to bind HTTP server
 
 | `port`
-| `2449`
+| `8080`
-| Port to bind gRPC server
+| Port to bind HTT server
 
+| `websocket`
+| `true`
+| Enable WebSocket Proxy
 
 | `tls`
 |
-| Setup TLS configuration for the Proxy server. See <<tls>> section
+| Setup TLS configuration for the Proxy server.
+See <<tls>> section
 
 | `routes`
 |
-a| Routing paths for Proxy.The proxy will handle requests as `https://${HOST}:${PORT}/${ROUTE_ID}` (or `http://` if TLS is not enabled)
+a| Routing paths for Proxy.
+The proxy will handle requests as `https://${HOST}:${PORT}/${ROUTE_ID}` (or `http://` if TLS is not enabled).
+For WebSocket it's `wss` / `ws`, accordingly.
 |===
 
 .Route config
@@ -663,14 +670,14 @@ rpc:
 ----
 
 | `ws.url`
-| Websocket URL to connect to.
+| WebSocket URL to connect to.
 Optional, but optimizes performance if it's available.
 
 | `ws.origin`
-| HTTP `Origin` if required by Websocket remote server.
+| HTTP `Origin` if required by WebSocket remote server.
 
 | `ws.basic-auth` + ...
-| Websocket Basic Auth configuration, if required by the remote server
+| WebSocket Basic Auth configuration, if required by the remote server
 
 | `ws.frameSize`
 | WebSocket frame size limit.

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/config/ProxyConfigReader.kt

@@ -58,6 +58,9 @@ class ProxyConfigReader : YamlConfigReader(), ConfigReader<ProxyConfig> {
         getValueAsBool(input, "enabled")?.let {
             config.enabled = it
         }
+        getValueAsBool(input, "websocket")?.let {
+            config.websocketEnabled = it
+        }
         val currentRoutes = HashSet<String>()
         getList<MappingNode>(input, "routes")?.let { routes ->
             config.routes = routes.value.map { route ->

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

@@ -130,7 +130,7 @@ class AccessHandlerGrpc(
     ): ServerCall.Listener<ReqT> {
         return process(
             call, headers, next,
-            EventsBuilder.NativeSubscribe() as EventsBuilder.RequestReply<*, ReqT, RespT>
+            EventsBuilder.NativeSubscribe(Events.Channel.GRPC) as EventsBuilder.RequestReply<*, ReqT, RespT>
         )
     }
 

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

@@ -8,6 +8,7 @@ import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
 import reactor.netty.http.server.HttpServerRequest
+import reactor.netty.http.websocket.WebsocketInbound
 import java.time.Instant
 import java.util.concurrent.locks.ReentrantLock
 import kotlin.concurrent.withLock
@@ -25,6 +26,9 @@ class AccessHandlerHttp(
 
     companion object {
         private val log = LoggerFactory.getLogger(AccessHandlerHttp::class.java)
+
+        private val NO_SUBSCRIBE = NoOnSubscriptionHandler()
+        private val NO_REQUEST = NoOpHandler()
     }
 
     /**
@@ -38,11 +42,21 @@ class AccessHandlerHttp(
 
     interface HandlerFactory {
         fun create(req: HttpServerRequest, blockchain: Chain): RequestHandler
+        fun start(req: WebsocketInbound, blockchain: Chain): WsHandlerFactory
+    }
+
+    interface WsHandlerFactory {
+        fun call(): RequestHandler
+        fun subscribe(): SubscriptionHandler
     }
 
     class NoOpFactory : HandlerFactory {
         override fun create(req: HttpServerRequest, blockchain: Chain): RequestHandler {
-            return NoOpHandler()
+            return NO_REQUEST
+        }
+
+        override fun start(req: WebsocketInbound, blockchain: Chain): WsHandlerFactory {
+            return NO_REQUEST
         }
     }
 
@@ -50,6 +64,10 @@ class AccessHandlerHttp(
         override fun create(req: HttpServerRequest, blockchain: Chain): RequestHandler {
             return StandardHandler(accessLogWriter, req, blockchain)
         }
+
+        override fun start(req: WebsocketInbound, blockchain: Chain): WsHandlerFactory {
+            return StandardWsHandlerFactory(accessLogWriter, req, blockchain)
+        }
     }
 
     interface RequestHandler {
@@ -58,7 +76,12 @@ class AccessHandlerHttp(
         fun onResponse(callResult: NativeCall.CallResult)
     }
 
-    class NoOpHandler : RequestHandler {
+    interface SubscriptionHandler {
+        fun onRequest(request: Pair<String, ByteArray?>)
+        fun onResponse(msgSize: Long)
+    }
+
+    class NoOpHandler : RequestHandler, WsHandlerFactory {
         override fun close() {
         }
 
@@ -67,37 +90,47 @@ class AccessHandlerHttp(
 
         override fun onResponse(callResult: NativeCall.CallResult) {
         }
+
+        override fun call(): RequestHandler {
+            return this
+        }
+
+        override fun subscribe(): SubscriptionHandler {
+            return NO_SUBSCRIBE
+        }
     }
 
-    class StandardHandler(
+    class NoOnSubscriptionHandler : SubscriptionHandler {
-        private val accessLogWriter: AccessLogWriter,
+        override fun onRequest(request: Pair<String, ByteArray?>) {
-        private val httpRequest: HttpServerRequest,
-        private val blockchain: Chain
-    ) : RequestHandler {
-
-        private var request: BlockchainOuterClass.NativeCallRequest? = null
-        private val responses = ArrayList<NativeCall.CallResult>()
-        private val updateLock = ReentrantLock()
-
-        override fun close() {
-            if (request == null) {
-                return
-            }
-            val responseTime = Instant.now()
-            val builder = EventsBuilder.NativeCall()
-            builder.withChain(blockchain.id)
-            builder.start(httpRequest)
-            builder.onRequest(request!!)
-            responses
-                .map {
-                    builder.onReply(it, Events.Channel.JSONRPC).also { item ->
-                        // since for JSON RPC you get a single response then the timestamp of all items included in it must have the same timestamp
-                        item.ts = responseTime
-                    }
-                }
-                .let(accessLogWriter::submit)
         }
 
+        override fun onResponse(msgSize: Long) {
+        }
+    }
+
+    class StandardWsHandlerFactory(
+        private val accessLogWriter: AccessLogWriter,
+        private val wsRequest: WebsocketInbound,
+        private val blockchain: Chain
+    ) : WsHandlerFactory {
+
+        override fun call(): RequestHandler {
+            return WsRequestHandler(accessLogWriter, wsRequest, blockchain)
+        }
+
+        override fun subscribe(): SubscriptionHandler {
+            return WsSubscriptionHandler(accessLogWriter, wsRequest, blockchain)
+        }
+    }
+
+    abstract class AbstractRequestHandler(
+        private val accessLogWriter: AccessLogWriter,
+        private val channel: Events.Channel
+    ) : RequestHandler {
+        protected var request: BlockchainOuterClass.NativeCallRequest? = null
+        protected val responses = ArrayList<NativeCall.CallResult>()
+        protected val updateLock = ReentrantLock()
+
         override fun onRequest(request: BlockchainOuterClass.NativeCallRequest) {
             this.request = request
         }
@@ -107,5 +140,75 @@ class AccessHandlerHttp(
                 responses.add(callResult)
             }
         }
+
+        fun onClose(builder: EventsBuilder.NativeCall) {
+            val responseTime = Instant.now()
+            responses
+                .map {
+                    builder.onReply(it, channel).also { item ->
+                        // since for JSON RPC you get a single response then the timestamp of all items included in it must have the same timestamp
+                        item.ts = responseTime
+                    }
+                }
+                .let(accessLogWriter::submit)
+        }
+    }
+
+    class StandardHandler(
+        accessLogWriter: AccessLogWriter,
+        private val httpRequest: HttpServerRequest,
+        private val blockchain: Chain
+    ) : RequestHandler, AbstractRequestHandler(accessLogWriter, Events.Channel.JSONRPC) {
+
+        override fun close() {
+            if (request == null) {
+                return
+            }
+            val builder = EventsBuilder.NativeCall()
+            builder.withChain(blockchain.id)
+            builder.start(httpRequest)
+            builder.onRequest(request!!)
+            onClose(builder)
+        }
+    }
+
+    class WsRequestHandler(
+        accessLogWriter: AccessLogWriter,
+        private val wsRequest: WebsocketInbound,
+        private val blockchain: Chain
+    ) : RequestHandler, AbstractRequestHandler(accessLogWriter, Events.Channel.WSJSONRPC) {
+
+        override fun close() {
+            if (request == null) {
+                return
+            }
+            val builder = EventsBuilder.NativeCall()
+            builder.withChain(blockchain.id)
+            builder.start(wsRequest)
+            builder.onRequest(request!!)
+            onClose(builder)
+        }
+    }
+
+    class WsSubscriptionHandler(
+        private val accessLogWriter: AccessLogWriter,
+        private val wsRequest: WebsocketInbound,
+        private val blockchain: Chain
+    ) : SubscriptionHandler {
+
+        private var builder: EventsBuilder.NativeSubscribeHttp? = null
+
+        override fun onRequest(request: Pair<String, ByteArray?>) {
+            val builder = EventsBuilder.NativeSubscribeHttp(Events.Channel.WSJSONRPC, blockchain)
+            builder.start(wsRequest)
+            builder.onRequest(request)
+            this.builder = builder
+        }
+
+        override fun onResponse(msgSize: Long) {
+            builder
+                ?.onReply(msgSize)
+                ?.let(accessLogWriter::submit)
+        }
     }
 }

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

@@ -28,7 +28,7 @@ class Events {
     }
 
     enum class Channel {
-        GRPC, JSONRPC
+        GRPC, JSONRPC, WSJSONRPC
     }
 
     abstract class Base(

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

@@ -21,9 +21,11 @@ import io.emeraldpay.grpc.Chain
 import io.grpc.Attributes
 import io.grpc.Grpc
 import io.grpc.Metadata
+import io.netty.handler.codec.http.HttpHeaders
 import org.apache.commons.lang3.StringUtils
 import org.slf4j.LoggerFactory
 import reactor.netty.http.server.HttpServerRequest
+import reactor.netty.http.websocket.WebsocketInbound
 import java.net.InetAddress
 import java.net.InetSocketAddress
 import java.time.Instant
@@ -44,12 +46,16 @@ class EventsBuilder {
         fun start(request: HttpServerRequest)
     }
 
+    interface StartingWsRequest {
+        fun start(request: WebsocketInbound)
+    }
+
     interface RequestReply<E, Req, Resp> : StartingHttp2Request {
         fun onRequest(msg: Req)
         fun onReply(msg: Resp): E
     }
 
-    abstract class Base<T> : StartingHttp2Request, StartingHttp1Request {
+    abstract class Base<T> : StartingHttp2Request, StartingHttp1Request, StartingWsRequest {
         companion object {
             private val remoteIpHeaders = listOf(
                 "x-real-ip",
@@ -136,17 +142,9 @@ class EventsBuilder {
 
         override fun start(request: HttpServerRequest) {
             val headers = request.requestHeaders()
-            val userAgent = headers.get("user-agent")
+            val userAgent = getUserAgent(headers)
-                ?.let(this@Base::clean)
-                ?: ""
             val ips = ArrayList<InetAddress>()
-            remoteIpHeaders.forEach { key ->
+            extractIps(headers, ips)
-                headers.get(key)?.let {
-                    it.trim().ifEmpty { null }
-                        ?.let(this@Base::toInetAddress)
-                        ?.let(ips::add)
-                }
-            }
             request.remoteAddress()?.let { addr ->
                 ips.add(addr.address)
             }
@@ -161,6 +159,53 @@ class EventsBuilder {
                 )
         }
 
+        override fun start(request: WebsocketInbound) {
+            val headers = request.headers()
+            val userAgent = getUserAgent(headers)
+            val ips = ArrayList<InetAddress>()
+            extractIps(headers, ips)
+            // class WebsocketServerOperations, which is an implementation for the Websocket server connection, has a remoteAddress method
+            // But the class, and it's parent HttpServerOperations, are both private and cannot be used directly,
+            // so we try to access the field via reflection when it's possible
+            val remoteAddress: InetSocketAddress? = request.javaClass.methods
+                .find { it.name == "remoteAddress" }
+                ?.let {
+                    if (it.canAccess(request) || it.trySetAccessible()) {
+                        it.invoke(request) as InetSocketAddress
+                    } else {
+                        null
+                    }
+                }
+            remoteAddress?.let { addr ->
+                ips.add(addr.address)
+            }
+            val ip = findBestIp(ips)?.hostAddress ?: ""
+            this.requestDetails = this.requestDetails
+                .copy(
+                    remote = Events.Remote(
+                        ips = ips.map { it.hostAddress },
+                        ip = ip,
+                        userAgent = userAgent
+                    )
+                )
+        }
+
+        fun getUserAgent(headers: HttpHeaders): String {
+            return headers.get("user-agent")
+                ?.let(this@Base::clean)
+                ?: ""
+        }
+
+        fun extractIps(headers: HttpHeaders, ips: MutableList<InetAddress>) {
+            remoteIpHeaders.forEach { key ->
+                headers.get(key)?.let {
+                    it.trim().ifEmpty { null }
+                        ?.let(this@Base::toInetAddress)
+                        ?.let(ips::add)
+                }
+            }
+        }
+
         fun withChain(chain: Int): T {
             this.chainId = chain
             this.chain = Chain.byId(chainId)
@@ -301,7 +346,9 @@ class EventsBuilder {
         }
     }
 
-    class NativeSubscribe :
+    class NativeSubscribe(
+        val channel: Events.Channel
+    ) :
         Base<NativeSubscribe>(),
         RequestReply<Events.NativeSubscribe, BlockchainOuterClass.NativeSubscribeRequest, BlockchainOuterClass.NativeSubscribeReplyItem> {
         var item: Events.NativeSubscribeItemDetails? = null
@@ -331,6 +378,42 @@ class EventsBuilder {
         }
     }
 
+    class NativeSubscribeHttp(
+        val channel: Events.Channel,
+        chain: Chain,
+    ) :
+        Base<NativeSubscribeHttp>(),
+        RequestReply<Events.NativeSubscribe, Pair<String, ByteArray?>, Long> {
+        var item: Events.NativeSubscribeItemDetails? = null
+        val replies = HashMap<Int, Events.NativeSubscribeReplyDetails>()
+
+        init {
+            withChain(chain.id)
+        }
+
+        override fun getT(): NativeSubscribeHttp {
+            return this
+        }
+
+        override fun onRequest(msg: Pair<String, ByteArray?>) {
+            this.item = Events.NativeSubscribeItemDetails(
+                msg.first,
+                msg.second?.size?.toLong() ?: 0L
+            )
+        }
+
+        override fun onReply(msg: Long): Events.NativeSubscribe {
+            return Events.NativeSubscribe(
+                request = requestDetails,
+                blockchain = chain,
+                nativeSubscribe = item!!,
+                payloadSizeBytes = msg,
+                id = UUID.randomUUID(),
+                channel = channel
+            )
+        }
+    }
+
     class Describe :
         Base<Describe>(),
         RequestReply<Events.Describe, BlockchainOuterClass.DescribeRequest, BlockchainOuterClass.DescribeResponse> {

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.dshackle.rpc.NativeSubscribe
-import io.emeraldpay.etherjar.rpc.RpcException
 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
 
@@ -54,11 +41,12 @@ import kotlin.concurrent.write
  */
 class ProxyServer(
     private var config: ProxyConfig,
-    private val readRpcJson: ReadRpcJson,
+    readRpcJson: ReadRpcJson,
-    private val writeRpcJson: WriteRpcJson,
+    writeRpcJson: WriteRpcJson,
-    private val nativeCall: NativeCall,
+    nativeCall: NativeCall,
+    nativeSubscribe: NativeSubscribe,
     private val tlsSetup: TlsSetup,
-    private val accessHandler: AccessHandlerHttp.HandlerFactory
+    accessHandler: AccessHandlerHttp.HandlerFactory
 ) {
 
     companion object {
@@ -92,12 +80,20 @@ class ProxyServer(
         StandardRequestMetrics()
     }
 
+    private val httpHandler = HttpHandler(readRpcJson, writeRpcJson, nativeCall, accessHandler, requestMetrics)
+    private val wsHandler: WebsocketHandler? = if (config.websocketEnabled) {
+        WebsocketHandler(readRpcJson, writeRpcJson, nativeCall, nativeSubscribe, accessHandler, requestMetrics)
+    } else null
+
     fun start() {
         if (!config.enabled) {
             log.debug("Proxy server is not enabled")
             return
         }
-        log.info("Listening Proxy on ${config.host}:${config.port}")
+        log.info("Start HTTP JSON RPC Proxy on ${connectAddress("http")}")
+        if (config.websocketEnabled) {
+            log.info("Start Websocket JSON RPC Proxy on ${connectAddress("ws")}")
+        }
         var serverBuilder = HttpServer.create()
             .doOnChannelInit { _, channel, _ ->
                 channel.pipeline().addFirst(errorHandler)
@@ -116,88 +112,17 @@ class ProxyServer(
 
     fun setupRoutes(routes: HttpServerRoutes) {
         config.routes.forEach { routeConfig ->
-            routes.post("/" + routeConfig.id, proxy(routeConfig))
+            // TODO implement a manual handling of the routes and WS upgrade to have a better control over the connection and improve the access logging
+            routes.post("/" + routeConfig.id, httpHandler.proxy(routeConfig))
+            if (config.websocketEnabled && wsHandler != null) {
+                routes.ws("/" + routeConfig.id, wsHandler.proxy(routeConfig))
+            }
         }
     }
 
-    fun execute(chain: Chain, call: ProxyCall, handler: AccessHandlerHttp.RequestHandler): Publisher<String> {
+    fun connectAddress(baseSchema: String): String {
-        // return empty response for empty request
+        val schema = if (config.tls != null) baseSchema + "s" else baseSchema
-        if (call.items.isEmpty()) {
+        return "$schema://${config.host}:${config.port}"
-            return if (call.type == ProxyCall.RpcType.BATCH) {
-                Mono.just("[]")
-            } else {
-                Mono.just("")
-            }
-        }
-        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 = extract(type, data)
-                list = objectMapper.readerFor(MutableList::class.java).readValue(data)
+            return convertMapToNativeCall(type, list)
-            } 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
         } 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,215 @@
+/**
+ * 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 com.google.protobuf.ByteString
+import io.emeraldpay.api.proto.BlockchainOuterClass
+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.reactivestreams.Publisher
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.core.publisher.Sinks
+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,
+    private val accessHandler: AccessHandlerHttp.HandlerFactory,
+    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 n.toString(16)
+    }
+
+    fun proxy(routeConfig: ProxyConfig.Route): BiFunction<WebsocketInbound, WebsocketOutbound, Publisher<Void>> {
+        return BiFunction { req, resp ->
+            // each connection keeps a list of subscription controllers
+            val control = HashMap<String, Sinks.One<Boolean>>()
+
+            val requests: Flux<RequestJson<Any>> = req.aggregateFrames()
+                .receiveFrames()
+                .map { ByteBufInputStream(it.content()).readAllBytes() }
+                .flatMap(this@WebsocketHandler::parseRequest)
+
+            val eventHandler = accessHandler.start(req, routeConfig.blockchain)
+
+            val responses = respond(routeConfig.blockchain, control, 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,
+        control: MutableMap<String, Sinks.One<Boolean>>,
+        requests: Flux<RequestJson<Any>>,
+        eventHandlerFactory: AccessHandlerHttp.WsHandlerFactory
+    ): Flux<String> {
+        return requests.flatMap { call ->
+            val method = call.method
+
+            if (method == "eth_subscribe") {
+                val methodParams = splitMethodParams(call.params)
+                if (methodParams != null) {
+                    val eventHandler: AccessHandlerHttp.SubscriptionHandler = eventHandlerFactory.subscribe()
+                    val subscriptionId = nextSubscriptionId()
+                    eventHandler.onRequest(
+                        methodParams.let { mp ->
+                            // TODO ineffective to encode the params each time just to get size, ideally should get a reference to the original JSON bytes
+                            // but it doesn't happen very ofter, only on initial subscribe only for logs with filter
+                            Pair(mp.first, mp.second?.let { Global.objectMapper.writeValueAsBytes(it) })
+                        }
+                    )
+                    val currentControl = Sinks.one<Boolean>()
+                    control[subscriptionId] = currentControl
+                    // 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))
+                        }
+                        .takeUntilOther(currentControl.asMono())
+                    Flux.concat(Mono.just(start), responses)
+                        .map { Global.objectMapper.writeValueAsString(it) }
+                        .doOnNext {
+                            eventHandler.onResponse(it.length.toLong())
+                        }
+                } else {
+                    // TODO should it produce a 404 to the AccessLog?
+                    Mono.empty()
+                }
+            } else if (method == "eth_unsubscribe") {
+                val id = call.params?.getOrNull(0) ?: ""
+
+                // put it to the Access Log with fake id=0 (it doesn't matter, except the later reference)
+                val eventHandler: AccessHandlerHttp.RequestHandler = eventHandlerFactory.call()
+                eventHandler.onRequest(
+                    BlockchainOuterClass.NativeCallRequest.newBuilder()
+                        .setChainValue(blockchain.id)
+                        .addItems(
+                            BlockchainOuterClass.NativeCallItem.newBuilder()
+                                .setId(0)
+                                .setMethod("eth_unsubscribe")
+                                .setPayload(ByteString.copyFromUtf8("[\"$id\"]"))
+                                .build()
+                        )
+                        .build()
+                )
+
+                val p = control.remove(id.toString())
+                val success = p?.tryEmitValue(true)?.isSuccess ?: false
+                val response = ResponseJson<Boolean, Any>().also {
+                    it.id = call.id
+                    it.result = success
+                }
+                Mono.just(response)
+                    .map { Global.objectMapper.writeValueAsString(it) }
+                    .doOnNext { eventHandler.onResponse(NativeCall.CallResult.ok(0, it.toByteArray())) }
+                    .doFinally { eventHandler.close() }
+            } else {
+                val eventHandler: AccessHandlerHttp.RequestHandler = eventHandlerFactory.call()
+                val proxyCall = readRpcJson.convertToNativeCall(ProxyCall.RpcType.SINGLE, listOf(call))
+                Mono.from(execute(blockchain, proxyCall, eventHandler))
+                    // thought the event handler is used in execute
+                    // it still needs to be closed at the end, so it can render the logs
+                    .doFinally { eventHandler.close() }
+            }
+        }
+    }
+
+    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/proxy/WriteRpcJson.kt

@@ -63,9 +63,6 @@ open class WriteRpcJson {
                         Mono.empty()
                     }
                 }
-                .onErrorContinue { t, _ ->
-                    log.warn("Failed to convert to JSON", t)
-                }
         }
     }
 

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

src/test/groovy/io/emeraldpay/dshackle/config/ProxyConfigReaderSpec.groovy

@@ -30,6 +30,7 @@ class ProxyConfigReaderSpec extends Specification {
 
         then:
         act.enabled
+        act.websocketEnabled
         act.port == 8080
         act.host == '127.0.0.1'
         act.routes.size() == 1
@@ -39,6 +40,17 @@ class ProxyConfigReaderSpec extends Specification {
         }
     }
 
+    def "Read proxy config with websocket disabled"() {
+        setup:
+        def config = this.class.getClassLoader().getResourceAsStream("dshackle-proxy-no-ws.yaml")
+        when:
+        def act = reader.read(config)
+
+        then:
+        act.enabled
+        !act.websocketEnabled
+    }
+
     def "Read proxy config with two elements"() {
         setup:
         def config = this.class.getClassLoader().getResourceAsStream("dshackle-proxy-two.yaml")

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

src/test/groovy/io/emeraldpay/dshackle/proxy/HttpHandlerSpec.groovy

@@ -0,0 +1,123 @@
+/**
+ * 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 com.google.protobuf.ByteString
+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.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.emeraldpay.etherjar.rpc.RpcException
+import io.emeraldpay.grpc.Chain
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.test.StepVerifier
+import spock.lang.Specification
+
+import java.time.Duration
+import java.util.function.Function
+
+class HttpHandlerSpec extends Specification {
+
+    def "Calls access log handler"() {
+        setup:
+        def reqItem = BlockchainOuterClass.NativeCallItem.newBuilder()
+                .setId(1)
+                .setMethod("test_test")
+                .setPayload(ByteString.copyFromUtf8("[]"))
+                .build()
+        def respItem = new NativeCall.CallResult(1, "100".bytes, null)
+        def req = BlockchainOuterClass.NativeCallRequest.newBuilder()
+                .setChain(Common.ChainRef.CHAIN_ETHEREUM)
+                .addItems(reqItem)
+                .build()
+
+        NativeCall nativeCall = Mock(NativeCall) {
+            1 * nativeCallResult(_) >> Flux.fromIterable([respItem])
+        }
+        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:
+        handler.execute(Chain.ETHEREUM, [reqItem], accessHandler)
+                .blockLast()
+
+        then:
+        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))
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/proxy/ProxyServerSpec.groovy

@@ -1,136 +1,92 @@
-/**
- * Copyright (c) 2020 ETCDEV GmbH
- * Copyright (c) 2020 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 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.AuthConfig
 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.rpc.NativeSubscribe
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
-import io.emeraldpay.etherjar.rpc.RpcException
 import io.emeraldpay.grpc.Chain
-import reactor.core.publisher.Flux
+import reactor.netty.http.server.HttpServerRoutes
-import reactor.core.publisher.Mono
-import reactor.test.StepVerifier
 import spock.lang.Specification
 
-import java.time.Duration
-import java.util.function.Function
-
 class ProxyServerSpec extends Specification {
 
-    def "Uses NativeCall"() {
+    def "Setup routes"() {
         setup:
-        NativeCall nativeCall = Mock(NativeCall)
+        def config1 = new ProxyConfig()
-        def predefined = { a -> Flux.just("hello") } as Function
+        config1.routes = [
-
+                new ProxyConfig.Route("test", Chain.ETHEREUM)
-        WriteRpcJson writeRpcJson = Mock {
+        ]
-            1 * toJsons(_) >> predefined
+        def proxyServer = new ProxyServer(
-        }
+                config1,
-
+                new ReadRpcJson(), new WriteRpcJson(),
-        ProxyServer server = new ProxyServer(
+                Stub(NativeCall), Stub(NativeSubscribe),
-                new ProxyConfig(),
+                Stub(TlsSetup), new AccessHandlerHttp.NoOpFactory()
-                new ReadRpcJson(),
-                writeRpcJson,
-                nativeCall,
-                new TlsSetup(TestingCommons.fileResolver()),
-                new AccessHandlerHttp.NoOpFactory()
         )
 
-        def call = new ProxyCall(ProxyCall.RpcType.SINGLE)
+        def routes = Mock(HttpServerRoutes)
-        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())
+        proxyServer.setupRoutes(routes)
 
         then:
-        1 * nativeCall.nativeCallResult(_) >> Flux.just(new NativeCall.CallResult(1, "".bytes, null))
+        1 * routes.post("/test", _)
-        StepVerifier.create(act)
+        1 * routes.ws("/test", _)
-                .expectNext("hello")
-                .expectComplete()
-                .verify(Duration.ofSeconds(1))
     }
 
-    def "Return error on invalid request"() {
+    def "Setup routes when WS is disabled"() {
         setup:
-        ReadRpcJson read = Mock(ReadRpcJson) {
+        def config1 = new ProxyConfig()
-            1 * apply(_) >> { throw new RpcException(-32123, "test", new JsonRpcResponse.NumberId(4)) }
+        config1.websocketEnabled = false
-        }
+        config1.routes = [
-        def server = new ProxyServer(
+                new ProxyConfig.Route("test", Chain.ETHEREUM)
-                Stub(ProxyConfig),
+        ]
-                read,
+        def proxyServer = new ProxyServer(
-                Stub(WriteRpcJson), Stub(NativeCall), Stub(TlsSetup),
+                config1,
-                new AccessHandlerHttp.NoOpFactory()
+                new ReadRpcJson(), new WriteRpcJson(),
+                Stub(NativeCall), Stub(NativeSubscribe),
+                Stub(TlsSetup), new AccessHandlerHttp.NoOpFactory()
         )
+
+        def routes = Mock(HttpServerRoutes)
         when:
-        def act = server.processRequest(Chain.ETHEREUM, Mono.just("".bytes), new AccessHandlerHttp.NoOpHandler())
+        proxyServer.setupRoutes(routes)
-                .map { new String(it.array()) }
+
         then:
-        StepVerifier.create(act)
+        1 * routes.post("/test", _)
-                .expectNext('{"jsonrpc":"2.0","id":4,"error":{"code":-32123,"message":"test"}}')
+        0 * routes.ws(_, _)
-                .expectComplete()
-                .verify(Duration.ofSeconds(1))
     }
 
-    def "Calls access log handler"() {
+    def "Generate Connect Address"() {
-        setup:
+        def config1 = new ProxyConfig()
-        def reqItem = BlockchainOuterClass.NativeCallItem.newBuilder()
+        config1.host = "192.168.0.1"
-                .setId(1)
+        config1.port = 1000
-                .setMethod("test_test")
+        def proxyServer = new ProxyServer(
-                .setPayload(ByteString.copyFromUtf8("[]"))
+                config1,
-                .build()
+                new ReadRpcJson(), new WriteRpcJson(),
-        def respItem = new NativeCall.CallResult(1, "100".bytes, null)
+                Stub(NativeCall), Stub(NativeSubscribe),
-        def req = BlockchainOuterClass.NativeCallRequest.newBuilder()
+                Stub(TlsSetup), new AccessHandlerHttp.NoOpFactory()
-                .setChain(Common.ChainRef.CHAIN_ETHEREUM)
-                .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()
         )
-
         when:
-        server.processRequest(Chain.ETHEREUM, Mono.just("".bytes), handler)
+        def act = proxyServer.connectAddress("http")
-                .blockLast()
-
         then:
-        1 * handler.onRequest(req)
+        act == "http://192.168.0.1:1000"
-        1 * handler.onResponse(respItem)
+    }
+
+    def "Generate Connect Address with TLS"() {
+        def config1 = new ProxyConfig()
+        config1.host = "192.168.0.1"
+        config1.port = 1000
+        config1.tls = new AuthConfig.ServerTlsAuth()
+        def proxyServer = new ProxyServer(
+                config1,
+                new ReadRpcJson(), new WriteRpcJson(),
+                Stub(NativeCall), Stub(NativeSubscribe),
+                Stub(TlsSetup), new AccessHandlerHttp.NoOpFactory()
+        )
+        when:
+        def act = proxyServer.connectAddress("ws")
+        then:
+        act == "wss://192.168.0.1:1000"
     }
 }

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

src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy

@@ -0,0 +1,157 @@
+/**
+ * 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 reactor.core.publisher.Sinks
+import spock.lang.Specification
+
+import java.time.Duration
+
+class WebsocketHandlerSpec extends Specification {
+
+    def requestHandlerFactory = new AccessHandlerHttp.NoOpFactory()
+    def requestHandler = new AccessHandlerHttp.NoOpHandler()
+
+    def "Parse standard RPC request"() {
+        setup:
+        def handler = new WebsocketHandler(
+                new ReadRpcJson(), Stub(WriteRpcJson), Stub(NativeCall), Stub(NativeSubscribe), requestHandlerFactory, 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), requestHandlerFactory, 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), requestHandlerFactory, 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), requestHandlerFactory, Stub(ProxyServer.RequestMetricsFactory)
+        )
+
+        def request = new RequestJson("foo_test", [], 2)
+        when:
+        def act = handler.respond(Chain.ETHEREUM, new HashMap<String, Sinks.One<Boolean>>(), 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, requestHandlerFactory, Stub(ProxyServer.RequestMetricsFactory)
+        )
+
+        def request = new RequestJson("eth_subscribe", ["foo_test"], 2)
+        when:
+        def act = handler.respond(Chain.ETHEREUM, new HashMap<String, Sinks.One<Boolean>>(), Flux.just(request), requestHandler)
+                .collectList()
+                .block(Duration.ofSeconds(1))
+        then:
+        act[0] == '{"jsonrpc":"2.0","id":2,"result":"1"}'
+        act[1] == '{"jsonrpc":"2.0","method":"eth_subscription","params":{"result":{"foo":1},"subscription":"1"}}'
+        act[2] == '{"jsonrpc":"2.0","method":"eth_subscription","params":{"result":{"foo":2},"subscription":"1"}}'
+        act.size() == 3
+    }
+
+    def "Unsubscribe"() {
+        setup:
+
+        def handler = new WebsocketHandler(
+                new ReadRpcJson(), new WriteRpcJson(), Stub(NativeCall), Stub(NativeSubscribe), requestHandlerFactory, Stub(ProxyServer.RequestMetricsFactory)
+        )
+
+        def control = new HashMap<String, Sinks.One<Boolean>>()
+        Sinks.One<Boolean> sink = Sinks.one();
+        control["5"] = sink
+        def request = new RequestJson("eth_unsubscribe", ["5"], 0)
+        when:
+        def act = handler.respond(Chain.ETHEREUM, control, Flux.just(request), requestHandler)
+                .single()
+                .block(Duration.ofSeconds(1))
+        def sinkResponse = sink.asMono().block()
+        then:
+        act == '{"jsonrpc":"2.0","id":0,"result":true}'
+        sinkResponse != null
+    }
+
+    def "Unsubscribe when no subscription"() {
+        setup:
+
+        def handler = new WebsocketHandler(
+                new ReadRpcJson(), new WriteRpcJson(), Stub(NativeCall), Stub(NativeSubscribe), requestHandlerFactory, Stub(ProxyServer.RequestMetricsFactory)
+        )
+
+        def control = new HashMap<String, Sinks.One<Boolean>>()
+        def request = new RequestJson("eth_unsubscribe", ["5"], 0)
+        when:
+        def act = handler.respond(Chain.ETHEREUM, control, Flux.just(request), requestHandler)
+                .single()
+                .block(Duration.ofSeconds(1))
+        then:
+        act == '{"jsonrpc":"2.0","id":0,"result":false}'
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/proxy/WriteRpcJsonSpec.groovy

@@ -16,10 +16,8 @@
  */
 package io.emeraldpay.dshackle.proxy
 
-import com.google.protobuf.ByteString
-import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.rpc.NativeCall
-import io.emeraldpay.dshackle.test.TestingCommons
+import org.jetbrains.annotations.NotNull
 import reactor.core.publisher.Flux
 import spock.lang.Specification
 
@@ -143,4 +141,27 @@ class WriteRpcJsonSpec extends Specification {
         act[1] == '{"jsonrpc":"2.0","id":11,"error":{"code":-32002,"message":"oops"}}'
         act[2] == '{"jsonrpc":"2.0","id":15,"result":{"hash": "0x2484f459dc"}}'
     }
+
+    def "Write JSON RPC error on exception"() {
+        setup:
+        def writer = new WriteRpcJson() {
+            @Override
+            String toJson(@NotNull ProxyCall call, @NotNull NativeCall.CallResult response) {
+                throw new NativeCall.CallFailure(1, new IllegalStateException("TEST"))
+            }
+        }
+
+        def call = new ProxyCall(ProxyCall.RpcType.SINGLE)
+        call.ids[1] = 10
+        def data = [
+                new NativeCall.CallResult(1, '"0x1"'.bytes, null),
+        ]
+        when:
+        def act = Flux.fromIterable(data)
+                .transform(writer.toJsons(call))
+                .collectList()
+                .block(Duration.ofSeconds(1))
+        then:
+        act[0] == '{"jsonrpc":"2.0","id":10,"error":{"code":-32003,"message":"TEST"}}'
+    }
 }

src/test/resources/dshackle-proxy-no-ws.yaml

@@ -0,0 +1,6 @@
+proxy:
+  port: 8080
+  websocket: false
+  routes:
+    - id: ethereum
+      blockchain: ethereum