AWS EventBridge

The Apache Pekko Connectors AWS EventBridge connector provides Apache Pekko Stream flows and sinks to publish to AWS EventBridge event buses.

Artifacts

sbt

libraryDependencies += "org.apache.pekko" %% "pekko-connectors-aws-event-bridge" % "1.0.2"

Maven

<properties>
  <scala.binary.version>2.13</scala.binary.version>
</properties>
<dependencies>
  <dependency>
    <groupId>org.apache.pekko</groupId>
    <artifactId>pekko-connectors-aws-event-bridge_${scala.binary.version}</artifactId>
    <version>1.0.2</version>
  </dependency>
</dependencies>

Gradle

def versions = [
  ScalaBinary: "2.13"
]
dependencies {
  implementation "org.apache.pekko:pekko-connectors-aws-event-bridge_${versions.ScalaBinary}:1.0.2"
}

The table below shows direct dependencies of this module and the second tab shows all libraries it depends on transitively.

Setup

Prepare an ActorSystem.

Scala

copysourceimplicit val system: ActorSystem = ActorSystem()

Java

copysourceimport org.apache.pekko.actor.ActorSystem;
ActorSystem system = ActorSystem.create();

This connector requires an EventBridge instance to communicate with AWS EventBridge.

It is your code’s responsibility to call close to free any resources held by the client. In this example it will be called when the actor system is terminated.

Scala

copysourceimport java.net.URI

import com.github.pjfanning.pekkohttpspi.PekkoHttpClient
import software.amazon.awssdk.auth.credentials.{ AwsBasicCredentials, StaticCredentialsProvider }
import software.amazon.awssdk.regions.Region
import software.amazon.awssdk.services.eventbridge.EventBridgeAsyncClient

implicit val awsEventBridgeClient: EventBridgeAsyncClient =
  EventBridgeAsyncClient
    .builder()
    .credentialsProvider(StaticCredentialsProvider.create(AwsBasicCredentials.create("x", "x")))
    .endpointOverride(URI.create(endEndpoint))
    .region(Region.EU_CENTRAL_1)
    .httpClient(PekkoHttpClient.builder().withActorSystem(system).build())
    .build()

system.registerOnTermination(awsEventBridgeClient.close())

Java

copysourceimport java.net.URI;
import com.github.pjfanning.pekkohttpspi.PekkoHttpClient;
import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.eventbridge.EventBridgeAsyncClient;

final EventBridgeAsyncClient awsClient =
    EventBridgeAsyncClient.builder()
        .credentialsProvider(
            StaticCredentialsProvider.create(AwsBasicCredentials.create("x", "x")))
        .endpointOverride(URI.create(endpoint))
        .region(Region.EU_CENTRAL_1)
        .httpClient(PekkoHttpClient.builder().withActorSystem(system).build())
        .build();

system.registerOnTermination(() -> awsClient.close());

The example above uses Apache Pekko HTTP as the default HTTP client implementation. For more details about the HTTP client, configuring request retrying and best practices for credentials, see AWS client configuration.

Publish messages to AWS EventBridge Event Bus

Create a PutEventsEntry-accepting sink, publishing to an event bus.

Scala

copysourceSource
  .single(PutEventsRequestEntry.builder().detail("string").build())
  .runWith(EventBridgePublisher.sink())

Java

copysourceSource.single(detailEntry("message"))
    .runWith(EventBridgePublisher.sink(eventBridgeClient), system);

Create a sink that accepts PutEventsRequestEntries to be published to an Event Bus.

Scala

copysourceSource
  .single(PutEventsRequest.builder().entries(PutEventsRequestEntry.builder().detail("string").build()).build())
  .runWith(EventBridgePublisher.publishSink())

Java

copysourceSource.single(detailPutEventsRequest("message"))
    .runWith(EventBridgePublisher.publishSink(eventBridgeClient), system);

You can also build flow stages which publish messages to Event Bus and then forward PutEventsResponse further down the stream.

Flow for PutEventEntry.

Scala

copysourceSource
  .single(PutEventsRequestEntry.builder().detail("string").build())
  .via(EventBridgePublisher.flow())
  .runWith(Sink.foreach(res => println(res)))

Java

copysourceSource.single(detailEntry("message"))
    .via(EventBridgePublisher.flow(eventBridgeClient))
    .runWith(Sink.foreach(res -> System.out.println(res)), system);

Flow for PutEventsRequest.

Scala

copysourceSource
  .single(PutEventsRequest.builder().entries(PutEventsRequestEntry.builder().detail("string").build()).build())
  .via(EventBridgePublisher.publishFlow())
  .runWith(Sink.foreach(res => println(res)))

Java

copysourceSource.single(detailPutEventsRequest("message"))
    .via(EventBridgePublisher.publishFlow(eventBridgeClient))
    .runWith(Sink.foreach(res -> System.out.println(res)), system);

Flow supporting a list of PutEventEntry objects.

Messages published in a batch using EventBridgePublisher.flowSeqEventBridgePublisher.flowSeq are not published in an “all or nothing” manner. Event Bridge will process each event independently. Retries of the failed messages in the PutEventsResponse are not yet implemented.

Integration testing

For integration testing without connecting directly to Amazon EventBridge, Apache Pekko Connectors uses Localstack, which comes as a docker image - and has a corresponding service amazoneventbridge in the docker-compose.yml file. Which needs to be started before running the integration tests docker compose up amazoneventbridge.