Hi, I have struggled a lot to get a Flow implemented by an Actor ask
pattern. This is the essence of my code:
def flickrRdfToJson(flickrPhotoUri: URI): Future[UriQueryResponse] = {
Logger.debug(s"Flickr rdf to json for $flickrPhotoUri")
val queryActor =
context.actorOf(FullSubjectQueryActor.props(tripleStoreActor))
(queryActor ? UriQueryRequest(flickrPhotoUri)).mapTo[UriQueryResponse]
}
I added this function to a Flow:
val queryFlow: Flow[URI, UriQueryResponse, Unit] =
Flow[URI].mapAsync(1)(flickrRdfToJson)
I verified that the actor sends back a UriQueryResponse, but it is not picked
up by the flow. There are no dead messages.
Is there a reason that Flows do not work with Futures from Actor ask patterns?
If not, could I see a working example?
My whole stream:
val source =
Source.actorPublisher[JsObject](FlickrSearchActorPublisher.byDate(date))
val exifFlow: Flow[JsObject, JsObject, Unit] =
Flow[JsObject].mapAsync(4)(searchResultsToExif)
val triplesFlow: Flow[JsObject, URI, Unit] =
Flow[JsObject].mapAsync(1)(flickrExifToTriples)
val queryFlow: Flow[URI, UriQueryResponse, Unit] =
Flow[URI].mapAsync(1)(flickrRdfToJson)
val indexFlow: Flow[UriQueryResponse, Boolean, Unit] =
Flow[UriQueryResponse].map(flickrJsonToIndex)
val sinkCount: Sink[Boolean, Future[Int]] = Sink.fold(0) { (count: Int,
nextElem: Boolean) => println(s"next: $nextElem"); if (nextElem) count + 1 else
count }
val runnableFlow =
source.via(exifFlow).via(triplesFlow).via(queryFlow).via(indexFlow).toMat(sinkCount)(Keep.right)
val futureInt = runnableFlow.run()
futureInt.foreach(println)
I think the other function implementations are not relevant as I pinpointed the
problem to be at the reception of the UriQueryResponse message.
Thanks! Jeroen
--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ:
>>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
