rssh
diff --git a/‎src/main/scala/gopher/channels/InputOutput.scala
Lines changed: 65 additions & 2 deletions b/‎src/main/scala/gopher/channels/InputOutput.scala
Lines changed: 65 additions & 2 deletions
@@ -1,8 +1,9 @@
 package gopher.channels
 
-import gopher.FlowTermination
+import gopher.{ChannelClosedException, FlowTermination, GopherAPI}
 
-import scala.concurrent.Future
+import scala.concurrent.{ExecutionContext, Future}
+import scala.util.{Failure, Success}
 
 trait InputOutput[A,B] extends Input[B] with Output[A]
 {
@@ -25,6 +26,59 @@ trait InputOutput[A,B] extends Input[B] with Output[A]
 
     override def filter(p: B=>Boolean):InputOutput[A,B] = new FilteredIO(p)
 
+
+
+    class CompositionIO[C](other:InputOutput[B,C]) extends InputOutput[A,C]
+    {
+        private val s = new ForeverSelectorBuilder(){
+            override def api = thisInputOutput.api
+        }
+
+        s.readingWithFlowTerminationAsync(thisInputOutput,
+            {(ec:ExecutionContext,ft:FlowTermination[Unit],a:B) =>
+                implicit val iec = ec
+                other.awrite(a) map (_ => ())
+            }
+        )
+
+        protected val internalFuture = s.go
+
+
+        // TODO: think, maybe we need intercept cbread here ?
+        override def cbread[D](f: (ContRead[C, D]) => Option[(ContRead.In[C]) => Future[Continuated[D]]], ft: FlowTermination[D]): Unit =
+        {
+            if (checkNotCompleted(ft)) other.cbread(f,ft)
+        }
+
+        override def cbwrite[B](f: (ContWrite[A, B]) => Option[(A, Future[Continuated[B]])], ft: FlowTermination[B]): Unit =
+        {
+            if (checkNotCompleted(ft)) thisInputOutput.cbwrite(f,ft)
+        }
+
+        private def checkNotCompleted[D](ft:FlowTermination[D]): Boolean =
+        {
+            if (internalFuture.isCompleted) {
+                implicit val ec = api.gopherExecutionContext
+                internalFuture.onComplete{
+                    case Failure(ex) => ft.doThrow(ex)
+                    case Success(_) => ft.doThrow(new ChannelClosedException())
+                }
+                false
+            } else true
+        }
+
+        override val api: GopherAPI = thisInputOutput.api
+    }
+
+    //TODO: add test suite.
+    def compose[C](other:InputOutput[B,C]):InputOutput[A,C] =
+      new CompositionIO[C](other)
+
+    /**
+      * Synonym for this.compose(other)
+      */
+    def |>[C](other:InputOutput[B,C]):InputOutput[A,C] = this.compose(other)
+
 }
 
 
@@ -43,5 +97,14 @@ trait CloseableInputOutput[A,B] extends InputOutput[A,B] with CloseableInput[B]
 
     override def filter(p: B=>Boolean): CloseableInputOutput[A,B] = new FilteredIOC(p)
 
+    class CompositionIOC[C](other: InputOutput[B,C]) extends CompositionIO(other)
+                                                      with DoneSignalDelegate[C]
+                                                      with CloseableInputOutput[A,C]
+
+    override def compose[C](other: InputOutput[B,C]):CloseableInputOutput[A,C] =
+      new CompositionIOC[C](other)
+
+    override def |>[C](other:InputOutput[B,C]):CloseableInputOutput[A,C] = this.compose(other)
+
 
 }