Skip to content
Advertisement

Sending data to an Akka Stream

I’m attempting to send an ArrayList to an Akka stream for processing. The stream processes the list as follows:

1. Sum the list
2. Square the result of the sum.

I have defined the code below to try to achieve this result:

public class SumListStream {

    final static akka.actor.ActorSystem system = akka.actor.ActorSystem.create("StreamsExamples");

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        int bufferSize = 100;
        final Source<Integer, ActorRef> source =
                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(),
                        bufferSize,
                        OverflowStrategy.dropHead());

        ActorRef actorRef = source
                        .fold(0, (aggr, next) -> aggr + next)
                        .map(x -> x * x)
                        .to(Sink.foreach(x -> System.out.println("got: " + x)))
                        .run(system);

        actorRef.tell(Arrays.asList(1,2,3), ActorRef.noSender());

        actorRef.tell(
                new akka.actor.Status.Success(CompletionStrategy.draining()), ActorRef.noSender());

    }
}

Sending data to the stream using : actorRef.tell(Arrays.asList(1,2,3), ActorRef.noSender()); does not appear to have any impact as the result of executing the above code is:

10:08:25.896 [StreamsExamples-akka.actor.default-dispatcher-5] INFO akka.event.slf4j.Slf4jLogger - Slf4jLogger started

Have I implemented the source correctly?

Advertisement

Answer

I think the first error is that you were sending a list to an actor that expects an Integer, as per the source typing. Send it an integer at a time, e.g.:

Arrays.asList(1, 2, 3).forEach(i -> actorRef.tell(i, ActorRef.noSender()));

Then, fold waits for the source to terminate before emitting its result. You set the actor to expect Done to terminate, but you are sending it Status.Success. Try:

actorRef.tell(Done.done(), ActorRef.noSender());

I am not an expert with Akka, but with these 2 modifications, your code yields the expected result (1 + 2 + 3)^2 = 36. You are never stopping the actor system, so the program runs forever, but the principles are there.

Advertisement