Dataflow programming in Oz language

ForCollect.oz

declare
proc {ForCollect Xs P Ys}
    Acc={NewCell Ys}
    proc {C X} R2 in
        @Acc=X|R2 Acc:=R2
    end
in
    for X in Xs do
        {P C X}
    end
    @Acc=nil
end

local R in
   {ForCollect [1 2 3 4 5] proc {$ C I} {C I*I} end R}
   {Browse R}
end

NotPrime.oz

fun {Prod S N}
   local Producer in
      fun {Producer L}
	 if L == N then N|nil else L|{Producer L+1} end
      end
      {Producer S}
   end
end
fun {NotPrime S1 S2}
   case S1 of nil then nil
   [] H|TS1 then
      if {List.member H S2} then
	 {NotPrime TS1 S2}
      else
	 H|{NotPrime TS1 S2}
      end
   end
end
fun {Sieve Xs}
   case Xs of nil then nil
   [] X|Xr then X|{Sieve thread {Filter Xr X} end}
   end
end
fun {Filter Xs K}
   case Xs of X|Xr then
      if X mod K \= 0 then X|{Filter Xr K}
      else {Filter Xr K}
      end
   else nil
   end
end

stateless_agent.oz

declare Math NewAgent0
proc {Math M}
   case M
   of add(N M A) then A=N+M
   [] mul(N M A) then A=N*M
   end
end
fun {NewAgent0 Process}
   Port Stream
in
   Port={NewPort Stream}
   thread {ForAll Stream Process} end
   Port
end

local Ans1 Ans2 R in
   R = {NewAgent0 Math}
   {Send R add(10 10 Ans1)}
   {Browse Ans1}
   {Send R mul(10 10 Ans2)}
   {Browse Ans2}
end

Stream.oz

declare X0 X1 X2 X3 in
thread X1=X0+1 end
thread X3=X1+X2 end
{Browse [X0 X1 X2 X3]}

X0=1
X2=3

declare Disp
proc {Disp S}
   case S of X|S2 then {Browse X} {Disp S2} end
end

declare S
thread {Disp S} end

declare S2 in S=a|b|c|S2

declare S3 in S2=d|e|f|S3

declare Prod
fun {Prod N}
   {Delay 1000} N|{Prod N+1}
end

declare S
thread S={Prod 1} end
thread {Disp S} end

declare Trans
fun {Trans S}
   case S of X|S2 then X*X|{Trans S2} end
end
declare S1 S2
thread S1={Prod 1} end
thread S2={Trans S1} end
thread {Disp S2} end

declare Producer Consumer Filter Display
fun {Producer N}
   local Prod in
      fun {Prod L}
	 if L == N then N|nil else L|{Prod L+1} end
      end
      {Prod 1}
   end
end
fun {Filter S}
   case S of nil then nil
   [] X|S2 then
      if X mod 2 == 0 then {Filter S2} else X|{Filter S2} end
   end
end
fun {Consumer S}
   local Sum in
      fun {Sum L Acc}
	 case L of nil then Acc
	 [] X|L2 then {Sum L2 Acc+X}
	 end
      end
      {Sum S 0}
   end
end
local S1 S2 S3 in
   thread S1={Producer 20} end
   thread S2={Filter S1} end
   thread S3={Consumer S2} end
   {Browse S3}
end

{Browse {Consumer {Filter {Producer 20}}}}

declare Counter
fun {Counter InS}
   local EnrichDict Producer Consumer Dict in
      Dict = {NewCell dict()}
      fun {EnrichDict Dict C}
	 if {Value.hasFeature Dict C} then
	    local Count in 
	       Count = Dict.C + 1
	       {Record.adjoinAt Dict C Count}
	    end
	 else {Record.adjoin Dict dict(C: 1)}
	 end
      end
      fun {Producer S}
	 case S of nil then nil
	 [] C|S2 then C|{Producer S2}
	 end
      end
      fun {Consumer S}
	 case S of nil then nil
	 [] C|S2 then
	    Dict := {EnrichDict @Dict C}
	    {Record.toListInd @Dict}|{Consumer S2}
	 end
      end
      local S1 S2 in
	 thread S1 = {Producer InS} end
	 thread S2 = {Consumer S1} end
	 S2
      end
   end
end

   
local InS X in
    X = {Counter InS}
    InS = a|b|a|c|_
    {Browse X}
end


{Browse {Counter [a b a c d]}}

{Browse {HasFeature dict(a: 1) b}}