Update RxFlow codegen to use ObservableSink.

Author: JoshRosen

README.md

@@ -12,7 +12,7 @@ The command line interface to the Bloom compiler is implemented by the
 During development, the easiest way to run the compiler is through `sbt run`;
 use
 
-    sbt run --infile <sourcefile> --target [dataflow|rxflow]
+    sbt "run --infile <sourcefile> --target [dataflow|rxflow]"
 
 to compile `sourcefile` and generate either a GraphViz .DOT file representing
 a dataflow graph (for the `dataflow` target) or Javascript code (for the

compiler/src/main/scala/edu/berkeley/cs/boom/bloomscala/codegen/js/RxFlowCodeGenerator.scala

@@ -50,8 +50,6 @@ object RxFlowCodeGenerator extends DataflowCodeGenerator with JsCodeGeneratorUti
         inputPort.name match {
           case "deltaIn" => ".insert"
         }
-      case OutputElement(collection) =>
-        empty  // Since outputs are currently implemented as RxJs Sbjects.
       case _ => ".input"
     })
   }
@@ -94,15 +92,14 @@ object RxFlowCodeGenerator extends DataflowCodeGenerator with JsCodeGeneratorUti
   }
 
   private def buildOutputs(graph: DataflowGraph): Doc = {
-    // TODO: this should create an Rx observable, not a Subject.
     val outputs = graph.outputs.values.map { output =>
-      (elemName(output), "new" <+> methodCall("rx", "Subject") <+>
+      (elemName(output), "new" <+> methodCall("rxflow", "ObservableSink") <+>
         comment(BloomPrettyPrinter.pretty(output.collection)))
     }.toMap
 
     "var" <+> "outputs" <+> equal <+> mapLiteral(outputs) <> semi <@@>
       graph.outputs.values.map { output =>
-        "this" <> dot <> output.collection.name <+> equal <+> elemRef(output) <> semi
+        "this" <> dot <> output.collection.name <+> equal <+> elemRef(output) <> dot <> "output" <> semi
       }.foldLeft(empty)(_ <@@> _)
   }