eagle-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From h..@apache.org
Subject [37/55] [abbrv] [partial] incubator-eagle git commit: [EAGLE-46] Rename package name as "org.apache.eagle"
Date Thu, 19 Nov 2015 10:47:44 GMT
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/FilterBoltWrapper.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/FilterBoltWrapper.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/FilterBoltWrapper.scala
deleted file mode 100644
index cf9b53f..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/FilterBoltWrapper.scala
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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 eagle.datastream
-
-import java.util
-
-import backtype.storm.task.{OutputCollector, TopologyContext}
-import backtype.storm.topology.OutputFieldsDeclarer
-import backtype.storm.topology.base.BaseRichBolt
-import backtype.storm.tuple.{Fields, Tuple}
-import org.slf4j.LoggerFactory
-
-case class FilterBoltWrapper[T](fn : T => Boolean) extends BaseRichBolt{
-  val LOG = LoggerFactory.getLogger(FilterBoltWrapper.getClass)
-  var _collector : OutputCollector = null
-
-  override def prepare(stormConf: util.Map[_, _], context: TopologyContext, collector: OutputCollector): Unit = {
-    _collector = collector
-  }
-
-  override def execute(input : Tuple): Unit = {
-    input.getValue(0) match {
-      case v:T =>
-        if(fn(v)){
-          _collector.emit(input, input.getValues)
-          _collector.ack(input)
-        }
-    }
-  }
-
-  override def declareOutputFields(declarer : OutputFieldsDeclarer): Unit ={
-    declarer.declare(new Fields(OutputFieldNameConst.FIELD_PREFIX + "0"))
-  }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/GraphPrinter.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/GraphPrinter.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/GraphPrinter.scala
deleted file mode 100644
index 24afdbb..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/GraphPrinter.scala
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- *
- *  * 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 eagle.datastream
-
-import org.jgrapht.experimental.dag.DirectedAcyclicGraph
-import org.slf4j.LoggerFactory
-import scala.collection.JavaConversions._
-
-object GraphPrinter {
-  private val LOG = LoggerFactory.getLogger(GraphPrinter.getClass)
-  def print(dag: DirectedAcyclicGraph[StreamProducer, StreamConnector]): Unit ={
-    val iter = dag.iterator()
-    while(iter.hasNext) {
-      val current = iter.next()
-      dag.outgoingEdgesOf(current).foreach(edge => {
-        LOG.info(edge.from + "{" + edge.from.parallelism + "}" +" => " + edge.to + "{" + edge.to.parallelism + "}" + " with groupByFields " + edge.groupByFields)
-      })
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/JavaStormBoltWrapper.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/JavaStormBoltWrapper.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/JavaStormBoltWrapper.scala
deleted file mode 100644
index 13be3c7..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/JavaStormBoltWrapper.scala
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * 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 eagle.datastream
-
-import java.util
-
-import backtype.storm.task.{OutputCollector, TopologyContext}
-import backtype.storm.topology.OutputFieldsDeclarer
-import backtype.storm.topology.base.BaseRichBolt
-import backtype.storm.tuple.{Fields, Tuple}
-import org.slf4j.LoggerFactory
-
-import scala.collection.JavaConverters._
-
-case class JavaStormBoltWrapper(worker : JavaStormStreamExecutor[EagleTuple]) extends BaseRichBolt{
-  val LOG = LoggerFactory.getLogger(StormBoltWrapper.getClass)
-  var _collector : OutputCollector = null
-
-  override def prepare(stormConf: util.Map[_, _], context: TopologyContext, collector: OutputCollector): Unit = {
-    _collector = collector
-    worker.init
-  }
-
-  override def execute(input : Tuple): Unit ={
-    worker.flatMap(input.getValues, new Collector[EagleTuple](){
-      def collect(t: EagleTuple): Unit ={
-        _collector.emit(input, t.getList.asJava)
-      }
-    })
-    _collector.ack(input)
-  }
-
-  override def declareOutputFields(declarer : OutputFieldsDeclarer): Unit ={
-    val fields = worker.fields
-    LOG.info("output fields for worker " + worker + " : " + fields.toList)
-    declarer.declare(new Fields(fields:_*))
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/MapBoltWrapper.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/MapBoltWrapper.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/MapBoltWrapper.scala
deleted file mode 100644
index 7407c32..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/MapBoltWrapper.scala
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * 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 eagle.datastream
-
-import java.util
-
-import backtype.storm.task.{OutputCollector, TopologyContext}
-import backtype.storm.topology.OutputFieldsDeclarer
-import backtype.storm.topology.base.BaseRichBolt
-import backtype.storm.tuple.{Fields, Tuple}
-import org.slf4j.LoggerFactory
-
-/**
- * @since  9/29/15
- */
-case class MapBoltWrapper[T,R](num: Int, fn: T => R) extends BaseRichBolt {
-  val LOG = LoggerFactory.getLogger(FilterBoltWrapper.getClass)
-  var _collector : OutputCollector = null
-
-  override def declareOutputFields(declarer: OutputFieldsDeclarer): Unit = {
-    var fields = new util.ArrayList[String]()
-    var i : Int = 0;
-    while(i < num){
-      fields.add(OutputFieldNameConst.FIELD_PREFIX + i)
-      i += 1
-    }
-    declarer.declare(new Fields(fields))
-  }
-
-  override def execute(input: Tuple): Unit = {
-    val size = input.size()
-    var values : AnyRef = null
-    size match {
-      case 1 => values = scala.Tuple1(input.getValue(0))
-      case 2 => values = scala.Tuple2(input.getValue(0), input.getValue(1))
-      case 3 => values = scala.Tuple3(input.getValue(0), input.getValue(1), input.getValue(2))
-      case 4 => values = scala.Tuple4(input.getValue(0), input.getValue(1), input.getValue(2), input.getValue(3))
-      case _ => throw new IllegalArgumentException
-    }
-    val output = fn(values.asInstanceOf[T])
-    output match {
-      case scala.Tuple1(a) => _collector.emit(input, util.Arrays.asList(a.asInstanceOf[AnyRef]))
-      case scala.Tuple2(a, b) => _collector.emit(input, util.Arrays.asList(a.asInstanceOf[AnyRef], b.asInstanceOf[AnyRef]))
-      case scala.Tuple3(a, b, c) => _collector.emit(input, util.Arrays.asList(a.asInstanceOf[AnyRef], b.asInstanceOf[AnyRef], c.asInstanceOf[AnyRef]))
-      case scala.Tuple4(a, b, c, d) => _collector.emit(input, util.Arrays.asList(a.asInstanceOf[AnyRef], b.asInstanceOf[AnyRef], c.asInstanceOf[AnyRef], d.asInstanceOf[AnyRef]))
-      case a => _collector.emit(input, util.Arrays.asList(a.asInstanceOf[AnyRef]))
-    }
-    _collector.ack(input)
-  }
-
-  override def prepare(stormConf: util.Map[_, _], context: TopologyContext, collector: OutputCollector): Unit = {
-    _collector = collector
-  }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/NodeNameSelector.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/NodeNameSelector.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/NodeNameSelector.scala
deleted file mode 100644
index f579881..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/NodeNameSelector.scala
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * 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 eagle.datastream
-
-case class NodeNameSelector(producer : StreamProducer) {
-  def getName : String = {
-    producer.name match {
-      case null => producer.toString
-      case _ => producer.name
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/OutputFieldNameConst.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/OutputFieldNameConst.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/OutputFieldNameConst.scala
deleted file mode 100644
index b00f149..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/OutputFieldNameConst.scala
+++ /dev/null
@@ -1,21 +0,0 @@
-/*
- * 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 eagle.datastream
-
-object OutputFieldNameConst {
-  val FIELD_PREFIX = "f"
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/SpoutProxy.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/SpoutProxy.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/SpoutProxy.scala
deleted file mode 100644
index 7bed261..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/SpoutProxy.scala
+++ /dev/null
@@ -1,59 +0,0 @@
-/*
- * 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 eagle.datastream
-
-import backtype.storm.spout.SpoutOutputCollector
-import backtype.storm.task.TopologyContext
-import backtype.storm.topology.OutputFieldsDeclarer
-import backtype.storm.topology.base.BaseRichSpout
-import backtype.storm.tuple.Fields
-
-/**
- * Declare delegated BaseRichSpout with given field names
- *
- * @param delegate delegated BaseRichSpout
- * @param outputFields given field names
- */
-case class SpoutProxy(delegate: BaseRichSpout, outputFields: java.util.List[String]) extends BaseRichSpout{
-  def open(conf: java.util.Map[_, _], context: TopologyContext, collector: SpoutOutputCollector) {
-    this.delegate.open(conf, context, collector)
-  }
-
-  def nextTuple {
-    this.delegate.nextTuple
-  }
-
-  override def ack(msgId: AnyRef) {
-    this.delegate.ack(msgId)
-  }
-
-  override def fail(msgId: AnyRef) {
-    this.delegate.fail(msgId)
-  }
-
-  override def deactivate {
-    this.delegate.deactivate
-  }
-
-  override def declareOutputFields(declarer: OutputFieldsDeclarer) {
-    declarer.declare(new Fields(outputFields))
-  }
-
-  override def close {
-    this.delegate.close
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormBoltFactory.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormBoltFactory.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormBoltFactory.scala
deleted file mode 100644
index c35b3e9..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormBoltFactory.scala
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- *
- *  * 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 eagle.datastream
-
-import backtype.storm.topology.base.BaseRichBolt
-import com.typesafe.config.Config
-
-object StormBoltFactory {
-  def getBoltWrapper(graph: AbstractStreamProducerGraph, producer : StreamProducer, config : Config) : BaseRichBolt = {
-    producer match{
-      case FlatMapProducer(id, worker) => {
-        if(worker.isInstanceOf[JavaStormStreamExecutor[EagleTuple]]){
-          worker.asInstanceOf[JavaStormStreamExecutor[EagleTuple]].prepareConfig(config)
-          JavaStormBoltWrapper(worker.asInstanceOf[JavaStormStreamExecutor[EagleTuple]])
-        }else if(worker.isInstanceOf[StormStreamExecutor[EagleTuple]]){
-          worker.asInstanceOf[StormStreamExecutor[EagleTuple]].prepareConfig(config)
-          StormBoltWrapper(worker.asInstanceOf[StormStreamExecutor[EagleTuple]])
-        }else {
-          throw new UnsupportedOperationException
-        }
-      }
-      case FilterProducer(id, fn) => {
-        FilterBoltWrapper(fn)
-      }
-      case MapProducer(id, n, fn) => {
-        MapBoltWrapper(n, fn)
-      }
-      case _ => throw new UnsupportedOperationException
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormBoltWrapper.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormBoltWrapper.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormBoltWrapper.scala
deleted file mode 100644
index 5a953c9..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormBoltWrapper.scala
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * 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 eagle.datastream
-
-import java.util
-
-import backtype.storm.task.{OutputCollector, TopologyContext}
-import backtype.storm.topology.OutputFieldsDeclarer
-import backtype.storm.topology.base.BaseRichBolt
-import backtype.storm.tuple.{Fields, Tuple}
-import org.slf4j.LoggerFactory
-
-import scala.collection.JavaConverters._
-
-case class StormBoltWrapper(worker : StormStreamExecutor[EagleTuple]) extends BaseRichBolt{
-  val LOG = LoggerFactory.getLogger(StormBoltWrapper.getClass)
-  var _collector : OutputCollector = null
-
-  override def prepare(stormConf: util.Map[_, _], context: TopologyContext, collector: OutputCollector): Unit = {
-    _collector = collector
-    worker.init
-  }
-
-  override def execute(input : Tuple): Unit = {
-    try {
-      worker.flatMap(input.getValues.asScala, new Collector[EagleTuple] {
-        override def collect(t: EagleTuple): Unit = {
-          _collector.emit(input, t.getList.asJava)
-        }
-      })
-    }catch{
-      case ex: Exception => {
-        LOG.error("fail executing", ex)
-        _collector.fail(input)
-        throw new RuntimeException(ex)
-      }
-    }
-    _collector.ack(input)
-  }
-
-  override def declareOutputFields(declarer : OutputFieldsDeclarer): Unit ={
-    val fields = worker.fields
-    LOG.info("output fields for worker " + worker + " : " + fields.toList)
-    declarer.declare(new Fields(fields:_*))
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormExecutorForAlertWrapper.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormExecutorForAlertWrapper.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormExecutorForAlertWrapper.scala
deleted file mode 100644
index fd3e2e5..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormExecutorForAlertWrapper.scala
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- *
- *  * 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 eagle.datastream
-
-import java.util
-
-import com.typesafe.config.Config
-
-case class StormExecutorForAlertWrapper(delegate: StormStreamExecutor[Tuple2[String, util.SortedMap[AnyRef, AnyRef]]], streamName: String)
-  extends StormStreamExecutor3[String, String, util.SortedMap[Object, Object]]{
-  override def prepareConfig(config: Config): Unit = {
-    delegate.prepareConfig(config)
-  }
-
-  override def init: Unit = {
-    delegate.init
-  }
-
-  override def flatMap(input: Seq[AnyRef], collector: Collector[Tuple3[String, String, util.SortedMap[Object, Object]]]): Unit = {
-    delegate.flatMap(input, new Collector[Tuple2[String, util.SortedMap[AnyRef, AnyRef]]] {
-      override def collect(r: Tuple2[String, util.SortedMap[AnyRef, AnyRef]]): Unit = {
-        collector.collect(Tuple3(r.f0, streamName, r.f1))
-      }
-    })
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormSpoutFactory.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormSpoutFactory.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormSpoutFactory.scala
deleted file mode 100644
index be4ee81..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormSpoutFactory.scala
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- *
- *  * 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 eagle.datastream
-
-import java.util
-
-import backtype.storm.topology.base.BaseRichSpout
-import com.typesafe.config.Config
-
-object StormSpoutFactory {
-  /**
-   * @param config context configuration
-   * @param sourceProducer source producer
-   * @return
-   */
-  def createSpout(config: Config, sourceProducer: StormSourceProducer) : BaseRichSpout = {
-    val numFields = sourceProducer.numFields
-    if(numFields <= 0) {
-      sourceProducer.source
-    }else{
-      var i = 0
-      val ret = new util.ArrayList[String]
-      while(i < numFields){
-        ret.add(OutputFieldNameConst.FIELD_PREFIX + i)
-        i += 1
-      }
-      SpoutProxy(sourceProducer.source, ret)
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormStreamDAG.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormStreamDAG.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormStreamDAG.scala
deleted file mode 100644
index 8fe9695..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormStreamDAG.scala
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- *
- *  * 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 eagle.datastream
-
-import org.jgrapht.experimental.dag.DirectedAcyclicGraph
-
-import scala.collection.JavaConverters._
-import scala.collection.{JavaConversions, mutable}
-
-/**
- * wrapper of DAG, used for storm topology compiler
- */
-class StormStreamDAG(graph: DirectedAcyclicGraph[StreamProducer, StreamConnector]) extends AbstractStreamProducerGraph {
-  var nodeMap: mutable.Map[String, StreamProducer] = null
-
-  override def addEdge(from: StreamProducer, to: StreamProducer, streamConnector: StreamConnector): Unit = {
-    graph.addEdge(from, to, streamConnector)
-  }
-
-  override def addVertex(producer: StreamProducer): Unit = {
-    graph.addVertex(producer)
-  }
-
-  override def iterator(): Iterator[StreamProducer] = {
-    JavaConversions.asScalaIterator(graph.iterator())
-  }
-
-  override def isSource(v: StreamProducer): Boolean = {
-    graph.inDegreeOf(v) match {
-      case 0 => true
-      case _ => false
-    }
-  }
-
-  override def outgoingEdgesOf(v: StreamProducer): scala.collection.mutable.Set[StreamConnector] = {
-    JavaConversions.asScalaSet(graph.outgoingEdgesOf(v))
-  }
-
-  override def getNodeByName(name: String): Option[StreamProducer] = {
-    nodeMap.get(name)
-  }
-
-  def setNodeMap(nodeMap: mutable.Map[String, StreamProducer]): Unit = {
-    this.nodeMap = nodeMap
-  }
-
-  override def incomingVertexOf(v: StreamProducer): scala.collection.mutable.Set[StreamProducer] = {
-    val set = mutable.Set[StreamProducer]()
-    graph.incomingEdgesOf(v).asScala.foreach(e => set += graph.getEdgeSource(e))
-    set
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormStreamDAGTransformer.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormStreamDAGTransformer.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormStreamDAGTransformer.scala
deleted file mode 100644
index 5966387..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormStreamDAGTransformer.scala
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- *
- *  * 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 eagle.datastream
-
-import org.jgrapht.experimental.dag.DirectedAcyclicGraph
-import scala.collection.mutable
-
-/**
- * convert generic DAG data structure to Storm specific DAG data structure for easy topology compiler
- */
-object StormStreamDAGTransformer {
-  /**
-   * Transform DirectedAcyclicGraph[StreamProducer, StreamConnector] into StormStreamDAG
-   *
-   * @param dag DirectedAcyclicGraph[StreamProducer, StreamConnector]
-   * @return StormStreamDAG
-   */
-  def transform(dag: DirectedAcyclicGraph[StreamProducer, StreamConnector]) : StormStreamDAG = {
-    val stormDAG = new StormStreamDAG(dag)
-    val nodeMap = mutable.HashMap[String, StreamProducer]()
-    val iter = dag.iterator()
-    while(iter.hasNext){
-      val sp = iter.next()
-      nodeMap.put(sp.name, sp)
-    }
-    stormDAG.setNodeMap(nodeMap)
-    stormDAG
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormTopologyCompiler.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormTopologyCompiler.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormTopologyCompiler.scala
deleted file mode 100644
index de6cc9f..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormTopologyCompiler.scala
+++ /dev/null
@@ -1,98 +0,0 @@
-/*
- *
- *  * 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 eagle.datastream
-
-import java.util
-
-import backtype.storm.topology.base.BaseRichBolt
-import backtype.storm.topology.{BoltDeclarer, TopologyBuilder}
-import backtype.storm.tuple.Fields
-import com.typesafe.config.Config
-import org.slf4j.LoggerFactory
-
-case class StormTopologyCompiler(config: Config, graph: AbstractStreamProducerGraph) extends AbstractTopologyCompiler{
-  val LOG = LoggerFactory.getLogger(StormTopologyCompiler.getClass)
-  val boltCache = scala.collection.mutable.Map[StreamProducer, StormBoltWrapper]()
-
-  override def buildTopology: AbstractTopologyExecutor ={
-    val builder = new TopologyBuilder();
-    val iter = graph.iterator()
-    val boltDeclarerCache = scala.collection.mutable.Map[String, BoltDeclarer]()
-    while(iter.hasNext){
-      val from = iter.next()
-      val fromName = from.name
-      if(graph.isSource(from)){
-        val spout = StormSpoutFactory.createSpout(config, from.asInstanceOf[StormSourceProducer])
-        builder.setSpout(fromName, spout, from.parallelism)
-        LOG.info("Spout name : " + fromName + " with parallelism " + from.parallelism)
-      } else {
-        LOG.info("Bolt name:" + fromName)
-      }
-
-      val edges = graph.outgoingEdgesOf(from)
-      edges.foreach(sc => {
-        val toName = sc.to.name
-        var boltDeclarer : BoltDeclarer = null
-        val toBolt = createBoltIfAbsent(toName)
-        boltDeclarerCache.get(toName) match{
-          case None => {
-            var finalParallelism = 1
-            graph.getNodeByName(toName) match {
-              case Some(p) => finalParallelism = p.parallelism
-              case None => finalParallelism = 1
-            }
-            boltDeclarer = builder.setBolt(toName, toBolt, finalParallelism);
-            LOG.info("created bolt " + toName + " with parallelism " + finalParallelism)
-            boltDeclarerCache.put(toName, boltDeclarer)
-          }
-          case Some(bt) => boltDeclarer = bt
-        }
-        sc.groupByFields match{
-          case Nil => boltDeclarer.shuffleGrouping(fromName)
-          case p => boltDeclarer.fieldsGrouping(fromName, new Fields(fields(p)))
-        }
-        LOG.info("bolt connected " + fromName + "->" + toName + " with groupby fields " + sc.groupByFields)
-      })
-    }
-    new StormTopologyExecutorImpl(builder.createTopology, config)
-  }
-
-  def fields(fields : Seq[Int]): java.util.List[String] ={
-    val ret = new util.ArrayList[String]
-    fields.map(n => ret.add(OutputFieldNameConst.FIELD_PREFIX + n))
-    ret
-  }
-
-  def createBoltIfAbsent(name : String) : BaseRichBolt = {
-    val producer = graph.getNodeByName(name)
-    producer match{
-      case Some(p) => createBoltIfAbsent(graph, p)
-      case None => throw new IllegalArgumentException("please check bolt name " + name)
-    }
-  }
-
-  def createBoltIfAbsent(graph: AbstractStreamProducerGraph, producer : StreamProducer): BaseRichBolt ={
-    boltCache.get(producer) match{
-      case Some(bolt) => bolt
-      case None => {
-        StormBoltFactory.getBoltWrapper(graph, producer, config)
-      }
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormTopologyExecutorImpl.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormTopologyExecutorImpl.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormTopologyExecutorImpl.scala
deleted file mode 100644
index 656495d..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StormTopologyExecutorImpl.scala
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * 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 eagle.datastream
-
-import backtype.storm.generated.StormTopology
-import backtype.storm.utils.Utils
-import backtype.storm.{Config, LocalCluster, StormSubmitter}
-import storm.trident.spout.RichSpoutBatchExecutor
-
-case class StormTopologyExecutorImpl(topology: StormTopology, config: com.typesafe.config.Config) extends AbstractTopologyExecutor {
-  @throws(classOf[Exception])
-  def execute {
-    val localMode: Boolean = config.getString("envContextConfig.mode").equalsIgnoreCase("local")
-    val conf: Config = new Config
-    conf.put(RichSpoutBatchExecutor.MAX_BATCH_SIZE_CONF, Int.box(64 * 1024))
-    conf.put(Config.TOPOLOGY_RECEIVER_BUFFER_SIZE, Int.box(8))
-    conf.put(Config.TOPOLOGY_TRANSFER_BUFFER_SIZE, Int.box(32))
-    conf.put(Config.TOPOLOGY_EXECUTOR_RECEIVE_BUFFER_SIZE, Int.box(16384))
-    conf.put(Config.TOPOLOGY_EXECUTOR_SEND_BUFFER_SIZE, Int.box(16384))
-
-    val topologyName = config.getString("envContextConfig.topologyName")
-    if (!localMode) {
-      StormSubmitter.submitTopologyWithProgressBar(topologyName, conf, topology)
-    }
-    else {
-      val cluster: LocalCluster = new LocalCluster
-      cluster.submitTopology(topologyName, conf, topology)
-      Utils.sleep(Integer.MAX_VALUE)
-      cluster.killTopology(topologyName)
-      cluster.shutdown
-    }
-  }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamAlertExpansion.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamAlertExpansion.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamAlertExpansion.scala
deleted file mode 100644
index e6a4012..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamAlertExpansion.scala
+++ /dev/null
@@ -1,192 +0,0 @@
-/*
- *
- *  * 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 eagle.datastream
-
-import java.util
-
-import com.typesafe.config.Config
-import eagle.alert.dao.AlertDefinitionDAOImpl
-import eagle.executor.AlertExecutorCreationUtils
-import org.jgrapht.experimental.dag.DirectedAcyclicGraph
-import org.slf4j.LoggerFactory
-
-import scala.collection.JavaConversions._
-import scala.collection.mutable.ListBuffer
-
-/**
- * The constraints for alert is:
- * 1. only 3 StreamProducers can be put immediately before MapProducer, FlatMapProducer, StreamUnionProducer
- * 2. For StreamUnionProducer, the only supported unioned producers are MapProducer and FlatMapProducer
- * 3. the output for MapProducer and FlatMapProducer is 2-field tuple, key and value, key is string, value has to be SortedMap
- * 4. the framework will wrapper original MapProducer and FlatMapProducer to emit 3-field tuple, {key, streamName and value}
- * 5. the framework will automatically partition traffic with first field
- *
- *
- * 2 steps
- * step 1: wrapper previous StreamProducer with one more field "streamName"
- * step 2: partition alert executor by policy partitioner class
- */
-
-class StreamAlertExpansion(config: Config) extends StreamDAGExpansion(config) {
-  val LOG = LoggerFactory.getLogger(classOf[StreamAlertExpansion])
-
-  override def expand(dag: DirectedAcyclicGraph[StreamProducer, StreamConnector]): Unit ={
-    val iter = dag.iterator()
-    val toBeAddedEdges = new ListBuffer[StreamConnector]
-    val toBeRemovedVertex = new ListBuffer[StreamProducer]
-    while(iter.hasNext) {
-      val current = iter.next()
-      dag.outgoingEdgesOf(current).foreach(edge => {
-        val child = edge.to
-        onIteration(toBeAddedEdges, toBeRemovedVertex, dag, current, child)
-      })
-    }
-    // add back edges
-    toBeAddedEdges.foreach(e => {
-      dag.addVertex(e.from)
-      dag.addVertex(e.to)
-      dag.addEdge(e.from, e.to, e)
-    })
-    toBeRemovedVertex.foreach(v => dag.removeVertex(v))
-  }
-
-  def onIteration(toBeAddedEdges: ListBuffer[StreamConnector], toBeRemovedVertex: ListBuffer[StreamProducer],
-               dag: DirectedAcyclicGraph[StreamProducer, StreamConnector], current: StreamProducer, child: StreamProducer): Unit = {
-    child match {
-      case AlertStreamSink(id, upStreamNames, alertExecutorId, withConsumer) => {
-        /**
-         * step 1: wrapper previous StreamProducer with one more field "streamName"
-         * for AlertStreamSink, we check previous StreamProducer and replace that
-         */
-        val newStreamProducers = new ListBuffer[StreamProducer]
-        current match {
-          case StreamUnionProducer(id, others) => {
-            val incomingEdges = dag.incomingEdgesOf(current)
-            incomingEdges.foreach(e => newStreamProducers += replace(toBeAddedEdges, toBeRemovedVertex, dag, e.from, upStreamNames.get(0)))
-            var i: Int = 1
-            others.foreach(o => {
-              newStreamProducers += replace(toBeAddedEdges, toBeRemovedVertex, dag, o, upStreamNames.get(i))
-              i += 1
-            })
-          }
-          case _: FlatMapProducer[AnyRef, AnyRef] => {
-            newStreamProducers += replace(toBeAddedEdges, toBeRemovedVertex, dag, current, upStreamNames.get(0))
-          }
-          case _: MapProducer => {
-            newStreamProducers += replace(toBeAddedEdges, toBeRemovedVertex, dag, current, upStreamNames.get(0))
-          }
-          case s: StreamProducer if dag.inDegreeOf(s) == 0 => {
-            newStreamProducers += replace(toBeAddedEdges, toBeRemovedVertex, dag, current, upStreamNames.get(0))
-          }
-          case p@_ => throw new IllegalStateException(s"$p can not be put before AlertStreamSink, only StreamUnionProducer,FlatMapProducer and MapProducer are supported")
-        }
-
-        /**
-         * step 2: partition alert executor by policy partitioner class
-         */
-        val alertExecutors = AlertExecutorCreationUtils.createAlertExecutors(config, new AlertDefinitionDAOImpl(config), upStreamNames, alertExecutorId)
-        var alertProducers = new scala.collection.mutable.MutableList[StreamProducer]
-        alertExecutors.foreach(exec => {
-          val t = FlatMapProducer(UniqueId.incrementAndGetId(), exec).withName(exec.getAlertExecutorId() + "_" + exec.getPartitionSeq())
-          t.setConfig(config)
-          t.setGraph(dag)
-          alertProducers += t
-          newStreamProducers.foreach(newsp => toBeAddedEdges += StreamConnector(newsp, t).groupBy(Seq(0)))
-        })
-
-        // remove AlertStreamSink
-        toBeRemovedVertex += child
-
-        // add alert consumer if necessary
-        if (withConsumer) {
-          AlertExecutorConsumerUtils.setupAlertConsumers(toBeAddedEdges, alertProducers.toList)
-        }
-      }
-      case _ =>
-    }
-  }
-
-  private def replace(toBeAddedEdges: ListBuffer[StreamConnector], toBeRemovedVertex: ListBuffer[StreamProducer],
-                      dag: DirectedAcyclicGraph[StreamProducer, StreamConnector], current: StreamProducer, upStreamName: String) : StreamProducer= {
-    var newsp: StreamProducer = null
-    current match {
-      case _: FlatMapProducer[AnyRef, AnyRef] => {
-        val mapper = current.asInstanceOf[FlatMapProducer[_, _]].mapper
-        mapper match {
-          case a: JavaStormStreamExecutor[EagleTuple] => {
-            val newmapper = new JavaStormExecutorForAlertWrapper(a.asInstanceOf[JavaStormStreamExecutor[Tuple2[String, util.SortedMap[AnyRef, AnyRef]]]], upStreamName)
-            newsp = FlatMapProducer(UniqueId.incrementAndGetId(), newmapper)
-            newsp.setGraph(dag)
-            newsp.setConfig(config)
-          }
-          case b: StormStreamExecutor[EagleTuple] => {
-            val newmapper = StormExecutorForAlertWrapper(b.asInstanceOf[StormStreamExecutor[Tuple2[String, util.SortedMap[AnyRef, AnyRef]]]], upStreamName)
-            newsp = FlatMapProducer(UniqueId.incrementAndGetId(), newmapper)
-            newsp.setGraph(dag)
-            newsp.setConfig(config)
-          }
-          case _ => throw new IllegalArgumentException
-        }
-        // remove old StreamProducer and replace that with new StreamProducer
-        val incomingEdges = dag.incomingEdgesOf(current)
-        incomingEdges.foreach(e => toBeAddedEdges += StreamConnector(e.from, newsp))
-        val outgoingEdges = dag.outgoingEdgesOf(current)
-        outgoingEdges.foreach(e => toBeAddedEdges += StreamConnector(newsp, e.to))
-        toBeRemovedVertex += current
-      }
-      case _: MapProducer => {
-        val mapper = current.asInstanceOf[MapProducer].fn
-        val newfun: (AnyRef => AnyRef) = {
-          a => mapper(a) match {
-            case scala.Tuple2(x1, x2) => (x1, upStreamName, x2)
-            case _ => throw new IllegalArgumentException
-          }
-        }
-        current match {
-          case MapProducer(id, 2, fn) => newsp = MapProducer(UniqueId.incrementAndGetId(), 3, newfun)
-          case _ => throw new IllegalArgumentException
-        }
-        val incomingEdges = dag.incomingEdgesOf(current)
-        incomingEdges.foreach(e => toBeAddedEdges += StreamConnector(e.from, newsp))
-        val outgoingEdges = dag.outgoingEdgesOf(current)
-        outgoingEdges.foreach(e => toBeAddedEdges += StreamConnector(newsp, e.to))
-        toBeRemovedVertex += current
-      }
-      case s: StreamProducer if dag.inDegreeOf(s) == 0 => {
-        val fn:(AnyRef => AnyRef) = {
-          n => {
-            n match {
-              case scala.Tuple3 => n
-              case scala.Tuple2(x1,x2) => (x1,upStreamName,x2)
-              case scala.Tuple1(x1) => (if(x1 == null) null else x1.hashCode(),upStreamName,x1)
-              case _ => (if(n == null) null else n.hashCode(),upStreamName,n)
-            }
-          }
-        }
-        newsp = MapProducer(UniqueId.incrementAndGetId(),3,fn)
-        toBeAddedEdges += StreamConnector(current,newsp)
-        val outgoingEdges = dag.outgoingEdgesOf(current)
-        outgoingEdges.foreach(e => toBeAddedEdges += StreamConnector(newsp,e.to))
-      }
-      case _ => throw new IllegalArgumentException("Only FlatMapProducer and MapProducer can be replaced before AlertStreamSink")
-    }
-    newsp
-  }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamAppDSL.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamAppDSL.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamAppDSL.scala
deleted file mode 100644
index 0a2e5ba..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamAppDSL.scala
+++ /dev/null
@@ -1,120 +0,0 @@
-/*
- * 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 eagle.datastream
-
-import backtype.storm.topology.base.BaseRichSpout
-import com.typesafe.config._
-import eagle.dataproc.impl.storm.AbstractStormSpoutProvider
-import eagle.dataproc.util.ConfigOptionParser
-
-import scala.reflect.runtime.universe._
-
-/**
- * @since  11/6/15
- */
-trait ConfigContext{
-  def set(config:Config)
-  def config:Config
-
-  def set[T<:AnyRef](key:String,value:T): Unit = {
-    set(config.withValue(key,ConfigValueFactory.fromAnyRef(value)))
-  }
-
-  /**
-   *
-   * @param key config key
-   * @param default default value
-   * @tparam T return type
-   * @return
-   */
-  def get[T](key:String,default:T=null)(implicit tag:TypeTag[T]):T = {
-    if(config.hasPath(key)) {
-      get(key)
-    } else default
-  }
-
-  def get[T](key:String)(implicit tag: TypeTag[T]):T = tag.tpe match {
-    case STRING_TYPE => config.getString(key).asInstanceOf[T]
-    case TypeTag.Double => config.getDouble(key).asInstanceOf[T]
-    case TypeTag.Long => config.getLong(key).asInstanceOf[T]
-    case TypeTag.Int => config.getInt(key).asInstanceOf[T]
-    case TypeTag.Byte => config.getBytes(key).asInstanceOf[T]
-    case TypeTag.Boolean => config.getBoolean(key).asInstanceOf[T]
-    case NUMBER_TYPE => config.getNumber(key).asInstanceOf[T]
-    case OBJECT_TYPE => config.getObject(key).asInstanceOf[T]
-    case VALUE_TYPE => config.getValue(key).asInstanceOf[T]
-    case ANY_REF_TYPE => config.getAnyRef(key).asInstanceOf[T]
-    case INT_LIST_TYPE => config.getIntList(key).asInstanceOf[T]
-    case DOUBLE_LIST_TYPE => config.getDoubleList(key).asInstanceOf[T]
-    case BOOL_LIST_TYPE => config.getBooleanList(key).asInstanceOf[T]
-    case LONG_LIST_TYPE => config.getLongList(key).asInstanceOf[T]
-    case _ => throw new UnsupportedOperationException(s"$tag is not supported yet")
-  }
-
-  val STRING_TYPE = typeOf[String]
-  val NUMBER_TYPE = typeOf[Number]
-  val INT_LIST_TYPE = typeOf[List[Int]]
-  val BOOL_LIST_TYPE = typeOf[List[Boolean]]
-  val DOUBLE_LIST_TYPE = typeOf[List[Double]]
-  val LONG_LIST_TYPE = typeOf[List[Double]]
-  val OBJECT_TYPE = typeOf[ConfigObject]
-  val VALUE_TYPE = typeOf[ConfigValue]
-  val ANY_REF_TYPE = typeOf[AnyRef]
-}
-
-/**
- * Stream APP DSL
- * @tparam E
- */
-trait StreamApp[+E<:ExecutionEnvironment] extends App with ConfigContext{
-  private var _executed = false
-  private var _config:Config = null
-
-  override def config:Config = _config
-
-  override def set(config:Config) = _config = config
-
-  def env:E
-
-  def execute() {
-    env.execute()
-    _executed = true
-  }
-
-  override def main(args: Array[String]): Unit = {
-    _config = new ConfigOptionParser().load(args)
-    super.main(args)
-    if(!_executed) execute()
-  }
-}
-
-trait StormStreamApp extends StreamApp[StormExecutionEnvironment]{
-  private var _env:StormExecutionEnvironment = null
-  def source(sourceProvider: AbstractStormSpoutProvider) = {
-    val spout = sourceProvider.getSpout(config)
-    env.newSource(spout)
-  }
-
-  def source(spout:BaseRichSpout) = env.newSource(spout)
-
-  override def env:StormExecutionEnvironment = {
-    if(_env == null){
-      _env = ExecutionEnvironmentFactory.getStorm(config)
-    }
-    _env
-  }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamConnector.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamConnector.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamConnector.scala
deleted file mode 100644
index ceeb411..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamConnector.scala
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
- *
- *  * 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 eagle.datastream
-
-case class StreamConnector(from: StreamProducer, to: StreamProducer) {
-  var groupByFields : Seq[Int] = Nil
-
-  def groupBy(fields : Seq[Int]) : StreamConnector = {
-    groupByFields = fields
-    this
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamDAGExpansion.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamDAGExpansion.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamDAGExpansion.scala
deleted file mode 100644
index baa514e..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamDAGExpansion.scala
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- *
- *  * 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 eagle.datastream
-
-import com.typesafe.config.Config
-import org.jgrapht.experimental.dag.DirectedAcyclicGraph
-
-abstract class StreamDAGExpansion(config: Config) {
-  def expand(dag: DirectedAcyclicGraph[StreamProducer, StreamConnector])
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamGroupbyExpansion.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamGroupbyExpansion.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamGroupbyExpansion.scala
deleted file mode 100644
index 3bed891..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamGroupbyExpansion.scala
+++ /dev/null
@@ -1,59 +0,0 @@
-/*
- *
- *  * 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 eagle.datastream
-
-import com.typesafe.config.Config
-import org.jgrapht.experimental.dag.DirectedAcyclicGraph
-import org.slf4j.LoggerFactory
-import scala.collection.JavaConversions._
-import scala.collection.mutable.ListBuffer
-
-/**
- * Replace GroupByProducer(Vertex) with StreamConnector (Edge)
- * @param config context configuration
- */
-class StreamGroupbyExpansion(config: Config) extends StreamDAGExpansion(config){
-  val LOG = LoggerFactory.getLogger(classOf[StreamGroupbyExpansion])
-
-  override def expand(dag: DirectedAcyclicGraph[StreamProducer, StreamConnector]) = {
-    val iter = dag.iterator()
-    var toBeAddedEdges = new ListBuffer[StreamConnector]
-    var toBeRemovedVertex = new ListBuffer[StreamProducer]
-    while(iter.hasNext) {
-      val current = iter.next()
-      dag.outgoingEdgesOf(current).foreach(edge => {
-        val child = edge.to
-        child match {
-          case p : GroupByProducer => {
-            dag.outgoingEdgesOf(p).foreach(c2 => {
-              toBeAddedEdges += StreamConnector(current, c2.to).groupBy(p.fields)
-            })
-            toBeRemovedVertex += p
-          }
-          case _ =>
-        }
-      })
-    }
-
-    // add back edges
-    toBeAddedEdges.foreach(e => dag.addEdge(e.from, e.to, e))
-    toBeRemovedVertex.foreach(v => dag.removeVertex(v))
-  }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamNameExpansion.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamNameExpansion.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamNameExpansion.scala
deleted file mode 100644
index 40abfd8..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamNameExpansion.scala
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- *
- *  * 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 eagle.datastream
-
-import com.typesafe.config.Config
-import org.jgrapht.experimental.dag.DirectedAcyclicGraph
-import org.slf4j.LoggerFactory
-
-/**
- * to set name for each StreamProducer
- * 1. if name is given programatically, then use this name
- * 2. otherwise use name generated by scala internally
- */
-class StreamNameExpansion(config: Config) extends StreamDAGExpansion(config){
-  val LOG = LoggerFactory.getLogger(classOf[StreamNameExpansion])
-
-  override def expand(dag: DirectedAcyclicGraph[StreamProducer, StreamConnector]) = {
-    val iter = dag.iterator()
-    while(iter.hasNext){
-      val sp = iter.next()
-      sp.name = NodeNameSelector(sp).getName
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamParallelismConfigExpansion.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamParallelismConfigExpansion.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamParallelismConfigExpansion.scala
deleted file mode 100644
index c3c4533..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamParallelismConfigExpansion.scala
+++ /dev/null
@@ -1,55 +0,0 @@
-/*
- *
- *  * 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 eagle.datastream
-
-import java.util.regex.Pattern
-
-import com.typesafe.config.{ConfigValue, ConfigObject, Config}
-import org.jgrapht.experimental.dag.DirectedAcyclicGraph
-import org.slf4j.LoggerFactory
-import scala.collection.JavaConverters._
-
-class StreamParallelismConfigExpansion(config: Config) extends StreamDAGExpansion(config){
-  val LOG = LoggerFactory.getLogger(classOf[StreamParallelismConfigExpansion])
-
-  override def expand(dag: DirectedAcyclicGraph[StreamProducer, StreamConnector]) = {
-    val map = getParallelismMap(config)
-    val iter = dag.iterator()
-    while(iter.hasNext){
-      val streamProducer = iter.next()
-      if(streamProducer.name != null) {
-        map.foreach(tuple => {
-          tuple._1.matcher(streamProducer.name).find() match {
-            case true => streamProducer.parallelism = tuple._2
-            case false =>
-          }
-        })
-      }
-    }
-  }
-
-  private def getParallelismMap(config: Config) : Map[Pattern, Int]= {
-    val parallelismConfig: ConfigObject = config.getObject("envContextConfig.parallelismConfig")
-    LOG.info("Found parallelismConfig ? " + (if (parallelismConfig == null) "no" else "yes"))
-    parallelismConfig.asScala.toMap map {
-      case (name, value) => (Pattern.compile(name), value.asInstanceOf[ConfigValue].unwrapped().asInstanceOf[Int])
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamProducer.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamProducer.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamProducer.scala
deleted file mode 100644
index 8485e28..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamProducer.scala
+++ /dev/null
@@ -1,193 +0,0 @@
-/*
- *
- *  * 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 eagle.datastream
-
-import java.util
-import java.util.concurrent.atomic.AtomicInteger
-
-import backtype.storm.topology.base.BaseRichSpout
-import com.typesafe.config.Config
-import org.jgrapht.experimental.dag.DirectedAcyclicGraph
-import scala.collection.JavaConversions._
-import scala.collection.JavaConverters._
-
-/**
- * StreamProducer is the base class for all other concrete StreamProducer
- * It defines high level API for user to organize data stream flow
- *
- * StreamProducer is independent of execution environment
- */
-
-trait StreamProducer{
-  var name: String = null
-  var parallelism: Int = 1
-  var graph: DirectedAcyclicGraph[StreamProducer, StreamConnector] = null
-  var config: Config = null
-
-  private def incrementAndGetId() = UniqueId.incrementAndGetId()
-
-  def setGraph(graph: DirectedAcyclicGraph[StreamProducer, StreamConnector]): Unit = this.graph = graph
-  def getGraph: DirectedAcyclicGraph[StreamProducer, StreamConnector] = graph
-  def setConfig(config: Config) : Unit = this.config = config
-  def getConfig: Config = config
-
-  def filter(fn : AnyRef => Boolean): StreamProducer ={
-    val ret = FilterProducer(incrementAndGetId(), fn)
-    hookupDAG(graph, this, ret)
-    ret
-  }
-
-  def flatMap[T, R](mapper : FlatMapper[T, R]) : StreamProducer = {
-    val ret = FlatMapProducer(incrementAndGetId(), mapper)
-    hookupDAG(graph, this, ret)
-    ret
-  }
-
-  def map1(fn : AnyRef => AnyRef) : StreamProducer = {
-    val ret = MapProducer(incrementAndGetId(), 1, fn)
-    hookupDAG(graph, this, ret)
-    ret
-  }
-
-  def map2(fn : AnyRef => AnyRef) : StreamProducer = {
-    val ret = MapProducer(incrementAndGetId(), 2, fn)
-    hookupDAG(graph, this, ret)
-    ret
-  }
-
-  def map3(fn : AnyRef => AnyRef) : StreamProducer = {
-    val ret = MapProducer(incrementAndGetId(), 3, fn)
-    hookupDAG(graph, this, ret)
-    ret
-  }
-
-  def map4(fn : AnyRef => AnyRef) : StreamProducer = {
-    val ret = MapProducer(incrementAndGetId(), 4, fn)
-    hookupDAG(graph, this, ret)
-    ret
-  }
-
-  /**
-   * starting from 0, groupby operator would be upon edge of the graph
-   */
-  def groupBy(fields : Int*) : StreamProducer = {
-    // validate each field index is greater or equal to 0
-    fields.foreach(n => if(n<0) throw new IllegalArgumentException("field index should be always >= 0"))
-    val ret = GroupByProducer(incrementAndGetId(), fields)
-    hookupDAG(graph, this, ret)
-    ret
-  }
-
-  //groupBy java version, starting from 1
-  def groupBy(fields : java.util.List[Integer]) : StreamProducer = {
-    // validate each field index is greater or equal to 0
-    fields.foreach(n => if(n<0) throw new IllegalArgumentException("field index should be always >= 0"))
-    val ret = GroupByProducer(incrementAndGetId(), fields.asScala.toSeq.asInstanceOf[Seq[Int]])
-    hookupDAG(graph, this, ret)
-    ret
-  }
-
-  def streamUnion(others : Seq[StreamProducer]) : StreamProducer = {
-    val ret = StreamUnionProducer(incrementAndGetId(), others)
-    hookupDAG(graph, this, ret)
-    ret
-  }
-
-  /**
-   * alert is always sink of data flow
-   */
-  def alertWithConsumer(upStreamNames: util.List[String], alertExecutorId : String) = {
-    alert(upStreamNames, alertExecutorId, true)
-  }
-
-  def alertWithoutConsumer(upStreamNames: util.List[String], alertExecutorId : String) = {
-    alert(upStreamNames, alertExecutorId, false)
-  }
-
-  def alert(upStreamNames: util.List[String], alertExecutorId : String, withConsumer: Boolean=true) = {
-    val ret = AlertStreamSink(incrementAndGetId(), upStreamNames, alertExecutorId, withConsumer)
-    hookupDAG(graph, this, ret)
-  }
-
-  def alertWithConsumer(upStreamName: String, alertExecutorId : String): Unit ={
-    alert(util.Arrays.asList(upStreamName), alertExecutorId, true)
-  }
-
-  def alertWithoutConsumer(upStreamName: String, alertExecutorId : String): Unit ={
-    alert(util.Arrays.asList(upStreamName), alertExecutorId, false)
-  }
-
-  def hookupDAG(graph: DirectedAcyclicGraph[StreamProducer, StreamConnector], current: StreamProducer, next: StreamProducer) = {
-    current.getGraph.addVertex(next)
-    current.getGraph.addEdge(current, next, StreamConnector(current, next))
-    passOnContext(current, next)
-  }
-
-  private def passOnContext(current: StreamProducer, next: StreamProducer): Unit ={
-    next.graph = current.graph
-    next.config = current.config
-  }
-
-  /**
-   * can be set by programatically or by configuration
-   */
-  def withParallelism(parallelism : Int) : StreamProducer = {
-    this.parallelism = parallelism
-    this
-  }
-
-  def withName(name : String) : StreamProducer = {
-    this.name = name
-    this
-  }
-}
-
-case class FilterProducer(id: Int, fn : AnyRef => Boolean) extends StreamProducer
-
-case class FlatMapProducer[T, R](id: Int, var mapper: FlatMapper[T, R]) extends StreamProducer {
-  override def toString() = mapper.toString + "_" + id
-}
-
-case class MapProducer(id: Int, numOutputFields : Int, var fn : AnyRef => AnyRef) extends StreamProducer
-
-case class GroupByProducer(id: Int, fields : Seq[Int]) extends StreamProducer
-
-case class StreamUnionProducer(id: Int, others: Seq[StreamProducer]) extends StreamProducer
-
-case class StormSourceProducer(id: Int, source : BaseRichSpout) extends StreamProducer{
-  var numFields : Int = 0
-  /**
-    * rename outputfields to f0, f1, f2, ...
-   * if one spout declare some field names, those fields names will be modified
-   * @param n
-   */
-  def renameOutputFields(n : Int): StormSourceProducer ={
-    this.numFields = n
-    this
-  }
-}
-
-case class AlertStreamSink(id: Int, upStreamNames: util.List[String], alertExecutorId : String, withConsumer: Boolean=true) extends StreamProducer
-
-object UniqueId{
-  val id : AtomicInteger = new AtomicInteger(0);
-  def incrementAndGetId() : Int = {
-    id.incrementAndGet()
-  }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamUnionExpansion.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamUnionExpansion.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamUnionExpansion.scala
deleted file mode 100644
index b68d213..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/StreamUnionExpansion.scala
+++ /dev/null
@@ -1,62 +0,0 @@
-/*
- *
- *  * 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 eagle.datastream
-
-import com.typesafe.config.Config
-import org.jgrapht.experimental.dag.DirectedAcyclicGraph
-import org.slf4j.LoggerFactory
-import scala.collection.JavaConversions._
-import scala.collection.mutable.ListBuffer
-
-/**
- * union operator should be expanded
- */
-class StreamUnionExpansion(config: Config) extends StreamDAGExpansion(config){
-  val LOG = LoggerFactory.getLogger(classOf[StreamUnionExpansion])
-
-  override def expand(dag: DirectedAcyclicGraph[StreamProducer, StreamConnector]) = {
-    val iter = dag.iterator()
-    var toBeAddedEdges = new ListBuffer[StreamConnector]
-    var toBeRemovedVertex = new ListBuffer[StreamProducer]
-    while(iter.hasNext) {
-      val current = iter.next()
-      dag.outgoingEdgesOf(current).foreach(edge => {
-        val child = edge.to
-        val groupByFields = edge.groupByFields;
-        child match {
-          case StreamUnionProducer(id, others) => {
-            dag.outgoingEdgesOf(child).foreach(c2 => {
-              toBeAddedEdges += StreamConnector(current, c2.to).groupBy(groupByFields)
-              others.foreach(o => {
-                toBeAddedEdges += StreamConnector(o, c2.to).groupBy(groupByFields)
-              })
-            })
-            toBeRemovedVertex += child
-          }
-          case _ =>
-        }
-      })
-    }
-
-    // add back edges
-    toBeAddedEdges.foreach(e => dag.addEdge(e.from, e.to, e))
-    toBeRemovedVertex.foreach(v => dag.removeVertex(v))
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/UnionUtils.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/UnionUtils.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/UnionUtils.scala
deleted file mode 100644
index fe914f1..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/UnionUtils.scala
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- *
- *  * 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 eagle.datastream
-
-import java.util
-
-import scala.collection.JavaConverters._
-
-object UnionUtils {
-  def join(producers : StreamProducer*) : StreamProducer = {
-    producers.head.streamUnion(producers.drop(1))
-  }
-
-  def join(producers : java.util.List[StreamProducer]) : StreamProducer = {
-    val newList = new util.ArrayList(producers)
-    val head = newList.get(0)
-    newList.remove(0)
-    head.streamUnion(newList.asScala);
-  }
-
-  def join(producers : List[StreamProducer]) : StreamProducer = {
-    val head = producers.head
-    head.streamUnion(producers.tail);
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/kafka/JsonMessageDeserializer.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/kafka/JsonMessageDeserializer.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/kafka/JsonMessageDeserializer.scala
deleted file mode 100644
index 1e735c5..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/kafka/JsonMessageDeserializer.scala
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * 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 eagle.datastream.kafka
-
-import java.io.IOException
-import java.util
-import java.util.Properties
-
-import com.fasterxml.jackson.databind.ObjectMapper
-import eagle.dataproc.impl.storm.kafka.SpoutKafkaMessageDeserializer
-import org.slf4j.{Logger, LoggerFactory}
-
-/**
- * @since  11/6/15
- */
-case class JsonMessageDeserializer(props:Properties) extends SpoutKafkaMessageDeserializer{
-  private val objectMapper: ObjectMapper = new ObjectMapper
-  private val LOG: Logger = LoggerFactory.getLogger(classOf[JsonMessageDeserializer])
-
-  override def deserialize(bytes: Array[Byte]): AnyRef = {
-    var map: util.Map[String, _] = null
-    try {
-      map = objectMapper.readValue(bytes, classOf[util.TreeMap[String, _]])
-    } catch {
-      case e: IOException => {
-        LOG.error("Failed to deserialize json from: " + new String(bytes), e)
-      }
-    }
-    map
-  }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/kafka/KafkaStreamMonitor.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/kafka/KafkaStreamMonitor.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/kafka/KafkaStreamMonitor.scala
deleted file mode 100644
index bb8fb56..0000000
--- a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/eagle/datastream/kafka/KafkaStreamMonitor.scala
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
- * 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 eagle.datastream.kafka
-
-import eagle.dataproc.impl.storm.kafka.KafkaSourcedSpoutProvider
-import eagle.datastream.StormStreamApp
-
-/**
- * @since  11/6/15
- */
-class KafkaStreamMonitorApp extends StormStreamApp{
-  val streamName = get[String]("eagle.stream.name","eventStream")
-  val streamExecutorId = get[String]("eagle.stream.executor",s"${streamName}Executor")
-
-  set("dataSourceConfig.deserializerClass",classOf[JsonMessageDeserializer].getCanonicalName)
-
-  source(new KafkaSourcedSpoutProvider).renameOutputFields(1).withName(streamName)
-    .alertWithConsumer(streamName, streamExecutorId)
-}
-
-object KafkaStreamMonitor extends KafkaStreamMonitorApp
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/org/apache/eagle/datastream/AbstractStreamProducerGraph.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/org/apache/eagle/datastream/AbstractStreamProducerGraph.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/org/apache/eagle/datastream/AbstractStreamProducerGraph.scala
new file mode 100644
index 0000000..dc2c198
--- /dev/null
+++ b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/org/apache/eagle/datastream/AbstractStreamProducerGraph.scala
@@ -0,0 +1,29 @@
+/*
+ *
+ *  * 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.eagle.datastream
+
+trait AbstractStreamProducerGraph {
+  def addEdge(from: StreamProducer, to: StreamProducer, streamConnector: StreamConnector)
+  def addVertex(producer: StreamProducer)
+  def iterator() : Iterator[StreamProducer]
+  def isSource(v : StreamProducer) : Boolean
+  def outgoingEdgesOf(v : StreamProducer) : scala.collection.mutable.Set[StreamConnector]
+  def getNodeByName(name : String) : Option[StreamProducer]
+  def incomingVertexOf(v: StreamProducer) : scala.collection.mutable.Set[StreamProducer]
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/org/apache/eagle/datastream/AbstractTopologyCompiler.scala
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/org/apache/eagle/datastream/AbstractTopologyCompiler.scala b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/org/apache/eagle/datastream/AbstractTopologyCompiler.scala
new file mode 100644
index 0000000..8c53ed5
--- /dev/null
+++ b/eagle-core/eagle-data-process/eagle-stream-process-api/src/main/scala/org/apache/eagle/datastream/AbstractTopologyCompiler.scala
@@ -0,0 +1,21 @@
+/*
+ * 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.eagle.datastream
+
+trait AbstractTopologyCompiler{
+  def buildTopology : AbstractTopologyExecutor
+}



Mime
View raw message