Skip to content

refactor: reduce cognitive complexity of StarConnector invocation handler #2894

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hongwei1 wants to merge 1 commit into
base: develop
Choose a base branch
from
Open

refactor: reduce cognitive complexity of StarConnector invocation handler #2894

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
198 changes: 93 additions & 105 deletions obp-api/src/main/scala/code/bankconnectors/package.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ import net.liftweb.util.ThreadGlobal

import scala.concurrent.Future
import scala.reflect.runtime.universe.{MethodSymbol, Type, typeOf}
import scala.util.{Success => TrySuccess, Failure => TryFailure}
import scala.util.{Try, Success => TrySuccess, Failure => TryFailure}
import com.openbankproject.commons.util.{ApiVersion, ReflectUtils}
import com.openbankproject.commons.util.ReflectUtils._
import com.openbankproject.commons.util.Functions.Implicits._
Expand All @@ -46,115 +46,103 @@ package object bankconnectors extends MdcLoggable {
//this object is a empty Connector implementation, just for supply default args
object StubConnector extends Connector

// Record the outcome of a connector call: counters, plus optional detailed metric/trace persistence.
def recordConnectorInboundMetrics(connectorName: String, methodName: String, correlationId: String,
duration: Long, isSuccess: Boolean, args: Array[AnyRef]): Unit = {
ConnectorCountsRedis.incrementInbound(connectorName, methodName, isSuccess)
if (getPropsAsBoolValue("write_connector_metrics", false)) {
val params = extractKeyParams(args)
Future {
ConnectorMetricsProvider.metrics.vend.saveConnectorMetric(
connectorName, methodName, correlationId, now, duration, params, isSuccess)
}
}
}

def recordConnectorTrace(connectorName: String, methodName: String, method: Method, args: Array[AnyRef],
duration: Long, isSuccess: Boolean, result: Try[Any]): Unit = {
if (getPropsAsBoolValue("write_connector_trace", false)) {
val outbound = serializeOutboundArgs(method, args)
val inbound = serializeInboundResult(result)
val correlationId = getCorrelationId()
val (detailUserId, detailHttpVerb, detailApiUrl) = extractCallContextInfo(args)
val bankIdValue = extractBankIdFromArgs(args)
Future {
ConnectorTraceProvider.saveConnectorTrace(
correlationId, connectorName, methodName, bankIdValue,
outbound, inbound, now, duration, isSuccess,
detailUserId, detailHttpVerb, detailApiUrl)
}
}
}

// The empty Connector implements both: the $default$ accessors it inherits, and the
// members Connector itself does not declare. Routing the latter would look them up as
// connector calls - and NPE on the way, since args is null for a no-arg method.
def delegateToStub(method: Method, args: Array[AnyRef]): AnyRef = {
val connectorMethodResult = method.invoke(StubConnector, args:_*)
if (connectorMethodResult.isInstanceOf[Future[_]] && canOpenFuture(method.getName)) {
FutureUtil.futureWithLimits(connectorMethodResult.asInstanceOf[Future[_]], method.getName)
}
connectorMethodResult
}

def routeToConnector(method: Method, args: Array[AnyRef]): AnyRef = {
val methodName = method.getName
val argNameToValue: Array[(String, AnyRef)] = method.getParameters.map(_.getName).zip(args)
// TODO: getConnectorNameAndMethodRouting is also called inside invokeMethod.
// Consider refactoring invokeMethod to accept a pre-resolved connectorName to avoid the duplicate lookup.
val (_, connectorName) = getConnectorNameAndMethodRouting(methodName, argNameToValue)

// Extract correlationId from CallContext before entering any Future callback,
// because Lift's S.containerSession is unavailable in async contexts.
val correlationId: String = args.collectFirst {
case Some(cc: CallContext) => cc.correlationId
case Full(cc: CallContext) => cc.correlationId
}.getOrElse(getCorrelationId()) // fallback to Lift session if no CallContext in args

// Record outbound (before call)
ConnectorCountsRedis.incrementOutbound(connectorName, methodName)
val t0 = System.currentTimeMillis()

val (connectorMethodResult, methodSymbol) = invokeMethod(method, args)

// Track metrics for Future results
if (connectorMethodResult.isInstanceOf[Future[_]]) {
val future = connectorMethodResult.asInstanceOf[Future[Any]]
future.onComplete { result =>
val duration = System.currentTimeMillis() - t0
val isSuccess = result match {
case TrySuccess(value) => !isFailureBox(value)
case TryFailure(_) => false
}
recordConnectorInboundMetrics(connectorName, methodName, correlationId, duration, isSuccess, args)
recordConnectorTrace(connectorName, methodName, method, args, duration, isSuccess, result)
}
} else {
// Non-future (legacy Box) result - track synchronously
val duration = System.currentTimeMillis() - t0
val isSuccess = !isFailureBox(connectorMethodResult)
recordConnectorInboundMetrics(connectorName, methodName, correlationId, duration, isSuccess, args)
recordConnectorTrace(connectorName, methodName, method, args, duration, isSuccess, TrySuccess(connectorMethodResult))
}

if (connectorMethodResult.isInstanceOf[Future[_]] && canOpenFuture(method.getName)) {
FutureUtil.futureWithLimits(connectorMethodResult.asInstanceOf[Future[_]], method.getName)
}
logger.debug(s"do required field validation for ${methodSymbol.typeSignature}")
val apiVersion = ApiVersionHolder.getApiVersion
validateRequiredFields(connectorMethodResult, methodSymbol.returnType, apiVersion)
}

val intercept: InvocationHandler = new InvocationHandler {
override def invoke(proxy: AnyRef, method: Method, args: Array[AnyRef]): AnyRef = {
if (method.getReturnType.getName == "scala.concurrent.Future" && !canOpenFuture(method.getName)) {
throw new RuntimeException(ServiceIsTooBusy + s"Current Service(${method.getName})")
} else if (method.getName.contains("$default$") || ConnectorProxy.isInheritedMember(method)) {
delegateToStub(method, args)
} else {
if (method.getName.contains("$default$") || ConnectorProxy.isInheritedMember(method)) {
// The empty Connector implements both: the $default$ accessors it inherits, and the
// members Connector itself does not declare. Routing the latter would look them up as
// connector calls - and NPE on the way, since args is null for a no-arg method.
val connectorMethodResult = method.invoke(StubConnector, args:_*)
if (connectorMethodResult.isInstanceOf[Future[_]] && canOpenFuture(method.getName)) {
FutureUtil.futureWithLimits(connectorMethodResult.asInstanceOf[Future[_]], method.getName)
}
connectorMethodResult
} else {
val methodName = method.getName
val argNameToValue: Array[(String, AnyRef)] = method.getParameters.map(_.getName).zip(args)
// TODO: getConnectorNameAndMethodRouting is also called inside invokeMethod.
// Consider refactoring invokeMethod to accept a pre-resolved connectorName to avoid the duplicate lookup.
val (_, connectorName) = getConnectorNameAndMethodRouting(methodName, argNameToValue)

// Extract correlationId from CallContext before entering any Future callback,
// because Lift's S.containerSession is unavailable in async contexts.
val correlationId: String = args.collectFirst {
case Some(cc: CallContext) => cc.correlationId
case Full(cc: CallContext) => cc.correlationId
}.getOrElse(getCorrelationId()) // fallback to Lift session if no CallContext in args

// Record outbound (before call)
ConnectorCountsRedis.incrementOutbound(connectorName, methodName)
val t0 = System.currentTimeMillis()

val (connectorMethodResult, methodSymbol) = invokeMethod(method, args)

// Track metrics for Future results
if (connectorMethodResult.isInstanceOf[Future[_]]) {
val future = connectorMethodResult.asInstanceOf[Future[Any]]
future.onComplete { result =>
val duration = System.currentTimeMillis() - t0
val isSuccess = result match {
case TrySuccess(value) => !isFailureBox(value)
case TryFailure(_) => false
}

// Record inbound
ConnectorCountsRedis.incrementInbound(connectorName, methodName, isSuccess)

// Record detailed metric to DB
if (getPropsAsBoolValue("write_connector_metrics", false)) {
val params = extractKeyParams(args)
Future {
ConnectorMetricsProvider.metrics.vend.saveConnectorMetric(
connectorName, methodName, correlationId, now, duration, params, isSuccess)
}
}

// Record connector trace (outbound/inbound messages)
if (getPropsAsBoolValue("write_connector_trace", false)) {
val outbound = serializeOutboundArgs(method, args)
val inbound = serializeInboundResult(result)
val correlationId = getCorrelationId()
val (detailUserId, detailHttpVerb, detailApiUrl) = extractCallContextInfo(args)
val bankIdValue = extractBankIdFromArgs(args)
Future {
ConnectorTraceProvider.saveConnectorTrace(
correlationId, connectorName, methodName, bankIdValue,
outbound, inbound, now, duration, isSuccess,
detailUserId, detailHttpVerb, detailApiUrl)
}
}
}
} else {
// Non-future (legacy Box) result - track synchronously
val duration = System.currentTimeMillis() - t0
val isSuccess = !isFailureBox(connectorMethodResult)

ConnectorCountsRedis.incrementInbound(connectorName, methodName, isSuccess)

if (getPropsAsBoolValue("write_connector_metrics", false)) {
val params = extractKeyParams(args)
Future {
ConnectorMetricsProvider.metrics.vend.saveConnectorMetric(
connectorName, methodName, correlationId, now, duration, params, isSuccess)
}
}

// Record connector trace (outbound/inbound messages)
if (getPropsAsBoolValue("write_connector_trace", false)) {
val outbound = serializeOutboundArgs(method, args)
val inbound = serializeInboundResult(TrySuccess(connectorMethodResult))
val correlationId = getCorrelationId()
val (detailUserId, detailHttpVerb, detailApiUrl) = extractCallContextInfo(args)
val bankIdValue = extractBankIdFromArgs(args)
Future {
ConnectorTraceProvider.saveConnectorTrace(
correlationId, connectorName, methodName, bankIdValue,
outbound, inbound, now, duration, isSuccess,
detailUserId, detailHttpVerb, detailApiUrl)
}
}
}

if (connectorMethodResult.isInstanceOf[Future[_]] && canOpenFuture(method.getName)) {
FutureUtil.futureWithLimits(connectorMethodResult.asInstanceOf[Future[_]], method.getName)
}
logger.debug(s"do required field validation for ${methodSymbol.typeSignature}")
val apiVersion = ApiVersionHolder.getApiVersion
validateRequiredFields(connectorMethodResult, methodSymbol.returnType, apiVersion)
}
routeToConnector(method, args)
}
}
}
Expand Down
Loading