akka streams api playground

akka-streams-api-playground.scala

/**
* Copyright (C) 2014 Typesafe Inc. <http://www.typesafe.com>
*/
package akka.stream

import org.reactivestreams.{ Publisher, Subscriber }

class Experiments {

  trait Graph {
    def broadcast[I](from: OutputOpen[_, I], to: InputOpen[I, _]*): Graph = ???
    def run() = ???
  }

  object Graph {
    def apply(): Graph = ???
  }

  trait InputOpen[I, O]

  trait OutputOpen[I, O]

  trait Flow[In, Out] {
    def in: Input[In]

    def steps: List[Any]

    def out: Output[Out]
  }

  class Flow[I, O](val steps: List[Any]) extends InputOpen[I, O] with OutputOpen[I, O] {
    def map[O2](f: O ⇒ O2): Flow[I, O2] = ???

    def to(subscriber: Subscriber[O]): OutputConnectedFlow[I, O] = ???
    def to(subscriber: Output[O]): OutputConnectedFlow[I, O] = ???
    def from(publisher: Publisher[I]): InputConnectedFlow[I, O] = ???
    def from(publisher: Input[I]): InputConnectedFlow[I, O] = ???
  }

  class InputConnectedFlow[I, O](in: Input[I], steps: List[Any])
    extends OutputOpen[I, O] {
    def detachInput(): Flow[I, O] = ???
    def to(subscriber: Subscriber[O]): ConnectedFlow[I, O] = ???
    def to(subscriber: Output[O]): ConnectedFlow[I, O] = ???
  }

  class OutputConnectedFlow[I, O](in: Input[I], steps: List[Any])
    extends InputOpen[I, O] {
    def detachOutput(): Flow[I, O] = ???
    def from(publisher: Publisher[I]): ConnectedFlow[I, O] = ???
    def from(publisher: Input[I]): ConnectedFlow[I, O] = ???

  }

  class ConnectedFlow[In, Out](in: Input[In], steps: List[Any], out: Output[Out]) {

    def run() = ???

    def detachInput(): OutputConnectedFlow[In, Out] = ???

    def detachOutput(): InputConnectedFlow[In, Out] = ???
  }

  trait Output[Out]

  trait Input[In]

  object Input {
    def apply[In](): Flow[In, In] = ???
  }

  val pipeline: Flow[Int, String] = Input[Int]().map(_.toString)
  val sub: Subscriber[String] = ???
  val pub: Publisher[Int] = ???

  pipeline.from(pub).to(sub).run()
  pipeline.to(sub).from(pub).run()

  // simple
  Graph()
    .broadcast(pipeline.from(pub), pipeline.to(sub))
    .run()

  Graph()
    .broadcast(pipeline.from(pub), pipeline.from(pub)) // input connected, compile error
    .run()

  Graph()
    .broadcast(pipeline.from(pub), pipeline) // input connected, compile error
    .run()

  trait MyLib {
    type Sound
    type Video
    type X = Sound with Video

    def audio: Input[Sound]
    def video: Input[Video]
    def mix: Output[X]

    def run() =
      Graph()
        .broadcast(pipeline.from(audio), mix)
        .run()

  }

}