Merge remote-tracking branch 'origin/master'

piflow-bundle/src/main/scala/cn/piflow/conf/bean/FlowBean.scala

@@ -13,6 +13,7 @@ class FlowBean {
   var checkpoint : String = _
   var checkpointParentProcessId : String = _
   var runMode : String = _
+  var showData : String = _
   var stops : List[StopBean] = List()
   var paths : List[PathBean] = List()
 
@@ -25,6 +26,7 @@ class FlowBean {
     this.checkpoint = flowMap.getOrElse("checkpoint","").asInstanceOf[String]
     this.checkpointParentProcessId = flowMap.getOrElse("checkpointParentProcessId", "").asInstanceOf[String]
     this.runMode = flowMap.getOrElse("runMode","RUN").asInstanceOf[String]
+    this.showData = flowMap.getOrElse("showData","0").asInstanceOf[String]
 
     //construct StopBean List
     val stopsList = MapUtil.get(flowMap,"stops").asInstanceOf[List[Map[String, Any]]]

piflow-core/src/main/scala/cn/piflow/main.scala

@@ -378,13 +378,12 @@ class JobOutputStreamImpl() extends JobOutputStream with Logging {
 
   def getDataFrame(port: String) = mapDataFrame(port);
 
-  def showData() = {
+  def showData(count:Int) = {
 
       mapDataFrame.foreach(en => {
         val portName = if(en._1.equals("")) "default" else en._1
         println(portName + " port: ")
-
-        en._2.apply().show(PropertyUtil.getPropertyValue("data.show").toInt)
+        en._2.apply().show(count)
 
       })
   }
@@ -484,7 +483,10 @@ class ProcessImpl(flow: Flow, runnerContext: Context, runner: Runner, parentProc
           runnerListener.onJobCompleted(pe.getContext());
 
           //show data in log
-          outputs.showData()
+          val showDataCount = PropertyUtil.getPropertyValue("data.show").toInt
+          if(showDataCount > 0) {
+            outputs.showData(showDataCount)
+          }
         }
         catch {
           case e: Throwable =>
@@ -528,7 +530,10 @@ class ProcessImpl(flow: Flow, runnerContext: Context, runner: Runner, parentProc
             }
           }
           //show data in log
-          outputs.showData()
+          val showDataCount = PropertyUtil.getPropertyValue("data.show").toInt
+          if(showDataCount > 0) {
+            outputs.showData(showDataCount)
+          }
 
           //save data in debug mode
           if(flow.getRunMode() == FlowRunMode.DEBUG) {

piflow-server/pom.xml

@@ -10,6 +10,12 @@
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>piflow-server</artifactId>
+
+    <properties>
+        <akka.version>2.4.14</akka.version>
+        <akka.http.version>10.0.0</akka.http.version>
+    </properties>
+
     <dependencies>
         <dependency>
             <groupId>piflow</groupId>
@@ -30,40 +36,31 @@
         <dependency>
             <groupId>com.typesafe.akka</groupId>
             <artifactId>akka-remote_2.11</artifactId>
-            <version>2.3.14</version>
+            <version>${akka.version}</version>
         </dependency>
 
         <dependency>
             <groupId>com.typesafe.akka</groupId>
             <artifactId>akka-actor_2.11</artifactId>
-            <version>2.3.14</version>
+            <version>${akka.version}</version>
         </dependency>
         <dependency>
             <groupId>com.typesafe.akka</groupId>
             <artifactId>akka-slf4j_2.11</artifactId>
-            <version>2.3.14</version>
+            <version>${akka.version}</version>
         </dependency>
         <dependency>
             <groupId>com.typesafe.akka</groupId>
-            <artifactId>akka-stream-experimental_2.11</artifactId>
-            <version>2.0.4</version>
+            <artifactId>akka-http_2.11</artifactId>
+            <version>${akka.http.version}</version>
         </dependency>
         <dependency>
             <groupId>com.typesafe.akka</groupId>
-            <artifactId>akka-http-core-experimental_2.11</artifactId>
-            <version>2.0.4</version>
-        </dependency>
-        <dependency>
-            <groupId>com.typesafe.akka</groupId>
-            <artifactId>akka-http-experimental_2.11</artifactId>
-            <version>2.0.4</version>
-        </dependency>
-        <dependency>
-            <groupId>com.typesafe.akka</groupId>
-            <artifactId>akka-http-spray-json-experimental_2.11</artifactId>
-            <version>2.0.4</version>
+            <artifactId>akka-http-spray-json_2.11</artifactId>
+            <version>${akka.http.version}</version>
         </dependency>
 
+
     </dependencies>
 
 

piflow-server/src/main/resources/application.conf

@@ -0,0 +1,10 @@
+PiFlowHTTPService{
+  shutdown-timeout:300s
+}
+
+akka {
+  http {
+    idle-timeout = 600 s
+    request-timeout = 200 s
+  }
+}

piflow-server/src/main/scala/cn/piflow/api/HTTPService.scala

@@ -19,7 +19,8 @@ import spray.json.DefaultJsonProtocol
 
 
 object HTTPService extends DefaultJsonProtocol with Directives with SprayJsonSupport{
-  implicit val system = ActorSystem("HTTPService", ConfigFactory.load())
+  implicit val config = ConfigFactory.load()
+  implicit val system = ActorSystem("PiFlowHTTPService", config)
   implicit val materializer = ActorMaterializer()
   implicit val executionContext = system.dispatcher
   var processMap = Map[String, SparkAppHandle]()