[akka-streams] Flow mapAsync implemented with Actor ask pattern does not work

[Jeroen Kransen]

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

[Patrik Nordwall]

That should work. I see one problem with you example. In flickrRdfToJson you are using context of the enclosing actor to create the FullSubjectQueryActor. It is not allowed to use context from other threads than the actor's receive thread. In this case you are calling it from the stream execution thread.

Instead, create the actor in the enclosing actor's constructor (or preStart). If you need to create one actor per request there is some discussion related to that in https://groups.google.com/d/msg/akka-user/-VF0ZeIt054/CW6PWO8z5LwJ

Example and documentation: http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0/scala/stream-integrations.html#Integrating_with_External_Services

Cheers,

Patrik

--
>>>>>>>>>> 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 akka-user+...@googlegroups.com.
To post to this group, send email to akka...@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

--

Patrik Nordwall
Typesafe -  Reactive apps on the JVM
Twitter: @patriknw

[Jeroen Kransen]

Thanks a lot! I guess there is no way for the actor system to detect and warn for this erroneous pattern? Or hopefully I am the only one making this error :-)

Op dinsdag 22 september 2015 10:02:48 UTC+2 schreef Patrik Nordwall: