@@ -28,7 +28,7 @@ import org.springframework.beans.factory.annotation.Autowired
|
||||
import org.springframework.stereotype.Service
|
||||
|
||||
@Service
|
||||
class TlsSetup(
|
||||
open class TlsSetup(
|
||||
@Autowired val fileResolver: FileResolver
|
||||
) {
|
||||
|
||||
|
||||
@@ -21,7 +21,7 @@ import io.emeraldpay.grpc.Chain
|
||||
/**
|
||||
* Configure HTTP Proxy to Upstreams
|
||||
*/
|
||||
class ProxyConfig {
|
||||
open class ProxyConfig {
|
||||
|
||||
companion object {
|
||||
public const val CONFIG_ID = "parsed.proxy"
|
||||
|
||||
@@ -18,14 +18,19 @@ 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.rpc.NativeCall
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
|
||||
import io.infinitape.etherjar.rpc.RpcException
|
||||
import io.netty.buffer.ByteBuf
|
||||
import io.netty.buffer.Unpooled
|
||||
import io.netty.handler.ssl.SslContextBuilder
|
||||
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.DisposableServer
|
||||
import reactor.netty.http.server.HttpServer
|
||||
@@ -90,15 +95,27 @@ class ProxyServer(
|
||||
}
|
||||
}
|
||||
|
||||
fun processRequest(chain: Common.ChainRef, request: Mono<ByteArray>): Flux<ByteBuf> {
|
||||
return request.map(readRpcJson)
|
||||
.flatMapMany { call -> execute(chain, call) }
|
||||
.onErrorResume(RpcException::class.java) { err ->
|
||||
val id = err.details?.let {
|
||||
if (it is JsonRpcResponse.Id) it else JsonRpcResponse.IntId(-1)
|
||||
} ?: JsonRpcResponse.IntId(-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>> {
|
||||
val chain = Common.ChainRef.forNumber(routeConfig.blockchain.id)
|
||||
return BiFunction { req, resp ->
|
||||
val results = req.receive()
|
||||
val request = req.receive()
|
||||
.aggregate()
|
||||
.asByteArray()
|
||||
.map(readRpcJson)
|
||||
.flatMapMany { call -> execute(chain, call) }
|
||||
.map { Unpooled.wrappedBuffer(it.toByteArray()) }
|
||||
val results = processRequest(chain, request)
|
||||
resp.addHeader(HttpHeaders.CONTENT_TYPE, "application/json")
|
||||
.send(results)
|
||||
}
|
||||
|
||||
@@ -20,6 +20,7 @@ import com.fasterxml.jackson.databind.ObjectMapper
|
||||
import com.google.protobuf.ByteString
|
||||
import io.emeraldpay.api.proto.BlockchainOuterClass
|
||||
import io.emeraldpay.dshackle.Global
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
|
||||
import io.infinitape.etherjar.rpc.RpcException
|
||||
import io.infinitape.etherjar.rpc.RpcResponseError
|
||||
import io.infinitape.etherjar.rpc.json.RequestJson
|
||||
@@ -36,8 +37,7 @@ import java.util.stream.Collectors
|
||||
* Reader for JSON RPC request
|
||||
*/
|
||||
@Service
|
||||
open class ReadRpcJson(
|
||||
) : Function<ByteArray, ProxyCall> {
|
||||
open class ReadRpcJson() : Function<ByteArray, ProxyCall> {
|
||||
|
||||
companion object {
|
||||
private val log = LoggerFactory.getLogger(ReadRpcJson::class.java)
|
||||
@@ -49,18 +49,21 @@ open class ReadRpcJson(
|
||||
|
||||
init {
|
||||
jsonExtractor = Function { json ->
|
||||
if ("2.0" != json["jsonrpc"]) {
|
||||
throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "Unsupported JSON RPC version")
|
||||
}
|
||||
if (json["id"] == null) {
|
||||
throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "ID not set")
|
||||
throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "ID is not set")
|
||||
}
|
||||
val id = json["id"]
|
||||
if ("2.0" != json["jsonrpc"]) {
|
||||
if (json["jsonrpc"] == null) {
|
||||
throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "jsonrpc version is not set", id?.let { JsonRpcResponse.Id.from(it) })
|
||||
}
|
||||
throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "Unsupported JSON RPC version: " + json["jsonrpc"].toString(), id?.let { JsonRpcResponse.Id.from(it) })
|
||||
}
|
||||
if (!(json["method"] != null && json["method"] is String)) {
|
||||
throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "ID not set")
|
||||
throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "Method is not set", id?.let { JsonRpcResponse.Id.from(it) })
|
||||
}
|
||||
if (json.containsKey("params") && json["params"] !is List<*>) {
|
||||
throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "Params must be an array")
|
||||
throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "Params must be an array", id?.let { JsonRpcResponse.Id.from(it) })
|
||||
}
|
||||
RequestJson<Any>(
|
||||
json["method"].toString(),
|
||||
|
||||
@@ -164,6 +164,19 @@ class JsonRpcResponse(
|
||||
override fun isInt(): Boolean {
|
||||
return true
|
||||
}
|
||||
|
||||
override fun equals(other: Any?): Boolean {
|
||||
if (this === other) return true
|
||||
if (other !is IntId) return false
|
||||
|
||||
if (id != other.id) return false
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
override fun hashCode(): Int {
|
||||
return id
|
||||
}
|
||||
}
|
||||
|
||||
class StringId(val id: String) : Id {
|
||||
@@ -178,6 +191,20 @@ class JsonRpcResponse(
|
||||
override fun isInt(): Boolean {
|
||||
return false
|
||||
}
|
||||
|
||||
override fun equals(other: Any?): Boolean {
|
||||
if (this === other) return true
|
||||
if (other !is StringId) return false
|
||||
|
||||
if (id != other.id) return false
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
override fun hashCode(): Int {
|
||||
return id.hashCode()
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
class ResponseJsonSerializer : JsonSerializer<JsonRpcResponse>() {
|
||||
|
||||
Reference in New Issue
Block a user