fold not working as expected using akka streams

Below is code I’ve written to try an output the sum of each akka message received which has been edited from this guide :

https://doc.akka.io/docs/akka/current/stream/stream-flows-and-basics.html

import akka.Done;
import akka.NotUsed;
import akka.actor.ActorRef;
import akka.actor.typed.ActorSystem;
import akka.actor.typed.javadsl.Behaviors;
import akka.japi.Pair;
import akka.stream.CompletionStrategy;
import akka.stream.OverflowStrategy;
import akka.stream.javadsl.Flow;
import akka.stream.javadsl.Sink;
import akka.stream.javadsl.Source;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

public class AkkaSourceTesting {

    public static void main(String args[]){

        ActorSystem actorSystem = ActorSystem.create(Behaviors.empty() , "actorSystem");

        Source<Integer, ActorRef> matValuePoweredSource =
                Source.actorRef(
                        elem -> {
                            // complete stream immediately if we send it Done
                            if (elem == Done.done()) return Optional.of(CompletionStrategy.immediately());
                            else return Optional.empty();
                        },
                        // never fail the stream because of a message
                        elem -> Optional.empty(),
                        100,
                        OverflowStrategy.fail());

        Pair<ActorRef, Source<Integer, NotUsed>> actorRefSourcePair =
                matValuePoweredSource.preMaterialize(actorSystem);

        actorRefSourcePair.first().tell(1, ActorRef.noSender());
        actorRefSourcePair.first().tell(1, ActorRef.noSender());
        actorRefSourcePair.first().tell(1, ActorRef.noSender());

        Flow<Integer, Integer, NotUsed> groupedFlow = Flow.of(Integer.class)
                .grouped(2)
                .map(value -> {
                    List<Integer> newList = new ArrayList<>(value);
                    return newList;
                })
                .mapConcat(value -> value);

        // pass source around for materialization
        actorRefSourcePair.second().via(Flow.of(Integer.class).via(groupedFlow).fold(0, (res, element) -> res + element)).runWith(Sink.foreach(System.out::println), actorSystem);

    }
}

The fold operation seems to cause nothing to outputted on the console.

However if I use

actorRefSourcePair.second().via(Flow.of(Integer.class).via(groupedFlow).map(x -> x * 2)).runWith(Sink.foreach(System.out::println), actorSystem);

instead of

actorRefSourcePair.second().via(Flow.of(Integer.class).via(groupedFlow).fold(0, (res, element) -> res + element)).runWith(Sink.foreach(System.out::println), actorSystem);
    

Then the following is outputted :

2
2

I’m attempting to group a list and perform a fold operation on each of the groups but the fold is not even being executed. Have I missed a step ?

Answer

Flow.fold does not emit a value until the upstream completes.

Also note that your groupedFlow is an identity flow: it could be removed without changing anything:

• grouped takes each successive pair of elements and bundles them into a List
• The map stage converts that List to an ArrayList
• mapConcat unwraps the ArrayList and emits the elements

The clearest expression of what you’re looking for (a stream of the sum of pairs of successive groups of 2) in Java is probably along the lines of

actorRefSourcePair.second()
    .grouped(2)
    .map(twoList -> twoList.stream().reduce(0, Integer::sum))
    .runWith(Sink.foreach(System.out::println), actorSystem);