make MockSQLSourceCompositor support cofinguration

allwefantasy · allwefantasy · commit cf94e140ed26 · 2016-11-21T14:23:21.000+08:00
diff --git a/src/main/java/streaming/core/compositor/spark/ss/source/MockSQLSourceCompositor.scala b/src/main/java/streaming/core/compositor/spark/ss/source/MockSQLSourceCompositor.scala
@@ -2,6 +2,7 @@ package streaming.core.compositor.spark.ss.source
 
 import java.util
 
+import net.sf.json.JSONArray
 import org.apache.log4j.Logger
 import org.apache.spark.sql.execution.streaming.MemoryStream
 import serviceframework.dispatcher.{Compositor, Processor, Strategy}
@@ -22,13 +23,17 @@ class MockSQLSourceCompositor[T] extends Compositor[T] with CompositorHelper {
     this._configParams = configParams
   }
 
+  def data = {
+    _configParams(0).map(f => f._2.asInstanceOf[JSONArray].map(k => k.asInstanceOf[String]).toSeq).toSeq
+  }
+
   override def result(alg: util.List[Processor[T]], ref: util.List[Strategy[T]], middleResult: util.List[T], params: util.Map[Any, Any]): util.List[T] = {
     val sparkSSRt = sparkStructuredStreamingRuntime(params)
     val sparkSession = sparkSSRt.sparkSession
     import sparkSession.implicits._
     implicit val sqlContext = sparkSession.sqlContext
-    val inputData = MemoryStream[Int]
-    inputData.addData(Seq(1, 2, 3, 4))
+    val inputData = MemoryStream[String]
+    inputData.addData(data.flatMap(f=>f).seq)
     val df = inputData.toDS()
     List(df.asInstanceOf[T])
   }
diff --git a/src/main/resources-debug/test/ss-test.json b/src/main/resources-debug/test/ss-test.json
@@ -8,10 +8,7 @@
     "compositor": [
       {
         "name": "ss.source.mock",
-        "params": [
-          {
-          }
-        ]
+        "params": [{"duration1":["1","2","3"]}]
       },
       {
         "name": "ss.table",

Original file line number	Diff line number	Diff line change
`@@ -8,10 +8,7 @@`
`8`	`8`	`"compositor": [`
`9`	`9`	`{`
`10`	`10`	`"name": "ss.source.mock",`
`11`		`- "params": [`
`12`		`- {`
`13`		`- }`
`14`		`- ]`
	`11`	`+ "params": [{"duration1":["1","2","3"]}]`
`15`	`12`	`},`
`16`	`13`	`{`
`17`	`14`	`"name": "ss.table",`