package com.blizzard.content.filter.chatreplay.sink
import com.blizzard.content.filter.chatreplay.config.EXECUTION
import com.blizzard.content.filter.chatreplay.config.ExternalServices
import com.blizzard.content.filter.chatreplay.config.INTERACTION
import com.blizzard.content.filter.chatreplay.config.METRICS_PREFIX
import com.blizzard.content.filter.chatreplay.config.MODE
import com.blizzard.content.filter.chatreplay.config.SERVICETARGET
import com.blizzard.content.filter.chatreplay.config.TRANSFORM
import com.blizzard.content.filter.chatreplay.config.TRANSPORT
import com.blizzard.content.filter.chatreplay.domain.TransformResponse
import com.blizzard.content.filter.chatreplay.domain.failedTransformResponse
import com.blizzard.content.filter.chatreplay.domain.log
import com.blizzard.content.filter.chatreplay.domain.parseId
import com.blizzard.content.filter.chatreplay.metrics.toMap
import com.blizzard.content.filter.chatreplay.metrics.toTags
import com.blizzard.content.filter.chatreplay.source.LogStreamService
import com.newrelic.api.agent.NewRelic
import io.micrometer.core.instrument.MeterRegistry
import io.micrometer.core.instrument.Metrics
import io.micrometer.core.instrument.Tag
import io.micrometer.core.instrument.Tags
import kotlinx.coroutines.CoroutineExceptionHandler
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onCompletion
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.onEmpty
import kotlinx.coroutines.flow.onStart
import kotlinx.coroutines.flow.retryWhen
import kotlinx.coroutines.flow.transform
import kotlinx.coroutines.launch
import kotlinx.coroutines.reactive.awaitFirstOrNull
import kotlinx.coroutines.reactor.awaitSingle
import org.slf4j.LoggerFactory
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
import org.springframework.boot.context.event.ApplicationReadyEvent
import org.springframework.context.event.EventListener
import org.springframework.core.NestedRuntimeException
import org.springframework.messaging.rsocket.RSocketRequester
import org.springframework.messaging.rsocket.retrieveFlow
import org.springframework.stereotype.Service
import org.springframework.util.MimeTypeUtils
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.util.retry.Retry
import java.time.Duration.ofMillis
import kotlin.random.Random
const val ROUTE_TRANSFORM_CHANNEL = "replace-channel-controller"
const val ROUTE_TRANSFORM_REQUEST_RESPONSE = "replace-request-response-controller"
const val PATH_RSOCKET = "/rsocket-api"
/**
* - [https://rsocket.io]
* - [https://docs.spring.io/spring-boot/docs/2.4.3/reference/html/spring-boot-features.html#boot-features-rsocket]
* - [https://docs.spring.io/spring/docs/5.2.9.RELEASE/spring-framework-reference/web-reactive.html#rsocket-spring]
*
* See file rest-client/metrics.http
*/
@Service
@ConditionalOnProperty("chat-replay.subscribers.rsocket-service.enable")
class OchRSocketService(
private val rSocketReqesterBuilder: RSocketRequester.Builder,
private val logStreamService: LogStreamService,
private val registry: MeterRegistry,
externalProperties: ExternalServices,
) {
private val cff = externalProperties.och.contentFilterFunctions
private val log = LoggerFactory.getLogger(OchRSocketService::class.java)
private val localTags = Tags.of(Tag.of(EXECUTION, "remote"), Tag.of(SERVICETARGET, "och"), Tag.of(TRANSPORT, "rsocket"), Tag.of(MODE, "streamlistener"), Tag.of(INTERACTION, "request-channel"))
private val transformedCounter = Metrics.counter(TRANSFORM, localTags.and("transformed", "true"))
private val untransformedCounter = Metrics.counter(TRANSFORM, localTags.and("transformed", "false"))
private val retryCounter = Metrics.counter("$METRICS_PREFIX.rsocket.request-channel.retry", localTags)
private val logStreamRetryCounter = Metrics.counter("$METRICS_PREFIX.rsocket.request-channel.logstream.retry", localTags)
private val rSocketScope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.IO + CoroutineExceptionHandler { _, e ->
log.error("Exception from rSocketScope handler: ${e.message}")
})
private val handlingScope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default + CoroutineExceptionHandler { _, e ->
log.error("Exception from handlingScope handler: ${e.message}")
})
private var requesterMono = Mono.empty<RSocketRequester>()
@EventListener(condition = "@environment.getProperty('chat-replay.subscribers.rsocket-service.enable')")
fun onApplicationEvent(readyEvent: ApplicationReadyEvent) {
log.info("ApplicationReady: ${javaClass.simpleName}")
launchChannelConnection()
}
private fun requesterMono(): Mono<RSocketRequester> {
log.info("Caching RSocketRequester for REQUEST_CHANNEL")
return Mono.just(rSocketReqesterBuilder
.rsocketConnector { it.configureConnector(log, localTags, registry) }
.dataMimeType(MimeTypeUtils.APPLICATION_JSON)
.websocket(resolveWsUri(cff).also {
log.info("RSocket REQUEST_CHANNEL connecting to $it")
}))
}
private fun launchChannelConnection() =
rSocketScope.launch {
log.info("Starting RSocket Channel connection...")
/* If we initialize this connection before app is ready, we can "steal" the
connection from the previously running instance. And if this app fails to start up,
no one is happy.
*/
if (requesterMono.awaitFirstOrNull() == null) {
requesterMono = requesterMono()
}
requesterMono.awaitSingle()
.route(ROUTE_TRANSFORM_CHANNEL)
.data(logStreamService.logStream()
.retryWhen(Retry.backoff(Long.MAX_VALUE, ofMillis(100)).doBeforeRetry {
log.error("Retrying [logStream] attempt [${it.totalRetries()}] due to ${it.failure()} : ${it.failure().message}")
logStreamRetryCounter.increment()
})
.map { chatLog -> chatLog.text ?: "" }.addReactiveMetrics())
/* End of RSocket config, beginning of Kotlin Suspending Flow processing */
.retrieveFlow<TransformResponse>()
.retryWhen { cause, attempt ->
log.error("Retrying attempt [$attempt] due to $cause : ${cause.message}")
retryCounter.increment()
true
}
.flowOn(Dispatchers.IO)
.onEach(::incrementCounters)
.onEach { if (log.isDebugEnabled) log.debug("Handling : ${it.parseId()} ${it.log()}") }
.transform {
try {
//occasionallyThrowException(/* To demonstrate error-handling */)
emit(it)
} catch (o_O: Exception) {
recordError(o_O, "Error: ${o_O.message}")
// Skip, continue
}
}
.catch /*Some other exception */{ o_O ->
recordError(o_O, "Actual exception: ${o_O.message}")
emit(failedTransformResponse)
}
.onStart { log.info("RSocket REQUEST_CHANNEL connected to ${cff.host}:${cff.port}$PATH_RSOCKET") }
.onCompletion { log.info("Stream complete") }
.onEmpty { log.info("onEmpty") }
.launchIn(handlingScope)
}
private fun occasionallyThrowException(): Unit =
if (Random.nextInt(10) == 0) throw RuntimeException("Random Exception") else Unit
private fun incrementCounters(response: TransformResponse) {
when {
response.wasModified -> transformedCounter.increment()
else -> untransformedCounter.increment()
}
}
private fun recordError(t: Throwable, s: String) {
log.error("$s: ${t.message}")
val contextTags: MutableMap<String, String> = mutableMapOf(
"class" to t.javaClass.simpleName,
"message" to "${t.message}",
)
if (t is NestedRuntimeException) {
contextTags += mutableMapOf(
"cause.class" to t.mostSpecificCause.javaClass.simpleName,
"cause.message" to "${t.mostSpecificCause.message}",
)
}
Metrics.counter("$TRANSFORM.error", localTags + contextTags.toTags()).increment()
NewRelic.noticeError(t, localTags.toMap() + contextTags)
}
/**
* - [https://projectreactor.io/docs/core/milestone/reference/#_publisher_metrics]
* - [https://projectreactor.io/docs/core/milestone/reference/#metrics]
*
* These metrics result in metric names of:
*
* name + ".onNext.delay"
* name + ".requested"
* name + ".subscribed"
*
*/
private fun <T : Any> Flux<T>.addReactiveMetrics(): Flux<T> {
var flux = name(TRANSFORM)
localTags.forEach {
flux = flux.tag(it.key, it.value)
}
return flux.metrics()
}
}