add getIcon for ConfigurableStop

piflow-bundle/pom.xml

@@ -26,6 +26,27 @@
             <artifactId>classutil_2.11</artifactId>
             <version>1.3.0</version>
         </dependency>
+        <dependency>
+            <groupId>com.sksamuel.scrimage</groupId>
+            <artifactId>scrimage-core_2.11</artifactId>
+            <version>2.1.0</version>
+        </dependency>
+        <dependency>
+            <groupId>com.sksamuel.scrimage</groupId>
+            <artifactId>scrimage-io-extra_2.11</artifactId>
+            <version>2.1.0</version>
+        </dependency>
+        <dependency>
+            <groupId>com.sksamuel.scrimage</groupId>
+            <artifactId>scrimage-filters_2.11</artifactId>
+            <version>2.1.0</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+            <version>1.7.25</version>
+        </dependency>
 
     </dependencies>
 

piflow-bundle/src/main/scala/cn/piflow/bundle/common/Fork.scala

@@ -22,4 +22,6 @@ class Fork extends ConfigurableStop{
   }
 
   override def getPropertyDescriptor(): List[PropertyDescriptor] = ???
+
+  override def getIcon(): Array[Byte] = ???
 }

piflow-bundle/src/main/scala/cn/piflow/bundle/common/Merge.scala

@@ -6,6 +6,8 @@ import cn.piflow.{JobContext, JobInputStream, JobOutputStream, ProcessContext}
 
 class Merge extends ConfigurableStop{
 
+  var inports : List[String] = _
+
   def perform(in: JobInputStream, out: JobOutputStream, pec: JobContext): Unit = {
     out.write(in.ports().map(in.read(_)).reduce((x, y) => x.union(y)));
   }
@@ -19,4 +21,6 @@ class Merge extends ConfigurableStop{
   }
 
   override def getPropertyDescriptor(): List[PropertyDescriptor] = ???
+
+  override def getIcon(): Array[Byte] = ???
 }

piflow-bundle/src/main/scala/cn/piflow/bundle/common/SelectField.scala

@@ -36,6 +36,8 @@ class SelectField extends ConfigurableStop {
   }
 
   override def getPropertyDescriptor(): List[PropertyDescriptor] = ???
+
+  override def getIcon(): Array[Byte] = ???
 }
 
 

piflow-bundle/src/main/scala/cn/piflow/bundle/csv/CSVParser.scala

@@ -62,5 +62,7 @@ class CSVParser extends ConfigurableStop{
   }
 
   override def getPropertyDescriptor(): List[PropertyDescriptor] = ???
+
+  override def getIcon(): Array[Byte] = ???
 }
 

piflow-bundle/src/main/scala/cn/piflow/bundle/hive/PutHiveStreaming.scala

@@ -32,4 +32,6 @@ class PutHiveStreaming extends ConfigurableStop {
   }
 
   override def getPropertyDescriptor(): List[PropertyDescriptor] = ???
+
+  override def getIcon(): Array[Byte] = ???
 }

piflow-bundle/src/main/scala/cn/piflow/bundle/hive/SelectHiveQL.scala

@@ -1,9 +1,10 @@
 package cn.piflow.bundle.hive
 
 import cn.piflow._
+import cn.piflow.conf.util.ImageUtil
 import cn.piflow.conf.ConfigurableStop
 import cn.piflow.conf.bean.PropertyDescriptor
-import cn.piflow.conf.util.MapUtil
+import cn.piflow.conf.util.{ImageUtil, MapUtil}
 import org.apache.spark.sql.SparkSession
 
 
@@ -36,6 +37,10 @@ class SelectHiveQL extends ConfigurableStop {
     descriptor = hiveQL :: descriptor
     descriptor
   }
+
+  override def getIcon(): Array[Byte] = {
+    ImageUtil.getImage("./src/main/resources/selectHiveQL.jpg")
+  }
 }
 
 

piflow-bundle/src/main/scala/cn/piflow/bundle/jdbc/JDBCRead.scala

@@ -43,4 +43,6 @@ class JDBCRead extends ConfigurableStop  {
   }
 
   override def getPropertyDescriptor(): List[PropertyDescriptor] = ???
+
+  override def getIcon(): Array[Byte] = ???
 }

piflow-bundle/src/main/scala/cn/piflow/bundle/jdbc/JDBCWrite.scala

@@ -38,4 +38,6 @@ class JDBCWrite extends ConfigurableStop{
   }
 
   override def getPropertyDescriptor(): List[PropertyDescriptor] = ???
+
+  override def getIcon(): Array[Byte] = ???
 }

piflow-bundle/src/main/scala/cn/piflow/bundle/json/JsonParser.scala

@@ -32,6 +32,8 @@ class JsonPathParser extends ConfigurableStop{
   }
 
   override def getPropertyDescriptor(): List[PropertyDescriptor] = ???
+
+  override def getIcon(): Array[Byte] = ???
 }
 
 class JsonStringParser extends ConfigurableStop{
@@ -57,4 +59,6 @@ class JsonStringParser extends ConfigurableStop{
   }
 
   override def getPropertyDescriptor(): List[PropertyDescriptor] = ???
+
+  override def getIcon(): Array[Byte] = ???
 }

piflow-bundle/src/main/scala/cn/piflow/bundle/json/JsonSave.scala

@@ -27,4 +27,6 @@ class JsonSave extends ConfigurableStop{
   }
 
   override def getPropertyDescriptor(): List[PropertyDescriptor] = ???
+
+  override def getIcon(): Array[Byte] = ???
 }

piflow-bundle/src/main/scala/cn/piflow/bundle/xml/XmlParser.scala

@@ -41,4 +41,6 @@ class XmlParser extends ConfigurableStop {
   }
 
   override def getPropertyDescriptor(): List[PropertyDescriptor] = ???
+
+  override def getIcon(): Array[Byte] = ???
 }

piflow-bundle/src/main/scala/cn/piflow/bundle/xml/XmlSave.scala

@@ -25,4 +25,6 @@ class XmlSave extends ConfigurableStop{
   }
 
   override def getPropertyDescriptor(): List[PropertyDescriptor] = ???
+
+  override def getIcon(): Array[Byte] = ???
 }

piflow-bundle/src/test/scala/cn/piflow/bundle/ClassFindTest.scala

@@ -1,6 +1,6 @@
 package cn.piflow.bundle
 
-import cn.piflow.bundle.util.ClassUtil.{findAllConfigurableStop, findConfigurableStop}
+import cn.piflow.conf.util.ClassUtil.{findAllConfigurableStop, findConfigurableStop}
 import org.junit.Test
 
 class ClassFindTest {
@@ -46,8 +46,9 @@ class ClassFindTest {
     val stop = findConfigurableStop(bundle)
     stop match {
       case Some(x) => {
-        val propertiesDescList = x.getPropertyDescriptor()
-        propertiesDescList.foreach(println(_))
+        println("Find " + x.getClass.toString + "!!!!!!!!")
+        //val propertiesDescList = x.getPropertyDescriptor()
+        //propertiesDescList.foreach(println(_))
       }
       case _ => println("Can not find : " + bundle)
     }
@@ -56,7 +57,7 @@ class ClassFindTest {
   @Test
   def testFindAllConfigurableStop() = {
     val allConfigurableStopList = findAllConfigurableStop()
-    println("\n\n\n" + allConfigurableStopList)
+    allConfigurableStopList.foreach( x => println(x))
   }
 
 }

piflow-bundle/src/test/scala/cn/piflow/bundle/FlowTest.scala

@@ -10,7 +10,7 @@ import scala.util.parsing.json.JSON
 
 class FlowTest {
 
-  /*@Test
+  @Test
   def testFlow(): Unit ={
 
     //parse flow json
@@ -40,7 +40,7 @@ class FlowTest {
 
     process.awaitTermination();
     spark.close();
-  }*/
+  }
 
   @Test
   def testFlow2json() = {

piflow-bundle/src/test/scala/cn/piflow/bundle/ImageTest.scala

@@ -0,0 +1,25 @@
+package cn.piflow.bundle
+
+import java.io.File
+
+import cn.piflow.conf.util.ImageUtil.{getImage, saveImage}
+import org.junit.Test
+
+class ImageTest {
+
+  @Test
+  def testGetImage() : Unit = {
+    //println(new File(".").getAbsolutePath)
+    val imagePath : String = "./src/test/resources/test.jpg"
+    val imageSavePath : String = "./src/test/resources/test_copy.jpg"
+
+    val imageByte = getImage(imagePath)
+
+    println(imageByte)
+
+    saveImage(imageByte, imageSavePath)
+
+  }
+
+
+}

piflow-conf/src/main/scala/cn/piflow/conf/ConfigurableStop.scala

@@ -10,4 +10,6 @@ trait ConfigurableStop extends Stop{
 
   def getPropertyDescriptor() : List[PropertyDescriptor]
 
+  def getIcon() : Array[Byte]
+
 }

piflow-conf/src/main/scala/cn/piflow/conf/bean/StopBean.scala

@@ -1,7 +1,7 @@
 package cn.piflow.conf.bean
 
 import cn.piflow.conf.ConfigurableStop
-import cn.piflow.conf.util.MapUtil
+import cn.piflow.conf.util.{ClassUtil, MapUtil}
 
 class StopBean {
   var uuid : String = _
@@ -18,9 +18,18 @@ class StopBean {
 
   def constructStop() : ConfigurableStop = {
     //val stop = Class.forName(this.bundle).getConstructor(classOf[Map[String, String]]).newInstance(this.properties)
-    val stop = Class.forName(this.bundle).newInstance()
+    /*val stop = Class.forName(this.bundle).newInstance()
     stop.asInstanceOf[ConfigurableStop].setProperties(this.properties)
-    stop.asInstanceOf[ConfigurableStop]
+    stop.asInstanceOf[ConfigurableStop]*/
+    val stop : Option[ConfigurableStop] = ClassUtil.findConfigurableStop(this.bundle)
+    stop match {
+      case Some(s) => {
+        s.asInstanceOf[ConfigurableStop].setProperties(this.properties)
+        s.asInstanceOf[ConfigurableStop]
+      }
+      case _ => throw new ClassNotFoundException(this.bundle + " is not found!!!")
+    }
+
   }
 
 }

piflow-conf/src/main/scala/cn/piflow/conf/util/ClassUtil.scala

@@ -1,14 +1,10 @@
-package cn.piflow.bundle.util
-
-
+package cn.piflow.conf.util
 
 import java.io.File
 
 import cn.piflow.conf.ConfigurableStop
 import org.clapper.classutil.ClassFinder
 
-import scala.util.control.Breaks
-
 
 object ClassUtil {
 
@@ -40,7 +36,7 @@ object ClassUtil {
     val finder = ClassFinder(getJarFile(classpathFile))
     val classes = finder.getClasses
     val classMap = ClassFinder.classInfoMap(classes)
-    val plugins = ClassFinder.concreteSubclasses("cn.piflow.conf.ConfigurableStop",classMap)
+    val plugins = ClassFinder.concreteSubclasses(configurableStopClass,classMap)
     plugins.foreach{
       pluginString =>
         //println(pluginString.name)

piflow-conf/src/main/scala/cn/piflow/conf/util/ImageUtil.scala

@@ -0,0 +1,19 @@
+package cn.piflow.conf.util
+
+import java.io.{BufferedInputStream, FileInputStream}
+
+import com.sksamuel.scrimage.Image
+
+
+object ImageUtil {
+
+  def getImage(imagePath:String) : Array[Byte] = {
+    val input = new BufferedInputStream(new FileInputStream(imagePath))
+    Image.fromStream(input).bytes
+  }
+
+  def saveImage(imageBytes :  Array[Byte], savePath : String) = {
+    Image(imageBytes).output(savePath)
+  }
+
+}