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

imcrease gRPC message size limit

Browse Source
This commit is contained in:
Maksim Fomenkov
2022-11-30 18:03:31 +04:00
parent 81f825f1c1
commit a55bd2305f
1 changed files with 9 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt
View File

@@ -29,7 +29,6 @@ import io.emeraldpay.dshackle.upstream.DefaultUpstream
import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics
import io.grpc.ManagedChannelBuilder
import io.grpc.netty.NettyChannelBuilder import io.grpc.netty.NettyChannelBuilder
import io.micrometer.core.instrument.Counter import io.micrometer.core.instrument.Counter
import io.micrometer.core.instrument.Metrics import io.micrometer.core.instrument.Metrics
@@ -70,21 +69,21 @@ class GrpcUpstreams(
private val lock = ReentrantLock() private val lock = ReentrantLock()
fun start(): Flux<UpstreamChangeEvent> { fun start(): Flux<UpstreamChangeEvent> {
val channel: ManagedChannelBuilder<*> = if (auth != null && StringUtils.isNotEmpty(auth.ca)) { val chanelBuilder = NettyChannelBuilder.forAddress(host, port)
NettyChannelBuilder.forAddress(host, port) // some messages are very large. many of them in megabytes, some even in gigabytes (ex. ETH Traces)
// some messages are very large. many of them in megabytes, some even in gigabytes (ex. ETH Traces) .maxInboundMessageSize(Int.MAX_VALUE)
.maxInboundMessageSize(Int.MAX_VALUE) .enableRetry()
.maxRetryAttempts(3)
if (auth != null && StringUtils.isNotEmpty(auth.ca)) {
chanelBuilder
.useTransportSecurity() .useTransportSecurity()
.enableRetry()
.maxRetryAttempts(3)
.sslContext(withTls(auth)) .sslContext(withTls(auth))
} else { } else {
log.warn("Using insecure connection to $host:$port") log.warn("Using insecure connection to $host:$port")
ManagedChannelBuilder.forAddress(host, port) chanelBuilder.usePlaintext()
.usePlaintext()
} }
val client = ReactorBlockchainGrpc.newReactorStub(channel.build()) val client = ReactorBlockchainGrpc.newReactorStub(chanelBuilder.build())
this.client = client this.client = client
val statusSubscription = AtomicReference<Disposable>() val statusSubscription = AtomicReference<Disposable>()