Skip to content

Commit

Permalink
Merge cd7b847 into 9a79320
Browse files Browse the repository at this point in the history
  • Loading branch information
@djfreels
djfreels committed Oct 15, 2019
2 parents 9a79320 + cd7b847 commit 6fa7263
Show file tree
Hide file tree
Showing 3 changed files with 104 additions and 46 deletions.
111 changes: 71 additions & 40 deletions spark-pipeline-engine/src/main/scala/com/acxiom/pipeline/utils/ReflectionUtils.scala
View file
Expand Up @@ -15,13 +15,13 @@ object ReflectionUtils {
private val logger = Logger.getLogger(getClass)

/**
* This function will attempt to find and instantiate the named class with the given parameters.
*
* @param className The fully qualified class name
* @param parameters The parameters to pass to the constructor or None
* @return An instantiated class.
*/
def loadClass(className: String, parameters: Option[Map[String, Any]] = None): Any = {
* This function will attempt to find and instantiate the named class with the given parameters.
*
* @param className The fully qualified class name
* @param parameters The parameters to pass to the constructor or None
* @return An instantiated class.
*/
def loadClass(className: String, parameters: Option[Map[String, Any]] = None, validateParameterTypes: Boolean = false): Any = {
val mirror = ru.runtimeMirror(getClass.getClassLoader)
val moduleClass = mirror.staticClass(className)
val module = mirror.staticModule(className)
Expand All @@ -32,18 +32,18 @@ object ReflectionUtils {
val method = getMethodBySymbol(symbols.head, parameters.getOrElse(Map[String, Any]()), Some(symbols))
classMirror.reflectConstructor(method)
classMirror.reflectConstructor(method)(mapMethodParameters(method.paramLists.head, parameters.getOrElse(Map[String, Any]()), mirror,
mirror.reflect(mirror.reflectModule(module)), symbols.head.asTerm.fullName, method.typeSignature, None, None, None)
mirror.reflect(mirror.reflectModule(module)), symbols.head.asTerm.fullName, method.typeSignature, None, None, None, validateParameterTypes)
: _*)
}
}

/**
* This function will execute the PipelineStep function using the provided parameter values.
*
* @param step The step to execute
* @param parameterValues A map of named parameter values to map to the step function parameters.
* @return The result of the step function execution.
*/
* This function will execute the PipelineStep function using the provided parameter values.
*
* @param step The step to execute
* @param parameterValues A map of named parameter values to map to the step function parameters.
* @return The result of the step function execution.
*/
def processStep(step: PipelineStep,
pipeline: Pipeline,
parameterValues: Map[String, Any],
Expand Down Expand Up @@ -73,7 +73,8 @@ object ReflectionUtils {
val ts = stepObject.symbol.typeSignature
// Get the parameters this method requires
val parameters = method.paramLists.head
val params = mapMethodParameters(parameters, parameterValues, mirror, stepObject, funcName, ts, Some(pipelineContext), step.id, pipeline.id)
val validateTypes = pipelineContext.globals.getOrElse(Map()).getOrElse("validateStepParameterTypes", false).asInstanceOf[Boolean]
val params = mapMethodParameters(parameters, parameterValues, mirror, stepObject, funcName, ts, Some(pipelineContext), step.id, pipeline.id, validateTypes)
logger.info(s"Executing step $objName.$funcName")
logger.debug(s"Parameters: $params")
// Invoke the method
Expand All @@ -95,13 +96,13 @@ object ReflectionUtils {
}

/**
* execute a function on an existing object by providing the function name and parameters (in expected order)
*
* @param obj the object with the function that needs to be run
* @param funcName the name of the function on the object to run
* @param params the parameters required for the function (in proper order)
* @return the results of the executed function
*/
* execute a function on an existing object by providing the function name and parameters (in expected order)
*
* @param obj the object with the function that needs to be run
* @param funcName the name of the function on the object to run
* @param params the parameters required for the function (in proper order)
* @return the results of the executed function
*/
def executeFunctionByName(obj: Any, funcName: String, params: List[Any]): Any = {
val mirror = ru.runtimeMirror(getClass.getClassLoader)
val reflectedObj = mirror.reflect(obj)
Expand All @@ -111,15 +112,15 @@ object ReflectionUtils {
}

/**
* This function will attempt to extract the value assigned to the field name from the given entity. The field can
* contain "." character to denote sub objects. If the field does not exist or is empty a None object will be
* returned. This function does not handle collections.
*
* @param entity The entity containing the value.
* @param fieldName The name of the field to extract
* @param extractFromOption Setting this to true will see if the value is an option and extract the sub value.
* @return The value from the field or an empty string
*/
* This function will attempt to extract the value assigned to the field name from the given entity. The field can
* contain "." character to denote sub objects. If the field does not exist or is empty a None object will be
* returned. This function does not handle collections.
*
* @param entity The entity containing the value.
* @param fieldName The name of the field to extract
* @param extractFromOption Setting this to true will see if the value is an option and extract the sub value.
* @return The value from the field or an empty string
*/
def extractField(entity: Any, fieldName: String, extractFromOption: Boolean = true): Any = {
if (fieldName == "") {
entity
Expand Down Expand Up @@ -187,7 +188,8 @@ object ReflectionUtils {
ts: ru.Type,
pipelineContext: Option[PipelineContext],
stepId: Option[String],
pipelineId: Option[String]) = {
pipelineId: Option[String],
validateParameterTypes: Boolean) = {
parameters.zipWithIndex.map { case (param, pos) =>
val name = param.name.toString
logger.debug(s"Mapping parameter $name")
Expand All @@ -209,17 +211,19 @@ object ReflectionUtils {
}

val finalValue = if (pipelineContext.isDefined) {
pipelineContext.get.security.secureParameter(getFinalValue(optionType, value))
pipelineContext.get.security.secureParameter(getFinalValue(param.asTerm.typeSignature, value))
} else {
getFinalValue(optionType, value)
getFinalValue(param.asTerm.typeSignature, value)
}

val finalValueType = finalValue match {
case v: Option[_] =>
if (v.asInstanceOf[Option[_]].isEmpty) "None" else s"Some(${v.asInstanceOf[Option[_]].get.getClass.getSimpleName})"
case _ => finalValue.getClass.getSimpleName
}
validateParamTypeAssignment(runtimeMirror, param, optionType, finalValue, finalValueType, funcName, stepId, pipelineId)
if (validateParameterTypes) {
validateParamTypeAssignment(runtimeMirror, param, optionType, finalValue, finalValueType, funcName, stepId, pipelineId)
}

logger.debug(s"Mapping parameter to method $funcName,paramName=$name,paramType=${param.typeSignature}," +
s"valueType=$finalValueType,value=$finalValue")
Expand Down Expand Up @@ -264,11 +268,32 @@ object ReflectionUtils {
}
}

private def getFinalValue(optionType: Boolean, value: Any): Any = {
private def getFinalValue(paramType: ru.Type, value: Any): Any = {
val optionType = paramType.toString.contains("Option[")
if (optionType && !value.isInstanceOf[Option[_]]) {
Some(value)
} else if (!optionType && value.isInstanceOf[Option[_]] && value.asInstanceOf[Option[_]].isDefined) {
value.asInstanceOf[Option[_]].get
Some(wrapValueInCollection(paramType, optionType, value))
} else if (!optionType && value.isInstanceOf[Option[_]]) {
if (value.asInstanceOf[Option[_]].isDefined) {
wrapValueInCollection(paramType, optionType, value.asInstanceOf[Option[_]].get)
} else {
value
}
} else {
wrapValueInCollection(paramType, optionType, value)
}
}

private def wrapValueInCollection(paramType: ru.Type, isOption: Boolean, value: Any): Any = {
val collectionType = isCollection(isOption, paramType)
val typeName = if (isOption) paramType.typeArgs.head.toString else paramType.toString
if (collectionType) {
typeName match {
case t if t.contains("List[") =>
if (!value.isInstanceOf[List[_]]) List(value)
case t if t.contains("Seq[") =>
if (!value.isInstanceOf[Seq[_]]) Seq(value)
case _ => value
}
} else {
value
}
Expand Down Expand Up @@ -320,7 +345,7 @@ object ReflectionUtils {
if (parameterValues.contains(name)) {
val paramType = Class.forName(param.typeSignature.typeSymbol.fullName)
val optionType = param.typeSignature.typeSymbol.fullName.contains("Option")
val instanceClass = getFinalValue(optionType, parameterValues(name)).getClass
val instanceClass = getFinalValue(param.asTerm.typeSignature, parameterValues(name)).getClass
val instanceType = if (instanceClass.getName == "java.lang.Boolean") Class.forName("scala.Boolean") else instanceClass
parameterValues.contains(name) && paramType.isAssignableFrom(instanceType)
} else {
Expand All @@ -330,4 +355,10 @@ object ReflectionUtils {

matches.length
}

private def isCollection(optionType: Boolean, paramType: ru.Type): Boolean = if (optionType) {
paramType.typeArgs.head <:< typeOf[Seq[_]]
} else {
paramType <:< typeOf[Seq[_]]
}
}
8 changes: 6 additions & 2 deletions spark-pipeline-engine/src/test/scala/com/acxiom/pipeline/MockStepObject.scala
View file
Expand Up @@ -15,12 +15,16 @@ object MockStepObject {
default
}

def mockStepFunctionWithListParams(list: List[String], seq: Seq[Int], arrayList: java.util.ArrayList[String]): String ={
s"${list.headOption},${seq.headOption},${if(arrayList.isEmpty) None else Some(arrayList.get(0))}"
}

def mockStepFunctionAnyResponse(string: String): String = {
string
}

def mockStepFunctionWithOptionalGenericParams(list: Option[Seq[String]]): String ={
list.getOrElse(List("chicken")).headOption.getOrElse("chicken")
def mockStepFunctionWithOptionalGenericParams(string: Option[String]): String ={
string.getOrElse("chicken")
}

def mockStepFunctionWithPrimitives(i: Int, l: Long, d: Double, f: Float, c: Char, by: Option[Byte], s: Short, a: Any): Int ={
Expand Down
31 changes: 27 additions & 4 deletions spark-pipeline-engine/src/test/scala/com/acxiom/pipeline/utils/ReflectionUtilsTests.scala
View file
@@ -1,13 +1,16 @@
package com.acxiom.pipeline.utils

import java.util

import com.acxiom.pipeline.{PipelineStepResponse, _}
import org.scalatest.FunSpec

class ReflectionUtilsTests extends FunSpec {
private val FIVE = 5
describe("ReflectionUtil - processStep") {
val pipeline = Pipeline(Some("TestPipeline"))
val pipelineContext = PipelineContext(None, None, None, PipelineSecurityManager(), PipelineParameters(),
val globals = Map[String, Any]("validateStepParameterTypes" -> true)
val pipelineContext = PipelineContext(None, None, Some(globals), PipelineSecurityManager(), PipelineParameters(),
Some(List("com.acxiom.pipeline.steps", "com.acxiom.pipeline")), PipelineStepMapper(), None, None)
it("Should process step function") {
val step = PipelineStep(None, None, None, None, None, Some(EngineMeta(Some("MockStepObject.mockStepFunction"))))
Expand Down Expand Up @@ -53,6 +56,16 @@ class ReflectionUtilsTests extends FunSpec {
assert(response.asInstanceOf[PipelineStepResponse].primaryReturn.get == "chicken")
}

it("Should wrap values in a List, Seq, or Array if passing a single element to a collection") {
val step = PipelineStep(None, None, None, None, None,
Some(EngineMeta(Some("MockStepObject.mockStepFunctionWithListParams"))))
val response = ReflectionUtils.processStep(step, pipeline,
Map[String, Any]("list" -> "l1", "seq" -> 1, "arrayList" -> new util.ArrayList()), pipelineContext)
assert(response.isInstanceOf[PipelineStepResponse])
assert(response.asInstanceOf[PipelineStepResponse].primaryReturn.isDefined)
assert(response.asInstanceOf[PipelineStepResponse].primaryReturn.get == "Some(l1),Some(1),None")
}

it("Should return an informative error if a step function is not found") {
val step = PipelineStep(None, None, None, None, None,
Some(EngineMeta(Some("MockStepObject.typo"))))
Expand All @@ -62,14 +75,24 @@ class ReflectionUtilsTests extends FunSpec {
assert(thrown.getMessage == "typo is not a valid function!")
}

it("Should respect the validateStepParameterTypes global"){
val step = PipelineStep(Some("chicken"), None, None, None, None,
Some(EngineMeta(Some("MockStepObject.mockStepFunctionWithOptionalGenericParams"))))
val thrown = intercept[ClassCastException] {
ReflectionUtils.processStep(step, pipeline, Map[String, Any]("string" -> 1), pipelineContext.setGlobal("validateStepParameterTypes", false))
}
val message = "java.lang.Integer cannot be cast to java.lang.String"
assert(thrown.getMessage == message)
}

it("Should return an informative error if the parameter types do not match function params"){
val step = PipelineStep(Some("chicken"), None, None, None, None,
Some(EngineMeta(Some("MockStepObject.mockStepFunctionWithOptionalGenericParams"))))
val thrown = intercept[PipelineException] {
ReflectionUtils.processStep(step, pipeline, Map[String, Any]("list" -> 1), pipelineContext)
ReflectionUtils.processStep(step, pipeline, Map[String, Any]("string" -> 1), pipelineContext)
}
val message = "Failed to map value [Some(1)] of type [Some(Integer)] to paramName [list] of" +
" type [Option[Seq[String]]] for method [mockStepFunctionWithOptionalGenericParams] in step [chicken] in pipeline [TestPipeline]"
val message = "Failed to map value [Some(1)] of type [Some(Integer)] to paramName [string] of" +
" type [Option[String]] for method [mockStepFunctionWithOptionalGenericParams] in step [chicken] in pipeline [TestPipeline]"
assert(thrown.getMessage == message)
}

Expand Down

0 comments on commit 6fa7263

Please sign in to comment.