pipe-async.edh

# producer procedures are syntatically very appealing to be used in data pipeline
# definition and assembling


# %{

# %%
producer add3( intake, outlet ) {
  for x from intake do outlet <- {
    x+3
  }
} @=> outlet <-nil # finally mark end-of-stream

# %%
producer mul5( intake, outlet ) {
  for x from intake do outlet <- {
    x*5
  }
} @=> outlet <-nil # finally mark end-of-stream

# %%
producer somePipeline ( intake, outlet ) {
  for y from (
    intake | add3 | mul5
  ) do outlet <- y
} @=> outlet <-nil # finally mark end-of-stream

# %}

# %%
producer someStreamIn( bias, outlet ) {
  outlet <- 3+bias
  outlet <- 11+bias
  outlet <- 2+bias
  outlet <-nil # mark end-of-stream
}

# %%
console.print$ ' * the input series:'
for y from someStreamIn( 300 ) do {
  console.print( y )
}

# %%
console.print$ ' * the output series:'
for y from someStreamIn( 300 ) | somePipeline do {
  console.print( y )
}


# %{

# %%
# producer based pipelines can be fed lazily / interactively,
# due to the async nature
feedIn = chan
perceive (
  feedIn | add3 | mul5
) { y } -> {
  # note the output to terminal via console.print can be blocked by repl,
  # so better to use console log here
  console.info<| 'pipe yielded a value: ' ++ y
}


# %%
feedIn <- 3 + 300

# %%
feedIn <- 11 + 300

# %%
feedIn <- 2 + 300

# %}


# %%
# wait a tiny bit to see all async induced logs, in case this file
# is run by runedh instead of interactively
for _ from console.everyMillis( 10 ) do { break }