Apache Pekko

https://pekko.apache.org/docs/pekko/current/
Admin
Apache Pekko is an open-source toolkit for building highly concurrent, distributed, and resilient...
Tokens:666,372
Snippets:5,172
Trust Score:10
Update:3 months ago
Show doc for...
Tokens:
Raw
### Start Media Driver with Configuration File

Source: https://pekko.apache.org/docs/pekko/current/remoting-artery.html

Command to start the media driver using a specific properties file.

```bash
java io.aeron.driver.MediaDriver config/aeron.properties
```

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

### Aeron Properties File Example

Source: https://pekko.apache.org/docs/pekko/current/remoting-artery.html

Example configuration file for tuning Aeron media driver settings.

```properties
aeron.mtu.length=16384
aeron.socket.so_sndbuf=2097152
aeron.socket.so_rcvbuf=2097152
aeron.rcv.buffer.length=16384
aeron.rcv.initial.window.length=2097152
agrona.disable.bounds.checks=true

aeron.threading.mode=SHARED_NETWORK

# low latency settings
#aeron.threading.mode=DEDICATED
#aeron.sender.idle.strategy=org.agrona.concurrent.BusySpinIdleStrategy
#aeron.receiver.idle.strategy=org.agrona.concurrent.BusySpinIdleStrategy

# use same director in pekko.remote.artery.advanced.aeron-dir config
# of the Pekko application
aeron.dir=/dev/shm/aeron
```

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

### Start Aeron Media Driver

Source: https://pekko.apache.org/docs/pekko/current/remoting-artery.html

Command to start the standalone Aeron media driver process.

```bash
java io.aeron.driver.MediaDriver
```

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

### Create the ActorSystem Entry Point

Source: https://pekko.apache.org/docs/pekko/current/typed/guide/tutorial_2.html

Initializes the ActorSystem and starts the top-level supervisor actor.

```scala
copysourcepackage 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")
  }

}
```

```java
copysourcepackage 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");
  }
}
```

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

### Complete Stream Pipeline

Source: https://pekko.apache.org/docs/pekko/current/stream/stream-quickstart.html

Full example combining system initialization, transformation, and execution.

```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);
```

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

### Initialize ActorSystem

Source: https://pekko.apache.org/docs/pekko/current/stream/stream-quickstart.html

Boilerplate code to start an ActorSystem for running streams.

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

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

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

### Initialize the Actor System

Source: https://pekko.apache.org/docs/pekko/current/typed/event-stream.html

Starts the actor system using the defined listener behavior.

```java
ActorSystem<Command> system =
    ActorSystem.create(AllDeadLettersListenerBehavior.create(), "AllDeadLettersListener");
```

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

### Example Console Output

Source: https://pekko.apache.org/docs/pekko/current/typed/actors.html

Sample log output generated by the actor system interaction.

```text
[INFO] [03/13/2018 15:50:05.814] [hello-pekko.actor.default-dispatcher-4] [pekko://hello/user/greeter] Hello World!
[INFO] [03/13/2018 15:50:05.815] [hello-pekko.actor.default-dispatcher-4] [pekko://hello/user/greeter] Hello Pekko!
[INFO] [03/13/2018 15:50:05.815] [hello-pekko.actor.default-dispatcher-2] [pekko://hello/user/World] Greeting 1 for World
[INFO] [03/13/2018 15:50:05.815] [hello-pekko.actor.default-dispatcher-4] [pekko://hello/user/Pekko] Greeting 1 for Pekko
[INFO] [03/13/2018 15:50:05.815] [hello-pekko.actor.default-dispatcher-5] [pekko://hello/user/greeter] Hello World!
[INFO] [03/13/2018 15:50:05.815] [hello-pekko.actor.default-dispatcher-5] [pekko://hello/user/greeter] Hello Pekko!
[INFO] [03/13/2018 15:50:05.815] [hello-pekko.actor.default-dispatcher-4] [pekko://hello/user/World] Greeting 2 for World
[INFO] [03/13/2018 15:50:05.815] [hello-pekko.actor.default-dispatcher-5] [pekko://hello/user/greeter] Hello World!
[INFO] [03/13/2018 15:50:05.815] [hello-pekko.actor.default-dispatcher-4] [pekko://hello/user/Pekko] Greeting 2 for Pekko
[INFO] [03/13/2018 15:50:05.816] [hello-pekko.actor.default-dispatcher-5] [pekko://hello/user/greeter] Hello Pekko!
[INFO] [03/13/2018 15:50:05.816] [hello-pekko.actor.default-dispatcher-4] [pekko://hello/user/World] Greeting 3 for World
[INFO] [03/13/2018 15:50:05.816] [hello-pekko.actor.default-dispatcher-6] [pekko://hello/user/Pekko] Greeting 3 for Pekko
```

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

### Execute Multi-JVM Run Command

Source: https://pekko.apache.org/docs/pekko/current/multi-jvm-testing.html

Example output when running a multi-node application test.

```text
> multi-jvm:run sample.Sample
...
[info] * sample.Sample
[JVM-1] Hello from node 1
[JVM-2] Hello from node 2
[JVM-3] Hello from node 3
[success] Total time: ...
```

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

### Implementing a basic actor

Source: https://pekko.apache.org/docs/pekko/current/actors.html

Provides a complete example of an actor implementation with message handling.

```Scala
copysourceimport org.apache.pekko
import pekko.actor.Actor
import pekko.actor.Props
import pekko.event.Logging

class MyActor extends Actor {
  val log = Logging(context.system, this)

  def receive = {
    case "test" => log.info("received test")
    case _      => log.info("received unknown message")
  }
}
```

```Java
copysourceimport org.apache.pekko.actor.AbstractActor;
import org.apache.pekko.event.Logging;
import org.apache.pekko.event.LoggingAdapter;

public class MyActor extends AbstractActor {
  private final LoggingAdapter log = Logging.getLogger(getContext().getSystem(), this);

  @Override
  public Receive createReceive() {
    return receiveBuilder()
        .match(
            String.class,
            s -> {
              log.info("Received String message: {}", s);
            })
        .matchAny(o -> log.info("received unknown message"))
        .build();
  }
}
```

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

### Handle Get Responses

Source: https://pekko.apache.org/docs/pekko/current/distributed-data.html

Examples for processing GetSuccess, GetFailure, and NotFound responses.

```scala
case g @ GetSuccess(Counter1Key, req) =>
  val value = g.get(Counter1Key).value
case NotFound(Counter1Key, req) => // key counter1 does not exist
```

```java
b.match(
        GetSuccess.class,
        a -> a.key().equals(counter1Key),
        a -> {
          GetSuccess<PNCounter> g = a;
          BigInteger value = g.dataValue().getValue();
        })
    .match(
        NotFound.class,
        a -> a.key().equals(counter1Key),
        a -> {
          // key counter1 does not exist
        });
```

```scala
case g @ GetSuccess(Set1Key, req) =>
  val elements = g.get(Set1Key).elements
case GetFailure(Set1Key, req) =>
// read from 3 nodes failed within 1.second
case NotFound(Set1Key, req) => // key set1 does not exist
```

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

### Define Source.range integers

Source: https://pekko.apache.org/docs/pekko/current/stream/operators/Source/range.html

Examples of creating ranges with different start, end, and step values.

```java
copysourceimport org.apache.pekko.Done;
import org.apache.pekko.NotUsed;
import org.apache.pekko.actor.ActorSystem;
import org.apache.pekko.actor.testkit.typed.javadsl.ManualTime;
import org.apache.pekko.actor.testkit.typed.javadsl.TestKitJunitResource;
import org.apache.pekko.stream.javadsl.Source;

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

Source<Integer, NotUsed> sourceStepFive = Source.range(1, 100, 5);

Source<Integer, NotUsed> sourceStepNegative = Source.range(100, 1, -1);
```

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

### Implementing a Counter Actor with ReplicatorMessageAdapter

Source: https://pekko.apache.org/docs/pekko/current/typed/distributed-data.html

This example shows a complete actor implementation using ReplicatorMessageAdapter to manage a GCounter, including message handling for updates, gets, and subscriptions.

```java
    private static class InternalGetResponse implements InternalCommand {
      final Replicator.GetResponse<GCounter> rsp;
      final ActorRef<Integer> replyTo;

      InternalGetResponse(Replicator.GetResponse<GCounter> rsp, ActorRef<Integer> replyTo) {
        this.rsp = rsp;
        this.replyTo = replyTo;
      }
    }

    private static final class InternalSubscribeResponse implements InternalCommand {
      final Replicator.SubscribeResponse<GCounter> rsp;

      InternalSubscribeResponse(Replicator.SubscribeResponse<GCounter> rsp) {
        this.rsp = rsp;
      }
    }

    public static Behavior<Command> create(Key<GCounter> key) {
      return Behaviors.setup(
          ctx ->
              DistributedData.withReplicatorMessageAdapter(
                  (ReplicatorMessageAdapter<Command, GCounter> replicatorAdapter) ->
                      new Counter(ctx, replicatorAdapter, key)));
    }

    // adapter that turns the response messages from the replicator into our own protocol
    private final ReplicatorMessageAdapter<Command, GCounter> replicatorAdapter;
    private final SelfUniqueAddress node;
    private final Key<GCounter> key;

    private int cachedValue = 0;

    private Counter(
        ActorContext<Command> context,
        ReplicatorMessageAdapter<Command, GCounter> replicatorAdapter,
        Key<GCounter> key) {
      super(context);

      this.replicatorAdapter = replicatorAdapter;
      this.key = key;

      final SelfUniqueAddress node = DistributedData.get(context.getSystem()).selfUniqueAddress();

      this.node = DistributedData.get(context.getSystem()).selfUniqueAddress();

      this.replicatorAdapter.subscribe(this.key, InternalSubscribeResponse::new);
    }

    @Override
    public Receive<Command> createReceive() {
      return newReceiveBuilder()
          .onMessage(Increment.class, this::onIncrement)
          .onMessage(InternalUpdateResponse.class, msg -> Behaviors.same())
          .onMessage(GetValue.class, this::onGetValue)
          .onMessage(GetCachedValue.class, this::onGetCachedValue)
          .onMessage(Unsubscribe.class, this::onUnsubscribe)
          .onMessage(InternalGetResponse.class, this::onInternalGetResponse)
          .onMessage(InternalSubscribeResponse.class, this::onInternalSubscribeResponse)
          .build();
    }

    private Behavior<Command> onIncrement(Increment cmd) {
      replicatorAdapter.askUpdate(
          askReplyTo ->
              new Replicator.Update<>(
                  key,
                  GCounter.empty(),
                  Replicator.writeLocal(),
                  askReplyTo,
                  curr -> curr.increment(node, 1)),
          InternalUpdateResponse::new);

      return this;
    }

    private Behavior<Command> onGetValue(GetValue cmd) {
      replicatorAdapter.askGet(
          askReplyTo -> new Replicator.Get<>(key, Replicator.readLocal(), askReplyTo),
          rsp -> new InternalGetResponse(rsp, cmd.replyTo));

      return this;
    }

    private Behavior<Command> onGetCachedValue(GetCachedValue cmd) {
      cmd.replyTo.tell(cachedValue);
      return this;
    }

    private Behavior<Command> onUnsubscribe(Unsubscribe cmd) {
      replicatorAdapter.unsubscribe(key);
      return this;
    }

    private Behavior<Command> onInternalGetResponse(InternalGetResponse msg) {
      if (msg.rsp instanceof Replicator.GetSuccess) {
        int value = ((Replicator.GetSuccess<?>) msg.rsp).get(key).getValue().intValue();
        msg.replyTo.tell(value);
        return this;
      } else {
        // not dealing with failures
        return Behaviors.unhandled();
      }
    }

    private Behavior<Command> onInternalSubscribeResponse(InternalSubscribeResponse msg) {
      if (msg.rsp instanceof Replicator.Changed) {
        GCounter counter = ((Replicator.Changed<?>) msg.rsp).get(key);
        cachedValue = counter.getValue().intValue();
        return this;
      } else {
        // no deletes
        return Behaviors.unhandled();
      }
    }
  }
```

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

### Binary Compatibility Versioning Examples

Source: https://pekko.apache.org/docs/pekko/current/common/binary-compatibility-rules.html

Illustrates valid and invalid version upgrade paths for binary compatibility.

```text
OK:  1.4.0 --> 1.5.x
OK:  1.5.0 --> 1.6.x
NO:  1.x.y --x 2.x.y
OK:  2.0.0 --> 2.0.1 --> ... --> 2.0.n
OK:  2.0.n --> 2.1.0 --> ... --> 2.1.n
OK:  2.1.n --> 2.2.0 ...
     ...
```

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

### Implement an Image Converter Worker with ConsumerController

Source: https://pekko.apache.org/docs/pekko/current/typed/reliable-delivery.html

This example demonstrates a consumer actor that registers with a ServiceKey and processes tasks delivered via ConsumerController. It requires sending a Start message to the controller and confirming each processed message.

```scala
import org.apache.pekko
import pekko.actor.typed.scaladsl.Behaviors
import pekko.actor.typed.Behavior
import pekko.actor.typed.delivery.ConsumerController
import pekko.actor.typed.receptionist.ServiceKey

object ImageConverter {
  sealed trait Command
  final case class ConversionJob(resultId: UUID, fromFormat: String, toFormat: String, image: Array[Byte])
  private case class WrappedDelivery(d: ConsumerController.Delivery[ConversionJob]) extends Command

  val serviceKey = ServiceKey[ConsumerController.Command[ConversionJob]]("ImageConverter")

  def apply(): Behavior[Command] = {
    Behaviors.setup { context =>
      val deliveryAdapter =
        context.messageAdapter[ConsumerController.Delivery[ConversionJob]](WrappedDelivery(_))
      val consumerController =
        context.spawn(ConsumerController(serviceKey), "consumerController")
      consumerController ! ConsumerController.Start(deliveryAdapter)

      Behaviors.receiveMessage {
        case WrappedDelivery(delivery) =>
          val image = delivery.message.image
          val fromFormat = delivery.message.fromFormat
          val toFormat = delivery.message.toFormat
          // convert image...
          // store result with resultId key for later retrieval

          // and when completed confirm
          delivery.confirmTo ! ConsumerController.Confirmed

          Behaviors.same
      }

    }
  }

}
```

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

### Using Source.completionStage in Java

Source: https://pekko.apache.org/docs/pekko/current/stream/operators/Source/completionStage.html

Demonstrates creating a source from a completed CompletionStage and consuming it with a foreach sink.

```java
import java.util.concurrent.CompletionStage;
import java.util.concurrent.CompletableFuture;

import org.apache.pekko.NotUsed;
import org.apache.pekko.Done;
import org.apache.pekko.actor.typed.ActorSystem;
import org.apache.pekko.stream.javadsl.*;

CompletionStage<Integer> stage = CompletableFuture.completedFuture(10);

Source<Integer, NotUsed> source = Source.completionStage(stage);

Sink<Integer, CompletionStage<Done>> sink = Sink.foreach(i -> System.out.println(i.toString()));

source.runWith(sink, system); // 10
```

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

### Start ClusterSingletonManager

Source: https://pekko.apache.org/docs/pekko/current/cluster-singleton.html

Start the ClusterSingletonManager on each node to manage the singleton actor lifecycle.

```scala
system.actorOf(
  ClusterSingletonManager.props(
    singletonProps = Props(classOf[Consumer], queue, testActor),
    terminationMessage = End,
    settings = ClusterSingletonManagerSettings(system).withRole("worker")),
  name = "consumer")
```

```java
final ClusterSingletonManagerSettings settings =
    ClusterSingletonManagerSettings.create(system).withRole("worker");

system.actorOf(
    ClusterSingletonManager.props(
        Props.create(Consumer.class, () -> new Consumer(queue, testActor)),
        TestSingletonMessages.end(),
        settings),
    "consumer");
```

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

### Example A/AAAA Record Lookup

Source: https://pekko.apache.org/docs/pekko/current/discovery/index.html

Example output of a dig command for an A record lookup.

```bash
dig a-double.pekko.test

; <<>> DiG 9.11.3-RedHat-9.11.3-6.fc28 <<>> a-double.pekko.test
;; global options: +cmd
;; Got answer:
;; ->>HEADER<<- opcode: QUERY, status: NOERROR, id: 11983
;; flags: qr aa rd ra; QUERY: 1, ANSWER: 2, AUTHORITY: 1, ADDITIONAL: 2

;; OPT PSEUDOSECTION:
; EDNS: version: 0, flags:; udp: 4096
; COOKIE: 16e9815d9ca2514d2f3879265b28bad05ff7b4a82721edd0 (good)
;; QUESTION SECTION:
;a-double.pekko.test.            IN      A

;; ANSWER SECTION:
a-double.pekko.test.     86400   IN      A       192.168.1.21
a-double.pekko.test.     86400   IN      A       192.168.1.22
```

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

### Enable Eager Initialization of Persistence Plugins

Source: https://pekko.apache.org/docs/pekko/current/persistence-plugins.html

Configure plugins to start automatically by adding the Persistence extension and specifying the plugin IDs in the auto-start configuration keys.

```hocon
pekko {

  extensions = [org.apache.pekko.persistence.Persistence]

  persistence {

    journal {
      plugin = "pekko.persistence.journal.leveldb"
      auto-start-journals = ["org.apache.pekko.persistence.journal.leveldb"]
    }

    snapshot-store {
      plugin = "pekko.persistence.snapshot-store.local"
      auto-start-snapshot-stores = ["pekko.persistence.snapshot-store.local"]
    }

  }

}

```

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

### Start Cluster Sharding Region

Source: https://pekko.apache.org/docs/pekko/current/cluster-sharding.html

Register entity types with the ClusterSharding extension to start the ShardRegion.

```Scala
val counterRegion: ActorRef = ClusterSharding(system).start(
  typeName = "Counter",
  entityProps = Props[Counter](),
  settings = ClusterShardingSettings(system),
  extractEntityId = extractEntityId,
  extractShardId = extractShardId)
```

```Java
import org.apache.pekko.japi.Option;
import org.apache.pekko.cluster.sharding.ClusterSharding;
import org.apache.pekko.cluster.sharding.ClusterShardingSettings;

Option<String> roleOption = Option.none();
ClusterShardingSettings settings = ClusterShardingSettings.create(system);
ActorRef startedCounterRegion =
    ClusterSharding.get(system)
        .start("Counter", Props.create(Counter.class), settings, messageExtractor);
```

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

### Log Configuration on Startup

Source: https://pekko.apache.org/docs/pekko/current/logging.html

Logs the complete configuration at INFO level when the actor system starts. Avoid using this in production as it may expose sensitive values.

```hocon
pekko {
  # Log the complete configuration at INFO level when the actor system is started.
  # We do not recommend using this logging in production environments as it can include sensitive values.
  # This is useful when you are uncertain of what configuration is used.
  log-config-on-start = on
}
```

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

### Example SRV Record Lookup

Source: https://pekko.apache.org/docs/pekko/current/discovery/index.html

Example output of a dig command for an SRV record lookup.

```bash
dig srv _service._tcp.pekko.test

; <<>> DiG 9.11.3-RedHat-9.11.3-6.fc28 <<>> srv service.tcp.pekko.test
;; global options: +cmd
;; Got answer:
;; ->>HEADER<<- opcode: QUERY, status: NOERROR, id: 60023
;; flags: qr aa rd ra; QUERY: 1, ANSWER: 2, AUTHORITY: 1, ADDITIONAL: 5

;; OPT PSEUDOSECTION:
; EDNS: version: 0, flags:; udp: 4096
; COOKIE: 5ab8dd4622e632f6190f54de5b28bb8fb1b930a5333c3862 (good)
;; QUESTION SECTION:
;service.tcp.pekko.test.         IN      SRV

;; ANSWER SECTION:
_service._tcp.pekko.test.  86400   IN      SRV     10 60 5060 a-single.pekko.test.
_service._tcp.pekko.test.  86400   IN      SRV     10 40 5070 a-double.pekko.test.
```

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

### Create and use a Group Router

Source: https://pekko.apache.org/docs/pekko/current/typed/routers.html

Demonstrates initializing a ServiceKey, registering an actor with the Receptionist, and spawning a group router to handle messages.

```scala
val serviceKey = ServiceKey[Worker.Command]("log-worker")

Behaviors.setup[Unit] { ctx =>
  // this would likely happen elsewhere - if we create it locally we
  // can just as well use a pool
  val worker = ctx.spawn(Worker(), "worker")
  ctx.system.receptionist ! Receptionist.Register(serviceKey, worker)

  val group = Routers.group(serviceKey)
  val router = ctx.spawn(group, "worker-group")

  // the group router will stash messages until it sees the first listing of registered
  // services from the receptionist, so it is safe to send messages right away
  (0 to 10).foreach { n =>
    router ! Worker.DoLog(s"msg $n")
  }

  Behaviors.empty
}
```

```java
ServiceKey<Worker.Command> serviceKey = ServiceKey.create(Worker.Command.class, "log-worker");

Behaviors.setup(
    context -> {

      // this would likely happen elsewhere - if we create it locally we
      // can just as well use a pool
      ActorRef<Worker.Command> worker = context.spawn(Worker.create(), "worker");
      context.getSystem().receptionist().tell(Receptionist.register(serviceKey, worker));

      GroupRouter<Worker.Command> group = Routers.group(serviceKey);
      ActorRef<Worker.Command> router = context.spawn(group, "worker-group");

      // the group router will stash messages until it sees the first listing of
      // registered
      // services from the receptionist, so it is safe to send messages right away
      for (int i = 0; i < 10; i++) {
        router.tell(new Worker.DoLog("msg " + i));
      }

      return Behaviors.empty();
    });
```

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

### Using Source.queue in Java

Source: https://pekko.apache.org/docs/pekko/current/stream/actor-interop.html

Demonstrates creating a bounded queue source and offering elements to it within a Java stream pipeline.

```java
int bufferSize = 10;
int elementsToProcess = 5;

BoundedSourceQueue<Integer> sourceQueue =
    Source.<Integer>queue(bufferSize)
        .throttle(elementsToProcess, Duration.ofSeconds(3))
        .map(x -> x * x)
        .to(Sink.foreach(x -> System.out.println("got: " + x)))
        .run(system);

Source<Integer, NotUsed> source = Source.from(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));

source.map(x -> sourceQueue.offer(x)).runWith(Sink.ignore(), system);
```

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

### Define and run a simple stream in Java

Source: https://pekko.apache.org/docs/pekko/current/stream/stream-flows-and-basics.html

Demonstrates creating a source from a list and running it with a fold sink.

```java
final Source<Integer, NotUsed> source =
    Source.from(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
final Sink<Integer, CompletionStage<Integer>> sink = Sink.fold(0, Integer::sum);

// materialize the flow, getting the Sink's materialized value
final CompletionStage<Integer> sum = source.runWith(sink, system);
```

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

### Actor Path Anchor Examples

Source: https://pekko.apache.org/docs/pekko/current/general/addressing.html

Examples of actor path strings showing both local and remote addressing formats.

```text
"pekko://my-sys/user/service-a/worker1"               // purely local
"pekko://my-sys@host.example.com:5678/user/service-b" // remote
```

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

### Acquire and Manage a Lease

Source: https://pekko.apache.org/docs/pekko/current/coordination.html

Demonstrates how to initialize, acquire, check, and release a distributed lease.

```scala
val lease = LeaseProvider(system).getLease("<name of the lease>", "docs-lease", "owner")
val acquired: Future[Boolean] = lease.acquire()
val stillAcquired: Boolean = lease.checkLease()
val released: Future[Boolean] = lease.release()
```

```java
Lease lease =
    LeaseProvider.get(system).getLease("<name of the lease>", "jdocs-lease", "<owner name>");
CompletionStage<Boolean> acquired = lease.acquire();
boolean stillAcquired = lease.checkLease();
CompletionStage<Boolean> released = lease.release();
```

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

### Constructing Replicator Get Requests

Source: https://pekko.apache.org/docs/pekko/current/typed/distributed-data.html

Shows alternative ways to construct Get requests for retrieving data from the Replicator.

```Scala
// alternative way to define the `createRequest` function
// Replicator.Get instance has a curried `apply` method
replicatorAdapter.askGet(Replicator.Get(key, Replicator.ReadLocal), value => InternalGetResponse(value, replyTo))

// that is the same as
replicatorAdapter.askGet(
  askReplyTo => Replicator.Get(key, Replicator.ReadLocal, askReplyTo),
  value => InternalGetResponse(value, replyTo))
```

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

### Initialize multiple ActorSystems with lifted configuration subtrees

Source: https://pekko.apache.org/docs/pekko/current/general/configuration.html

Demonstrates how to isolate specific configuration subtrees for different ActorSystem instances using withFallback and withOnlyPath.

```scala
val config = ConfigFactory.load()
val app1 = ActorSystem(rootBehavior, "MyApp1", config.getConfig("myapp1").withFallback(config))
val app2 =
  ActorSystem(rootBehavior, "MyApp2", config.getConfig("myapp2").withOnlyPath("pekko").withFallback(config))
```

```java
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));
```

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

### Create a Source from a CompletionStage in Java

Source: https://pekko.apache.org/docs/pekko/current/stream/operators/Source/completionStageSource.html

Demonstrates wrapping a CompletionStage that produces a Source, useful for deferred stream initialization.

```java
import org.apache.pekko.NotUsed;
import org.apache.pekko.stream.javadsl.Source;

import java.util.concurrent.CompletionStage;

public class CompletionStageSource {

  public static void sourceCompletionStageSource() {
    UserRepository userRepository = null; // an abstraction over the remote service
    Source<User, CompletionStage<NotUsed>> userCompletionStageSource =
        Source.completionStageSource(userRepository.loadUsers());
    // ...
  }

  interface UserRepository {
    CompletionStage<Source<User, NotUsed>> loadUsers();
  }

  static class User {}
}
```

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

### Source.range(start, end, step)

Source: https://pekko.apache.org/docs/pekko/current/stream/operators/Source/range.html

Creates a Source that emits integers from start to end. The step parameter is optional and defaults to 1.

```APIDOC
## Source.range(start, end, step)

### Description
Emits each integer in a range, with an option to take bigger steps than 1. The stream completes when the end of the range has been reached.

### Signature
`Source<Integer, NotUsed> Source.range(int start, int end)`
`Source<Integer, NotUsed> Source.range(int start, int end, int step)`

### Parameters
- **start** (int) - The starting value of the range.
- **end** (int) - The ending value of the range.
- **step** (int) - Optional. The increment or decrement value. Defaults to 1.

### Examples
```java
// Basic range
Source<Integer, NotUsed> source = Source.range(1, 100);

// Range with step
Source<Integer, NotUsed> sourceStepFive = Source.range(1, 100, 5);

// Negative step
Source<Integer, NotUsed> sourceStepNegative = Source.range(100, 1, -1);
```
```

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

### intersperse(start, element, end)

Source: https://pekko.apache.org/docs/pekko/current/stream/operators/Source-or-Flow/intersperse.html

The intersperse operator injects a start element, a separator element between stream items, and an end element into the stream.

```APIDOC
## Method Signature

`Source.intersperse(start, element, end)`
`Flow.intersperse(start, element, end)`

### Description
Intersperse the stream with a provided element, similar to `List.mkString`. It can optionally inject start and end marker elements into the stream.

### Parameters
- **start** (Any) - Optional - The element to emit at the beginning of the stream.
- **element** (Any) - Required - The element to inject between each upstream element.
- **end** (Any) - Optional - The element to emit at the end of the stream.

### Reactive Streams Semantics
- **emits**: When upstream emits an element, or before with the start element if provided.
- **backpressures**: When downstream backpressures.
- **completes**: When upstream completes.
```

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

### Combine custom configuration with the standard stack

Source: https://pekko.apache.org/docs/pekko/current/general/configuration.html

Demonstrates how to create a custom configuration object and merge it with the default configuration stack before initializing an ActorSystem.

```scala
// 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);
```

```java
// 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);
```

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

### Define FSM Transitions

Source: https://pekko.apache.org/docs/pekko/current/fsm.html

Use onTransition to trigger logic when the FSM state changes. The Scala example uses pattern matching, while the Java example uses the matchState builder.

```scala
copysourceonTransition {
  case Active -> Idle =>
    stateData match {
      case Todo(ref, queue) => ref ! Batch(queue)
      case _                => // nothing to do
    }
}
```

```java
copysourceonTransition(
    matchState(
            Active,
            Idle,
            () -> {
              // reuse this matcher
              final UnitMatch<Data> m =
                  UnitMatch.create(
                      matchData(
                          Todo.class,
                          todo ->
                              todo.getTarget().tell(new Batch(todo.getQueue()), getSelf())));
              m.match(stateData());
            })
        .state(
            Idle,
            Active,
            () -> {
              /* Do something here */
            }));
```

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

### Using Source.queue in Scala

Source: https://pekko.apache.org/docs/pekko/current/stream/actor-interop.html

Demonstrates creating a queue source with throttling and handling offer results using Scala Futures.

```scala
val bufferSize = 10
val elementsToProcess = 5

val queue = Source
  .queue[Int](bufferSize)
  .throttle(elementsToProcess, 3.second)
  .map(x => x * x)
  .toMat(Sink.foreach(x => println(s"completed $x")))(Keep.left)
  .run()

val source = Source(1 to 10)

implicit val ec = system.dispatcher
source
  .map(x => {
    queue.offer(x).map {
      case QueueOfferResult.Enqueued    => println(s"enqueued $x")
      case QueueOfferResult.Dropped     => println(s"dropped $x")
      case QueueOfferResult.Failure(ex) => println(s"Offer failed ${ex.getMessage}")
      case QueueOfferResult.QueueClosed => println("Source Queue closed")
    }
  })
  .runWith(Sink.ignore)
```

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

### Implement EventSourcedProducerQueue

Source: https://pekko.apache.org/docs/pekko/current/typed/reliable-delivery.html

Example of integrating EventSourcedProducerQueue into a WorkPullingProducerController.

```scala
copysourceimport org.apache.pekko
import pekko.persistence.typed.delivery.EventSourcedProducerQueue
import pekko.persistence.typed.PersistenceId

val durableQueue =
  EventSourcedProducerQueue[ImageConverter.ConversionJob](PersistenceId.ofUniqueId("ImageWorkManager"))
val durableProducerController = context.spawn(
  WorkPullingProducerController(
    producerId = "workManager",
    workerServiceKey = ImageConverter.serviceKey,
    durableQueueBehavior = Some(durableQueue)),
  "producerController")
```

```java
copysourceimport org.apache.pekko.persistence.typed.PersistenceId;
import org.apache.pekko.persistence.typed.delivery.EventSourcedProducerQueue;

Behavior<DurableProducerQueue.Command<ImageConverter.ConversionJob>> durableQueue =
    EventSourcedProducerQueue.create(PersistenceId.ofUniqueId("ImageWorkManager"));
ActorRef<WorkPullingProducerController.Command<ImageConverter.ConversionJob>>
    durableProducerController =
        context.spawn(
            WorkPullingProducerController.create(
                ImageConverter.ConversionJob.class,
                "workManager",
                ImageConverter.serviceKey,
                Optional.of(durableQueue)),
            "producerController");
```

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

### Initialize Producer and Consumer Controllers

Source: https://pekko.apache.org/docs/pekko/current/typed/reliable-delivery.html

Setup code for spawning producer and consumer controllers and registering the consumer to the producer controller.

```scala
val consumerController = context.spawn(ConsumerController[FibonacciConsumer.Command](), "consumerController")
context.spawn(FibonacciConsumer(consumerController), "consumer")

val producerId = s"fibonacci-${UUID.randomUUID()}"
val producerController = context.spawn(
  ProducerController[FibonacciConsumer.Command](producerId, durableQueueBehavior = None),
  "producerController")
context.spawn(FibonacciProducer(producerController), "producer")

consumerController ! ConsumerController.RegisterToProducerController(producerController)
```

```java
ActorRef<ConsumerController.Command<FibonacciConsumer.Command>> consumerController =
    context.spawn(ConsumerController.create(), "consumerController");
context.spawn(FibonacciConsumer.create(consumerController), "consumer");

String producerId = "fibonacci-" + UUID.randomUUID();
ActorRef<ProducerController.Command<FibonacciConsumer.Command>> producerController =
    context.spawn(
        ProducerController.create(
            FibonacciConsumer.Command.class, producerId, Optional.empty()),
        "producerController");
context.spawn(FibonacciProducer.create(producerController), "producer");

consumerController.tell(
    new ConsumerController.RegisterToProducerController<>(producerController));
```

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

### Configure Logback for Development

Source: https://pekko.apache.org/docs/pekko/current/logging.html

Sets up console output for INFO level and file output for DEBUG level logging.

```xml
<?xml version="1.0" encoding="UTF-8"?>
<configuration>

    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
            <level>INFO</level>
        </filter>
        <encoder>
            <pattern>[%date{ISO8601}] [%level] [%logger] [%marker] [%thread] - %msg MDC: {%mdc}%n</pattern>
        </encoder>
    </appender>

    <appender name="FILE" class="ch.qos.logback.core.FileAppender">
        <file>target/myapp-dev.log</file>
        <encoder>
            <pattern>[%date{ISO8601}] [%level] [%logger] [%marker] [%thread] - %msg MDC: {%mdc}%n</pattern>
        </encoder>
    </appender>

    <root level="DEBUG">
        <appender-ref ref="STDOUT"/>
        <appender-ref ref="FILE"/>
    </root>
</configuration>
```

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

### Define Subscriber Interfaces

Source: https://pekko.apache.org/docs/pekko/current/stream/reactive-streams-interop.html

Example definitions for author subscribers.

```scala
def alert: Subscriber[Author]
def storage: Subscriber[Author]
```

```java
Subscriber<Author> alert();
Subscriber<Author> storage();
```

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

### Constructing and testing a BidiFlow protocol stack

Source: https://pekko.apache.org/docs/pekko/current/stream/stream-graphs.html

Demonstrates creating a protocol stack using BidiFlow and testing it by joining the flow with its inverse.

```java
/* construct protocol stack
 *         +------------------------------------+
 *         | stack                              |
 *         |                                    |
 *         |  +-------+            +---------+  |
 *    ~>   O~~o       |     ~>     |         o~~O    ~>
 * Message |  | codec | ByteString | framing |  | ByteString
 *    <~   O~~o       |     <~     |         o~~O    <~
 *         |  +-------+            +---------+  |
 *         +------------------------------------+
 */
final BidiFlow<Message, ByteString, ByteString, Message, NotUsed> stack = codec.atop(framing);

// test it by plugging it into its own inverse and closing the right end
final Flow<Message, Message, NotUsed> pingpong =
    Flow.of(Message.class)
        .collect(
            PFBuilder.<Message, Message>create()
                .match(Ping.class, p -> new Pong(p.id))
                .build());
final Flow<Message, Message, NotUsed> flow = stack.atop(stack.reversed()).join(pingpong);
final CompletionStage<List<Message>> result =
    Source.from(Arrays.asList(0, 1, 2))
        .<Message>map(id -> new Ping(id))
        .via(flow)
        .grouped(10)
        .runWith(Sink.<List<Message>>head(), system);
assertArrayEquals(
    new Message[] {new Pong(0), new Pong(1), new Pong(2)},
    result.toCompletableFuture().get(1, TimeUnit.SECONDS).toArray(new Message[0]));
```

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

### Start ClusterSingletonProxy

Source: https://pekko.apache.org/docs/pekko/current/cluster-routing.html

Initialization of the ClusterSingletonProxy to delegate jobs to the master actor.

```scala
system.actorOf(
  ClusterSingletonProxy.props(
    singletonManagerPath = "/user/statsService",
    settings = ClusterSingletonProxySettings(system).withRole("compute")),
  name = "statsServiceProxy")
```

```java
ClusterSingletonProxySettings proxySettings =
    ClusterSingletonProxySettings.create(system).withRole("compute");
system.actorOf(
    ClusterSingletonProxy.props("/user/statsService", proxySettings), "statsServiceProxy");
```

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

### Restarting a source with backoff in Java

Source: https://pekko.apache.org/docs/pekko/current/stream/operators/RestartSource/onFailuresWithBackoff.html

Demonstrates creating a source that fails and restarts using RestartSource.onFailuresWithBackoff with defined backoff settings.

```java
Source<Creator<Integer>, NotUsed> flakySource =
    Source.from(
        Arrays.<Creator<Integer>>asList(
            () -> 1,
            () -> 2,
            () -> 3,
            () -> {
              throw new RuntimeException("darn");
            }));
Source<Creator<Integer>, NotUsed> forever =
    RestartSource.onFailuresWithBackoff(
        RestartSettings.create(Duration.ofSeconds(1), Duration.ofSeconds(10), 0.1),
        () -> flakySource);
forever.runWith(
    Sink.foreach((Creator<Integer> nr) -> system.log().info("{}", nr.create())), system);
```

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

### Custom Dispatcher Configuration

Source: https://pekko.apache.org/docs/pekko/current/typed/dispatchers.html

Example configuration for a custom dispatcher in application.conf.

```hocon
your-dispatcher {
  type = Dispatcher
  executor = "thread-pool-executor"
  thread-pool-executor {
    fixed-pool-size = 32
  }
  throughput = 1
}
```

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

### Define Initial Event Class

Source: https://pekko.apache.org/docs/pekko/current/serialization-jackson.html

The starting schema for the event class before any modifications.

```Scala
case class ItemAdded(shoppingCartId: String, productId: String, quantity: Int) extends MySerializable
```

```Java
public class ItemAdded implements MySerializable {
  public final String shoppingCartId;
  public final String productId;
  public final int quantity;

  public ItemAdded(String shoppingCartId, String productId, int quantity) {
    this.shoppingCartId = shoppingCartId;
    this.productId = productId;
    this.quantity = quantity;
  }
}
```

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

### Nesting setup, withStash, and withTimers

Source: https://pekko.apache.org/docs/pekko/current/typed/style-guide.html

Demonstrates how to nest multiple behavior wrappers to access dependencies like context, stash, and timers simultaneously.

```scala
def apply(): Behavior[Command] =
  Behaviors.setup[Command](context =>
    Behaviors.withStash(100)(stash =>
      Behaviors.withTimers { timers =>
        context.log.debug("Starting up")

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

```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 ...
                      })));
}
```

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

### Define Mailbox Configuration

Source: https://pekko.apache.org/docs/pekko/current/typed/mailboxes.html

Example HOCON configuration for a custom mailbox type.

```hocon
# SPDX-License-Identifier: Apache-2.0

my-app {
  my-special-mailbox {
    mailbox-type = "org.apache.pekko.dispatch.SingleConsumerOnlyUnboundedMailbox"
  }
}
```
Tokens:
Raw