Update apps for new API

WallarooLabs · Nov 3, 2018 · 2792c5f · 2792c5f
1 parent 0e83747
commit 2792c5f
Show file tree

Hide file tree

Showing 5 changed files with 38 additions and 23 deletions.
diff --git a/examples/pony/alphabet/_test/validate.py b/examples/pony/alphabet/_test/validate.py
@@ -17,6 +17,8 @@
 from json import loads
 from struct import calcsize, unpack
 
+print("Validating Alphabet")
+
 fmt = '>LsQ'
 def decoder(bs):
     return unpack(fmt, bs)[1:3]

diff --git a/examples/pony/celsius-kafka/celsius.pony b/examples/pony/celsius-kafka/celsius.pony
@@ -41,18 +41,19 @@ actor Main
     end
 
     try
-      let application = recover val
-        Application("Celsius Conversion App")
-          .new_pipeline[F32, F32]("Celsius Conversion",
-            KafkaSourceConfig[F32](ksource_clip.parse_options(env.args)?,
-            env.root as AmbientAuth, CelsiusKafkaDecoder))
-            .to[F32]({(): Multiply => Multiply})
-            .to[F32]({(): Add => Add})
-            .to_sink(KafkaSinkConfig[F32](FahrenheitEncoder,
-              ksink_clip.parse_options(env.args)?,
-              env.root as AmbientAuth))
+      let pipeline = recover val
+        let inputs = Wallaroo.source[F32]("Celsius Conversion",
+          KafkaSourceConfig[F32](ksource_clip.parse_options(env.args)?,
+          env.root as AmbientAuth, CelsiusKafkaDecoder))
+
+        inputs
+          .to[F32](Multiply)
+          .to[F32](Add)
+          .to_sink(KafkaSinkConfig[F32](FahrenheitEncoder,
+            ksink_clip.parse_options(env.args)?,
+            env.root as AmbientAuth))
       end
-      Startup(env, application, "celsius-conversion")
+      Wallaroo.build_application(env, "Celsius Conversion", pipeline)
     else
       @printf[I32]("Couldn't build topology\n".cstring())
     end

diff --git a/examples/pony/celsius/celsius.pony b/examples/pony/celsius/celsius.pony
@@ -31,17 +31,18 @@ use "wallaroo/core/topology"
 actor Main
   new create(env: Env) =>
     try
-      let application = recover val
-        Application("Celsius Conversion App")
-          .new_pipeline[F32, F32]("Celsius Conversion",
+      let pipeline = recover val
+        let inputs = Wallaroo.source[F32]("Celsius Conversion",
             TCPSourceConfig[F32].from_options(CelsiusDecoder,
               TCPSourceConfigCLIParser(env.args)?(0)?))
-            .to[F32]({(): Multiply => Multiply})
-            .to[F32]({(): Add => Add})
-            .to_sink(TCPSinkConfig[F32 val].from_options(FahrenheitEncoder,
-              TCPSinkConfigCLIParser(env.args)?(0)?))
+
+        inputs
+          .to[F32](Multiply)
+          .to[F32](Add)
+          .to_sink(TCPSinkConfig[F32 val].from_options(FahrenheitEncoder,
+            TCPSinkConfigCLIParser(env.args)?(0)?))
       end
-      Startup(env, application, "celsius-conversion")
+      Wallaroo.build_application(env, "Celsius Conversion", pipeline)
     else
       @printf[I32]("Couldn't build topology\n".cstring())
     end

diff --git a/examples/python/celsius_connectors/celsius.py b/examples/python/celsius_connectors/celsius.py
@@ -37,6 +37,7 @@ def application_setup(args):
     # ab.to_sink("fahrenheit_conversion")
     # return ab.build()
 
+    #!@
     values = wallaroo.source("convert",
                     wallaroo.???(decoder))
 

diff --git a/utils/data_receiver/data_receiver.pony b/utils/data_receiver/data_receiver.pony
@@ -64,11 +64,12 @@ actor Main
 
       // Start it up!
       let listener_addr = l_arg as Array[String]
+      let host = listener_addr(0)?
+      let port = listener_addr(1)?
       let tcp_auth = TCPListenAuth(env.root as AmbientAuth)
       TCPListener(tcp_auth,
-        ListenerNotify(env.out, env.err, input_mode, output_mode),
-        listener_addr(0)?,
-        listener_addr(1)?)
+        ListenerNotify(env.out, env.err, input_mode, output_mode, host, port),
+        host, port)
     else
       usage(env.out)
     end
@@ -91,16 +92,25 @@ class ListenerNotify is TCPListenNotify
   let _stderr: OutStream
   let _input_mode: InputMode
   let _output_mode: OutputMode
+  let _host: String
+  let _port: String
 
   new iso create(stdout: OutStream,
     stderr: OutStream,
     input_mode: InputMode,
-    output_mode: OutputMode)
+    output_mode: OutputMode,
+    host: String,
+    port: String)
   =>
     _stdout = stdout
     _stderr = stderr
     _input_mode = input_mode
     _output_mode = output_mode
+    _host = host
+    _port = port
+
+  fun ref listening(listen: TCPListener ref) =>
+    _stdout.print("Listening on " + _host + ":" + _port)
 
   fun ref not_listening(listen: TCPListener ref) =>
     _stderr.print("Unable to listen\n")