AkkaStream.java

import akka.NotUsed;
import akka.actor.ActorSystem;
import akka.japi.Creator;
import akka.japi.Pair;
import akka.stream.*;
import akka.stream.javadsl.*;
import scala.collection.immutable.Seq;
import scala.concurrent.duration.FiniteDuration;

import java.util.Iterator;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Stream;

/**
 * Created by fabiofumarola on 21/03/16.
 */
public class Tutorial {

    final ActorSystem system = ActorSystem.create("Tutorial");
    final Materializer materializer = ActorMaterializer.create(system);


    public void zipSource() {

        final Source<Integer, NotUsed> src1 = Source.fromIterator(() -> Stream.iterate(1, n -> n + 1).iterator());

        final Source<Integer, NotUsed> src2 = Source.
                fromIterator(() -> Stream.iterate(1000, n -> n + 1).iterator())
                .throttle(1, FiniteDuration.create(1, TimeUnit.SECONDS),1,ThrottleMode.shaping());


        final Flow<Integer, Integer, NotUsed> expand = Flow.of(Integer.class)
                .expand(i -> Stream.iterate(i, n -> n).iterator());

        final Graph<SourceShape<Pair<Integer, Integer>>, NotUsed> graph = GraphDSL.create(b -> {

            final FanInShape2<Integer, Integer, Pair<Integer, Integer>> zip = b.add(Zip.<Integer, Integer>create());
            final SourceShape<Integer> src1S = b.add(src1);
            final SourceShape<Integer> src2S = b.add(src2);
            final FlowShape<Integer, Integer> expandS = b.add(expand);

            b.from(src1S).toInlet(zip.in0());
            b.from(src2S).via(expandS).toInlet(zip.in1());

            return SourceShape.of(zip.out());

        });

        Source.fromGraph(graph)
                .runForeach(System.out::println, materializer);

    }

    public static void main(String[] args) {
        final Tutorial tutorial = new Tutorial();

        tutorial.zipSource();
    }
}