Merge pull request #48 from p2p-org/fix_gRPC_message_size_limit

Fix grpc message size limit

build.gradle

@@ -36,7 +36,7 @@ group = 'io.emeraldpay.dshackle'
 // Version schema:
 //    x.x.x for production, following SemVer model
 //    x.x.x-SNAPSHOT for development
-version = '0.15.0-SNAPSHOT'
+version = '0.16.0-SNAPSHOT'
 
 java {
     sourceCompatibility = JavaVersion.VERSION_13

src/main/kotlin/io/emeraldpay/dshackle/Defaults.kt

@@ -21,6 +21,7 @@ import java.time.Duration
 class Defaults {
 
     companion object {
+        const val maxMessageSize: Int = 32 * 1024 * 1024
         val timeout: Duration = Duration.ofSeconds(60)
         val timeoutInternal: Duration = timeout.dividedBy(4)
         val retryConnection: Duration = Duration.ofSeconds(10)

src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt

@@ -46,6 +46,7 @@ open class GrpcServer(
         log.info("Listening Native gRPC on ${mainConfig.host}:${mainConfig.port}")
         val serverBuilder = NettyServerBuilder
             .forAddress(InetSocketAddress(mainConfig.host, mainConfig.port))
+            .maxInboundMessageSize(Defaults.maxMessageSize)
             .let {
                 if (mainConfig.accessLogConfig.enabled) {
                     it.intercept(accessHandler)
@@ -65,7 +66,7 @@ open class GrpcServer(
         val server = serverBuilder.build()
         this.server = server
 
-        Thread { server.start() }.run()
+        server.start()
         log.info("GRPC Server started")
     }
 

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

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