GrpcConfiguration

GrpcConfiguration.kt

package com.company.service.configuration

import com.google.protobuf.MessageLite
import io.grpc.BindableService
import io.grpc.Channel
import io.grpc.netty.NettyChannelBuilder
import io.netty.handler.ssl.ApplicationProtocolConfig
import io.netty.handler.ssl.ApplicationProtocolConfig.Protocol.ALPN
import io.netty.handler.ssl.ApplicationProtocolConfig.SelectedListenerFailureBehavior.ACCEPT
import io.netty.handler.ssl.ApplicationProtocolConfig.SelectorFailureBehavior.NO_ADVERTISE
import io.netty.handler.ssl.ApplicationProtocolNames.HTTP_2
import io.netty.handler.ssl.SslContextBuilder
import io.netty.handler.ssl.SslProvider
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.reactive.awaitSingle
import kotlinx.coroutines.reactor.asCoroutineContext
import kotlinx.coroutines.reactor.mono
import org.springframework.beans.factory.annotation.Value
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.context.annotation.Profile
import org.springframework.http.MediaType
import org.springframework.web.reactive.function.server.coRouter
import reactor.core.publisher.Mono
import reactor.util.context.Context
import java.security.cert.X509Certificate
import java.util.Optional
import javax.net.ssl.X509TrustManager
import kotlin.reflect.full.callSuspend
import kotlin.reflect.full.memberFunctions

@Configuration
internal class GrpcConfiguration {

    companion object {
        private val protobuf = arrayOf(
            MediaType.parseMediaType("application/x-protobuf"),
            MediaType.parseMediaType("application/protobuf"),
            MediaType.parseMediaType("application/grpc-web-text")
        )
    }

    @Bean
    fun grpcChannel(@Value("\${server.port:8080}") serverPort: Int): Channel =
        NettyChannelBuilder.forTarget("localhost:$serverPort")
            .sslContext(
                SslContextBuilder.forClient()
                    .sslProvider(SslProvider.JDK)
                    .applicationProtocolConfig(
                        ApplicationProtocolConfig(
                            ALPN, NO_ADVERTISE, ACCEPT, HTTP_2
                        )
                    )
                    .trustManager(
                        object : X509TrustManager {
                            override fun checkClientTrusted(p0: Array<out X509Certificate>?, p1: String?) {}
                            override fun checkServerTrusted(p0: Array<out X509Certificate>?, p1: String?) {}
                            override fun getAcceptedIssuers(): Array<X509Certificate> = arrayOf()
                        }
                    ).build()
            )
            .build()

    @Bean
    fun grpcRouter(bindableServices: Optional<List<BindableService>>) =
        if (!bindableServices.isPresent) null else coRouter {
            accept(*protobuf).nest {
                bindableServices.get().forEach { bindableService ->
                    val serviceDefinition = bindableService.bindService()
                    val serviceName = serviceDefinition.serviceDescriptor.name
                    serviceDefinition.methods.forEach { methodDefinition ->
                        val methodName = methodDefinition.methodDescriptor.bareMethodName
                        POST("$serviceName/$methodName").invoke { request ->
                            ok()
                                .contentType(protobuf[0])
                                .bodyValue(
                                    request.body { inputMessage, _ ->
                                        inputMessage.body.flatMap {
                                            Mono.just(
                                                methodDefinition.methodDescriptor
                                                    .parseRequest(it.asInputStream()) as Any
                                            )
                                        }
                                    }.flatMap { requestData ->
                                        Mono.deferContextual { context ->
                                            mono(
                                                Dispatchers.Unconfined +
                                                    Context.of(context).asCoroutineContext()
                                            ) {
                                                val method = bindableService::class.memberFunctions
                                                    .find { it.name.equals(methodName, ignoreCase = true) }
                                                val result = method!!
                                                    .callSuspend(bindableService, requestData) as MessageLite
                                                result.toByteArray()
                                            }
                                        }
                                    }.awaitSingle()
                                ).awaitSingle()
                        }
                    }
                }
            }
        }
}