/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.gearpump.akkastream.materializer

import akka.actor.ActorSystem
import akka.stream.impl.StreamLayout.Module
import akka.stream.impl.Timers.{Completion, DelayInitial, Idle, IdleInject, IdleTimeoutBidi, Initial}
import akka.stream.impl.fusing.{Batch, Collect, Delay, Drop, DropWhile, DropWithin, Filter, FlattenMerge, Fold, GraphStageModule, GroupBy, GroupedWithin, Intersperse, LimitWeighted, Log, MapAsync, MapAsyncUnordered, PrefixAndTail, Recover, Reduce, Scan, Split, StatefulMapConcat, SubSink, SubSource, Take, TakeWhile, TakeWithin, Map => FMap}
import akka.stream.impl.fusing.GraphStages.{MaterializedValueSource, SimpleLinearGraphStage, SingleSource, TickSource}
import akka.stream.impl.io.IncomingConnectionStage
import akka.stream.impl.{HeadOptionStage, Stages, Throttle, Unfold, UnfoldAsync}
import akka.stream.scaladsl.{Balance, Broadcast, Concat, Interleave, Merge, MergePreferred, MergeSorted, ModuleExtractor, Unzip, Zip, ZipWith2}
import akka.stream.stage.AbstractStage.PushPullGraphStageWithMaterializedValue
import akka.stream.stage.GraphStage
import org.apache.gearpump.akkastream.GearAttributes
import org.apache.gearpump.akkastream.GearpumpMaterializer.Edge
import org.apache.gearpump.akkastream.module.{GroupByModule, ProcessorModule, ReduceModule, SinkBridgeModule, SinkTaskModule, SourceBridgeModule, SourceTaskModule}
import org.apache.gearpump.akkastream.task.{BalanceTask, BatchTask, BroadcastTask, ConcatTask, DelayInitialTask, DropWithinTask, FlattenMergeTask, FoldTask, GraphTask, GroupedWithinTask, InterleaveTask, MapAsyncTask, MergeTask, SingleSourceTask, SinkBridgeTask, SourceBridgeTask, StatefulMapConcatTask, TakeWithinTask, ThrottleTask, TickSourceTask, Zip2Task}
import org.apache.gearpump.akkastream.task.TickSourceTask.{INITIAL_DELAY, INTERVAL, TICK}
import org.apache.gearpump.cluster.UserConfig
import org.apache.gearpump.streaming.dsl.plan.functions.FlatMapper
import org.apache.gearpump.streaming.dsl.plan.{ChainableOp, DataSinkOp, DataSourceOp, Direct, GroupByOp, MergeOp, Op, OpEdge, ProcessorOp, Shuffle}
import org.apache.gearpump.streaming.dsl.scalaapi.StreamApp
import org.apache.gearpump.streaming.dsl.scalaapi.functions.FlatMapFunction
import org.apache.gearpump.streaming.dsl.window.api.CountWindow
import org.apache.gearpump.streaming.dsl.window.impl.GroupAlsoByWindow
import org.apache.gearpump.streaming.{ProcessorId, StreamApplication}
import org.apache.gearpump.util.Graph
import org.slf4j.LoggerFactory

import scala.concurrent.Promise
import scala.concurrent.duration.FiniteDuration

/**
 * [[RemoteMaterializerImpl]] will materialize the [[Graph[Module, Edge]] to a Gearpump
 * Streaming Application.
 *
 * @param graph Graph
 * @param system ActorSystem
 */
class RemoteMaterializerImpl(graph: Graph[Module, Edge], system: ActorSystem) {

  import RemoteMaterializerImpl._

  type ID = String
  private implicit val actorSystem = system

  private def uuid: String = {
    java.util.UUID.randomUUID.toString
  }

  def materialize: (StreamApplication, Map[Module, ProcessorId]) = {
    val (opGraph, ids) = toOpGraph
    val app: StreamApplication = new StreamApp("app", system, UserConfig.empty, opGraph)
    val processorIds = resolveIds(app, ids)

    val updatedApp = updateJunctionConfig(processorIds, app)
    (removeIds(updatedApp), processorIds)
  }

  private def updateJunctionConfig(processorIds: Map[Module, ProcessorId],
      app: StreamApplication): StreamApplication = {
    val config = junctionConfig(processorIds)

    val dag = app.dag.mapVertex { vertex =>
      val processorId = vertex.id
      val newConf = vertex.taskConf.withConfig(config(processorId))
      vertex.copy(taskConf = newConf)
    }
    new StreamApplication(app.name, app.inputUserConfig, dag)
  }

  private def junctionConfig(processorIds: Map[Module, ProcessorId]):
  Map[ProcessorId, UserConfig] = {
    val updatedConfigs = graph.vertices.flatMap { vertex =>
      buildShape(vertex, processorIds)
    }.toMap
    updatedConfigs
  }

  private def buildShape(vertex: Module, processorIds: Map[Module, ProcessorId]):
  Option[(ProcessorId, UserConfig)] = {
    def inProcessors(vertex: Module): List[ProcessorId] = {
      vertex.shape.inlets.flatMap { inlet =>
        graph.incomingEdgesOf(vertex).find(
          _._2.to == inlet).map(_._1
        ).flatMap(processorIds.get)
      }.toList
    }
    def outProcessors(vertex: Module): List[ProcessorId] = {
      vertex.shape.outlets.flatMap { outlet =>
        graph.outgoingEdgesOf(vertex).find(
          _._2.from == outlet).map(_._3
        ).flatMap(processorIds.get)
      }.toList
    }
    processorIds.get(vertex).map(processorId => {
      (processorId, UserConfig.empty.
        withValue(GraphTask.OUT_PROCESSORS, outProcessors(vertex)).
        withValue(GraphTask.IN_PROCESSORS, inProcessors(vertex)))
    })
  }

  private def resolveIds(app: StreamApplication, ids: Map[Module, ID]):
  Map[Module, ProcessorId] = {
    ids.flatMap { kv =>
      val (module, id) = kv
      val processorId = app.dag.vertices.find { processor =>
        processor.taskConf.getString(id).isDefined
      }.map(_.id)
      processorId.map((module, _))
    }
  }

  private def removeIds(app: StreamApplication): StreamApplication = {
    val graph = app.dag.mapVertex { processor =>
      val conf = removeId(processor.taskConf)
      processor.copy(taskConf = conf)
    }
    new StreamApplication(app.name, app.inputUserConfig, graph)
  }

  private def removeId(conf: UserConfig): UserConfig = {
    conf.filter { kv =>
      kv._2 != RemoteMaterializerImpl.TRACKABLE
    }
  }

  private def toOpGraph: (Graph[Op, OpEdge], Map[Module, ID]) = {
    var matValues = collection.mutable.Map.empty[Module, ID]
    val opGraph = graph.mapVertex[Op] { module =>
      val name = uuid
      val conf = UserConfig.empty.withString(name, RemoteMaterializerImpl.TRACKABLE)
      matValues += module -> name
      val parallelism = GearAttributes.count(module.attributes)
      val op = module match {
        case source: SourceTaskModule[_] =>
          val updatedConf = conf.withConfig(source.conf)
          DataSourceOp(source.source, parallelism, updatedConf, "source")
        case sink: SinkTaskModule[_] =>
          val updatedConf = conf.withConfig(sink.conf)
          DataSinkOp(sink.sink, parallelism, updatedConf, "sink")
        case sourceBridge: SourceBridgeModule[_, _] =>
          ProcessorOp(classOf[SourceBridgeTask], parallelism = 1, conf, "source")
        case processor: ProcessorModule[_, _, _] =>
          val updatedConf = conf.withConfig(processor.conf)
          ProcessorOp(processor.processor, parallelism, updatedConf, "source")
        case sinkBridge: SinkBridgeModule[_, _] =>
          ProcessorOp(classOf[SinkBridgeTask], parallelism, conf, "sink")
        case groupBy: GroupByModule[_, _] =>
          GroupByOp(GroupAlsoByWindow(groupBy.groupBy, CountWindow.apply(1).accumulating),
            parallelism, "groupBy", conf)
        case reduce: ReduceModule[_] =>
          reduceOp(reduce.f, conf)
        case graphStage: GraphStageModule =>
          translateGraphStageWithMaterializedValue(graphStage, parallelism, conf)
        case _ =>
          null
      }
      if (op == null) {
        throw new UnsupportedOperationException(
          module.getClass.toString + " is not supported with RemoteMaterializer"
        )
      }
      op
    }.mapEdge[OpEdge] { (n1, edge, n2) =>
      n2 match {
        case chainableOp: ChainableOp[_, _]
          if !n1.isInstanceOf[ProcessorOp[_]] && !n2.isInstanceOf[ProcessorOp[_]] =>
          Direct
        case _ =>
          Shuffle
      }
    }
    (opGraph, matValues.toMap)
  }

  private def translateGraphStageWithMaterializedValue(module: GraphStageModule,
      parallelism: Int, conf: UserConfig): Op = {
    module.stage match {
      case tickSource: TickSource[_] =>
        val tick: AnyRef = tickSource.tick.asInstanceOf[AnyRef]
        val tiConf = conf.withValue[FiniteDuration](INITIAL_DELAY, tickSource.initialDelay).
          withValue[FiniteDuration](INTERVAL, tickSource.interval).
          withValue(TICK, tick)
        ProcessorOp(classOf[TickSourceTask[_]], parallelism, tiConf, "tickSource")
      case graphStage: GraphStage[_] =>
        translateGraphStage(module, parallelism, conf)
      case headOptionStage: HeadOptionStage[_] =>
        headOptionOp(headOptionStage, conf)
      case pushPullGraphStageWithMaterializedValue:
        PushPullGraphStageWithMaterializedValue[_, _, _, _] =>
        translateSymbolic(pushPullGraphStageWithMaterializedValue, conf)
    }
  }

  private def translateGraphStage(module: GraphStageModule,
      parallelism: Int, conf: UserConfig): Op = {
    module.stage match {
      case balance: Balance[_] =>
        ProcessorOp(classOf[BalanceTask], parallelism, conf, "balance")
      case batch: Batch[_, _] =>
        val batchConf = conf.withValue[_ => Long](BatchTask.COST, batch.costFn).
          withLong(BatchTask.MAX, batch.max).
          withValue[(_, _) => _](BatchTask.AGGREGATE, batch.aggregate).
          withValue[_ => _](BatchTask.SEED, batch.seed)
        ProcessorOp(classOf[BatchTask[_, _]],
          parallelism, batchConf, "batch")
      case broadcast: Broadcast[_] =>
        val name = ModuleExtractor.unapply(broadcast).map(_.attributes.nameOrDefault()).get
        ProcessorOp(classOf[BroadcastTask], parallelism, conf, name)
      case collect: Collect[_, _] =>
        collectOp(collect.pf, conf)
      case concat: Concat[_] =>
        ProcessorOp(classOf[ConcatTask], parallelism, conf, "concat")
      case delayInitial: DelayInitial[_] =>
        val dIConf = conf.withValue[FiniteDuration](
          DelayInitialTask.DELAY_INITIAL, delayInitial.delay)
        ProcessorOp(classOf[DelayInitialTask[_]], parallelism, dIConf, "delayInitial")
      case dropWhile: DropWhile[_] =>
        dropWhileOp(dropWhile.p, conf)
      case flattenMerge: FlattenMerge[_, _] =>
        ProcessorOp(classOf[FlattenMergeTask], parallelism, conf, "flattenMerge")
      case fold: Fold[_, _] =>
        val foldConf = conf.withValue(FoldTask.ZERO, fold.zero.asInstanceOf[AnyRef]).
          withValue(FoldTask.AGGREGATOR, fold.f)
        ProcessorOp(classOf[FoldTask[_, _]], parallelism, foldConf, "fold")
      case groupBy: GroupBy[_, _] =>
        GroupByOp(GroupAlsoByWindow(groupBy.keyFor, CountWindow.apply(1).accumulating),
          groupBy.maxSubstreams, "groupBy", conf)
      case groupedWithin: GroupedWithin[_] =>
        val diConf = conf.withValue[FiniteDuration](GroupedWithinTask.TIME_WINDOW, groupedWithin.d).
          withInt(GroupedWithinTask.BATCH_SIZE, groupedWithin.n)
        ProcessorOp(classOf[GroupedWithinTask[_]], parallelism, diConf, "groupedWithin")
      case idleInject: IdleInject[_, _] =>
        // TODO
        null
      case idleTimeoutBidi: IdleTimeoutBidi[_, _] =>
        // TODO
        null
      case incomingConnectionStage: IncomingConnectionStage =>
        // TODO
        null
      case interleave: Interleave[_] =>
        val ilConf = conf.withInt(InterleaveTask.INPUT_PORTS, interleave.inputPorts).
          withInt(InterleaveTask.SEGMENT_SIZE, interleave.segmentSize)
        ProcessorOp(classOf[InterleaveTask], parallelism, ilConf, "interleave")
        null
      case intersperse: Intersperse[_] =>
        // TODO
        null
      case limitWeighted: LimitWeighted[_] =>
        // TODO
        null
      case map: FMap[_, _] =>
        mapOp(map.f, conf)
      case mapAsync: MapAsync[_, _] =>
        ProcessorOp(classOf[MapAsyncTask[_, _]],
          mapAsync.parallelism, conf.withValue(MapAsyncTask.MAPASYNC_FUNC, mapAsync.f), "mapAsync")
      case mapAsyncUnordered: MapAsyncUnordered[_, _] =>
        ProcessorOp(classOf[MapAsyncTask[_, _]],
          mapAsyncUnordered.parallelism,
          conf.withValue(MapAsyncTask.MAPASYNC_FUNC, mapAsyncUnordered.f), "mapAsyncUnordered")
      case materializedValueSource: MaterializedValueSource[_] =>
        // TODO
        null
      case merge: Merge[_] =>
        val mergeConf = conf.withBoolean(MergeTask.EAGER_COMPLETE, merge.eagerComplete).
          withInt(MergeTask.INPUT_PORTS, merge.inputPorts)
        ProcessorOp(classOf[MergeTask], parallelism, mergeConf, "merge")
      case mergePreferred: MergePreferred[_] =>
        MergeOp("mergePreferred", conf)
      case mergeSorted: MergeSorted[_] =>
        MergeOp("mergeSorted", conf)
      case prefixAndTail: PrefixAndTail[_] =>
        // TODO
        null
      case recover: Recover[_] =>
        // TODO
        null
      case scan: Scan[_, _] =>
        scanOp(scan.zero, scan.f, conf)
      case simpleLinearGraphStage: SimpleLinearGraphStage[_] =>
        translateSimpleLinearGraph(simpleLinearGraphStage, parallelism, conf)
      case singleSource: SingleSource[_] =>
        val singleSourceConf = conf.withValue[AnyRef](SingleSourceTask.ELEMENT,
          singleSource.elem.asInstanceOf[AnyRef])
        ProcessorOp(classOf[SingleSourceTask[_]], parallelism, singleSourceConf, "singleSource")
      case split: Split[_] =>
        // TODO
        null
      case statefulMapConcat: StatefulMapConcat[_, _] =>
        val func = statefulMapConcat.f
        val statefulMapConf =
          conf.withValue[() => _ => Iterable[_]](StatefulMapConcatTask.FUNC, func)
        ProcessorOp(classOf[StatefulMapConcatTask[_, _]], parallelism,
          statefulMapConf, "statefulMapConcat")
      case subSink: SubSink[_] =>
        // TODO
        null
      case subSource: SubSource[_] =>
        // TODO
        null
      case unfold: Unfold[_, _] =>
        // TODO
        null
      case unfoldAsync: UnfoldAsync[_, _] =>
        // TODO
        null
      case unzip: Unzip[_, _] =>
        // ProcessorOp(classOf[Unzip2Task[_, _, _]], parallelism,
        //   conf.withValue(
        //     Unzip2Task.UNZIP2_FUNCTION, Unzip2Task.UnZipFunction(unzip.unzipper)), "unzip")
        // TODO
        null
      case zip: Zip[_, _] =>
        zipWithOp(zip.zipper, conf)
      case zipWith2: ZipWith2[_, _, _] =>
        ProcessorOp(classOf[Zip2Task[_, _, _]],
          parallelism,
          conf.withValue(
            Zip2Task.ZIP2_FUNCTION, Zip2Task.ZipFunction(zipWith2.zipper)
          ), "zipWith2")
    }
  }

  private def translateSimpleLinearGraph(stage: SimpleLinearGraphStage[_],
      parallelism: Int, conf: UserConfig): Op = {
    stage match {
      case completion: Completion[_] =>
        // TODO
        null
      case delay: Delay[_] =>
        // TODO
        null
      case drop: Drop[_] =>
        dropOp(drop.count, conf)
      case dropWithin: DropWithin[_] =>
        val dropWithinConf =
          conf.withValue[FiniteDuration](DropWithinTask.TIMEOUT, dropWithin.timeout)
        ProcessorOp(classOf[DropWithinTask[_]],
          parallelism, dropWithinConf, "dropWithin")
      case filter: Filter[_] =>
        filterOp(filter.p, conf)
      case idle: Idle[_] =>
        // TODO
        null
      case initial: Initial[_] =>
        // TODO
        null
      case log: Log[_] =>
        logOp(log.name, log.extract, conf)
      case reduce: Reduce[_] =>
        reduceOp(reduce.f, conf)
      case take: Take[_] =>
        takeOp(take.count, conf)
      case takeWhile: TakeWhile[_] =>
        filterOp(takeWhile.p, conf)
      case takeWithin: TakeWithin[_] =>
        val takeWithinConf =
          conf.withValue[FiniteDuration](TakeWithinTask.TIMEOUT, takeWithin.timeout)
        ProcessorOp(classOf[TakeWithinTask[_]],
          parallelism, takeWithinConf, "takeWithin")
      case throttle: Throttle[_] =>
        val throttleConf = conf.withInt(ThrottleTask.COST, throttle.cost).
          withInt(ThrottleTask.MAX_BURST, throttle.maximumBurst).
          withValue[_ => Int](ThrottleTask.COST_CALC, throttle.costCalculation).
          withValue[FiniteDuration](ThrottleTask.TIME_PERIOD, throttle.per)
        ProcessorOp(classOf[ThrottleTask[_]],
          parallelism, throttleConf, "throttle")
    }
  }

  private def translateSymbolic(stage: PushPullGraphStageWithMaterializedValue[_, _, _, _],
      conf: UserConfig): Op = {
    stage match {
      case symbolicGraphStage: Stages.SymbolicGraphStage[_, _, _]
        if symbolicGraphStage.symbolicStage.attributes.equals(
          Stages.DefaultAttributes.buffer) => {
            // ignore the buffering operation
            identity("buffer", conf)
        }
    }
  }

}

object RemoteMaterializerImpl {
  final val NotApplied: Any => Any = _ => NotApplied

  def collectOp[In, Out](collect: PartialFunction[In, Out], conf: UserConfig): Op = {
    flatMapOp({ data: In =>
      collect.applyOrElse(data, NotApplied) match {
        case NotApplied => None
        case result: Any => Option(result)
      }
    }, "collect", conf)
  }

  def filterOp[In](filter: In => Boolean, conf: UserConfig): Op = {
    flatMapOp({ data: In =>
      if (filter(data)) Option(data) else None
    }, "filter", conf)
  }

  def headOptionOp[T](headOptionStage: HeadOptionStage[T], conf: UserConfig): Op = {
    val promise: Promise[Option[T]] = Promise()
    flatMapOp({ data: T =>
      data match {
        case None =>
          Some(promise.future.failed)
        case Some(d) =>
          promise.future.value
      }
    }, "headOption", conf)
  }

  def reduceOp[T](reduce: (T, T) => T, conf: UserConfig): Op = {
    var result: Option[T] = None
    val flatMap = { elem: T =>
      result match {
        case None =>
          result = Some(elem)
        case Some(r) =>
          result = Some(reduce(r, elem))
      }
      List(result)
    }
    flatMapOp(flatMap, "reduce", conf)
  }

  def zipWithOp[In1, In2](zipWith: (In1, In2) => (In1, In2), conf: UserConfig): Op = {
    val flatMap = { elem: (In1, In2) =>
      val (e1, e2) = elem
      val result: (In1, In2) = zipWith(e1, e2)
      List(result)
    }
    flatMapOp(flatMap, "zipWith", conf)
  }

  def zipWithOp2[In1, In2, Out](zipWith: (In1, In2) => Out, conf: UserConfig): Op = {
    val flatMap = { elem: (In1, In2) =>
      val (e1, e2) = elem
      val result: Out = zipWith(e1, e2)
      List(result)
    }
    flatMapOp(flatMap, "zipWith", conf)
  }

  def identity(description: String, conf: UserConfig): Op = {
    flatMapOp({ data: Any =>
      List(data)
    }, description, conf)
  }

  def mapOp[In, Out](map: In => Out, conf: UserConfig): Op = {
    val flatMap = (data: In) => List(map(data))
    flatMapOp (flatMap, conf)
  }

  def flatMapOp[In, Out](flatMap: In => Iterable[Out], conf: UserConfig): Op = {
    flatMapOp(flatMap, "flatmap", conf)
  }

  def flatMapOp[In, Out](fun: In => TraversableOnce[Out], description: String,
      conf: UserConfig): Op = {
    ChainableOp(new FlatMapper(FlatMapFunction[In, Out](fun), description), conf)
  }

  def conflateOp[In, Out](seed: In => Out, aggregate: (Out, In) => Out,
    conf: UserConfig): Op = {
    var agg = None: Option[Out]
    val flatMap = {elem: In =>
      agg = agg match {
        case None =>
          Some(seed(elem))
        case Some(value) =>
          Some(aggregate(value, elem))
      }
      List(agg.get)
    }
    flatMapOp (flatMap, "conflate", conf)
  }

  def foldOp[In, Out](zero: Out, fold: (Out, In) => Out, conf: UserConfig): Op = {
    var aggregator: Out = zero
    val map = { elem: In =>
      aggregator = fold(aggregator, elem)
      List(aggregator)
    }
    flatMapOp(map, "fold", conf)
  }

  def groupedOp(count: Int, conf: UserConfig): Op = {
    var left = count
    val buf = {
      val b = Vector.newBuilder[Any]
      b.sizeHint(count)
      b
    }

    val flatMap: Any => Iterable[Any] = {input: Any =>
      buf += input
      left -= 1
      if (left == 0) {
        val emit = buf.result()
        buf.clear()
        left = count
        Some(emit)
      } else {
        None
      }
    }
    flatMapOp(flatMap, conf: UserConfig)
  }

  def dropOp[T](number: Long, conf: UserConfig): Op = {
    var left = number
    val flatMap: T => Iterable[T] = {input: T =>
      if (left > 0) {
        left -= 1
        None
      } else {
        Some(input)
      }
    }
    flatMapOp(flatMap, "drop", conf)
  }

  def dropWhileOp[In](drop: In => Boolean, conf: UserConfig): Op = {
    flatMapOp({ data: In =>
      if (drop(data))  None else Option(data)
    }, "dropWhile", conf)
  }

  def logOp[T](name: String, extract: T => Any, conf: UserConfig): Op = {
    val flatMap = {elem: T =>
      LoggerFactory.getLogger(name).info(s"Element: {${extract(elem)}}")
      List(elem)
    }
    flatMapOp(flatMap, "log", conf)
  }

  def scanOp[In, Out](zero: Out, f: (Out, In) => Out, conf: UserConfig): Op = {
    var aggregator = zero
    var pushedZero = false

    val flatMap = {elem: In =>
      aggregator = f(aggregator, elem)

      if (pushedZero) {
        pushedZero = true
        List(zero, aggregator)
      } else {
        List(aggregator)
      }
    }
    flatMapOp(flatMap, "scan", conf)
  }

  def statefulMapOp[In, Out](f: In => Iterable[Out], conf: UserConfig): Op = {
    flatMapOp ({ data: In =>
      f(data)
    }, conf)
  }

  def takeOp(count: Long, conf: UserConfig): Op = {
    var left: Long = count

    val filter: Any => Iterable[Any] = {elem: Any =>
      left -= 1
      if (left > 0) Some(elem)
      else if (left == 0) Some(elem)
      else None
    }
    flatMapOp(filter, "take", conf)
  }

  /**
   * We use this attribute to track module to Processor
   *
   */
  val TRACKABLE = "track how module is fused to processor"
}