object StreamingApp {
  def main(args: Array[String]): Unit = {
    val params = new ParamsUtil(args)
    require(params.hasParam("streaming.name"), "Application name should be set")
    PlatformManager.getOrCreate.run(params)
  }
}

trait StreamingRuntime {
  def startRuntime: StreamingRuntime
  def destroyRuntime(stopGraceful: Boolean, stopContext: Boolean = false): Boolean
  def streamingRuntimeInfo: StreamingRuntimeInfo
  def resetRuntimeOperator(runtimeOperator: RuntimeOperator)
  def configureStreamingRuntimeInfo(streamingRuntimeInfo: StreamingRuntimeInfo)
  def awaitTermination
  def startThriftServer
  def startHttpServer
  def params: JMap[Any, Any]
}

  def platformNameMapping = Map[String, String](
    SPAKR_S_S -> "streaming.core.strategy.platform.SparkStructuredStreamingRuntime",
    SPAKR_STRUCTURED_STREAMING -> "streaming.core.strategy.platform.SparkStructuredStreamingRuntime",
    FLINK_STREAMING -> "streaming.core.strategy.platform.FlinkStreamingRuntime",
    SPAKR_STREAMING -> "streaming.core.strategy.platform.SparkStreamingRuntime",
    SPARK -> "streaming.core.strategy.platform.SparkRuntime"
  )

{
  "convert-multi-csv-to-json": {
    "desc": "测试",
    "strategy": "spark",
    "algorithm": [
      {
        "name": "testProcessor"
      }
    ],
    "ref": [],
    "compositor": [
      {
        "name": "testCompositor"
      }
    ],
    "configParams": {
    }
  }
}

class DefaultShortNameMapping extends ShortNameMapping {
  private val compositorNameMap: Map[String, String] = Map[String, String](
    "spark" -> "serviceframework.dispatcher.test.DefaultStrategy",
    "testProcessor" -> "serviceframework.dispatcher.test.TestProcessor",
    "testCompositor" -> "serviceframework.dispatcher.test.TestCompositor"
  )
  override def forName(shortName: String): String = {
    if (compositorNameMap.contains(shortName)) compositorNameMap(shortName)
    else shortName
  }
}

trait Strategy[T] extends ServiceInj{
  def processor:JList[Processor[T]]
  def ref:JList[Strategy[T]]
  def compositor:JList[Compositor[T]]
  def name:String
  def initialize(name:String,alg:JList[Processor[T]],ref:JList[Strategy[T]],com:JList[Compositor[T]],params:JMap[Any,Any])
  def result(params:JMap[Any,Any]):JList[T]
  def configParams:util.Map[Any, Any]
  def stop = {}
}

  def createStrategy(name: String, desc: JMap[_, _]): Option[Strategy[T]] = {
    if (_strategies.contains(name)) return None;
    // 实例化策略，如果有shortName，则先获取fullName，并通过Class.forName实例化具体的策略类
    val strategy = Class.forName(shortNameMapping.forName(desc.get("strategy").asInstanceOf[String])).newInstance().asInstanceOf[Strategy[T]]
    // 读取配置信息，并实例化为Map[Any,Any]
    val configParams: JMap[Any, Any] = if (desc.containsKey("configParams")) desc.get("configParams").asInstanceOf[JMap[Any, Any]] else new java.util.HashMap()
    // 初始化策略，需要创建算法、引用、组合器
    strategy.initialize(name, createAlgorithms(desc), createRefs(desc), createCompositors(desc), configParams)
    _strategies.put(name, strategy)
    Option(strategy)
  }

  // 创建算法。一个策略由0个或者多个算法提供结果
  private def createAlgorithms(jobJMap: JMap[String, Any]): JList[Processor[T]] = {
    if (!jobJMap.contains("algorithm") && !jobJMap.contains("processor")) return new AList[Processor[T]]()
    val processors = if (jobJMap.contains("algorithm")) jobJMap("algorithm") else jobJMap("processor")
    processors.asInstanceOf[JList[JMap[String, Any]]].map {
      alg =>
        val name = shortName2FullName(alg)
        val processor = Class.forName(name).newInstance().asInstanceOf[Processor[T]]
        val params: JList[JMap[String, Any]] = if (alg.contains("params")) alg("params").asInstanceOf[JList[JMap[String, Any]]] else new AList[JMap[String, Any]]()
        processor.initialize(name, params)
        processor
    }
  }

  // 创建组合器，可以多个，按顺序调用。有点类似过滤器链。第一个过滤器会接受算法或者策略的结果。后续的组合器就只能处理上一阶段的组合器吐出的结果
  private def createCompositors(jobJMap: JMap[String, Any]): JList[Compositor[T]] = {
    if (!jobJMap.contains("compositor")) return new AList()
    val compositors = jobJMap.get("compositor")
    compositors.asInstanceOf[JList[JMap[String, Any]]].map {
      f =>
        val compositor = Class.forName(shortName2FullName(f)).newInstance().asInstanceOf[Compositor[T]]
        val params: JList[JMap[String, Any]] = if (f.contains("params")) f.get("params").asInstanceOf[JList[JMap[String, Any]]] else new AList[JMap[String, Any]]()
        compositor.initialize(f.get("typeFilter").asInstanceOf[JList[String]], params)
        compositor
    }
  }

  def dispatch(params: JMap[Any, Any]): JList[T] = {
    findStrategies(clientType) match {
      case Some(strategies) =>
        strategies.flatMap { f => f.result(params) }
    }
  }

class SparkRuntime(_params: JMap[Any, Any]) extends StreamingRuntime with PlatformManagerListener {
  override def startRuntime: StreamingRuntime = this

  var sparkSession: SparkSession = createRuntime
  def createRuntime = {
    //...创建SparkSession，这里会根据参数判断是否支持Hive、Carbondata
  }

  params.put("_session_", sparkSession) //将SparkSession放入params中
  registerUDF  

  override def params: JMap[Any, Any] = _params
}

class SparkStreamingRuntime(_params: JMap[Any, Any]) extends StreamingRuntime with PlatformManagerListener { self =>
  var streamingContext: StreamingContext = createRuntime
  def createRuntime = {
    //创建StreamingContext，并将SparkSession放入params中
  }

  override def startRuntime = {
    streamingContext.start()
    this
  }
  override def awaitTermination = streamingContext.awaitTermination()
}

class SparkStreamingStrategy[T] extends Strategy[T] with DebugTrait with JobStrategy {
  var _ref: util.List[Strategy[T]] = _
  var _compositor: util.List[Compositor[T]] = _
  var _processor: util.List[Processor[T]] = _
  var _configParams: util.Map[Any, Any] = _

  def result(params: util.Map[Any, Any]): util.List[T] = {
    ref.foreach { r => r.result(params) } // 先执行ref
    if (compositor != null && compositor.size() > 0) {
      // 第一个Compositor, 产生第一个中间结果
      var middleR = compositor.get(0).result(processor, ref, null, params)
      // 将新的中间结果运用到下一个Compositor
      // 第一个Compositor的结果运用到第二个的输入, 第二个Compositor的结果运用到第三个Compositor的输入...
      // 所以不同Compositor是链式执行的
      for (i <- 1 until compositor.size()) {
        middleR = compositor.get(i).result(processor, ref, middleR, params)
      }
      middleR
    } else new util.ArrayList[T]()
  }  
}