StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

solution: basic implementation for Proxy endpoint

Browse Source
This commit is contained in:
Igor Artamonov
2020-03-19 23:34:32 -04:00
parent 7455949541
commit 26f8dd294e
27 changed files with 1309 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
src/main/kotlin/io/emeraldpay/dshackle/DshackleEnvironment.kt
View File

@@ -15,6 +15,8 @@
*/
package io.emeraldpay.dshackle
import io.emeraldpay.dshackle.config.ProxyConfig
import io.emeraldpay.dshackle.config.ProxyConfigReader
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.config.YamlPropertiesFactoryBean
import org.springframework.core.env.*
@@ -23,7 +25,6 @@ import org.springframework.core.io.Resource
import org.springframework.core.io.support.ResourcePropertySource
import java.io.File
import java.util.*
import javax.annotation.PostConstruct
const val DEFAULT_CONFIG = "/etc/dshackle/dshackle.yaml"
const val LOCAL_CONFIG = "./dshackle.yaml"
@@ -40,18 +41,25 @@ open class DshackleEnvironment: StandardEnvironment() {
propertySources.addLast(ResourcePropertySource("version.properties"))
}
open fun mainConfig(): PropertySource<*> {
open fun getResource(): File? {
var target = File(DEFAULT_CONFIG)
if (!isAcceptedConfig(target)) {
target = File(LOCAL_CONFIG)
if (!isAcceptedConfig(target)) {
log.error("Configuration is not found neither at $DEFAULT_CONFIG nor $LOCAL_CONFIG")
return PropertySource.named("mainConfig")
return null
}
}
target = target.normalize()
val loadedProperties = this.loadYaml(FileSystemResource(target))
return target
}
open fun mainConfig(): PropertySource<*> {
val target = getResource() ?: return PropertySource.named("mainConfig")
val resource = FileSystemResource(target)
val loadedProperties = this.loadYaml(resource)
loadedProperties["configPath"] = target.absolutePath
loadedProperties[ProxyConfig.CONFIG_ID] = extractYamlProxy(resource)
return PropertiesPropertySource("mainConfig", loadedProperties)
}
@@ -62,6 +70,11 @@ open class DshackleEnvironment: StandardEnvironment() {
return factory.getObject()!!
}
protected fun extractYamlProxy(resource: Resource): ProxyConfig? {
val reader = ProxyConfigReader()
return reader.read(resource.inputStream)
}
protected fun isAcceptedConfig(target: File): Boolean {
return target.exists() && target.isFile
}

2
src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt
View File

@@ -45,7 +45,7 @@ open class GrpcServer(
log.info("Starting GRPC Server...")
log.debug("Running with DEBUG LOGGING")
val port = env.getProperty("port", "2449").toInt()
log.info("Listening on 0.0.0.0:$port")
log.info("Listening Native on 0.0.0.0:$port")
val serverBuilder = NettyServerBuilder.forPort(port)
rpcs.forEach {
serverBuilder.addService(it)

55
src/main/kotlin/io/emeraldpay/dshackle/ProxyStarter.kt Normal file
View File

@@ -0,0 +1,55 @@
/**
* Copyright (c) 2020 ETCDEV GmbH
*
* 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
import io.emeraldpay.dshackle.config.ProxyConfig
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 org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.core.env.Environment
import org.springframework.stereotype.Service
import javax.annotation.PostConstruct
/**
* Starts HTTP proxy endpoint, if configured
*/
@Service
class ProxyStarter(
@Autowired private val env: Environment,
@Autowired private val readRpcJson: ReadRpcJson,
@Autowired private val writeRpcJson: WriteRpcJson,
@Autowired private val nativeCall: NativeCall
) {
companion object {
private val log = LoggerFactory.getLogger(ProxyStarter::class.java)
}
@PostConstruct
fun start() {
val config = env.getProperty(ProxyConfig.CONFIG_ID, ProxyConfig::class.java)
if (config == null) {
log.debug("Proxy server is not configured")
return
}
val server = ProxyServer(config, readRpcJson, writeRpcJson, nativeCall)
server.start()
}
}

28
src/main/kotlin/io/emeraldpay/dshackle/config/InvalidConfigException.kt Normal file
View File

@@ -0,0 +1,28 @@
/**
* Copyright (c) 2020 ETCDEV GmbH
*
* 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.config
import org.yaml.snakeyaml.error.Mark
open class InvalidConfigException(
message: String
) : Exception(message)
class InvalidConfigYamlException(
filename: String,
mark: Mark,
message: String
) : InvalidConfigException("Invalid YAML configuration ${message}, at ${filename}:${mark.line}")

56
src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfig.kt Normal file
View File

@@ -0,0 +1,56 @@
/**
* Copyright (c) 2020 ETCDEV GmbH
*
* 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.config
import io.emeraldpay.grpc.Chain
/**
* Configure HTTP Proxy to Upstreams
*/
class ProxyConfig {
companion object {
public const val CONFIG_ID = "parsed.proxy"
}
var enabled: Boolean = true
/**
* Host to bind server. Default: 127.0.0.1
*/
var host = "127.0.0.1"
/**
* Port to bind. Default: 8080
*/
var port: Int = 8080
/**
* List of available routes
*/
var routes: List<Route> = ArrayList()
class Route(
/**
* URL binding for the route. http://$host:$port/$id
*/
val id: String,
/**
* Blockchain to dispatch requests
*/
val blockchain: Chain
)
}

81
src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfigReader.kt Normal file
View File

@@ -0,0 +1,81 @@
/**
* Copyright (c) 2020 ETCDEV GmbH
*
* 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.config
import io.emeraldpay.grpc.Chain
import org.apache.commons.lang3.StringUtils
import org.slf4j.LoggerFactory
import org.yaml.snakeyaml.Yaml
import org.yaml.snakeyaml.nodes.MappingNode
import java.io.InputStream
import java.io.InputStreamReader
/**
* Read YAML config, part related to Proxy configuration
*/
class ProxyConfigReader : YamlConfigReader() {
companion object {
private val log = LoggerFactory.getLogger(ProxyConfigReader::class.java)
}
private var filename = "dshackle.yaml"
fun read(input: InputStream): ProxyConfig? {
val yaml = Yaml()
val configNode = asMappingNode(yaml.compose(InputStreamReader(input)))
return read(getMapping(configNode, "proxy"))
}
fun read(input: MappingNode?): ProxyConfig? {
if (input == null) {
return null
}
val config = ProxyConfig()
getValueAsString(input, "host")?.let {
config.host = it
}
getValueAsInt(input, "port")?.let {
config.port = it
}
getValueAsBool(input, "enabled")?.let {
config.enabled = it
}
val currentRoutes = HashSet<String>()
getList<MappingNode>(input, "routes")?.let { routes ->
config.routes = routes.value.map { route ->
val id = getValueAsString(route, "id")
if (id == null || StringUtils.isEmpty(id) || !StringUtils.isAlphanumeric(id)) {
throw InvalidConfigYamlException(filename, route.startMark, "Route id must be alphanumeric")
}
if (currentRoutes.contains(id)) {
throw InvalidConfigYamlException(filename, route.startMark, "Route id repeated: $id")
}
currentRoutes.add(id)
val blockchain = getValueAsString(route, "blockchain")
if (StringUtils.isEmpty(blockchain) || getBlockchain(blockchain!!) == Chain.UNSPECIFIED) {
throw InvalidConfigYamlException(filename, route.startMark, "Invalid blockchain or not specified")
}
ProxyConfig.Route(id, getBlockchain(blockchain))
}
}
if (config.routes.isEmpty()) {
return null
}
return config
}
}

90
src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt
View File

@@ -29,10 +29,9 @@ import java.lang.IllegalArgumentException
import java.net.URI
import java.time.Duration
class UpstreamsConfigReader {
class UpstreamsConfigReader : YamlConfigReader() {
private val log = LoggerFactory.getLogger(UpstreamsConfigReader::class.java)
private val envVariables = EnvVariables()
fun read(input: InputStream): UpstreamsConfig {
val yaml = Yaml()
@@ -218,91 +217,4 @@ class UpstreamsConfigReader {
}
}
private fun hasAny(mappingNode: MappingNode?, key: String): Boolean {
if (mappingNode == null) {
return false
}
return mappingNode.value
.stream()
.filter { n -> n.keyNode is ScalarNode }
.filter { n ->
val sn = n.keyNode as ScalarNode
key == sn.value
}.count() > 0
}
private fun <T> getValue(mappingNode: MappingNode?, key: String, type: Class<T>): T? {
if (mappingNode == null) {
return null
}
return mappingNode.value
.stream()
.filter { n -> n.keyNode is ScalarNode && type.isAssignableFrom(n.valueNode.javaClass) }
.filter { n ->
val sn = n.keyNode as ScalarNode
key == sn.value
}
.map { n -> n.valueNode as T }
.findFirst().let {
if (it.isPresent) {
it.get()
} else {
null
}
}
}
private fun getMapping(mappingNode: MappingNode?, key: String): MappingNode? {
return getValue(mappingNode, key, MappingNode::class.java)
}
private fun getValue(mappingNode: MappingNode?, key: String): ScalarNode? {
return getValue(mappingNode, key, ScalarNode::class.java)
}
private fun <T> getList(mappingNode: MappingNode?, key: String): CollectionNode<T>? {
val value = getValue(mappingNode, key, CollectionNode::class.java) ?: return null
return value as CollectionNode<T>
}
private fun getListOfString(mappingNode: MappingNode?, key: String): List<String>? {
return getList<ScalarNode>(mappingNode, key)?.value
?.map { it.value }
?.map(envVariables::postProcess)
}
private fun getValueAsString(mappingNode: MappingNode?, key: String): String? {
return getValue(mappingNode, key)?.let {
return@let it.value
}?.let(envVariables::postProcess)
}
private fun getValueAsInt(mappingNode: MappingNode?, key: String): Int? {
return getValue(mappingNode, key)?.let {
return@let if (it.isPlain) {
it.value.toIntOrNull()
} else {
null
}
}
}
private fun getValueAsBool(mappingNode: MappingNode?, key: String): Boolean? {
return getValue(mappingNode, key)?.let {
return@let if (it.isPlain) {
it.value?.toLowerCase() == "true"
} else {
null
}
}
}
private fun asMappingNode(node: Node): MappingNode {
return if (MappingNode::class.java.isAssignableFrom(node.javaClass)) {
node as MappingNode
} else {
throw IllegalArgumentException("Not a map")
}
}
}

123
src/main/kotlin/io/emeraldpay/dshackle/config/YamlConfigReader.kt Normal file
View File

@@ -0,0 +1,123 @@
/**
* Copyright (c) 2020 ETCDEV GmbH
*
* 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.config
import io.emeraldpay.grpc.Chain
import org.yaml.snakeyaml.nodes.CollectionNode
import org.yaml.snakeyaml.nodes.MappingNode
import org.yaml.snakeyaml.nodes.Node
import org.yaml.snakeyaml.nodes.ScalarNode
open class YamlConfigReader {
private val envVariables = EnvVariables()
protected fun hasAny(mappingNode: MappingNode?, key: String): Boolean {
if (mappingNode == null) {
return false
}
return mappingNode.value
.stream()
.filter { n -> n.keyNode is ScalarNode }
.filter { n ->
val sn = n.keyNode as ScalarNode
key == sn.value
}.count() > 0
}
private fun <T> getValue(mappingNode: MappingNode?, key: String, type: Class<T>): T? {
if (mappingNode == null) {
return null
}
return mappingNode.value
.stream()
.filter { n -> n.keyNode is ScalarNode && type.isAssignableFrom(n.valueNode.javaClass) }
.filter { n ->
val sn = n.keyNode as ScalarNode
key == sn.value
}
.map { n -> n.valueNode as T }
.findFirst().let {
if (it.isPresent) {
it.get()
} else {
null
}
}
}
protected fun getMapping(mappingNode: MappingNode?, key: String): MappingNode? {
return getValue(mappingNode, key, MappingNode::class.java)
}
private fun getValue(mappingNode: MappingNode?, key: String): ScalarNode? {
return getValue(mappingNode, key, ScalarNode::class.java)
}
protected fun <T> getList(mappingNode: MappingNode?, key: String): CollectionNode<T>? {
val value = getValue(mappingNode, key, CollectionNode::class.java) ?: return null
return value as CollectionNode<T>
}
protected fun getListOfString(mappingNode: MappingNode?, key: String): List<String>? {
return getList<ScalarNode>(mappingNode, key)?.value
?.map { it.value }
?.map(envVariables::postProcess)
}
protected fun getValueAsString(mappingNode: MappingNode?, key: String): String? {
return getValue(mappingNode, key)?.let {
return@let it.value
}?.let(envVariables::postProcess)
}
protected fun getValueAsInt(mappingNode: MappingNode?, key: String): Int? {
return getValue(mappingNode, key)?.let {
return@let if (it.isPlain) {
it.value.toIntOrNull()
} else {
null
}
}
}
protected fun getValueAsBool(mappingNode: MappingNode?, key: String): Boolean? {
return getValue(mappingNode, key)?.let {
return@let if (it.isPlain) {
it.value?.toLowerCase() == "true"
} else {
null
}
}
}
protected fun asMappingNode(node: Node): MappingNode {
return if (MappingNode::class.java.isAssignableFrom(node.javaClass)) {
node as MappingNode
} else {
throw IllegalArgumentException("Not a map")
}
}
// ----
fun getBlockchain(id: String): Chain {
return Chain.values().find { chain ->
chain.name == id.toUpperCase()
|| chain.chainCode.toUpperCase() == id.toUpperCase()
|| chain.id.toString() == id
} ?: Chain.UNSPECIFIED
}
}

57
src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyCall.kt Normal file
View File

@@ -0,0 +1,57 @@
/**
* Copyright (c) 2020 ETCDEV GmbH
*
* 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 org.slf4j.LoggerFactory
/**
* JSON RPC call to the proxy
*/
class ProxyCall(
/**
* Type of the request. The response format depends on it
*/
val type: RpcType
) {
companion object {
private val log = LoggerFactory.getLogger(ProxyCall::class.java)
}
/**
* Mapping from our internal ids to user provided JSON RPC ids.
*/
val ids = HashMap<Int, Any>()
/**
* Content of the request
*/
val items: MutableList<BlockchainOuterClass.NativeCallItem> = ArrayList()
enum class RpcType {
/**
* One item request passed as Object
*/
SINGLE,
/**
* Batch passed as Array of Object. It may be one-element array, i.e., single request, though response
* must be formatted as an Array
*/
BATCH
}
}

95
src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt Normal file
View File

@@ -0,0 +1,95 @@
/**
* Copyright (c) 2020 ETCDEV GmbH
*
* 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.config.ProxyConfig
import io.emeraldpay.dshackle.rpc.NativeCall
import io.netty.buffer.Unpooled
import org.reactivestreams.Publisher
import org.slf4j.LoggerFactory
import org.springframework.http.HttpHeaders
import reactor.core.publisher.Mono
import reactor.netty.DisposableServer
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.function.BiFunction
/**
* HTTP Proxy Server
*/
class ProxyServer(
private var config: ProxyConfig,
private val readRpcJson: ReadRpcJson,
private val writeRpcJson: WriteRpcJson,
private val nativeCall: NativeCall
) {
companion object {
private val log = LoggerFactory.getLogger(ProxyServer::class.java)
}
fun start() {
if (!config.enabled) {
log.debug("Proxy server is not enabled")
return
}
log.info("Listening Proxy on ${config.host}:${config.port}")
val server: DisposableServer = HttpServer.create()
.host(config.host)
.port(config.port)
.route(this::setupRoutes)
.bindNow()
}
fun setupRoutes(routes: HttpServerRoutes) {
config.routes.forEach { routeConfig ->
routes.post("/" + routeConfig.id, proxy(routeConfig))
}
}
fun execute(chain: Common.ChainRef, call: ProxyCall): Publisher<String> {
val request = BlockchainOuterClass.NativeCallRequest.newBuilder()
.setChain(chain)
.addAllItems(call.items)
.build()
val jsons = nativeCall
.nativeCall(Mono.just(request))
.transform(writeRpcJson.toJsons(call))
return if (call.type == ProxyCall.RpcType.SINGLE) {
jsons.next()
} else {
jsons.transform(writeRpcJson.asArray())
}
}
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()
.aggregate()
.asByteArray()
.map(readRpcJson)
.flatMapMany { call -> execute(chain, call) }
.map { Unpooled.wrappedBuffer(it.toByteArray()) }
resp.addHeader(HttpHeaders.CONTENT_TYPE, "application/json")
.send(results)
}
}
}

145
src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt Normal file
View File

@@ -0,0 +1,145 @@
/**
* Copyright (c) 2020 ETCDEV GmbH
*
* 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.fasterxml.jackson.databind.ObjectMapper
import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.infinitape.etherjar.rpc.RpcException
import io.infinitape.etherjar.rpc.RpcResponseError
import io.infinitape.etherjar.rpc.json.RequestJson
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
import java.io.IOException
import java.util.*
import java.util.function.Function
import java.util.stream.Collectors
/**
* Reader for JSON RPC request
*/
@Service
open class ReadRpcJson(
@Autowired private val objectMapper: ObjectMapper
) : Function<ByteArray, ProxyCall> {
companion object {
private val log = LoggerFactory.getLogger(ReadRpcJson::class.java)
private val spaces = " \n\t".toByteArray()
}
private val jsonExtractor: Function<Map<*, *>, RequestJson<Any>>
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")
}
val id = json["id"]
if (!(json["method"] != null && json["method"] is String)) {
throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "ID not set")
}
if (json.containsKey("params") && json["params"] !is List<*>) {
throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "Params must be an array")
}
RequestJson<Any>(
json["method"].toString(),
json["params"] as List<*>,
id
)
}
}
/**
* Read fist non-space character, which supposed to start actual JSON part of the request
*/
@Throws(IOException::class)
fun getStartOfJson(buf: ByteArray): Byte {
val count = buf.size
var i = 0
//if cannot find anything in the first 255 bytes, just consider it as invalid
while (i < 256 && i < count) {
if (buf[i] != spaces[0] && buf[i] != spaces[1] && buf[i] != spaces[2]) {
return buf[i]
}
i++
}
throw IllegalArgumentException("Invalid input")
}
/**
* Check the type of the payload, based on the format (first character at this case)
*/
@Throws(IOException::class)
fun getType(data: ByteArray): ProxyCall.RpcType {
val first = try {
getStartOfJson(data)
} catch (e: IllegalArgumentException) {
throw RpcException(RpcResponseError.CODE_INVALID_JSON, "Empty JSON")
}
if (first == '{'.toByte()) {
return ProxyCall.RpcType.SINGLE
} else if (first == '['.toByte()) {
return ProxyCall.RpcType.BATCH
}
throw RpcException(RpcResponseError.CODE_INVALID_JSON, "Failed to parse JSON")
}
/**
* Convert payload to the proxy call details
*/
override fun apply(data: ByteArray): ProxyCall {
val list: MutableList<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
} catch (e: RpcException) {
throw e
} catch (e: Exception) {
log.error("Parse Error: " + e.message)
throw RpcException(RpcResponseError.CODE_INVALID_JSON, e.message)
}
}
}

92
src/main/kotlin/io/emeraldpay/dshackle/proxy/WriteRpcJson.kt Normal file
View File

@@ -0,0 +1,92 @@
/**
* Copyright (c) 2020 ETCDEV GmbH
*
* 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.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.infinitape.etherjar.rpc.RpcResponseError
import io.infinitape.etherjar.rpc.json.ResponseJson
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.lang.StringBuilder
import java.time.Duration
import java.util.function.Function
/**
* Writer for JSON RPC requests
*/
@Service
open class WriteRpcJson(
@Autowired private val objectMapper: ObjectMapper
) {
companion object {
private val log = LoggerFactory.getLogger(WriteRpcJson::class.java)
}
/**
* Convert Dshackle protobuf based responses to JSON RPC formatted as strings
*/
open fun toJsons(call: ProxyCall): Function<Flux<BlockchainOuterClass.NativeCallReplyItem>, Flux<String>> {
return Function { flux ->
flux.flatMap { response ->
val json = ResponseJson<Any, Any>()
if (!call.ids.containsKey(response.id)) {
log.warn("ID wasn't requested: ${response.id}")
return@flatMap Flux.empty<String>()
}
json.id = call.ids[response.id]
if (response.succeed) {
val payload = objectMapper.readValue(response.payload.toByteArray(), ResponseJson::class.java)
if (payload.error != null) {
json.error = payload.error
} else {
json.result = payload.result
}
} else {
json.error = RpcResponseError(-32002, response.errorMessage)
}
Flux.just(objectMapper.writeValueAsString(json))
}.onErrorContinue { t, u ->
log.warn("Failed to convert to JSON", t)
}
}
}
/**
* Format response as JSON Array, for Batch requests
*/
fun asArray(): Function<Flux<String>, Flux<String>> {
return Function { flux ->
val body = flux.zipWith(Flux.concat(Mono.just(false), Flux.just(true).repeat()))
.map {
if (it.t2) {
"," + it.t1
} else {
it.t1
}
}
Flux.concat(
Mono.just("["),
body,
Mono.just("]")
)
}
}
}

6
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
View File

@@ -33,7 +33,7 @@ import reactor.util.function.Tuples
import java.lang.Exception
@Service
class NativeCall(
open class NativeCall(
@Autowired private val upstreams: Upstreams,
@Autowired private val objectMapper: ObjectMapper
) {
@@ -42,8 +42,8 @@ class NativeCall(
open fun nativeCall(requestMono: Mono<BlockchainOuterClass.NativeCallRequest>): Flux<BlockchainOuterClass.NativeCallReplyItem> {
return requestMono.flatMapMany(this::prepareCall)
.map(this::setupCallParams)
.parallel()
.map(this::setupCallParams)
.parallel()
.flatMap(this::fetch)
.sequential()
.map(this::buildResponse)