Pekko tutorial

This repository contains a collection of runnable and self-contained examples from various Pekko Streams, Pekko Connectors and Pekko HTTP tutorials, blogs, and postings.

Akka vs Pekko
As of umbrella release 22.10 Akka has changed the licensing model and then for the 2.7.x release back to Apache 2.0.

Apache Pekko is the open source alternative. A BIG Thank you to the committed Pekko committers.

The branch migrate_pekko contains the migration (with a few losses). This is the only maintained branch. Currently we tentatively migrate to Scala 3 and the Pekko 2.x milestone builds.

Project Description

"It's working!" a colleague used to shout across the office when yet another proof of concept was running its first few hundred meters along the happy path, aware that the real work started right there. This repo aims to provide you with exactly this feeling by offering a collection of runnable examples in Scala and Java.

Getting Started

Prerequisites

• Java: Minimum is Java 17, we currently run on Java 21. Either distribution works: Temurin or GraalVM. CI builds and tests are on Java 17 (Temurin).
• sbt: the build uses sbt 2.x. The version is pinned in project/build.properties, and any recent sbt runner downloads it automatically.
• Docker

Installation

1. Clone the repository: git clone https://github.com/pbernet/akka_streams_tutorial.git
2. Navigate to the project directory: cd akka_streams_tutorial
3. Compile the project: sbt compile
4. Run the tests: sbt testFull

Running Examples

Each example class contains instructions on how to run it from the IDE. Most examples are throttled and provide a verbose log, by searching the log you see what is happening. Some examples deliberately throw RuntimeException e.g. to show recovery behaviour.

Examples Overview

Featured examples with complex workflows:

• Element deduplication
• Windturbine example
• Apache Kafka WordCount
• HL7 V2 over TCP via Kafka to Websockets
• Analyse Wikipedia edits live stream
• Movie subtitle translation via LLMs
• Local RAG chat with PDF docs

Many examples deal with shared state management. While most Pekko Streams operators are stateless, the samples in package sample.stream_shared_state show some trickier stateful operators in action.

The *Echo example series implement round trips e.g. HttpFileEcho and WebsocketEcho

Using testcontainers allows running realistic integration test scenarios with just one click:

• OIDCKeycloak
• ClickhousedbIT
• InfluxdbIT
• SlickIT

Examples of integrating AWS services with Pekko Connectors:

• KinesisEcho
• FirehoseEcho
• DynamoDBEcho
• SqsEcho
• S3Echo

Run them via the corresponding IT test classes locally in ministack/S3Proxy or against your AWS account.

Other example resources

• Maintained pekko-streams examples are in pekko-stream-tests and the Streams Cookbook
• Getting started guides: Streams Quickstart Guide and this popular stackoverflow article
• The doc chapters Modularity, Composition and Hierarchy and Design Principles behind Apache Pekko Streams provide useful background
• The concept of running streams using materialized values is also explained in this blog, this video and in this stackoverflow article
• Maintained Pekko Connectors examples

Element deduplication

Dropping identical (consecutive or non-consecutive) elements in an unbounded stream:

• DeduplicateConsecutiveElements using the sliding operator
• Dedupe shows the squbs Deduplicate GraphStage which allows to dedupe both types

The following use case uses a local caffeine cache to avoid duplicate HTTP file downloads:

• Process a stream of incoming messages with reoccurring TRACE_ID
• For the first message: download a .zip file from a FileServer and add TRACE_ID→Path to the local cache
• For subsequent messages with the same TRACE_ID: fetch file from cache to avoid duplicate downloads per TRACE_ID
• Use time based cache eviction to get rid of old downloads

Class	Description
FileServer	Local HTTP FileServer for non-idempotent file download simulation
LocalFileCacheCaffeine	Pekko streams client flow, with cache implemented with caffeine

Windturbine example

Working sample from the blog series 1-4 from Colin Breck where classic Actors are used to model shared state, life-cycle management and fault-tolerance in combination with Pekko Streams. Colin Breck explains these concepts and more in the 2017 Reactive Summit talk Islands in the Stream: Integrating Akka Streams and Akka Actors

Class	Description
SimulateWindTurbines	Starts n clients which feed measurements to the server
WindTurbineServer	Start server which collects measurements

The clients communicate via websockets with the WindTurbineServer. After a restart of SimulateWindTurbines the clients are able to resume. Shutting down the WindTurbineServer results in reporting to the clients that the server is not reachable. After restarting WindTurbineServer the clients are able to resume. Since there is no persistence, the processing just continuous.

Apache Kafka WordCount

The ubiquitous word count with an additional message count. A message is a sequence of words. Start the classes in the order below and watch the console output.

Class	Description
KafkaServerEmbedded	Uses Embedded Kafka (= an in-memory Kafka instance). No persistence on restart
WordCountProducer	pekko-streams-kafka client which feeds random words to topic wordcount-input
WordCountKStreams.java	Kafka Streams DSL client to count words and messages and feed the results to wordcount-output and messagecount-output topics. Contains additional interactive queries which should yield the same results WordCountConsumer
WordCountConsumer	pekko-streams-kafka client which consumes aggregated results from topic wordcount-output and messagecount-output

HL7 V2 over TCP via Kafka to Websockets

The PoC in package alpakka.tcp_to_websockets is from the E-Health domain, relaying HL7 V2 text messages in some kind of "Trophy" across these stages:

Hl7TcpClient → Hl7Tcp2Kafka → KafkaServer → Kafka2Websocket → WebsocketServer

The focus is on resilience (= try not to lose messages during the restart of the stages). However, currently messages may reach the WebsocketServer unordered (due to retry in Hl7TcpClient) and in-flight messages may get lost (upon re-start of WebsocketServer).

Start each stage separately in the IDE, or together via the integration test AlpakkaTrophySpec

Analyse Wikipedia edits live stream

Respect the Wikimedia API rate limits. WikipediaEditsAnalyser#nerProcessingFlow throttles to 180 requests/min and drops the oldest edits when the backlog grows.

See which persons appear in English Wikipedia articles while they are being edited, and chat about them. WikipediaEditsAnalyser consumes the Wikipedia edits SSE stream, finds persons with two NER engines and makes the results searchable and chattable.

flowchart LR
  SSE["Wikipedia edits<br/>SSE stream"] --> F["Filter<br/>enwiki, no bots"]
  F --> T["Throttle<br/>180/min"]
  T --> W["Fetch extract<br/>Wikipedia API"]
  W --> L["Local NER<br/>GLiNER, fallback OpenNLP"]
  L --> R["Remote NER<br/>OpenAI<br/>gpt-6-luna"]
  R --> OS[("OpenSearch<br/>one doc per article")]
  R --> E[("Embeddings<br/>in-memory")]
  OS <-- person search --> GUI["Web GUI<br/>http://localhost:8080/assistant"]
  E <-- RAG chat --> GUI
  click GUI href "http://localhost:8080/assistant" _blank

Loading

1. Start Docker and the GLiNER service: docker compose up -d gliner-ner in docker/
2. Set the environment variable OPENAI_API_KEY
3. Run WikipediaEditsAnalyser and wait 20s until the Web GUI opens

Movie subtitle translation via LLMs

SubtitleTranslator translates all blocks of an English source .srt file to a target language using LLMs via LangChain4j

Pekko streams helps with:

• Workflow modelling
• Scene splitting to session windows. All blocks of a scene are grouped in one session and then translated in one API call
• Throttling to not exceed the OpenAI API rate limits
• Continuous writing of translated blocks to the target file to avoid data loss on NW failure

Local RAG chat with PDF docs

LocalRagApplication launches a RAG system (RagEngine) that indexes local PDF documents into a PostgreSQL/pgvector store and serves two endpoints:

• Web Chat UI at http://localhost:8090/rag - interactive Q&A with source attribution and optional Cohere reranking
• MCP Server at http://localhost:8091/mcp - RAG chat as Model Context Protocol tools to use with Claude, Amp, etc.

Ingestion pipeline: Local PDF files → Docling HybridChunker → BGE-small-en embedding → pgvector

Retrieval pipeline: Request → pgvector similarity search → optional Cohere reranking → LLM answer generation

Requires running local Docker instances PostgreSQL with pgvector and docling-serve ( see docker/docker-compose-rag.yml) as well as env vars OPENAI_API_KEY, COHERE_API_KEY (see RagEngine)

The default local PDF files are 3 downloaded publications form https://cartographicperspectives.org. Suitable prompts:

• What is the definition of a map?
• Is there an article about 9/11?