dynaml-pipes: Improvements to pipes api

 - DataPipe2 3 and 4 are now more compatible with DataPipe in
composition operations

Author: mandar2812

dynaml-pipes/src/main/scala-2.11/io/github/mandar2812/dynaml/pipes/DataPipe.scala

@@ -20,14 +20,19 @@ package io.github.mandar2812.dynaml.pipes
 
 import scalaxy.streams.optimize
 
+trait DataPipeConvertible[-Source, +Destination] {
+  def toPipe: (Source) => Destination
+}
+
+
 /**
   * Top level trait representing an
   * abstract pipe that defines a transformation
   * between two data types, i.e. [[Source]] and [[Destination]]
   * @author mandar2812 on 18/11/15.
   *
   * */
-trait DataPipe[-Source, +Destination] extends Serializable {
+trait DataPipe[-Source, +Destination] extends DataPipeConvertible[Source, Destination] with Serializable {
 
   self =>
 
@@ -47,11 +52,11 @@ trait DataPipe[-Source, +Destination] extends Serializable {
     * [[Source]] -> [[Further]]
     *
     * */
-  def >[Further](that: DataPipe[Destination, Further]):
-  DataPipe[Source, Further] = {
-    val runFunc = (d: Source) => that.run(self.run(d))
-    DataPipe(runFunc)
-  }
+  def >[Further](that: DataPipeConvertible[Destination, Further]) =
+    DataPipe((d: Source) => that.toPipe(self.run(d)))
+
+  /*def >[Result1, Result2](that: BifurcationPipe[Destination, Result1, Result2])
+  : BifurcationPipe[Source, Result1, Result2] = DataPipe((x: Source) => that.run(self.run(x)))*/
 
   /**
     * Represents the composition of two
@@ -72,6 +77,8 @@ trait DataPipe[-Source, +Destination] extends Serializable {
   def >-<[OtherSource, OtherDestination](that: DataPipe[OtherSource, OtherDestination])
   :DataPipe2[Source, OtherSource, (Destination, OtherDestination)] =
     DataPipe2((d1: Source, d2: OtherSource) => (self(d1), that(d2)))
+
+  override def toPipe: Source => Destination = self.run _
 }
 
 object DataPipe {
@@ -87,8 +94,7 @@ object DataPipe {
     }
   }
 
-  def apply[S1, D1, S2, D2](pipe1: DataPipe[S1, D1],
-                            pipe2: DataPipe[S2, D2]): ParallelPipe[S1, D1, S2, D2] =
+  def apply[S1, D1, S2, D2](pipe1: DataPipe[S1, D1], pipe2: DataPipe[S2, D2]): ParallelPipe[S1, D1, S2, D2] =
     ParallelPipe(pipe1.run, pipe2.run)
 
   def apply[S, D1, D2](func: (S) => (D1, D2)):
@@ -106,8 +112,6 @@ object DataPipe {
   }
 }
 
-
-
 trait ParallelPipe[-Source1, +Result1, -Source2, +Result2]
   extends DataPipe[(Source1, Source2), (Result1, Result2)] {
 
@@ -133,6 +137,14 @@ object ParallelPipe {
 trait BifurcationPipe[-Source, +Result1, +Result2]
   extends DataPipe[Source, (Result1, Result2)] {
 
+  self =>
+
+  def >[FinalResult](other: DataPipe2[Result1, Result2, FinalResult]): DataPipe[Source, FinalResult] =
+    DataPipe((input: Source) => {
+      val (x, y) = self.run(input)
+      other.run(x, y)
+    })
+
 }
 
 trait SideEffectPipe[I] extends DataPipe[I, Unit] {
@@ -141,12 +153,12 @@ trait SideEffectPipe[I] extends DataPipe[I, Unit] {
 
 object BifurcationPipe {
 
-  def apply[Source,
-  Destination1,
-  Destination2](pipe1: DataPipe[Source, Destination1],
-                pipe2: DataPipe[Source, Destination2]):
-  BifurcationPipe[Source, Destination1, Destination2] = {
+  def apply[Source, Destination1, Destination2](f: (Source) => (Destination1, Destination2)) = DataPipe(f)
 
+  def apply[Source, Destination1, Destination2](
+    pipe1: DataPipe[Source, Destination1],
+    pipe2: DataPipe[Source, Destination2]):
+  BifurcationPipe[Source, Destination1, Destination2] = {
     DataPipe((x: Source) => (pipe1.run(x), pipe2.run(x)))
   }
 }

dynaml-pipes/src/main/scala-2.11/io/github/mandar2812/dynaml/pipes/DataPipe2.scala

@@ -23,7 +23,10 @@ package io.github.mandar2812.dynaml.pipes
   *
   * Data Pipes representing functions of 2 arguments
   */
-trait DataPipe2[-Source1, -Source2, +Result] extends Serializable {
+trait DataPipe2[-Source1, -Source2, +Result] extends
+  DataPipeConvertible[(Source1, Source2), Result] with
+  Serializable {
+
   self =>
 
   def run(data1: Source1, data2: Source2): Result
@@ -33,6 +36,7 @@ trait DataPipe2[-Source1, -Source2, +Result] extends Serializable {
   def >[Result2](otherPipe: DataPipe[Result, Result2]): DataPipe2[Source1, Source2, Result2] =
     DataPipe2((d1: Source1, d2:Source2) => otherPipe.run(self.run(d1, d2)))
 
+  override def toPipe: ((Source1, Source2)) => Result = (x: (Source1, Source2)) => self.run(x._1, x._2)
 }
 
 object DataPipe2 {

dynaml-pipes/src/main/scala-2.11/io/github/mandar2812/dynaml/pipes/DataPipe3.scala

@@ -23,7 +23,9 @@ package io.github.mandar2812.dynaml.pipes
   *
   * Data Pipes representing functions of 3 arguments
   */
-trait DataPipe3[-Source1, -Source2, -Source3, +Result] extends Serializable {
+trait DataPipe3[-Source1, -Source2, -Source3, +Result] extends
+  DataPipeConvertible[(Source1, Source2, Source3), Result] with
+  Serializable {
   self =>
 
   def run(data1: Source1, data2: Source2, data3: Source3): Result
@@ -33,6 +35,8 @@ trait DataPipe3[-Source1, -Source2, -Source3, +Result] extends Serializable {
   def >[Result2](otherPipe: DataPipe[Result, Result2]): DataPipe3[Source1, Source2, Source3, Result2] =
     DataPipe3((d1: Source1, d2:Source2, d3: Source3) => otherPipe.run(self.run(d1, d2, d3)))
 
+  override def toPipe: ((Source1, Source2, Source3)) => Result =
+    (x: (Source1, Source2, Source3)) => self.run(x._1, x._2, x._3)
 }
 
 object DataPipe3 {

dynaml-pipes/src/main/scala-2.11/io/github/mandar2812/dynaml/pipes/DataPipe4.scala

@@ -23,7 +23,9 @@ package io.github.mandar2812.dynaml.pipes
   *
   * Data Pipes representing functions of 4 arguments
   */
-trait DataPipe4[-Source1, -Source2, -Source3, -Source4, +Result] extends Serializable {
+trait DataPipe4[-Source1, -Source2, -Source3, -Source4, +Result] extends
+  DataPipeConvertible[(Source1, Source2, Source3, Source4), Result] with
+  Serializable {
   self =>
 
   def run(data1: Source1, data2: Source2, data3: Source3, data4: Source4): Result
@@ -33,6 +35,9 @@ trait DataPipe4[-Source1, -Source2, -Source3, -Source4, +Result] extends Seriali
   def >[Result2](otherPipe: DataPipe[Result, Result2]): DataPipe4[Source1, Source2, Source3, Source4, Result2] =
     DataPipe4((d1: Source1, d2:Source2, d3: Source3, d4: Source4) => otherPipe.run(self.run(d1, d2, d3, d4)))
 
+  override def toPipe: ((Source1, Source2, Source3, Source4)) => Result =
+    (x: (Source1, Source2, Source3, Source4)) => self.run(x._1, x._2, x._3, x._4)
+
 }
 
 object DataPipe4 {