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 On Fri, Sep 18, 2015 at 8:11 PM, Jeroen Kransen <[email protected]> wrote: > 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. > -- Patrik Nordwall Typesafe <http://typesafe.com/> - Reactive apps on the JVM Twitter: @patriknw -- >>>>>>>>>> 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.
