Handler/Sink refactor, added Pipe trait

src/main/scala/outwatch/Handler.scala

@@ -0,0 +1,19 @@
+package outwatch
+
+import cats.effect.IO
+import outwatch.Sink.{ObservableSink, SubjectSink}
+
+object Handler {
+
+  /**
+    * This function also allows you to create initial values for your newly created Handler.
+    * This is equivalent to calling `startWithMany` with the given values.
+    * @param seeds a sequence of initial values that the Handler will emit.
+    * @tparam T the type parameter of the elements
+    * @return the newly created Handler.
+    */
+  def create[T](seeds: T*): IO[Pipe[T, T]] = Pipe.create(seeds:_*)
+
+  def create[T]: IO[Pipe[T, T]] = Pipe.create[T]
+}
+

src/main/scala/outwatch/Pipe.scala

@@ -0,0 +1,67 @@
+package outwatch
+
+import cats.effect.IO
+import outwatch.Sink.{ObservableSink, SubjectSink}
+import rxscalajs.Observable
+
+
+trait PipeOps[-I, +O] { self : Pipe[I, O]  =>
+
+  def mapSink[I2](f: I2 => I): Pipe[I2, O] = Pipe(redirectMap(f), self)
+
+  def mapSource[O2](f: O => O2): Pipe[I, O2] = Pipe(self, self.map(f))
+
+  def mapPipe[I2, O2](f: I2 => I)(g: O => O2): Pipe[I2, O2] = Pipe(self.redirectMap(f), self.map(g))
+
+
+  def collectSink[I2](f: PartialFunction[I2, I]): Pipe[I2, O] = Pipe(self.redirect(_.collect(f)), self)
+
+  def collectSource[O2](f: PartialFunction[O, O2]): Pipe[I, O2] = Pipe(self, self.collect(f))
+
+  def collectPipe[I2, O2](f: PartialFunction[I2, I])(g: PartialFunction[O, O2]): Pipe[I2, O2] = Pipe(
+    self.redirect(_.collect(f)), self.collect(g)
+  )
+
+
+  def filterSource(f: O => Boolean): Pipe[I, O] = Pipe(self, self.filter(f))
+
+
+  def transformSink[I2](f: Observable[I2] => Observable[I]): Pipe[I2, O] = Pipe(self.redirect(f), self)
+
+  def transformSource[O2](f: Observable[O] => Observable[O2]): Pipe[I, O2] = Pipe(self, f(self))
+
+  def transformPipe[I2, O2](f: Observable[I2] => Observable[I])(g: Observable[O] => Observable[O2]): Pipe[I2, O2] =
+    Pipe(self.redirect(f), g(self))
+}
+
+
+object Pipe {
+  private[outwatch] def apply[I, O](sink: Sink[I], source: Observable[O]): Pipe[I, O] =
+    new ObservableSink[I, O](sink, source) with PipeOps[I, O]
+
+  // This is not in PipeOps, because I is contravariant
+  implicit class FilterSink[I, +O](pipe: Pipe[I, O]) {
+    def filterSink(f: I => Boolean): Pipe[I, O] = Pipe(pipe.redirect(_.filter(f)), pipe)
+  }
+
+  /**
+    * This function also allows you to create initial values for your newly created Pipe.
+    * This is equivalent to calling `startWithMany` with the given values.
+    * @param seeds a sequence of initial values that the Pipe will emit.
+    * @tparam T the type parameter of the elements
+    * @return the newly created Pipe.
+    */
+  def create[T](seeds: T*): IO[Pipe[T, T]] = create[T].map { pipe =>
+    if (seeds.nonEmpty) {
+      pipe.transformSource(_.startWithMany(seeds: _*))
+    }
+    else {
+      pipe
+    }
+  }
+
+  def create[T]: IO[Pipe[T, T]] = IO {
+    val sink = SubjectSink[T]()
+    Pipe(sink, sink)
+  }
+}

src/main/scala/outwatch/Sink.scala

@@ -1,11 +1,11 @@
 package outwatch
 
 import cats.effect.IO
-import outwatch.dom.Handler
 import rxscalajs.facade.SubjectFacade
 import rxscalajs.subscription.Subscription
 import rxscalajs.{Observable, Observer, Subject}
 
+
 import scala.scalajs.js
 
 sealed trait Sink[-T] extends Any {
@@ -47,12 +47,16 @@ sealed trait Sink[-T] extends Any {
 }
 
 object Sink {
-  private final case class SubjectSink[T]() extends Subject[T](new SubjectFacade) with Sink[T] {
-    override private[outwatch] def observer = this
+
+  // these classes need to be in this file, because Sink is sealed
+  private[outwatch] case class ObservableSink[-I, +O](
+    sink: Sink[I], source: Observable[O]
+  ) extends Observable[O](source.inner) with Sink[I]{
+    override private[outwatch] def observer = sink.observer
   }
-  private final case class ObservableSink[T]
-  (oldSink: Sink[T], stream: Observable[T]) extends Observable[T](stream.inner) with Sink[T] {
-    override private[outwatch] def observer = oldSink.observer
+
+  private[outwatch] final case class SubjectSink[T]() extends Subject[T](new SubjectFacade) with Sink[T] {
+    override private[outwatch] def observer = this
   }
 
   /**
@@ -78,27 +82,6 @@ object Sink {
     sink
   }
 
-  /**
-    * Creates a Handler that is both Observable and Sink.
-    * An Observable with Sink is an Observable that can also receive Events, i.e. it’s both a Source and a Sink of events.
-    * If you’re familiar with Rx, they’re very similar to Subjects.
-    * This function also allows you to create initial values for your newly created Handler.
-    * This is equivalent to calling `startWithMany` with the given values.
-    * @param seeds a sequence of initial values that the Handler will emit.
-    * @tparam T the type parameter of the elements
-    * @return the newly created Handler.
-    */
-  def createHandler[T](seeds: T*): IO[Handler[T]] = IO {
-    val handler = new SubjectSink[T]
-
-    if (seeds.nonEmpty) {
-      ObservableSink[T](handler, handler.startWithMany(seeds: _*))
-    }
-    else {
-      handler
-    }
-  }
-
 
   private def completionObservable[T](sink: Sink[T]): Option[Observable[Unit]] = {
     sink match {
@@ -122,7 +105,7 @@ object Sink {
     * @return the resulting sink, that will forward the values
     */
   def redirect[T,R](sink: Sink[T])(project: Observable[R] => Observable[T]): Sink[R] = {
-    val forward = Sink.createHandler[R]().unsafeRunSync()
+    val forward = SubjectSink[R]()
 
     completionObservable(sink)
       .fold(project(forward))(completed => project(forward).takeUntil(completed))
@@ -144,8 +127,8 @@ object Sink {
     * @return the two resulting sinks, that will forward the values
     */
   def redirect2[T,U,R](sink: Sink[T])(project: (Observable[R], Observable[U]) => Observable[T]): (Sink[R], Sink[U]) = {
-    val r = Sink.createHandler[R]().unsafeRunSync()
-    val u = Sink.createHandler[U]().unsafeRunSync()
+    val r = SubjectSink[R]()
+    val u = SubjectSink[U]()
 
     completionObservable(sink)
       .fold(project(r, u))(completed => project(r, u).takeUntil(completed))
@@ -170,9 +153,9 @@ object Sink {
   def redirect3[T,U,V,R](sink: Sink[T])
                        (project: (Observable[R], Observable[U], Observable[V]) => Observable[T])
                        :(Sink[R], Sink[U], Sink[V]) = {
-    val r = Sink.createHandler[R]().unsafeRunSync()
-    val u = Sink.createHandler[U]().unsafeRunSync()
-    val v = Sink.createHandler[V]().unsafeRunSync()
+    val r = SubjectSink[R]()
+    val u = SubjectSink[U]()
+    val v = SubjectSink[V]()
 
     completionObservable(sink)
       .fold(project(r, u, v))(completed => project(r, u, v).takeUntil(completed))
@@ -197,6 +180,4 @@ object Sink {
 
 }
 
-final case class ObserverSink[-T](observer: Observer[T]) extends AnyVal with Sink[T]
-
-
+final case class ObserverSink[-T](observer: Observer[T]) extends AnyVal with Sink[T]

src/main/scala/outwatch/dom/HandlerFactories.scala

@@ -0,0 +1,43 @@
+package outwatch.dom
+
+import cats.effect.IO
+import org.scalajs.dom.{ClipboardEvent, DragEvent, KeyboardEvent, MouseEvent}
+import outwatch.dom.helpers.InputEvent
+
+/**
+  * Trait containing event handlers, so they can be mixed in to other objects if needed.
+  */
+trait HandlerFactories {
+
+  @deprecated("Use Handler.inputEvents instead", "0.11.0")
+  def createInputHandler() = Pipe.create[InputEvent]
+  @deprecated("Use Handler.mouseEvents instead", "0.11.0")
+  def createMouseHandler() = Pipe.create[MouseEvent]
+  @deprecated("Use Handler.keyboardEvents instead", "0.11.0")
+  def createKeyboardHandler() = Pipe.create[KeyboardEvent]
+  @deprecated("Use Handler.dragEvents instead", "0.11.0")
+  def createDragHandler() = Pipe.create[DragEvent]
+  @deprecated("Use Handler.clipboardEvents instead", "0.11.0")
+  def createClipboardHandler() = Pipe.create[ClipboardEvent]
+
+  @deprecated("Use Handler.create[String] instead", "0.11.0")
+  def createStringHandler(defaultValues: String*) = Pipe.create[String](defaultValues: _*)
+  @deprecated("Use Handler.create[Boolean] instead", "0.11.0")
+  def createBoolHandler(defaultValues: Boolean*) = Pipe.create[Boolean](defaultValues: _*)
+  @deprecated("Use Handler.create[Double] instead", "0.11.0")
+  def createNumberHandler(defaultValues: Double*) = Pipe.create[Double](defaultValues: _*)
+
+  @deprecated("Use Handler.create[T] instead", "0.11.0")
+  def createHandler[T](defaultValues: T*): IO[Pipe[T, T]] = Pipe.create[T](defaultValues: _*)
+
+
+  implicit class HandlerCreateHelpers(handler: Handler.type) {
+    lazy val inputEvents = Pipe.create[InputEvent]
+    lazy val mouseEvents = Pipe.create[MouseEvent]
+    lazy val keyboardEvents = Pipe.create[KeyboardEvent]
+    lazy val dragEvents = Pipe.create[DragEvent]
+    lazy val clipboardEvents = Pipe.create[ClipboardEvent]
+  }
+
+}
+

src/main/scala/outwatch/dom/VDomModifier.scala

@@ -4,7 +4,7 @@ import cats.effect.IO
 import org.scalajs.dom._
 import outwatch.dom.helpers.DomUtils
 import scala.scalajs.js.|
-import rxscalajs.{Observable, Observer}
+import rxscalajs.Observer
 import snabbdom.{VNodeProxy, h}
 
 import scala.scalajs.js

src/main/scala/outwatch/dom/helpers/DomUtils.scala

@@ -2,7 +2,7 @@ package outwatch.dom.helpers
 
 import cats.effect.IO
 import org.scalajs.dom._
-import org.scalajs.dom.raw.HTMLInputElement
+import org.scalajs.dom.html
 import outwatch.dom._
 import rxscalajs.Observable
 import rxscalajs.subscription.Subscription
@@ -76,7 +76,7 @@ object DomUtils {
 
   private val valueSyncHook: (VNodeProxy, VNodeProxy) => Unit = (_, node) => {
     node.elm.foreach { elm =>
-      val input = elm.asInstanceOf[HTMLInputElement]
+      val input = elm.asInstanceOf[html.Input]
       if (input.value != input.getAttribute("value")) {
         input.value = input.getAttribute("value")
       }

src/main/scala/outwatch/dom/helpers/InputEvent.scala

@@ -1,12 +1,13 @@
 package outwatch.dom.helpers
 
-import org.scalajs.dom.raw.{Event, HTMLInputElement}
+import org.scalajs.dom.Event
+import org.scalajs.dom.html
 
 import scala.scalajs.js.annotation.ScalaJSDefined
 
 @ScalaJSDefined
 class InputEvent() extends Event {
   override def target = {
-    super.target.asInstanceOf[HTMLInputElement]
+    super.target.asInstanceOf[html.Input]
   }
 }

src/main/scala/outwatch/dom/package.scala

@@ -5,11 +5,23 @@ import cats.effect.IO
 import scala.language.implicitConversions
 
 
-package object dom extends Attributes with Tags with Handlers {
+package object dom extends Attributes with Tags with HandlerFactories {
 
   type VNode = IO[VNode_]
   type VDomModifier = IO[VDomModifier_]
 
+  type Observable[+A] = rxscalajs.Observable[A]
+  val Observable = rxscalajs.Observable
+
+  type Sink[-A] = outwatch.Sink[A]
+  val Sink = outwatch.Sink
+
+  type Pipe[-I, +O] = outwatch.Pipe[I, O]
+  val Pipe = outwatch.Pipe
+
+  type Handler[T] = outwatch.Handler[T]
+  val Handler = outwatch.Handler
+
   implicit def stringNode(string: String): VDomModifier = IO.pure(StringNode(string))
 
   implicit def optionIsEmptyModifier(opt: Option[VDomModifier]): VDomModifier = opt getOrElse IO.pure(EmptyVDomModifier)

src/main/scala/outwatch/package.scala

@@ -0,0 +1,6 @@
+import rxscalajs.Observable
+
+package object outwatch {
+  type Pipe[-I, +O] = Observable[O] with Sink[I] with PipeOps[I, O]
+  type Handler[T] = Pipe[T,T]
+}

src/main/scala/outwatch/util/Store.scala

@@ -1,7 +1,7 @@
 package outwatch.util
 
 import cats.effect.IO
-import outwatch.Sink
+import outwatch.{Pipe, Sink}
 import outwatch.dom._
 import outwatch.dom.helpers.STRef
 import rxscalajs.Observable
@@ -12,7 +12,7 @@ import scala.language.implicitConversions
 
 final case class Store[State, Action](initialState: State,
                                            reducer: (State, Action) => (State, Option[IO[Action]]),
-                                           handler: Observable[Action] with Sink[Action]) {
+                                           handler: Pipe[Action, Action]) {
   val sink: Sink[Action] = handler
   val source: Observable[State] = handler
     .scan(initialState)(fold)
@@ -35,19 +35,19 @@ final case class Store[State, Action](initialState: State,
 }
 
 object Store {
-  implicit def toSink[Action](store: Store[_, Action]): Sink[Action] = store.sink
-  implicit def toSource[State](store: Store[State, _]): Observable[State] = store.source
+  implicit def toHandler[State, Action](store: Store[State, Action]): Pipe[Action, State] =
+    Pipe(store.sink, store.source)
 
   private val storeRef = STRef.empty
 
   def renderWithStore[S, A](initialState: S, reducer: (S, A) => (S, Option[IO[A]]), selector: String, root: VNode): IO[Unit] = for {
-    handler <- createHandler[A]()
+    handler <- Pipe.create[A]()
     store <- IO(Store(initialState, reducer, handler))
     _ <- storeRef.asInstanceOf[STRef[Store[S, A]]].put(store)
     _ <- OutWatch.render(selector, root)
   } yield ()
 
-  def getStore[S, A]: IO[Store[S, A]] = 
+  def getStore[S, A]: IO[Store[S, A]] =
     storeRef.asInstanceOf[STRef[Store[S, A]]].getOrThrow(NoStoreException)
 
   private object NoStoreException extends

src/main/scala/outwatch/util/WebSocket.scala

@@ -1,7 +1,7 @@
 package outwatch.util
 
 import cats.effect.IO
-import org.scalajs.dom.raw.{CloseEvent, ErrorEvent, MessageEvent}
+import org.scalajs.dom.{CloseEvent, ErrorEvent, MessageEvent}
 import outwatch.Sink
 import rxscalajs.Observable