Graph API to complement Streams

[Jon Brisbin]

We've talked from early on about having a succinct and powerful Graph API that would allow you to create DAG-like [1] processing chains based on Predicates. A bit like Stream but more open, with the ability to specify Dispatchers, error handlers, and alternative paths.

I've been playing a little with some psuedo-code to see what that would look like. I wondered if anyone had any comments on this:

GraphSpec<T> graph = Graphs.compose()

.env(env)

.dispatcher(dispatcher)

.create("bad arg handler", new Consumer<State<Throwable>>(){})

.create("global err handler", new Consumer<State<Throwable>>(){});

// Provide a state object that passes through the graph.

NodeSpec<T> start = graph.first(new Function<T, State<T>>(){});

// Create a path of execution using the default Dispatcher.

// when() is basically a filter

start.when("condition name", new Predicate<State<T>>(){})

// if condition is true...

.then()

// ...follow this chain...

.swap(new Function<State<T>, State<V>>(){})

// ...and terminate.

.end(new Consumer<State<V>>(){}, "threadPool") 

// if condition is false...

.otherwise() 

// ...terminate here instead.

.end(new Consumer<State<T>>(){}); 

graph.get().accept(value);

If this was dumped to log or stdout, it would provide a pretty-printed map of the execution flow, printing out the user-defined names of nodes and describing the paths and how they're connected.

The API would also throw an Exception if a path was begun but never terminated, or if you specified a condition without an "otherwise", which would lead to items that don't pass the predicate being dropped (but you could explicitly configure things to be ignored).

Any thoughts, criticisms, etc…? :)

[1] - http://en.wikipedia.org/wiki/Directed_acyclic_graph

Thanks!

Jon Brisbin | Reactor Project Lead

http://about.me/jbrisbin | @j_brisbin

[Behrooz N]

Hi,

I really appreciate your idea. Just thinking out loud here:

• A DAG can be seen as a simplified version of finite state machine (FSM); as an example
• An FSM is very useful and handy in many situations that a decision should be made or a chain of execution should be built.
• If there would be a higher level DSL (could be also something in like Groovy) to be able to describe such FSM/DAG, then its usability is just multiplied.

Regards,

--
You received this message because you are subscribed to the Google Groups "reactor-framework" group.
To unsubscribe from this group and stop receiving emails from it, send an email to reactor-framew...@googlegroups.com.
For more options, visit https://groups.google.com/groups/opt_out.

--
-- Behrooz Nobakht

[Jon Brisbin]

I actually wrote some FSM code for Reactor quite some time ago but we took it out so we could implement it properly. This would be a continuation of that work, essentially.

A Graph's Nodes are in fact very much like an FSM's states. So I'm thinking that the combination of the two would be very useful. 

I'm also thinking of borrowing some of the semantics from Erlang's FSM like the init() method that returns a State object that's passed along (not that the method name will actually be "init" but you get the idea).

Thanks!

Jon Brisbin | Reactor Project Lead

http://about.me/jbrisbin | @j_brisbin

[Suminda Dharmasena]

May be you can borrow from FBP (http://en.wikipedia.org/wiki/Flow-based_programming). Also use http://noflojs.org/ editor files to create the network.

[Suminda Dharmasena]

Also may be to have Active Objects as nodes. Also Reactor can benefit from having a solid active object API.

[Suminda Dharmasena]

You should be able to keep some routes static and some dynamic.

Similar concept that might give you some ideas: https://github.com/lihaoyi/scala.rx

[张斌]

Hi,

Regarding the Graph API:

• Node<T> node(String name, Dispatcher dispatcher) always create a new Reactor instance,  is it too heavy or even necessary ?

• No way to retrieve the existing node through name ?
• How to create multiple routeTo conditions?  can i do it like this or do we have a better way?

         start.when(predicateA).routeTo(nodeA)

         start.when(predicateB).routeTo(nodeB)

         ...

         start.when(predicateN).routeTo(nodeN)

Thanks,

Bin

[Suminda Dharmasena]

How is this coming along?