### Twitter Stream Quickstart - First Sample

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/stream/stream-quickstart.md

This is the initial sample code for the Twitter Stream Quickstart. It demonstrates a basic stream setup.

```scala
val source = io.Source.fromInputStream(getClass.getResourceAsStream("/twitter.txt"))
val lines = source.getLines()
val stream = Source.fromIterator(() => lines).via(new Twitter).map(_.hashtags)

stream.runWith(Sink.seq).map(println(_))
```

--------------------------------

### Twitter Stream Quickstart - First Sample (Java)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/stream/stream-quickstart.md

This is the initial sample code for the Twitter Stream Quickstart in Java. It demonstrates a basic stream setup.

```java
final Source<String, NotUsed> source = FileIO.fromResource("twitter.txt").via(Framing.delimiter(ByteString.fromString("\n"), 256));
final Sink<String, CompletionStage<Done>> sink = // ...

source.via(new Twitter()).map(t -> t.hashtags()).runWith(sink, materializer);
```

--------------------------------

### Main Application Setup (Java)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/typed/dispatchers.md

Sets up the actors and sends messages to demonstrate the blocking dispatcher problem. This example highlights thread starvation.

```java
public class BlockingDispatcherTest {

  public static void main(String[] args) throws Exception {
    ActorSystem<Void> system = ActorSystem.create();

    ActorRef<String> printActor = system.spawn(PrintActor.create(), "print-actor");
    ActorRef<String> blockingActor = system.spawn(BlockingActor.create(), "blocking-actor");

    // send 100 messages to blocking actors and print actors
    for (int i = 0; i < 100; i++) {
      blockingActor.tell("blocking message " + i);
      printActor.tell("print message " + i);
    }

    // allow actors to process messages
    Thread.sleep(5000);
    system.terminate();
  }
}
```

--------------------------------

### Main Actor System Setup (Scala)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/typed/guide/tutorial_1.md

Sets up and runs the Akka actor system for the supervision experiment. It starts the supervising actor.

```scala
object ActorHierarchyExperiments {
  def main(args: Array[String]): Unit = {
    val testKit = ActorTestKit("ActorHierarchyExperiments")

    val supervisingActor = testKit.spawn(SupervisingActor(), "supervising-actor")
    testKit.log.info("supervised actor started")

    // The supervised actor will fail and be restarted
    // We expect to see "supervised actor will be restarted" log message
    // and the supervised actor will be started again.
    // The supervisor will log the failure.
    Thread.sleep(2000)

    testKit.shutdownTestKit()
  }
}
```

--------------------------------

### Auction Setup (Java)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/typed/replicated-eventsourcing-auction.md

Initializes the auction entity in Java with minimum bid and starting parameters. The minimum bid is modeled as an `initialBid`.

```java
BigDecimal initialBid = new BigDecimal(100);
var auctionEntity = Auction.apply("auction-1", initialBid, system)
    .withAquirer(system)
    .withJournalConfig(commonJournalConfig);
```

--------------------------------

### Main Application Setup (Scala)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/typed/dispatchers.md

Sets up the actors and sends messages to demonstrate the blocking dispatcher problem. This example highlights thread starvation.

```scala
object BlockingDispatcherSample extends App {

  val printActor = ActorSystem("print-actor").spawn(PrintActor(), "print-actor")
  val blockingActor = ActorSystem("blocking-actor").spawn(BlockingActor(), "blocking-actor")

  // send 100 messages to blocking actors and print actors
  for (i <- 1 to 100) {
    blockingActor ! s"blocking message $i"
    printActor ! s"print message $i"
  }
}
```

--------------------------------

### Main Actor System Setup (Java)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/typed/guide/tutorial_1.md

Sets up and runs the Akka actor system for the supervision experiment in Java. It starts the supervising actor.

```java
public class ActorHierarchyExperiments {

    public static void main(String[] args) {
        ActorSystem<Void> actorSystem = ActorSystem.create(SupervisingActor.create(), "ActorHierarchyExperiments");

        try {
            // Allow time for actor to start, fail, and restart
            Thread.sleep(2000);
        } finally {
            actorSystem.terminate();
        }
    }
}
```

--------------------------------

### Auction Setup (Scala)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/typed/replicated-eventsourcing-auction.md

Initializes the auction entity with minimum bid and starting parameters. The minimum bid is modeled as an `initialBid`.

```scala
val initialBid = BigDecimal(100)
val auctionEntity =
  Auction("auction-1", initialBid)
    .withAquirer(system)
    .withJournalConfig(commonJournalConfig)
```

--------------------------------

### Starting Destination Actors (Scala)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/distributed-pub-sub.md

Example of how to start destination actors on multiple nodes for receiving messages. These actors will receive messages sent to the specified path.

```scala
val sendDestination = system.actorOf(Props[SendDestination], "destination")

// The 'sendDestination' actor will receive messages sent to the 'topic1' topic.
// The topic name is a logical name and does not need to be unique across the cluster.
system.actorOf(Props[SendDestination], "destination")
```

--------------------------------

### Persistence TestKit Initialization (Scala)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/typed/persistence-testing.md

Example of initializing persistence using `PersistenceInit` in Scala. This is useful for coordinating plugin initialization when multiple ActorSystems might start concurrently.

```scala
PersistenceInit.initializeDefaultPlugins(system, materializer)
```

--------------------------------

### Lease Configuration Example (Scala)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/coordination.md

Example of how to configure a lease implementation in Scala, specifying the lease class and its properties.

```scala
akka.coordination.lease {
  lease-class = "com.example.MyLease"
  time-to-live = "infinite"
  recovery-timeout = 1m
}
```

--------------------------------

### Java: Counter with Parameters in a Setup Class

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/typed/style-guide.md

Groups parameters into a 'Setup' class for easier management in functional-style actors. It separates changing state ('n') from immutable setup parameters.

```java
class Setup { // immutable "constructor" parameters
  final String name;
  final Timer timer;
  final ActorContext<Counter> context;

  Setup(String name, Timer timer, ActorContext<Counter> context) {
    this.name = name;
    this.timer = timer;
    this.context = context;
  }

  final int n = 0;

  public Behavior<Counter> counter() {
    return Behaviors.receiveMessage(msg -> {
      // ...
      return Behaviors.same();
    });
  }
}

// This is not a runnable example, it is just a snippet to illustrate the point.
```

--------------------------------

### Lease Implementation Example (Java)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/coordination.md

Example of how to implement a lease in Java. Extend `akka.coordination.lease.javadsl.Lease` and implement the required methods.

```java
public class MyLease implements javadsl.Lease {

  public MyLease(LeaseSettings leaseSettings) {
  }

  @Override
  public CompletionStage<Boolean> acquire(String owner) {
    return CompletableFuture.completedFuture(false);
  }

  @Override
  public CompletionStage<Boolean> release(String owner) {
    return CompletableFuture.completedFuture(false);
  }

  @Override
  public CompletionStage<Boolean> checkLease(String owner) {
    return CompletableFuture.completedFuture(false);
  }

  @Override
  public CompletionStage<Boolean> release(String owner, String expectedOwner) {
    return CompletableFuture.completedFuture(false);
  }

  @Override
  public CompletionStage<Boolean> acquire(String owner, String expectedOwner) {
    return CompletableFuture.completedFuture(false);
  }

  @Override
  public CompletionStage<Boolean> checkLease(String owner, String expectedOwner) {
    return CompletableFuture.completedFuture(false);
  }
}
```

--------------------------------

### Start ActorSystem and Guardian Actor

Source: https://github.com/akka/akka-core/blob/main/samples/akka-sample-sharding-java/README.md

Initializes an ActorSystem and joins the cluster. The Guardian actor is then started to bootstrap the application.

```java
public class KillrWeather {

    public static void main(String[] args) throws Exception {
        // Starts ActorSystem and joins the cluster through configuration
        ActorSystem<Guardian.Command> system = ActorSystem.create(Guardian.create(), "killrweather");

        // Starts a Guardian actor for the system
        // The Guardian actor bootstraps the application to shard WeatherStation actors across the cluster nodes.
        // No explicit message is sent to the guardian, it is started implicitly by ActorSystem.create.
    }
}
```

--------------------------------

### Start a Supervisor for Entity Actors (Java)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/cluster-sharding.md

Begin the process of starting a supervisor actor for entity actors using Java. The procedure is analogous to starting any other entity actor.

```java
ClusterSharding.get(system).start("Counter",
    Counter.props(self), ClusterShardingSettings.create(system), Counter.extractShardId());

```

--------------------------------

### Lease Implementation Example (Scala)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/coordination.md

Example of how to implement a lease in Scala. Extend `akka.coordination.lease.scaladsl.Lease` and implement the required methods.

```scala
class MyLease(leaseSettings: LeaseSettings) extends scaladsl.Lease { 

  override def acquire(owner: String): Future[Boolean] = ???

  override def release(owner: String): Future[Boolean] = ???

  override def checkLease(owner: String): Future[Boolean] = ???

  override def release(owner: String, expectedOwner: String): Future[Boolean] = ???

  override def acquire(owner: String, expectedOwner: String): Future[Boolean] = ???

  override def checkLease(owner: String, expectedOwner: String): Future[Boolean] = ???
}
```

--------------------------------

### Non-nested Flow Example

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/stream/stream-composition.md

This example demonstrates a basic flow composition without explicit module nesting. It serves as a starting point for understanding how to combine operators.

```Scala
val nonNestedFlow = Flow.fromSinkAndSource(
  Sink.cancelled[Int](),
  Source.single(1).map(_ * -1)
)
```

```Java
final Sink<Integer, CompletionStage<Done>> sink = Sink.cancelled();
final Source<Integer, NotUsed> source = Source.single(1).map(i -> i * -1);
final Flow<Integer, Integer, NotUsed> nonNestedFlow = Flow.fromSinkAndSource(sink, source);
```

--------------------------------

### flatMapConcat Example (Java)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/stream/operators/Source-or-Flow/flatMapConcat.md

Java version of the flatMapConcat example, illustrating the same pattern of transforming input elements into Sources and concatenating them.

```java
import akka.actor.ActorSystem;
import akka.stream.javadsl.Source;
import akka.stream.javadsl.Flow;
import java.util.Arrays;
import java.util.List;

public class FlatMapConcat {

    public static void main(String[] args) throws Exception {
        ActorSystem system = ActorSystem.create("System");

        Source<Integer, ?> customerIds = Source.from(Arrays.asList(1, 2, 3));

        Source<String, ?> customerEvents = customerIds.flatMapConcat(customerId -> lookupCustomerEvents(customerId));

        customerEvents.runForeach(System.out::println, system);
    }

    public static Source<String, ?> lookupCustomerEvents(Integer customerId) {
        switch (customerId) {
            case 1:
                return Source.from(Arrays.asList("Customer " + customerId + " Event A", "Customer " + customerId + " Event B"));
            case 2:
                return Source.from(Arrays.asList("Customer " + customerId + " Event C", "Customer " + customerId + " Event D"));
            case 3:
                return Source.from(Arrays.asList("Customer " + customerId + " Event E", "Customer " + customerId + " Event F"));
            default:
                return Source.empty();
        }
    }
}
```

--------------------------------

### Start First Seed Node (Separate JVMs)

Source: https://github.com/akka/akka-core/blob/main/samples/akka-sample-sharding-java/README.md

Starts the first seed node for a multi-JVM cluster setup. The argument '2553' specifies the port for this seed node, corresponding to the configuration.

```bash
mvn -pl killrweather exec:java -Dexec.args="2553"
```

--------------------------------

### Define initial schema version

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/serialization-jackson.md

The starting schema definition for the event class.

```Scala
case class ItemAdded(itemId: String, quantity: Int)
```

```Java
public class ItemAdded {
  public final String itemId;
  public final int quantity;

  @JsonCreator
  public ItemAdded(String itemId, int quantity) {
    this.itemId = itemId;
    this.quantity = quantity;
  }
}
```

--------------------------------

### Java Sink.lastOption Example

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/stream/operators/Sink/lastOption.md

Demonstrates using Sink.lastOption in Java to get the last element of a stream.

```java
CompletionStage<Optional<Integer>> result = Source.from(Arrays.asList(1, 2, 3))
    .runWith(Sink.lastOption(), materializer);

result.whenComplete((optionalValue, throwable) -> {
  if (throwable != null) {
    System.err.println("Error: " + throwable.getMessage());
  } else if (optionalValue.isPresent()) {
    System.out.println("Last element was: " + optionalValue.get()); // Last element was: 3
  } else {
    System.out.println("Stream was empty");
  }
});
```

--------------------------------

### Define Structural Changes for Customer Class

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/serialization-jackson.md

Examples of old and new class structures for migration.

```Scala
@@snip [Customer.scala](/akka-serialization-jackson/src/test/scala/doc/akka/serialization/jackson/v1/Customer.scala) { #structural }
```

```Java
@@snip [Customer.java](/akka-serialization-jackson/src/test/java/jdoc/akka/serialization/jackson/v1/Customer.java) { #structural }
```

```Scala
@@snip [Customer.scala](/akka-serialization-jackson/src/test/scala/doc/akka/serialization/jackson/v2a/Customer.scala) { #structural }
```

```Java
@@snip [Customer.java](/akka-serialization-jackson/src/test/java/jdoc/akka/serialization/jackson/v2a/Customer.java) { #structural }
```

--------------------------------

### Scala Sink.lastOption Example

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/stream/operators/Sink/lastOption.md

Demonstrates using Sink.lastOption in Scala to get the last element of a stream.

```scala
val result = Source(List(1, 2, 3)).runWith(Sink.lastOption)

result.onComplete { 
  case Success(Some(last)) => println(s"Last element was: $last") // Last element was: 3
  case Success(None)       => println("Stream was empty")
  case Failure(ex)         => println(s"Error: ${ex.getMessage}")
}
```

--------------------------------

### Interactive sbt Shell Example

Source: https://github.com/akka/akka-core/blob/main/CONTRIBUTING.md

Demonstrates entering commands directly into the interactive sbt shell for potentially faster execution and convenience.

```shell
% sbt
[info] Set current project to default (in build file:/.../akka/project/plugins/)
[info] Set current project to akka (in build file:/.../akka/)
> compile
...
> test
...
```

--------------------------------

### Rename Class for Serialization

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/serialization-jackson.md

Examples of old and new class definitions for renaming.

```Scala
@@snip [OrderAdded.scala](/akka-serialization-jackson/src/test/scala/doc/akka/serialization/jackson/v1/OrderAdded.scala) { #rename-class }
```

```Java
@@snip [OrderAdded.java](/akka-serialization-jackson/src/test/java/jdoc/akka/serialization/jackson/v1/OrderAdded.java) { #rename-class }
```

```Scala
@@snip [OrderPlaced.scala](/akka-serialization-jackson/src/test/scala/doc/akka/serialization/jackson/v2a/OrderPlaced.scala) { #rename-class }
```

```Java
@@snip [OrderPlaced.java](/akka-serialization-jackson/src/test/java/jdoc/akka/serialization/jackson/v2a/OrderPlaced.java) { #rename-class }
```

--------------------------------

### Source.setup

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/stream/operators/Source-or-Flow/setup.md

Defers the creation of a Source until materialization and provides access to the Materializer and Attributes.

```APIDOC
## Source.setup

### Description
Defers the creation of a Source until materialization and provides access to the Materializer and Attributes. Typically used when access to the materializer is needed to run a different stream during the construction of a source. Can also be used to access the underlying ActorSystem from ActorMaterializer.

### Method
POST

### Endpoint
/akka/akka-core/setup

### Parameters
#### Request Body
- **factory** (Function) - Required - A factory function that takes `ActorMaterializer` and `Attributes` and returns a `Source`.

### Request Example
```json
{
  "factory": "(materializer, attributes) => Source.empty[T]"
}
```

### Response
#### Success Response (200)
- **Source** (akka.stream.scaladsl.Source) - The created Source.

#### Response Example
```json
{
  "source": "..."
}
```
```

--------------------------------

### Get tweet authors stream

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/stream/futures-interop.md

This snippet shows how to obtain a stream of tweet authors. This is the starting point for further stream processing.

```scala
val tweetAuthors: Source[Author, NotUsed] = tweets.map(_.author)
```

```java
Source<Author, NotUsed> tweetAuthors = tweets.map(tweet -> tweet.getAuthor());
```

--------------------------------

### Basic Usage of Read Journal (Java)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/persistence-query.md

Obtain an instance of a ReadJournal to issue queries. This example shows how to get the 'NoopJournal'.

```java
import akka.persistence.query.PersistenceQuery;
import akka.persistence.query.javadsl.ReadJournal;

// Obtain the ReadJournal instance
final ReadJournal journal = PersistenceQuery.getReadJournalFor(
    NoopJournal.class,
    NoopJournal.identifier());

// Use the journal instance for queries...
```

--------------------------------

### Persistence TestKit Initialization (Java)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/typed/persistence-testing.md

Example of initializing persistence using `PersistenceInit` in Java. This helps manage concurrent plugin initialization across multiple ActorSystems in a cluster.

```java
PersistenceInit.initializeDefaultPlugins(system, materializer);
```

--------------------------------

### Chatroom Main Actor Setup (Java)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/typed/actors.md

Sets up the main Actor system, spawning a chat room and a gabbler actor. It uses Behaviors.setup for deferred behavior creation and context.watch to monitor child actors for termination.

```java
import akka.actor.typed.ActorSystem;
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.Void;

public class Main {

  public static Behavior<Void> create() {
    return Behaviors.setup(context -> {
      ActorRef<ChatRoom.Event> chatRoom = context.spawn(ChatRoom.create(), "ChatRoom");
      context.spawn(Gabbler.create(chatRoom), "Gabbler");
      return Behaviors.receiveSignal((context1, signal) -> {
        if (signal == akka.actor.typed.PostStop) {
          context1.getLog().info("Main actor is stopping");
        }
        return Behaviors.same();
      });
    });
  }

  public static void main(String[] args) {
    ActorSystem<Void> system = ActorSystem.create(create(), "ChatRoomDemo");
  }
}

```

--------------------------------

### Basic Usage of Read Journal (Scala)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/persistence-query.md

Obtain an instance of a ReadJournal to issue queries. This example shows how to get the 'NoopJournal'.

```scala
import akka.persistence.query.PersistenceQuery
import akka.persistence.query.scaladsl.ReadJournal

// Obtain the ReadJournal instance
val journal = PersistenceQuery
  .apply(system)
  .readJournalFor[ReadJournal](NoopJournal.identifier)

// Use the journal instance for queries...
```

--------------------------------

### Java preStart Initialization Example

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/actors.md

Demonstrates initializing an actor using the preStart method in Java. This method is called only once during the initial creation of the ActorRef, ensuring that certain initializations, like creating child actors, happen only at the actor's birth.

```java
@Override
public void preStart() {
  super.preStart();
  // initialize here
}
```

--------------------------------

### Compile and Run Akka Java Example with Gradle

Source: https://github.com/akka/akka-core/blob/main/samples/akka-quickstart-java/README.md

Use this command to compile and execute the Akka Java example project when using Gradle for build management.

```bash
./gradlew run
```

--------------------------------

### Java Subscriber Actor

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/distributed-pub-sub.md

Example of a Java actor subscribing to a topic. Ensure the actor is started on multiple nodes to receive messages.

```java
import akka.actor.AbstractActor;
import akka.actor.ActorRef;
import akka.cluster.pubsub.DistributedPubSubMediator;

public class DistributedPubSubMediatorTest {

  public static class Subscriber extends AbstractActor {
    private final ActorRef sendTo; // Actor to send messages to

    public Subscriber(ActorRef sendTo) {
      this.sendTo = sendTo;
    }

    @Override
    public Receive createReceive() {
      return receiveBuilder()
        .match(DistributedPubSubMediator.SubscribeAck.class, msg -> {
          System.out.println("subscribed to content");
        })
        .match(String.class, msg -> {
          if (msg.equals("content")) {
            System.out.println("received content message");
            sendTo.forward(msg, getContext().getSystem());
          }
        })
        .matchEquals("unsubscribe", msg -> {
          getContext().getSystem().getUndeploy().tell(new DistributedPubSubMediator.Unsubscribe("content", getSelf()), getSelf());
        })
        .build();
    }

    @Override
    public void preStart() {
      // subscribe to the topic
      getContext().getSystem().getUndeploy().tell(new DistributedPubSubMediator.Subscribe("content", getSelf()), getSelf());
    }
  }
}
```

--------------------------------

### Scala Subscriber Actor

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/distributed-pub-sub.md

Example of a Scala actor subscribing to a topic. Ensure the actor is started on multiple nodes to receive messages.

```scala
import akka.cluster.pubsub.DistributedPubSubMediator.{Subscribe, SubscribeAck}
import akka.actor.Actor

object DistributedPubSubMediatorSpec {

  object Subscriber {
    case object Unsubscribe
  }

  class Subscriber extends Actor {
    import DistributedPubSubMediatorSpec.Subscriber._

    // subscribe to the topic
    DistributedPubSubMediator ! Subscribe("content", self)

    def receive = {
      case SubscribeAck(Subscribe("content", _, _)) =>
        // AlreadySubscribed is not needed, the message is published to all subscribers
        // when the subscription is acknowledged.
        println("subscribed to content")
      case "content" =>
        println("received content message")
      case Unsubscribe =>
        DistributedPubSubMediator ! Unsubscribe("content", self)
      case UnsubscribeAck(Subscribe("content", _, _)) =>
        println("unsubscribed from content")
    }
  }

}
```

--------------------------------

### Nesting setup, withTimers, and withStash in Java

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/typed/style-guide.md

Illustrates nesting setup, withTimers, and withStash in Java, showing how to combine these behavioral modifiers.

```java
Behaviors.setup(ctx ->
    Behaviors.withTimers(
        timers ->
            Behaviors.withStash(
                10,
                stash ->
                    // Actor logic here, using ctx, timers, and stash
                    Behaviors.receive(
                        Command.class,
                        (innerCtx, msg) -> {
                          // ... handle message
                          return Behaviors.same();
                        })))
    ));
```

--------------------------------

### Chatroom Main Actor Setup (Scala)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/typed/actors.md

Sets up the main Actor system, spawning a chat room and a gabbler actor. It uses Behaviors.setup for deferred behavior creation and context.watch to monitor child actors for termination.

```scala
import akka.actor.typed.ActorSystem
import akka.actor.typed.Behavior
import akka.actor.typed.javadsl.Behaviors

object Main {
  def apply(): Behavior[Void] = Behaviors.setup[Void] { ctx =>
    val chatRoom = ctx.spawn(ChatRoom(), "ChatRoom")
    ctx.spawn(Gabbler(chatRoom), "Gabbler")
    Behaviors.receiveSignal { case (ctx, akka.actor.typed.PostStop) =>
      ctx.log.info("Main actor is stopping")
      Behaviors.same
    }
  }

  def main(args: Array[String]): Unit = {
    ActorSystem[Void](Main(), "ChatRoomDemo")
  }
}

```

--------------------------------

### Start Actor System (Java)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/typed/guide/tutorial_2.md

Provides the main entry point for the Akka application in Java. It initializes the actor system and logs a startup message.

```java
public class IotMain {

  public static void main(String[] args) {
    ActorSystem<GuardianCommand> start = ActorSystem.create(IotSupervisor.create(null), "MyActorSystem");
    start.tell(IotApp.Start);
  }
}

```

--------------------------------

### Example: Database Rows to Source (Java)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/stream/operators/Source/fromPublisher.md

This example demonstrates creating a Source from a database client that supports Reactive Streams. It shows how to process database rows and extract names, with backpressure applied throughout the stream.

```java
import akka.actor.ActorSystem;
import akka.stream.Materializer;
import akka.stream.javadsl.JavaFlowSupport;
import akka.stream.javadsl.Source;
import java.util.concurrent.Flow;
import java.util.concurrent.CompletionStage;
import java.util.Arrays;
import java.util.List;

// Assume a database client that returns a Reactive Streams Publisher
// For example, a hypothetical client returning rows as a Publisher<Row>

// Hypothetical Row class
class Row {
    int id;
    String name;

    public Row(int id, String name) {
        this.id = id;
        this.name = name;
    }

    public String getName() {
        return name;
    }
}

// Hypothetical database client
class DatabaseClient {
    public Flow.Publisher<Row> queryAllRows() {
        // In a real scenario, this would return actual data from a database
        // For demonstration, we create a simple publisher
        List<Row> data = Arrays.asList(
            new Row(1, "Alice"),
            new Row(2, "Bob"),
            new Row(3, "Charlie")
        );
        
        return new Flow.Publisher<Row>() {
            @Override
            public void subscribe(Flow.Subscriber<? super Row> subscriber) {
                // Simple implementation for demonstration
                subscriber.onSubscribe(new Flow.Subscription() {
                    private boolean cancelled = false;
                    private int index = 0;

                    @Override
                    public void request(long n) {
                        if (cancelled) return;
                        for (long i = 0; i < n && index < data.size(); i++) {
                            subscriber.onNext(data.get(index++));
                        }
                        if (index == data.size()) {
                            subscriber.onComplete();
                        }
                    }

                    @Override
                    public void cancel() {
                        cancelled = true;
                    }
                });
            }
        };
    }
}

public class FromPublisherExample {

    public static void main(String[] args) throws Exception {
        ActorSystem system = ActorSystem.create("FromPublisherExample");
        Materializer materializer = Materializer.createMaterializer(system);

        DatabaseClient dbClient = new DatabaseClient();

        // Create a Source from the database client's Publisher
        Flow.Publisher<Row> dbPublisher = dbClient.queryAllRows();
        Source<Row, akka.NotUsed> rowsSource = JavaFlowSupport.Source.fromPublisher(dbPublisher);

        // Process the rows to extract names
        Source<String, akka.NotUsed> namesSource = rowsSource.map(Row::getName);

        // Consume the names (e.g., print them)
        CompletionStage<Void> resultFuture = namesSource.runForeach(name -> System.out.println("Processing name: " + name), materializer);

        resultFuture.whenComplete((done, ex) -> {
            system.terminate();
        });
    }
}
```

--------------------------------

### Initialize Counter Entity (Java)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/typed/cluster-sharding.md

Java example for initializing a basic counter entity with cluster sharding. This shows the equivalent setup in Java.

```java
import akka.actor.typed.Behavior;
import akka.cluster.sharding.typed.javadsl.Entity;

Entity<CounterCommand, CounterEvent> counter = Entity.create(
    CounterCommand.class, // Message class
    ctx -> Behaviors.receive((ctx, msg) -> {
      // Counter behavior implementation
      if (msg instanceof CounterCommand.Increment) {
        System.out.println("Got increment: " + ((CounterCommand.Increment) msg).count);
      }
      return Behaviors.same();
    })).withTagger(CounterEvent.class, // Event class
        event -> {
          List<String> list = new ArrayList<>();
          if (event instanceof CounterEvent.Incremented) {
            list.add("incremented");
          }
          return list;
        });
```

--------------------------------

### Build Documentation Locally

Source: https://github.com/akka/akka-core/blob/main/RELEASING.md

Generate the documentation locally using the sbt task `paradoxBrowse`.

```shell
sbt akka-docs/paradoxBrowse
```

--------------------------------

### Simple lazyFlow Example

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/stream/operators/Flow/lazyFlow.md

Demonstrates the basic usage of lazyFlow. Note that the creation of the inner flow happens after the source starts producing elements.

```scala
import akka.stream.scaladsl.Flow
import akka.stream.scaladsl.Source

val flow = Flow.lazyFlow[Int, String, Unit](() => {
  println("Creating inner flow")
  Flow[Int].map(i => s"Element: $i")
})

val source = Source.tick(java.time.Duration.ofMillis(100), java.time.Duration.ofMillis(100), 1).take(5)

source.via(flow).runForeach(println)

// Output will show "Creating inner flow" after the first element is produced by the source.
```

--------------------------------

### Sink.headOption Java Example

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/stream/operators/Sink/headOption.md

Demonstrates the use of Sink.headOption with an empty source, resulting in an empty Optional value.

```java
CompletionStage<Optional<Integer>> result = Source.<Integer>empty().runWith(Sink.headOption(), materializer);

// result will be a CompletionStage completed with an empty Optional
```

--------------------------------

### Scala Auto-Pilot Example

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/testing.md

Install an AutoPilot in a TestKit to intercept and forward messages before they are enqueued for inspection. This is useful for verifying message flows in chains.

```scala
import akka.testkit.TestKitBase
import akka.testkit.TestKitBase.AutoPilot

class MyTestKit extends TestKitBase {
  // ...

  def myAutoPilot(): AutoPilot = (sender, msg) => {
    // ... process message ...
    // return the next autopilot
    // KeepRunning to retain the current one
    // NoAutoPilot to switch it off
    TestKitBase.KeepRunning
  }
}
```

--------------------------------

### Nesting setup, withTimers, and withStash in Scala

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/typed/style-guide.md

Demonstrates how to nest setup, withTimers, and withStash methods in Scala for managing actor dependencies and state.

```scala
Behaviors.setup[Command] {
  ctx =>
    val service = ctx.spawn(MyService())
    Behaviors.withTimers[Command] {
      timers =>
        Behaviors.withStash(10) {
          stash =>
            // Actor logic here, using ctx, service, timers, and stash
            Behaviors.receiveMessage[Command] { msg =>
              // ... handle message
              Behaviors.same
            }
        }
    }
}
```

--------------------------------

### Compile and Run Akka Java Example with Maven

Source: https://github.com/akka/akka-core/blob/main/samples/akka-quickstart-java/README.md

Use this command to compile and execute the Akka Java example project when using Maven for build management.

```bash
mvn compile exec:exec
```

--------------------------------

### Persistent Entity Usage (Scala)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/typed/cluster-sharding.md

Demonstrates how to initialize and use a persistent entity within a sharding setup in Scala. This involves creating the entity behavior and starting the sharding.

```scala
import akka.cluster.sharding.typed.scaladsl.{ClusterSharding, Entity, EntityTypeKey}
import akka.actor.typed.scaladsl.adapter.TypedActorSystemOps
import akka.actor.typed.ActorSystem

object HelloWorldShardingSetup {

  val TypeKey = EntityTypeKey[HelloWorldPersistentEntity.Command]("HelloWorld")

  def init(system: ActorSystem[_]): Unit = {
    ClusterSharding(system).init(
      Entity(
        TypeKey
      )(entityContext =>
        HelloWorldPersistentEntity(entityContext)
      )
    )
  }

  // Example of sending a message to the entity
  def sendMessage(system: ActorSystem[_]): Unit = {
    val sharding = ClusterSharding(system)
    val helloWorld = sharding.entityRefFor(TypeKey, "entity-1")

    // Using 'ask' pattern
    implicit val timeout: akka.util.Timeout = akka.util.Timeout.create(system.settings.config.getDuration("akka.cluster.sharding.retry-interval"))
    import akka.actor.typed.scaladsl.AskPattern._
    import system.executionContext

    helloWorld.ask(HelloWorldPersistentEntity.Greet("World", _)).foreach { reply =>
      println(s"Received reply: $reply")
    }

    // Using 'tell' pattern
    // helloWorld ! HelloWorldPersistentEntity.Greet("World", system.ignoreRef)
  }
}

```

--------------------------------

### Start Actor System (Scala)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/typed/guide/tutorial_2.md

Provides the main entry point for the Akka application in Scala. It creates the actor system and logs a startup message.

```scala
object IotApp {

  def main(args: Array[String]): Unit = {
    val guardian: ActorRef[GuardianCommand] =
      ActorSystem(IotSupervisor.apply(null), "MyActorSystem")

    guardian ! IotApp.Start
  }

  case object Start
}

```

--------------------------------

### Dead Cycle Example in Java

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/stream/stream-graphs.md

Illustrates a dead cycle in Java for Akka Streams, where processing halts because an initial element is missing. This requires injecting a starting element to proceed.

```java
RunnableGraph<NotUsed> graph = RunnableGraph.fromGraph(GraphDSL.create(builder -> {
    Source<Integer, NotUsed> source = Source.empty();
    Sink<Integer, NotUsed> cycle = Sink.ignore();

    // cycle up
    builder.from(source).to(cycle);
    // cycle back
    builder.from(cycle).to(source);

    return ClosedShape.completingShape();
}));
```

--------------------------------

### Basic PartitionHub Usage (Java)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/stream/stream-dynamic.md

Demonstrates the basic usage of PartitionHub in Java for routing elements. The producer is attached to the Sink, and consumers are attached to the resulting Source.

```java
Sink<String, Source<String, NotUsed>> partitionHubSink = PartitionHub.sink(Integer.class, (numConsumers, element) -> {
    // Example: route based on element's hash code
    return Math.abs(element.hashCode()) % numConsumers;
});

Source<String, NotUsed> producer = Source.tick(Duration.ofMillis(100), Duration.ofMillis(100), "msg");

RunnableGraph<Source<String, NotUsed>> materialized = RunnableGraph.fromGraph(GraphBuilder.create(builder -> {
    // The producer is attached to the partition hub sink
    producer.to(partitionHubSink).run(builder.getMaterializer());

    // Consumers can be attached to the source
    // For example, a consumer can be materialized like this:
    // Source<String, NotUsed> consumer1 = builder.add(new SourceShape<>(partitionHubSink.out()));
    // consumer1.mapAsync(1, System.out::println).runWith(Sink.ignore(), builder.getMaterializer());

    return partitionHubSink.out();
}));

Source<String, NotUsed> consumerSource = materialized.run(materializer);

// Materialize the source multiple times to attach consumers
consumerSource.mapAsync(1, System.out::println).runWith(Sink.ignore(), materializer);
consumerSource.mapAsync(1, System.out::println).runWith(Sink.ignore(), materializer);
```

--------------------------------

### Async Side Channel Example in Java

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/stream/stream-customize.md

Demonstrates an asynchronous side channel operator that starts dropping elements when a future completes. Use getAsyncCallback to acquire an AsyncCallback for external events.

```java
import akka.stream.stage.*;
import akka.stream.*;
import java.util.concurrent.CompletableFuture;

public class AsyncSideChannelStage<T> extends GraphStage<FlowShape<T, T>> {

    private final Inlet<T> in = new Inlet<>("AsyncSideChannelStage.in");
    private final Outlet<T> out = new Outlet<>("AsyncSideChannelStage.out");
    private final FlowShape<T, T> shape = new FlowShape<>(in, out);

    private final CompletableFuture<Void> futureToDropAfter;

    public AsyncSideChannelStage(CompletableFuture<Void> futureToDropAfter) {
        this.futureToDropAfter = futureToDropAfter;
    }

    @Override
    public FlowShape<T, T> shape() {
        return shape;
    }

    @Override
    public GraphStageLogic createLogicAndMaterializedValue(Attributes inheritedAttributes) {
        return new GraphStageLogic(shape) {
            private boolean dropped = false;
            private AsyncCallback<Void> asyncCallback;

            @Override
            public void preStart() {
                // It is recommended to use preStart() to acquire the AsyncCallback
                asyncCallback = getAsyncCallback(new Procedure<Void>() {
                    @Override
                    public void apply(Void aVoid) {
                        dropped = true;
                    }
                });

                // Trigger the callback when the future completes
                futureToDropAfter.thenRun(() -> asyncCallback.invoke(null));
            }

            { // Initialization block for handlers
                setHandler(in, new InHandler() {
                    @Override
                    public void onPush(T element) {
                        if (!dropped)
                            push(out, element);
                        else
                            // drop the element
                            pull(in);
                    }
                });

                setHandler(out, new OutHandler() {
                    @Override
                    public void onPull(Object element) {
                        pull(in);
                    }
                });
            }
        };
    }
}
```

--------------------------------

### Async Side Channel Example in Scala

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/stream/stream-customize.md

Demonstrates an asynchronous side channel operator that starts dropping elements when a future completes. Use getAsyncCallback to acquire an AsyncCallback for external events.

```scala
import akka.stream.stage.{GraphStage, GraphStageLogic, InHandler, OutHandler, AsyncCallback}
import akka.stream.{Attributes, FlowShape, Inlet, Outlet}
import scala.concurrent.Future

class AsyncSideChannelStage[T](futureToDropAfter: Future[Unit]) extends GraphStage[FlowShape[T, T]] {
  val in: Inlet[T] = Inlet("AsyncSideChannelStage.in")
  val out: Outlet[T] = Outlet("AsyncSideChannelStage.out")
  override val shape: FlowShape[T, T] = FlowShape(in, out)

  override def createLogicAndMaterializedValue(inheritedAttributes: Attributes): (GraphStageLogic, Unit) = {
    val logic = new GraphStageLogic(shape) {
      private var dropped = false
      private var asyncCallback: AsyncCallback[Unit] = _

      override def preStart(): Unit = {
        // It is recommended to use preStart() to acquire the AsyncCallback
        asyncCallback = getAsyncCallback[Unit](_ => dropped = true)

        // Trigger the callback when the future completes
        import scala.concurrent.ExecutionContext.Implicits.global
        futureToDropAfter.onComplete(result => asyncCallback.invoke(result.get))
      }

      setHandler(in, new InHandler {
        override def onPush(elem: T): Unit = {
          if (!dropped)
            push(out, elem)
          else
            // drop the element
            pull(in)
        }
      })

      setHandler(out, new OutHandler {
        override def onPull(element: Any): Unit = {
          pull(in)
        }
      })
    }
    (logic, ()) // Materialized value is Unit
  }
}
```

--------------------------------

### Spawn Actor with Setup - Java

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/typed/actor-lifecycle.md

Use Behaviors.setup to obtain the ActorContext for spawning child actors or accessing context.getSelf(). Ensure ActorContext methods are used only within the actor's message processing thread.

```java
Behaviors.setup(context -> {
  context.spawn(
      Greeter.create("World"),
      "greeter"
  );
  return Behaviors.receive(HelloWorld.Command.class)
      .onMessage(SayHello.class, command -> {
        command.getSender().tell("hello from the actor system");
        return Behaviors.same();
      })
      .build();
})
```

--------------------------------

### Create Cluster Singleton Manager

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/cluster-singleton.md

Start the ClusterSingletonManager on each node in the cluster, providing the Props of the singleton actor. This example limits the singleton to nodes tagged with the "worker" role.

```java
ClusterSingletonManager.props(
    Consumer.props(),
    // The same message that is used to stop the singleton actor
    // (in this case, TestSingletonMessages.end() message)
    TestSingletonMessages.end(),
    // The role of the nodes that the singleton actor can be started on
    "worker"
)
```

--------------------------------

### Build Documentation Locally

Source: https://github.com/akka/akka-core/blob/main/CONTRIBUTING.md

Use this command to build the documentation locally. The generated HTML can be found in akka-docs/target/paradox/site/main/index.html.

```shell
sbt
akka-docs/paradox
```

--------------------------------

### Create Cluster Singleton Manager

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/cluster-singleton.md

Start the ClusterSingletonManager on each node in the cluster, providing the Props of the singleton actor. This example limits the singleton to nodes tagged with the "worker" role.

```scala
ClusterSingletonManager.props(
  singletonProps = Props[Consumer],
  terminationMessage = End,
  role = Some("worker")
)
```

--------------------------------

### Collect 'b' elements and emit at end with statefulMapConcat (Scala)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/stream/operators/Source-or-Flow/statefulMapConcat.md

This Scala example collects elements starting with 'b' and emits them only when a special end element is encountered, using statefulMapConcat to manage the collection state.

```scala
val source = Source(List("apple", "banana", "apricot", "blueberry", "cherry", "end"))

val flow = Flow.statefulConcat[String, String] {
  var collectedB = List.empty[String]
  () => {
    element =>
      if (element == "end") {
        val result = collectedB
        collectedB = List.empty[String]
        result
      } else if (element.startsWith("b")) {
        collectedB = collectedB ::: List(element)
        List.empty[String]
      } else {
        List(element)
      }
  }
}

source.via(flow).runWith(Sink.seq).map(_.toList should be (List("apple", "apricot", "cherry", "banana", "blueberry")))

```

--------------------------------

### Java Example: Using Source.asSubscriber with a Database Client

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/stream/operators/Source/asSubscriber.md

This Java example demonstrates how to use Source.asSubscriber to create a Source from a Reactive Streams compatible database client. It shows how to query the database and process the rows, ensuring backpressure is applied.

```java
import akka.actor.ActorSystem;
import akka.stream.javadsl.Sink;
import akka.stream.javadsl.Source;
import java.util.concurrent.Flow;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.CompletableFuture;

public class AsSubscriber {

    public static void main(String[] args) throws Exception {
        ActorSystem system = ActorSystem.create("Sys");

        // Simulate a database client that returns a Publisher
        interface DatabaseClient {
            Flow.Publisher<String> queryRows();
        }

        // Simulate a database client implementation
        DatabaseClient dbClient = new DatabaseClient() {
            @Override
            public Flow.Publisher<String> queryRows() {
                // In a real application, this would query a database and return a Publisher
                // For demonstration, we return a simple sequence converted to a Publisher
                return Source.from(java.util.Arrays.asList("row1", "row2", "row3")).runWith(Sink.asPublisher(), system);
            }
        };

        // Create a Source from the database client's Publisher
        Source<String, Flow.Subscriber<String>> rowSource = Source.asSubscriber();

        // Materialize the Source to get the Subscriber and run the stream
        var result = rowSource.toMat(Sink.seq(), (subscriber, future) -> new Object[]{subscriber, future})
                .run(system);

        Flow.Subscriber<String> subscriber = (Flow.Subscriber<String>) result[0];
        CompletionStage<java.util.List<String>> done = (CompletionStage<java.util.List<String>>) result[1];

        // Attach the Subscriber to the database client's Publisher
        dbClient.queryRows().subscribe(subscriber);

        // Wait for the stream to complete and print the results
        done.whenComplete((value, throwable) -> {
            if (throwable != null) {
                System.err.println("Stream failed: " + throwable.getMessage());
            } else {
                System.out.println("Stream completed with: " + value);
            }
            system.terminate();
        });
    }
}
```

--------------------------------

### Run release preparation script

Source: https://github.com/akka/akka-core/blob/main/akka-docs/release-train-issue-template.md

Executes the script to update license information and version numbers across sample build files.

```bash
./scripts/release-prep.sh --commit-and-pr $VERSION$
```

--------------------------------

### Collect 'b' elements and emit at end with statefulMapConcat (Java)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/stream/operators/Source-or-Flow/statefulMapConcat.md

This Java example collects elements starting with 'b' and emits them upon encountering a special 'end' element. It utilizes statefulMapConcat to manage the state of collected elements.

```java
final Source<String, NotUsed> source = Source.from(Arrays.asList("apple", "banana", "apricot", "blueberry", "cherry", "end"));

final Flow<String, String, NotUsed> flow = Flow.statefulMapConcat(new Creator<Function<String, Iterable<String>>>() {
    @Override
    public Function<String, Iterable<String>> create() {
        return new Function<String, Iterable<String>>() {
            private List<String> collectedB = new ArrayList<>();

            @Override
            public Iterable<String> apply(String element) {
                if (element.equals("end")) {
                    List<String> result = new ArrayList<>(collectedB);
                    collectedB.clear();
                    return result;
                } else if (element.startsWith("b")) {
                    collectedB.add(element);
                    return Collections.emptyList();
                } else {
                    return Collections.singletonList(element);
                }
            }
        };
    }
});

source.via(flow).runWith(Sink.seq(), materializer).map(list -> {
    List<String> expected = Arrays.asList("apple", "apricot", "cherry", "banana", "blueberry");
    return list.equals(expected);
}).to(TestKit.runWith(system, duration));

```

--------------------------------

### Java TestKit Setup for Full Sample Test

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/testing.md

This Java snippet illustrates a comprehensive setup for testing Akka Classic actors using `TestKit`. It includes initializing the `ActorSystem`, obtaining the `testActor` reference, and performing message assertions.

```java
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.testkit.javadsl.TestKit;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import scala.concurrent.duration.FiniteDuration;

import java.util.concurrent.TimeUnit;

public class TestKitSampleTest {

  static ActorSystem system;

  @BeforeClass
  public static void setup() {
    system = ActorSystem.create("testSystem");
  }

  @AfterClass
  public static void teardown() {
    if (system != null) {
      system.terminate();
    }
  }

  @Test
  public void testIt() throws Exception {
    // Akka Classic Actor under test
    class SimpleActor extends akka.actor.Actor {
      public Receive createReceive() {
        return receiveBuilder()
            .matchString("ping", () -> {
              getSender().tell("pong", getSelf());
            })
            .build();
      }
    }

    final TestKit testKit = new TestKit(system);
    ActorRef simpleActor = system.actorOf(Props.create(SimpleActor.class));

    // "testActor" is a built-in test actor that you can use to send messages to
    // and receive replies from.
    // The "testActor" is implicitly used as the sender reference when you send
    // messages from the test procedure.
    simpleActor.tell("ping", testKit.getRef());
    testKit.expectMsg(new FiniteDuration(3, TimeUnit.SECONDS), "pong");
  }

}
```

--------------------------------

### Denylist functionality with statefulMapConcat (Scala)

Source: https://github.com/akka/akka-core/blob/main/akka-docs/src/main/paradox/stream/operators/Source-or-Flow/statefulMapConcat.md

This Scala example implements a denylist feature using statefulMapConcat. It adds elements starting with 'deny:word' to a private list and filters out subsequent occurrences of 'word' until the deny list is cleared.

```scala
val source = Source(List("apple", "banana", "deny:word", "orange", "word", "grape", "word", "kiwi"))

val flow = Flow.statefulConcat[String, String] {
  var denied = false
  () => {
    element =>
      if (element == "deny:word") {
        denied = true
        List.empty[String]
      } else if (denied && element == "word") {
        denied = false
        List.empty[String]
      } else {
        List(element)
      }
  }
}

source.via(flow).runWith(Sink.seq).map(_.toList should be (List("apple", "banana", "orange", "grape", "kiwi")))

```