-
Notifications
You must be signed in to change notification settings - Fork 56
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add BroadcastReader instead of BroadcastQuorum (#263)
- Loading branch information
1 parent
6f576fc
commit 674c69f
Showing
12 changed files
with
497 additions
and
207 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
45 changes: 0 additions & 45 deletions
45
src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
118 changes: 118 additions & 0 deletions
118
src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,118 @@ | ||
| package io.emeraldpay.dshackle.reader | ||
|
|
||
| import io.emeraldpay.dshackle.commons.BROADCAST_READER | ||
| import io.emeraldpay.dshackle.commons.SPAN_REQUEST_UPSTREAM_ID | ||
| import io.emeraldpay.dshackle.upstream.Selector | ||
| import io.emeraldpay.dshackle.upstream.Upstream | ||
| import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError | ||
| import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest | ||
| import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse | ||
| import io.emeraldpay.dshackle.upstream.signature.ResponseSigner | ||
| import org.slf4j.LoggerFactory | ||
| import org.springframework.cloud.sleuth.Tracer | ||
| import reactor.core.publisher.Mono | ||
| import java.util.concurrent.atomic.AtomicInteger | ||
|
|
||
| class BroadcastReader( | ||
| private val upstreams: List<Upstream>, | ||
| matcher: Selector.Matcher, | ||
| signer: ResponseSigner?, | ||
| private val tracer: Tracer | ||
| ) : RpcReader(signer) { | ||
| private val internalMatcher = Selector.MultiMatcher( | ||
| listOf(Selector.AvailabilityMatcher(), matcher) | ||
| ) | ||
|
|
||
| companion object { | ||
| private val log = LoggerFactory.getLogger(BroadcastReader::class.java) | ||
| } | ||
|
|
||
| override fun attempts(): AtomicInteger { | ||
| return AtomicInteger(1) | ||
| } | ||
|
|
||
| override fun read(key: JsonRpcRequest): Mono<Result> { | ||
| return Mono.just(upstreams) | ||
| .map { ups -> | ||
| ups.filter { internalMatcher.matches(it) }.map { execute(key, it) } | ||
| }.flatMap { | ||
| Mono.zip(it) { responses -> | ||
| analyzeResponses( | ||
| key, | ||
| getJsonRpcResponses(key.method, responses) | ||
| ) | ||
| }.onErrorResume { err -> | ||
| log.error("Broadcast error: ${err.message}") | ||
| Mono.error(handleError(null, 0, null)) | ||
| }.flatMap { broadcastResult -> | ||
| if (broadcastResult.result != null) { | ||
| Mono.just( | ||
| Result(broadcastResult.result, broadcastResult.signature, 0, null) | ||
| ) | ||
| } else { | ||
| val err = handleError(broadcastResult.error, key.id, null) | ||
| Mono.error(err) | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private fun analyzeResponses(key: JsonRpcRequest, jsonRpcResponses: List<BroadcastResponse>): BroadcastResult { | ||
| val errors = mutableListOf<JsonRpcError>() | ||
| jsonRpcResponses.forEach { | ||
| val response = it.jsonRpcResponse | ||
| if (response.hasResult()) { | ||
| val signature = getSignature(key, response, it.upstreamId) | ||
| return BroadcastResult(response.getResult(), null, signature) | ||
| } else if (response.hasError()) { | ||
| errors.add(response.error!!) | ||
| } | ||
| } | ||
|
|
||
| val error = errors.takeIf { it.isNotEmpty() }?.get(0) | ||
|
|
||
| return BroadcastResult(error) | ||
| } | ||
|
|
||
| private fun getJsonRpcResponses(method: String, responses: Array<Any>) = | ||
| responses | ||
| .map { response -> | ||
| (response as BroadcastResponse) | ||
| .also { r -> | ||
| if (r.jsonRpcResponse.hasResult()) { | ||
| log.info( | ||
| "Response for $method from upstream ${r.upstreamId}: ${String(r.jsonRpcResponse.getResult())}" | ||
| ) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private fun execute( | ||
| key: JsonRpcRequest, | ||
| upstream: Upstream | ||
| ): Mono<BroadcastResponse> = | ||
| SpannedReader( | ||
| upstream.getIngressReader(), tracer, BROADCAST_READER, mapOf(SPAN_REQUEST_UPSTREAM_ID to upstream.getId()) | ||
| ) | ||
| .read(key) | ||
| .map { BroadcastResponse(it, upstream.getId()) } | ||
| .onErrorResume { | ||
| log.warn("Error during execution ${key.method} from upstream ${upstream.getId()} with message - ${it.message}") | ||
| Mono.just( | ||
| BroadcastResponse(JsonRpcResponse(null, getError(key, it).error), upstream.getId()) | ||
| ) | ||
| } | ||
|
|
||
| private class BroadcastResponse( | ||
| val jsonRpcResponse: JsonRpcResponse, | ||
| val upstreamId: String | ||
| ) | ||
|
|
||
| private class BroadcastResult( | ||
| val result: ByteArray?, | ||
| val error: JsonRpcError?, | ||
| val signature: ResponseSigner.Signature? | ||
| ) { | ||
| constructor(error: JsonRpcError?) : this(null, error, null) | ||
| } | ||
| } |
81 changes: 81 additions & 0 deletions
81
src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReaderFactory.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,81 @@ | ||
| package io.emeraldpay.dshackle.reader | ||
|
|
||
| import io.emeraldpay.dshackle.quorum.CallQuorum | ||
| import io.emeraldpay.dshackle.quorum.QuorumRpcReader | ||
| import io.emeraldpay.dshackle.reader.RpcReader.Result | ||
| import io.emeraldpay.dshackle.upstream.Multistream | ||
| import io.emeraldpay.dshackle.upstream.Selector | ||
| import io.emeraldpay.dshackle.upstream.Upstream | ||
| import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError | ||
| import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException | ||
| import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest | ||
| import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse | ||
| import io.emeraldpay.dshackle.upstream.signature.ResponseSigner | ||
| import io.emeraldpay.etherjar.rpc.RpcException | ||
| import org.springframework.cloud.sleuth.Tracer | ||
| import java.util.concurrent.atomic.AtomicInteger | ||
|
|
||
| abstract class RpcReader( | ||
| private val signer: ResponseSigner?, | ||
| ) : Reader<JsonRpcRequest, Result> { | ||
| abstract fun attempts(): AtomicInteger | ||
|
|
||
| protected fun getError(key: JsonRpcRequest, err: Throwable) = | ||
| when (err) { | ||
| is RpcException -> JsonRpcException.from(err) | ||
| is JsonRpcException -> err | ||
| else -> JsonRpcException( | ||
| JsonRpcResponse.NumberId(key.id), | ||
| JsonRpcError(-32603, "Unhandled internal error: ${err.javaClass}: ${err.message}") | ||
| ) | ||
| } | ||
|
|
||
| protected fun handleError(error: JsonRpcError?, id: Int, resolvedBy: String?) = | ||
| error?.asException(JsonRpcResponse.NumberId(id), resolvedBy) | ||
| ?: JsonRpcException(JsonRpcResponse.NumberId(id), JsonRpcError(-32603, "Unhandled Upstream error"), resolvedBy) | ||
|
|
||
| protected fun getSignature(key: JsonRpcRequest, response: JsonRpcResponse, upstreamId: String) = | ||
| response.providedSignature | ||
| ?: if (key.nonce != null) { | ||
| signer?.sign(key.nonce, response.getResult(), upstreamId) | ||
| } else { | ||
| null | ||
| } | ||
|
|
||
| class Result( | ||
| val value: ByteArray, | ||
| val signature: ResponseSigner.Signature?, | ||
| val quorum: Int, | ||
| val resolvedBy: Upstream? | ||
| ) | ||
| } | ||
|
|
||
| interface RpcReaderFactory { | ||
|
|
||
| companion object { | ||
| fun default(): RpcReaderFactory { | ||
| return Default() | ||
| } | ||
| } | ||
|
|
||
| fun create(data: RpcReaderData): RpcReader | ||
|
|
||
| class Default : RpcReaderFactory { | ||
| override fun create(data: RpcReaderData): RpcReader { | ||
| if (data.method == "eth_sendRawTransaction") { | ||
| return BroadcastReader(data.multistream.getAll(), data.matcher, data.signer, data.tracer) | ||
| } | ||
| val apis = data.multistream.getApiSource(data.matcher) | ||
| return QuorumRpcReader(apis, data.quorum, data.signer, data.tracer) | ||
| } | ||
| } | ||
|
|
||
| data class RpcReaderData( | ||
| val multistream: Multistream, | ||
| val method: String, | ||
| val matcher: Selector.Matcher, | ||
| val quorum: CallQuorum, | ||
| val signer: ResponseSigner?, | ||
| val tracer: Tracer | ||
| ) | ||
| } |
Oops, something went wrong.