Page History

import akka.Done;
import akka.NotUsed;
import akka.actor.ActorSystem;
import akka.stream.ActorMaterializer;
import akka.stream.IOResult;
import akka.stream.ThrottleMode;
import akka.stream.javadsl.FileIO;
import akka.stream.javadsl.Flow;
import akka.stream.javadsl.RunnableGraph;
import akka.stream.javadsl.Sink;
import akka.stream.javadsl.Source;
import java.nio.file.Paths;
import java.time.Duration;
import java.util.Iterator;
import java.util.UUID;
import java.util.concurrent.CompletionStage;
import java.util.stream.Stream;

public class EventStorageApp {

    public static void main(String[] args) {
        ActorSystem system = ActorSystem.create();
        ActorMaterializer materializer = ActorMaterializer.create(system);

        Source<String, NotUsed> source = Source.fromIterator(() -> eventIterator());
        Sink<String, CompletionStage<IOResult>> sink = FileIO.toPath(Paths.get("/path/to/file"));
        Flow<String, String, NotUsed> throttle = Flow.of(String.class)
            .throttle(60, Duration.ofSeconds(1));

        RunnableGraph<CompletionStage<IOResult>> runnableGraph = source.via(throttle).to(sink);
        CompletionStage<IOResult> completion = runnableGraph.run(materializer);

        completion.thenAccept(result -> {
            System.out.println("File write result: " + result);
            system.terminate();
        });
    }

    private static Iterator<String> eventIterator() {
        return Stream.generate(() -> UUID.randomUUID().toString()).iterator();
    }

}