Custom Sink:with CompletionStage - onPush() only get one element, why?

[Guofeng Zhang]

Hi,

I create a custom sink following the sample in Custom stream processing, When the onPush() implemented like the sample:

 public void onPush() throws Exception {

     Integer element = grab(in);

     System.out.println(element);

     pull(in);

 }

It works well. 

But when I changed it using CompletionStage in the body:

public void onPush() throws Exception {

   Integer element = grab(in);

   func(element).whenComplete((v, e) -> {

      Logger.debug("onPush: pull again");

      pull(in);

   });

}

where func() return the value of CompletionStage..

I see one 'onPush: pull again' on the console, then no further element pulled. the program seems hung.

I can not figure out what's wrong with the above code. Your help is appreciated.

Akka 2.4.17, Play 2.5.14, Java 8-111

Guofeng

[Akka Team]

Hi,

You cannot access any inner state or call any methods on the GraphStageLogic from another thread, which is what you are doing when you are executing it in whenComplete. To do this you must use the async callback as described in this section of the docs: http://doc.akka.io/docs/akka/2.5.0/java/stream/stream-customize.html#Using_asynchronous_side-channels

--

Johan

Akka Team

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

[Guofeng Zhang]

It works, Thanks!