scalaz.stream Process & sequences of Futures

[Henry Story]

Hi,

I am trying to apply the scalaz.stream library to a problem space involving
actors fetching resources on the web. I am not sure if the Process is the
right tool for this yet at all, as I am exploring the library.

Introduction
------------

Just to build up the problem a bit I have a few functions that seem to work:

//get a URI off the web ( or locally )
def get(uri: Rdf#URI): Process[Future,LinkedDataResource[Rdf]] = {
val url = uri.fragmentLess
val ldr: Future[LinkedDataResource[Rdf]] =
rww.execute(getLDPR(url)).map{gr=>LinkedDataResource(url,PointedGraph(uri,gr))}
await(ldr){o=>emit(o)}
}

//follow a relation in a graph
def follow(rel: Rdf#URI): Process1[LinkedDataResource[Rdf],LinkedDataResource[Rdf]] = {
await1[LinkedDataResource[Rdf]].flatMap{ldr=>
emitAll((ldr.resource/rel).toSeq).map{pg=>LinkedDataResource(ldr.location,pg)}
}
}

So this allows one to the following to find the members of the group

get(tpacGroup) |> follow(foaf.member)


Sequences of Futures
--------------------

The above is nice but not very useful, as it can be done much more efficiently
without Processes. What we really want to do is see if the URIs found by following
a link in a document is defined locally in the graph under consideration or remotely.
If it is remote, we'd like to GET those, and follow on from there.

So here I want to define a function ~> like this

def ~>(rel: Rdf#URI): Process[...?] = {
await1[LinkedDataResource[Rdf]].flatMap{ldr=>
val res = (ldr.resource/rel)
val local_remote = res.groupBy {
pg =>
foldNode(pg.pointer)(
uri => if (uri.fragmentLess == ldr.location) "local" else "remote",
bnode => "local",
lit => "local"
)
}

emitAll(local_remote("local").map(pg=>LinkedDataResource(ldr.location,pg)).toSeq) ++ {
val x = get(local_remote("remote").map(pg=>pg.pointer.asInstanceOf[Rdf#URI]).toSeq)
x
}
}
}

but I am already not quite sure what the signature for it should be. Probably
Process1[LinkedDataResource[Rdf],LinkedDataResource[Rdf]] because that is what
the follow( ) function is defined as and I'd like this to be used in the same way.
But as it calls a the get(...) function with the following signature things don't work out.


def get(uris: Seq[Rdf#URI]): Process[Future,LinkedDataResource[Rdf]] = {
val futures = uris.map { uri =>
val url = uri.fragmentLess
val ldr = rww.execute(getLDPR(url)).map { gr => LinkedDataResource(url, PointedGraph(uri, gr))}
ldr
}
futures ....
}

I want this to GET asynchronously all the resources and treat the
result as a Stream so that the next value is available as soon as a remote
fetch is finished.

I did in fact get this to work with Enumerators but I was wondering
reading chapter 15 of the book on Functional Programming if I would
gain some advantages doing this with Processes.

https://github.com/w3c/banana-rdf/blob/master/ldp/src/main/scala/LinkedResource.scala#L42

Henry

Social Web Architect
http://bblfish.net/

[Paul Chiusano]

A couple things: 

First, it's cool that you are experimenting with scalaz-stream, but I just want it to be clear that the code is alpha quality, so don't be surprised if you find bugs or weirdness in the API!

* I wouldn't use Future, assuming that is the scalaz Future, since that doesn't bake in any error handling (and will fail if any exceptions are thrown in the Futures you are constructing). It's really more of a primitive if you wanted to define your own error handling strategy. Instead, I'd use scalaz.concurrent.Task.

* You can say Process.wrap { myFuture } instead of await(myFuture)(o => emit(o))

* I'm not totally following your example, but it sounds like you just want to allow your ~> function to evaluate Futures (Tasks). You can represent this as a Process[Future, LinkedDataResource[Rdf] => Future[List[LinkedDataResource[Rdf]]], which is type aliased in scalaz-stream as a Channel[Future, LinkedDataResource[Rdf], List[LinkedDataResource[Rdf]]]

Then you could write something like: 

get(tpacGroup) |> follow(foaf.member) through (~>(blah)) flatMap (emitAll)

The `through` function attaches a channel onto a `Process` - it is the effectful analogue of `|>`. Any time you have an effectful channel like this, there is some decision you get to make about the 'queueing' strategy you want to use - do you want to require each value sent to the channel to be evaluated before sending the next value is sent (that is what `through` does), or do you want to allow for some nondeterminism, with elements queueing up and potentially be executed in parallel. I am just starting to explore patterns for this second use case, but ultimately, scalaz-stream will have very good support for that sort of thing, in addition to allowing for out-of-order evaluation (like when you may place items in a Channel in some order, but want to allow the results to come out in any order, based on when they become available). You can look at `through_y`, `ChanneledProcess.enqueue` and `wye.byipWith` to see some stuff that I am playing with in this area. 

Paul

--
You received this message because you are subscribed to the Google Groups "scalaz" group.
To unsubscribe from this group and stop receiving emails from it, send an email to scalaz+un...@googlegroups.com.
To post to this group, send email to sca...@googlegroups.com.
Visit this group at http://groups.google.com/group/scalaz?hl=en.
For more options, visit https://groups.google.com/groups/opt_out.

[Henry Story]

On Sunday, 14 April 2013 18:05:19 UTC+2, Paul Chiusano wrote:

A couple things: 

First, it's cool that you are experimenting with scalaz-stream, but I just want it to be clear that the code is alpha quality, so don't be surprised if you find bugs or weirdness in the API!

Thanks for the warning. I'll keep playing around with it so that I can feel comfortable with it when it is stable, but I may use Enumerations in the mean time

for the project I am on, as I have some deadlines to respect. I'll publish the work I have done here on a branch.

 

* I wouldn't use Future, assuming that is the scalaz Future, since that doesn't bake in any error handling (and will fail if any exceptions are thrown in the Futures you are constructing). It's really more of a primitive if you wanted to define your own error handling strategy. Instead, I'd use scalaz.concurrent.Task.

In fact I needed to use scala.concurrent.Future, ( as my futures are getting evaluated by akka ) but I wrote it along the lines of Task to get things done.

 

* You can say Process.wrap { myFuture } instead of await(myFuture)(o => emit(o)) 

thanks

 

* I'm not totally following your example, but it sounds like you just want to allow your ~> function to evaluate Futures (Tasks). You can represent this as a Process[Future, LinkedDataResource[Rdf] => Future[List[LinkedDataResource[Rdf]]], which is type aliased in scalaz-stream as a Channel[Future, LinkedDataResource[Rdf], List[LinkedDataResource[Rdf]]]

Then you could write something like: 

get(tpacGroup) |> follow(foaf.member) through (~>(blah)) flatMap (emitAll)

The `through` function attaches a channel onto a `Process` - it is the effectful analogue of `|>`. Any time you have an effectful channel like this, there is some decision you get to make about the 'queueing' strategy you want to use - do you want to require each value sent to the channel to be evaluated before sending the next value is sent (that is what `through` does), or do you want to allow for some nondeterminism, with elements queueing up and potentially be executed in parallel.

I was really hoping to be able to use nondeterminism as much as possible, so that the first futures to return would be the ones to be looked at in the subsequent

pipe. When fetching things on the web it is impossible to tell which order things will be returned in, and in fact in RDF order is not important ( most of the time ).

 

I am just starting to explore patterns for this second use case, but ultimately, scalaz-stream will have very good support for that sort of thing, in addition to allowing for out-of-order evaluation (like when you may place items in a Channel in some order, but want to allow the results to come out in any order, based on when they become available). You can look at `through_y`, `ChanneledProcess.enqueue` and `wye.byipWith` to see some stuff that I am playing with in this area. 

Thanks for the tips.