Streaming.pudu

Pudu47 lines1.9 KB

Copy pathGitHub ↗

1/** @Examples.Streaming.Program — a stream consumed by a sink, collected, and pulled by a reader */2module Streaming34import Std.Io as Io5import PuduLangMediator.Context as Context6import PuduLangMediator.Mediator as Mediator7import PuduLangMediator as Messaging8import PuduLangMediator.Reader as Reader9import PuduLangMediator.Registration as Registration10import PuduLangMediator.Stream as Stream1112/// Streams the first numbers of the Fibonacci sequence three ways.13fn main() -> Int {14  let fibonacci: Stream.Kind[Int, Int, Str] = Stream.kind("fibonacci")15  let built = Mediator.build([16      Registration.streamHandler(&fibonacci, fn(count: Int, context: Context.Context, sink: Stream.Sink[Int]) -> Messaging.Outcome[(), Str] {17          var current = 018          var following = 119          var emitted = 020          while emitted < count {21            Context.check(&context) ?22            if !sink(current) { return Ok(()) }23            let sum = current + following24            current = following25            following = sum26            emitted = emitted + 127          }28          Ok(())29        }),30      Registration.streamBehavior(&fibonacci, fn(_count: Int, context: Context.Context, sink: Stream.Sink[Int], next: Stream.Next[Int, Str]) -> Messaging.Outcome[(), Str] {31          next(context, |item: Int| item % 2 == 1 || sink(item))32        })33    ])34  let mediator = match built {35    case Ok(found) => found36    case Err(invalid) => panic(Mediator.explain(&invalid))37  }38  let evens = Mediator.collect(&mediator, &fibonacci, 12)39  let _said = Io.writeLine("even Fibonacci numbers: " + Messaging.summarize(&evens))40  let reader = Mediator.reader(&mediator, &fibonacci, 1000, &Context.create(), 4)41  let first = Reader.next(&reader)42  let second = Reader.next(&reader)43  Reader.close(&reader)44  let _pulled = Io.writeLine("pulled " + show(first) + " then " + show(second))45  if evens == Ok([0, 2, 8, 34]) && first == Some(Ok(0)) && second == Some(Ok(2)) { 0 } else { 1 }46}47