Apache Pekko

https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/
Admin
Apache Pekko is a modern runtime for building distributed, concurrent, and resilient systems using...
Tokens:944,599
Snippets:6,124
Trust Score:10
Update:9 months ago
Show doc for...
Tokens:
Raw
### Complete Pekko Stream Example (Scala & Java)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/stream/stream-quickstart

Combines ActorSystem initialization, stream definition, transformation, and execution into a single, complete example. This demonstrates the end-to-end flow of creating and running a simple Pekko Stream.

```scala
implicit val system: ActorSystem = ActorSystem("reactive-tweets")

val authors: Source[Author, NotUsed] =
  tweets.filter(_.hashtags.contains(pekkoTag)).map(_.author)

authors.runWith(Sink.foreach(println))
```

```java
final ActorSystem system = ActorSystem.create("reactive-tweets");

final Source<Author, NotUsed> authors =
    tweets.filter(t -> t.hashtags().contains(PEKKO)).map(t -> t.author);

authors.runWith(Sink.foreach(a -> System.out.println(a)), system);
```

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

### Run Source to Console (Scala)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/stream/stream-quickstart

Runs the integer source and prints each emitted number to the console.

```scala
source.runForeach(i => println(i))
```

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

### Initialize ActorSystem (Java)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/stream/stream-quickstart

Defines a Java class with a main method to create an ActorSystem for Pekko Streams.

```java
public class Main {
  public static void main(String[] argv) {
    final ActorSystem system = ActorSystem.create("QuickStart");
    // Code here
  }
}
```

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

### Wiring Stream Components (Java)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/stream/stream-flows-and-basics

Demonstrates different methods for wiring together Sources, Sinks, and Flows in Java. Examples include explicit wiring, starting from a Source, and starting from a Sink.

```java
copysource
// Explicitly creating and wiring up a Source, Sink and Flow
Source.from(Arrays.asList(1, 2, 3, 4))
    .via(Flow.of(Integer.class).map(elem -> elem * 2))
    .to(Sink.foreach(System.out::println));

// Starting from a Source
final Source<Integer, NotUsed> source =
    Source.from(Arrays.asList(1, 2, 3, 4)).map(elem -> elem * 2);
source.to(Sink.foreach(System.out::println));

// Starting from a Sink
final Sink<Integer, NotUsed> sink =
    Flow.of(Integer.class).map(elem -> elem * 2).to(Sink.foreach(System.out::println));
Source.from(Arrays.asList(1, 2, 3, 4)).to(sink);
```

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

### Test Persistence with PersistenceTestKit (Java)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/typed/persistence-testing

A Java example demonstrating the setup for testing a persistent actor using `PersistenceTestKit`. It initializes the `TestKitJunitResource` and `PersistenceTestKit` for use in JUnit tests.

```java
import org.apache.pekko.actor.testkit.typed.javadsl.TestKitJunitResource;
import org.apache.pekko.persistence.testkit.PersistenceTestKitPlugin;
import org.apache.pekko.persistence.testkit.javadsl.PersistenceTestKit;
import com.typesafe.config.ConfigFactory;
import org.junit.ClassRule;

// Assuming Command is defined elsewhere
// public class PersistenceTestKitSampleTest extends AbstractJavaTest {

//   @ClassRule
//   public static final TestKitJunitResource testKit = 
//       new TestKitJunitResource(
//           PersistenceTestKitPlugin.getInstance()
//               .config()
//               .withFallback(ConfigFactory.defaultApplication()));

//   PersistenceTestKit persistenceTestKit = PersistenceTestKit.create(testKit.system());

//   // Test methods would go here
// }
```

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

### Wiring Stream Components (Scala)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/stream/stream-flows-and-basics

Demonstrates different methods for wiring together Sources, Sinks, and Flows in Scala. Examples include explicit wiring, starting from a Source, starting from a Sink, and broadcasting to an inline sink.

```scala
copysource
// Explicitly creating and wiring up a Source, Sink and Flow
Source(1 to 6).via(Flow[Int].map(_ * 2)).to(Sink.foreach(println(_)))

// Starting from a Source
val source = Source(1 to 6).map(_ * 2)
source.to(Sink.foreach(println(_)))

// Starting from a Sink
val sink: Sink[Int, NotUsed] = Flow[Int].map(_ * 2).to(Sink.foreach(println(_)))
Source(1 to 6).to(sink)

// Broadcast to a sink inline
val otherSink:
  Sink[Int, NotUsed] = 
  Flow[Int].alsoTo(Sink.foreach(println(_))).to(Sink.ignore)
Source(1 to 6).to(otherSink)
```

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

### Import Pekko Stream DSLs (Scala)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/stream/stream-quickstart

Imports the necessary Pekko Stream Scala DSL components for building streams.

```scala
import org.apache.pekko
import pekko.stream._
import pekko.stream.scaladsl._
```

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

### Java Test Kit Setup for Actor Testing

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/fault-tolerance

Configures a test setup in Java using Pekko's TestKit for testing actor systems. It extends AbstractJavaTest and utilizes JUnit annotations for setup and cleanup. The example includes setting up an ActorSystem and defining a timeout duration for asynchronous operations, preparing for actor interactions within tests.

```java
import org.apache.pekko.testkit.TestProbe;
import org.apache.pekko.testkit.ErrorFilter;
import org.apache.pekko.testkit.TestEvent;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.pekko.japi.Util.immutableSeq;
import static org.junit.Assert.assertEquals;
import org.apache.pekko.actor.ActorSystem;
import org.apache.pekko.testkit.TestKit;
import java.time.Duration;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;

public class FaultHandlingTest extends AbstractJavaTest {
  static ActorSystem system;
  scala.concurrent.duration.Duration timeout =
      scala.concurrent.duration.Duration.create(5, SECONDS);

  @BeforeClass
  public static void start() {
    system = ActorSystem.create("FaultHandlingTest", config);
  }

  @AfterClass
  public static void cleanup() {
    TestKit.shutdownActorSystem(system);
    system = null;
  }

  @Test
  public void mustEmploySupervisorStrategy() throws Exception {
    // code here
  }
}
```

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

### Import Core Pekko and Concurrent Utilities (Scala)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/stream/stream-quickstart

Imports core Pekko types, actor system, future utilities, and I/O related classes for stream execution.

```scala
import pekko.{ Done, NotUsed }
import pekko.actor.ActorSystem
import pekko.util.ByteString
import scala.concurrent._
import scala.concurrent.duration._
import java.nio.file.Paths
```

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

### Create ActorSystem and Start Application - Scala

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/typed/guide/tutorial_2

Sets up and starts the Apache Pekko ActorSystem with the IotSupervisor as the root actor. This Scala code snippet demonstrates the entry point for the application, initializing the actor system named 'iot-system'.

```scala
package com.example

import org.apache.pekko.actor.typed.ActorSystem

object IotApp {

  def main(args: Array[String]): Unit = {
    // Create ActorSystem and top level supervisor
    ActorSystem[Nothing](IotSupervisor(), "iot-system")
  }

}
```

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

### Count Elements in a Stream (Scala)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/stream/stream-quickstart

This Scala code snippet demonstrates how to count the number of elements in a finite stream using `Sink.fold` and `Flow.map`. It transforms each element into '1' and then sums these values to get a total count.

```scala
val count: Flow[Tweet, Int, NotUsed] = Flow[Tweet].map(_ => 1)

val sumSink: Sink[Int, Future[Int]] = Sink.fold[Int, Int](0)(_ + _)

val counterGraph: RunnableGraph[Future[Int]] = 
  tweets.via(count).toMat(sumSink)(Keep.right)

val sum: Future[Int] = counterGraph.run()

sum.foreach(c => println(s"Total tweets processed: $c"))
```

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

### Define a Pekko Stream Source (Scala & Java)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/stream/stream-quickstart

Defines a data stream using Pekko's Source. A Source represents a stream of elements that can be processed. The example shows a Source of 'Tweet' objects, with 'NotUsed' indicating no special materialized value.

```scala
val tweets: Source[Tweet, NotUsed]
```

```java
Source<Tweet, NotUsed> tweets;
```

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

### Create ActorSystem and Start Application - Java

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/typed/guide/tutorial_2

Initializes and launches the Apache Pekko ActorSystem, with IotSupervisor as the top-level actor. This Java code snippet represents the application's main entry point, creating an actor system named 'iot-system'.

```java
package com.example;

import org.apache.pekko.actor.typed.ActorSystem;

public class IotMain {

  public static void main(String[] args) {
    // Create ActorSystem and top level supervisor
    ActorSystem.create(IotSupervisor.create(), "iot-system");
  }
}
```

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

### Load Custom Configuration in Scala

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/general/configuration

Demonstrates how to load a custom configuration in Scala by parsing a string, falling back to the regular configuration stack, and then creating an ActorSystem with the combined configuration. This approach allows fine-grained control over configuration merging.

```scala
copysource// make a Config with just your special setting
val myConfig = ConfigFactory.parseString("something=somethingElse");
// load the normal config stack (system props,
// then application.conf, then reference.conf)
val regularConfig = ConfigFactory.load();
// override regular stack with myConfig
val combined = myConfig.withFallback(regularConfig);
// put the result in between the overrides
// (system props) and defaults again
val complete = ConfigFactory.load(combined);
// create ActorSystem
val system = ActorSystem(rootBehavior, "myname", complete);
```

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

### Initialize SnapshotTestKit with Plugin Configuration

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/typed/persistence-testing

Shows how to set up the `SnapshotTestKit` by configuring the Actor System with `PersistenceTestKitSnapshotPlugin`. This allows for testing snapshotting mechanisms in Pekko Persistence.

```scala
import org.apache.pekko.actor.typed.ActorSystem
import org.apache.pekko.persistence.testkit.PersistenceTestKitSnapshotPlugin
import org.apache.pekko.persistence.testkit.scaladsl.SnapshotTestKit
import com.typesafe.config.ConfigFactory

// Assuming SomeBehavior is defined elsewhere
// val yourConfiguration = ConfigFactory.defaultApplication()

// val system = ActorSystem[??? /*some behavior*/]("test-system", PersistenceTestKitSnapshotPlugin.config.withFallback(yourConfiguration))

// val testKit = SnapshotTestKit(system)
```

```java
import org.apache.pekko.actor.typed.ActorSystem;
import org.apache.pekko.persistence.testkit.PersistenceTestKitSnapshotPlugin;
import org.apache.pekko.persistence.testkit.javadsl.SnapshotTestKit;
import com.typesafe.config.ConfigFactory;

// Assuming SomeBehavior is defined elsewhere
// Config conf = 
//     PersistenceTestKitSnapshotPlugin.getInstance()
//         .config()
//         .withFallback(ConfigFactory.defaultApplication());

// ActorSystem<Command> system = ActorSystem.create(new SomeBehavior(), "example", conf);

// SnapshotTestKit testKit = SnapshotTestKit.create(system);
```

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

### Example ByteBufferSerializer Implementation (Java)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/remoting-artery

Presents a Java implementation of `ByteBufferSerializer`, inheriting from `SerializerWithStringManifest`. This example illustrates the setup for handling both ByteBuffer and array-based serialization, ensuring compatibility with the `SerializerWithStringManifest` interface.

```java
copysourceimport org.apache.pekko.serialization.ByteBufferSerializer;
import org.apache.pekko.serialization.SerializerWithStringManifest;

class ExampleByteBufSerializer extends SerializerWithStringManifest
    implements ByteBufferSerializer {

  @Override
  public int identifier() {
    return 1337;
  }

  @Override
  public String manifest(Object o) {
    return "serialized-" + o.getClass().getSimpleName();
  }

  @Override
  public byte[] toBinary(Object o) {
    // in production code, acquire this from a BufferPool
    final ByteBuffer buf = ByteBuffer.allocate(256);

    toBinary(o, buf);
    buf.flip();
    final byte[] bytes = new byte[buf.remaining()];
    buf.get(bytes);
    return bytes;
  }

  @Override
  public Object fromBinary(byte[] bytes, String manifest) {
    return fromBinary(ByteBuffer.wrap(bytes), manifest);
  }

  @Override
  public void toBinary(Object o, ByteBuffer buf) {
    // Implement actual serialization here
  }

  @Override
  public Object fromBinary(ByteBuffer buf, String manifest) {
    // Implement actual deserialization here
    return null;
  }
}
```

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

### Load Custom Configuration in Java

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/general/configuration

Illustrates how to load custom configuration in Java by parsing a string, falling back to the regular configuration stack, and creating an ActorSystem with the merged configuration. This provides a programmatic way to manage configuration layers.

```java
copysource// make a Config with just your special setting
Config myConfig = ConfigFactory.parseString("something=somethingElse");
// load the normal config stack (system props,
// then application.conf, then reference.conf)
Config regularConfig = ConfigFactory.load();
// override regular stack with myConfig
Config combined = myConfig.withFallback(regularConfig);
// put the result in between the overrides
// (system props) and defaults again
Config complete = ConfigFactory.load(combined);
// create ActorSystem
ActorSystem system = ActorSystem.create(rootBehavior, "myname", complete);
```

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

### Nesting with Supervise in Java

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/typed/style-guide

This Java example demonstrates nesting Behaviors.setup with Behaviors.supervise and Behaviors.withStash. The 'supervise' behavior restarts its wrapped behavior on failure, but not the behavior encapsulating it. Each restart initializes a new stash, discarding the prior one.

```java
public static Behavior<Command> create() {
  return Behaviors.setup(
      context -> {
        // only run on initial actor start, not on crash-restart
        context.getLog().info("Starting");

        return Behaviors.<Command>supervise(
                Behaviors.withStash(
                    100,
                    stash -> {
                      // every time the actor crashes and restarts a new stash is created
                      // (previous stash is lost)
                      context.getLog().debug("Starting up with stash");
                      // Behaviors.receiveMessage { ... }
                    }))
            .onFailure(RuntimeException.class, SupervisorStrategy.restart());
      });
}
```

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

### Log Actor System Configuration (Java)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/general/configuration

Demonstrates how to log the complete configuration of an Actor System in Java at INFO level upon startup. This is controlled by the `pekko.log-config-on-start` property and is useful for debugging.

```java
import org.apache.pekko.actor.ActorSystem;

// Assuming rootBehavior is defined elsewhere
ActorSystem<Void> system = ActorSystem.create("MySystem");
system.logConfiguration();
```

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

### Scala Actor with Parameter Object for Setup

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/typed/style-guide

Refactors the Scala actor example by introducing a private 'Setup' case class to encapsulate parameters like 'name', 'context', and 'timers'. This reduces the number of arguments passed to the 'counter' method, improving readability.

```Scala
copysource  
// this is better than previous example, but even better solution exists
object Counter {
  sealed trait Command
  case object Increment extends Command
  final case class IncrementRepeatedly(interval: FiniteDuration) extends Command
  final case class GetValue(replyTo: ActorRef[Value]) extends Command
  final case class Value(n: Int)

  private case class Setup(name: String, context: ActorContext[Command], timers: TimerScheduler[Command])

  def apply(name: String): Behavior[Command] = 
    Behaviors.setup { context => 
      Behaviors.withTimers { timers => 
        counter(Setup(name, context, timers), 0)
      }
    }

```

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

### Initialize PersistenceTestKit with Plugin Configuration

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/typed/persistence-testing

Demonstrates how to initialize the `PersistenceTestKit` in an Actor System, ensuring the `PersistenceTestKitPlugin` configuration is included. This setup is crucial for the testkit to function correctly.

```scala
import org.apache.pekko.actor.typed.ActorSystem
import org.apache.pekko.persistence.testkit.PersistenceTestKitPlugin
import org.apache.pekko.persistence.testkit.scaladsl.PersistenceTestKit
import com.typesafe.config.ConfigFactory

// Assuming SomeBehavior is defined elsewhere
// val yourConfiguration = ConfigFactory.defaultApplication()

// val system = 
//   ActorSystem[??? /*some behavior*/]("test-system", PersistenceTestKitPlugin.config.withFallback(yourConfiguration))

// val testKit = PersistenceTestKit(system)
```

```java
import org.apache.pekko.actor.typed.ActorSystem;
import org.apache.pekko.persistence.testkit.PersistenceTestKitPlugin;
import org.apache.pekko.persistence.testkit.javadsl.PersistenceTestKit;
import com.typesafe.config.ConfigFactory;

// Assuming SomeBehavior is defined elsewhere
// Config conf = 
//     PersistenceTestKitPlugin.getInstance()
//         .config()
//         .withFallback(ConfigFactory.defaultApplication());

// ActorSystem<Command> system = ActorSystem.create(new SomeBehavior(), "example", conf);

// PersistenceTestKit testKit = PersistenceTestKit.create(system);
```

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

### Snapshot Storage Policy Example (Java)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/typed/persistence-testing

This Java snippet shows an example of implementing a `ProcessingPolicy` for snapshot storage. It defines a `SampleSnapshotStoragePolicy` that intercepts `WriteSnapshot` operations and returns `StorageFailure`, demonstrating how to control snapshot persistence.

```java
import org.apache.pekko.persistence.testkit.SnapshotStorage;
import org.apache.pekko.persistence.testkit.SnapshotStorage.SnapshotOperation;
import org.apache.pekko.persistence.testkit.EventStorage.JournalPolicies.ProcessingResult;
import org.apache.pekko.persistence.testkit.EventStorage.JournalPolicies.StorageFailure;
import org.apache.pekko.persistence.testkit.EventStorage.JournalPolicies.ProcessingSuccess;
import org.apache.pekko.persistence.testkit.SnapshotStorage.SnapshotPolicies.WriteSnapshot;

class SampleSnapshotStoragePolicy implements SnapshotStorage.ProcessingPolicy<SnapshotOperation> {
    @Override
    public ProcessingResult tryProcess(String processId, SnapshotOperation processingUnit) {
        if (processingUnit instanceof WriteSnapshot) {
            // Emulate an exception during snapshot writing
            return StorageFailure.create(); 
        } else {
            // Allow other snapshot operations like reading or deleting
            return ProcessingSuccess.getInstance();
        }
    }
}

```

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

### Create an Integer Source (Scala)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/stream/stream-quickstart

Creates a Pekko Stream Source that emits integers from 1 to 100.

```scala
val source: Source[Int, NotUsed] = Source(1 to 100)
```

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

### Start Chat Room and Gabbler Actors using Scala and Java

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/typed/actors

This snippet demonstrates how to initialize an ActorSystem and spawn both the ChatRoom and Gabbler actors. It shows how to manage actor lifecycles and initiate communication between them. The Scala version uses `Behaviors.setup` and `context.spawn`, while the Java version uses `Behaviors.setup` and `context.spawn`.

```scala
copysourceobject Main {
  def apply(): Behavior[NotUsed] = 
    Behaviors.setup { context =>
      val chatRoom = context.spawn(ChatRoom(), "chatroom")
      val gabblerRef = context.spawn(Gabbler(), "gabbler")
      context.watch(gabblerRef)
      chatRoom ! ChatRoom.GetSession("ol’ Gabbler", gabblerRef)

      Behaviors.receiveSignal {
        case (_, Terminated(_)) =>
          Behaviors.stopped
      }
    }

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

}

```

```java
copysourcepublic class Main {
  public static Behavior<Void> create() {
    return Behaviors.setup(
        context -> {
          ActorRef<ChatRoom.RoomCommand> chatRoom = context.spawn(ChatRoom.create(), "chatRoom");
          ActorRef<ChatRoom.SessionEvent> gabbler = context.spawn(Gabbler.create(), "gabbler");
          context.watch(gabbler);
          chatRoom.tell(new ChatRoom.GetSession("ol’ Gabbler", gabbler));

          return Behaviors.receive(Void.class)
              .onSignal(Terminated.class, sig -> Behaviors.stopped())
              .build();
        });
  }

  public static void main(String[] args) {
    ActorSystem.create(Main.create(), "ChatRoomDemo");
  }
}

```

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

### Sample Lease Implementation in Java

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/coordination

A sample Java implementation of the `Lease` class. It shows how to override `acquire`, `release`, and `checkLease` using `CompletableFuture` for asynchronous operations. This provides a Java-based foundation for custom lease functionalities.

```Java
copysourcestatic class SampleLease extends Lease {

  private LeaseSettings settings;

  public SampleLease(LeaseSettings settings) {
    this.settings = settings;
  }

  @Override
  public LeaseSettings getSettings() {
    return settings;
  }

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

  @Override
  public CompletionStage<Boolean> acquire(Consumer<Optional<Throwable>> leaseLostCallback) {
    return CompletableFuture.completedFuture(true);
  }

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

  @Override
  public boolean checkLease() {
    return true;
  }
}
```

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

### Start ActorSystem and Send Messages

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/typed/actors

Creates and starts the Pekko ActorSystem using the Java API, then dispatches 'SayHello' messages to begin the actor interaction flow. This code demonstrates the entry point for running the actor application.

```java
final ActorSystem<HelloWorldMain.SayHello> system = 
    ActorSystem.create(HelloWorldMain.create(), "hello");

system.tell(new HelloWorldMain.SayHello("World"));
system.tell(new HelloWorldMain.SayHello("Pekko"));

```

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

### Start TodoService Initialization (Scala)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/typed/reliable-delivery

Handles the initial setup phase of the TodoService. It creates a message adapter for ShardingProducerController.RequestNext messages and starts the producer controller. It then transitions to the active state or handles immediate updates.

```scala
private def start(producerController: ActorRef[ShardingProducerController.Start[TodoList.Command]]): Behavior[Command] = {
    val requestNextAdapter: ActorRef[ShardingProducerController.RequestNext[TodoList.Command]] = context.messageAdapter(WrappedRequestNext.apply)
    producerController ! ShardingProducerController.Start(requestNextAdapter)

    Behaviors.receiveMessagePartial {
      case WrappedRequestNext(next) =>
        active(next)
      case UpdateTodo(_, _, _, replyTo) =>
        // not hooked up with shardingProducerController yet
        replyTo ! Rejected
        Behaviors.same
    }
  }
```

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

### Snapshot Store Plugin Configuration Example

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/persistence-journals

Demonstrates the minimal configuration required to activate a custom snapshot store plugin in Apache Pekko. It specifies the plugin path and the class name of the custom implementation, along with optional settings like the plugin dispatcher.

```hocon
# Path to the snapshot store plugin to be used
pekko.persistence.snapshot-store.plugin = "my-snapshot-store"

# My custom snapshot store plugin
my-snapshot-store {
  # Class name of the plugin.
  class = "docs.persistence.MySnapshotStore"
  # Dispatcher for the plugin actor.
  plugin-dispatcher = "pekko.actor.default-dispatcher"
}
```

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

### Install AutoPilot on TestProbe (Java)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/testing

Provides an example of installing an AutoPilot on a TestProbe in Java. The `run` method handles incoming messages, deciding whether to forward them or stop the auto-pilot, and returns the appropriate `AutoPilot` instance.

```java
new TestKit(system) {
  {
    final TestKit probe = new TestKit(system);
    // install auto-pilot
    probe.setAutoPilot(
        new TestActor.AutoPilot() {
          public AutoPilot run(ActorRef sender, Object msg) {
            sender.tell(msg, ActorRef.noSender());
            return noAutoPilot();
          }
        });
    // first one is replied to directly ...
    probe.getRef().tell("hello", getRef());
    expectMsgEquals("hello");
    // ... but then the auto-pilot switched itself off
    probe.getRef().tell("world", getRef());
    expectNoMessage();
  }
};
```

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

### Actor Context Import Example in Scala

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/actors

Demonstrates how to import members from the actor's context to avoid prefixing access with 'context.'. This example shows creating a child actor named 'myactor' and forwarding received messages to it.

```scala
class ContextActor extends Actor {
  import context._
  val myActor = actorOf(Props[MyActor](), name = "myactor")
  def receive: Receive = {
    case x => myActor ! x
  }
}
```

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

### Create HelloWorldMain Actor

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/typed/actors

Sets up the HelloWorldMain actor, which spawns the Greeter and HelloWorldBot actors and initiates the greeting interaction. It listens for SayHello messages to start the process.

```scala
object HelloWorldMain {

  final case class SayHello(name: String)

  def apply(): Behavior[SayHello] = 
    Behaviors.setup { context =>
      val greeter = context.spawn(HelloWorld(), "greeter")

      Behaviors.receiveMessage { message =>
        val replyTo = context.spawn(HelloWorldBot(max = 3), message.name)
        greeter ! HelloWorld.Greet(message.name, replyTo)
        Behaviors.same
      }
    }

}

```

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

### Initialize ActorSystem (Scala)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/stream/stream-quickstart

Defines a Scala object with an 'App' trait to initialize an implicit ActorSystem for Pekko Streams.

```scala
object Main extends App {
  implicit val system: ActorSystem = ActorSystem("QuickStart")
  // Code here
}
```

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

### Handle Stream Completion and Termination (Scala)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/stream/stream-quickstart

Runs the stream, captures the completion signal, and terminates the ActorSystem upon completion.

```scala
val done: Future[Done] = source.runForeach(i => println(i))

implicit val ec = system.dispatcher
done.onComplete(_ => system.terminate())
```

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

### Nesting Behaviors (setup, withStash, withTimers) in Java

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/typed/style-guide

This Java code snippet demonstrates nesting Behaviors.setup, Behaviors.withStash, and Behaviors.withTimers. It enables access to the actor context, stash, and timers within the combined behavior. The order of nesting is flexible, though 'setup' is commonly used as the outermost behavior.

```java
public static Behavior<Command> apply() {
  return Behaviors.setup(
      context ->
          Behaviors.withStash(
              100,
              stash ->
                  Behaviors.withTimers(
                      timers -> {
                        context.getLog().debug("Starting up");

                        // behavior using context, stash and timers ...
                      })));
}
```

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

### Create an Integer Source (Java)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/stream/stream-quickstart

Creates a Pekko Stream Source that emits integers from 1 to 100 using Java.

```java
final Source<Integer, NotUsed> source = Source.range(1, 100);
```

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

### Lift Actor System Configuration Subtree (Java)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/general/configuration

Demonstrates creating an Actor System using a specific configuration subtree in Java. The `withFallback` method is used to merge the subtree with the global configuration.

```java
import org.apache.pekko.actor.typed.ActorSystem;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;

Config config = ConfigFactory.load();
ActorSystem<Void> app1 =
    ActorSystem.create(rootBehavior, "MyApp1", config.getConfig("myapp1").withFallback(config));
ActorSystem<Void> app2 =
    ActorSystem.create(
        rootBehavior,
        "MyApp2",
        config.getConfig("myapp2").withOnlyPath("pekko").withFallback(config));
```

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

### Initialize Default Persistence Plugins (Java)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/typed/persistence-testing

This Java code snippet shows how to initialize default persistence plugins using `PersistenceInit`. It utilizes `Duration` for the timeout and returns a `CompletionStage<Done>`. The example then blocks until the initialization is complete using `toCompletableFuture().get()`.

```java
import org.apache.pekko.persistence.testkit.javadsl.PersistenceInit;
import org.apache.pekko.Done;

import java.time.Duration;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;

Duration timeout = Duration.ofSeconds(5);
CompletionStage<Done> done =
    PersistenceInit.initializeDefaultPlugins(testKit.system(), timeout);
done.toCompletableFuture().get(timeout.getSeconds(), TimeUnit.SECONDS);
```

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

### Import Pekko Stream DSLs (Java)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/stream/stream-quickstart

Imports the necessary Pekko Stream Java DSL components for building streams.

```java
import org.apache.pekko.stream.*;
import org.apache.pekko.stream.javadsl.*;
```

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

### Log Actor System Configuration (Scala)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/general/configuration

Shows how to log the complete configuration of an Actor System in Scala at INFO level upon startup. This is controlled by the `pekko.log-config-on-start` property and is useful for debugging.

```scala
import org.apache.pekko.actor.ActorSystem

// Assuming rootBehavior is defined elsewhere
val system = ActorSystem("MySystem")
system.logConfiguration()
```

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

### Start Destination Actors on Multiple Nodes (Java)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/distributed-pub-sub

Starts 'Destination' actors on multiple nodes within the cluster using Java. This allows for a distributed setup where a single path can receive messages on different cluster members.

```java
copysourcesystem.actorOf(Props.create(Destination.class), "destination");
// another node
system.actorOf(Props.create(Destination.class), "destination");
```

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

### Initialize Default Persistence Plugins (Scala)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/typed/persistence-testing

This Scala code snippet demonstrates how to initialize the default persistence plugins using `PersistenceInit`. It takes an `ActorSystem` and a timeout duration as input and returns a `Future[Done]`. The `Await.result` call ensures the initialization is complete before proceeding.

```scala
import org.apache.pekko.persistence.testkit.scaladsl.PersistenceInit

import scala.concurrent.Await
import scala.concurrent.Future
import scala.concurrent.duration._

val timeout = 5.seconds
val done: Future[Done] = PersistenceInit.initializeDefaultPlugins(system, timeout)
Await.result(done, timeout)
```

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

### Start ActorSystem and Send Messages

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/typed/actors

Initiates the Pekko ActorSystem with the HelloWorldMain behavior and sends initial 'SayHello' messages to trigger the actor interactions. This sets up the environment for the actors to communicate.

```scala
val system: ActorSystem[HelloWorldMain.SayHello] =
  ActorSystem(HelloWorldMain(), "hello")

system ! HelloWorldMain.SayHello("World")
system ! HelloWorldMain.SayHello("Pekko")

```

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

### Transform Source and Write to File (Scala)

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/stream/stream-quickstart

Transforms an integer source to calculate factorials, formats them as ByteString, and writes them to a file named 'factorials.txt'.

```scala
val factorials = source.scan(BigInt(1))((acc, next) => acc * next)

val result: Future[IOResult] =
  factorials.map(num => ByteString(s"$num\n")).runWith(FileIO.toPath(Paths.get("factorials.txt")))
```

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

### Constructing Protocol Stack in Java

Source: https://nightlies.apache.org/pekko/docs/pekko/main-snapshot/docs/stream/stream-graphs

Demonstrates how to build a complete protocol stack by composing a codec with the framing BidiFlow. This example shows the 'atop' operation for stacking flows.

```Java
final BidiFlow<Message, ByteString, ByteString, Message, NotUsed> stack = codec.atop(framing);
```
Tokens:
Raw