Google FCM

The Apache Pekko Connectors Google Firebase Cloud Messaging connector provides a way to send notifications with Firebase Cloud Messaging.

Artifacts

sbt

val PekkoVersion = "1.0.3"
val PekkoHttpVersion = "1.0.1"
libraryDependencies ++= Seq(
  "org.apache.pekko" %% "pekko-connectors-google-fcm" % "1.0.2",
  "org.apache.pekko" %% "pekko-stream" % PekkoVersion,
  "org.apache.pekko" %% "pekko-http" % PekkoHttpVersion,
  "org.apache.pekko" %% "pekko-http-spray-json" % PekkoHttpVersion
)

Maven

<properties>
  <pekko.version>1.0.3</pekko.version>
  <pekko.http.version>1.0.1</pekko.http.version>
  <scala.binary.version>2.13</scala.binary.version>
</properties>
<dependencies>
  <dependency>
    <groupId>org.apache.pekko</groupId>
    <artifactId>pekko-connectors-google-fcm_${scala.binary.version}</artifactId>
    <version>1.0.2</version>
  </dependency>
  <dependency>
    <groupId>org.apache.pekko</groupId>
    <artifactId>pekko-stream_${scala.binary.version}</artifactId>
    <version>${pekko.version}</version>
  </dependency>
  <dependency>
    <groupId>org.apache.pekko</groupId>
    <artifactId>pekko-http_${scala.binary.version}</artifactId>
    <version>${pekko.http.version}</version>
  </dependency>
  <dependency>
    <groupId>org.apache.pekko</groupId>
    <artifactId>pekko-http-spray-json_${scala.binary.version}</artifactId>
    <version>${pekko.http.version}</version>
  </dependency>
</dependencies>

Gradle

def versions = [
  PekkoVersion: "1.0.3",
  PekkoHttpVersion: "1.0.1",
  ScalaBinary: "2.13"
]
dependencies {
  implementation "org.apache.pekko:pekko-connectors-google-fcm_${versions.ScalaBinary}:1.0.2"
  implementation "org.apache.pekko:pekko-stream_${versions.ScalaBinary}:${versions.PekkoVersion}"
  implementation "org.apache.pekko:pekko-http_${versions.ScalaBinary}:${versions.PekkoHttpVersion}"
  implementation "org.apache.pekko:pekko-http-spray-json_${versions.ScalaBinary}:${versions.PekkoHttpVersion}"
}

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

Settings

The FCM connector shares its basic configuration with all the Google connectors in Apache Pekko Connectors. Additional FCM-specific configuration settings can be found in its own reference.conf. You can send test notifications (so called validate only). And you can set the number of maximum concurrent connections. There is a limitation in the docs; from one IP you can have maximum 1k pending connections, and you may need to configure pekko.http.host-connection-pool.max-open-requests in your application.conf.

Sending notifications

To send a notification message create your notification object, and send it!

Scala

copysourceimport org.apache.pekko
import pekko.actor.ActorSystem
import pekko.stream.connectors.google.firebase.fcm.FcmSettings
import pekko.stream.connectors.google.firebase.fcm.v1.models._
import pekko.stream.connectors.google.firebase.fcm.v1.scaladsl.GoogleFcm

val result1: Future[immutable.Seq[FcmResponse]] =
  Source
    .single(notification)
    .via(GoogleFcm.send(fcmConfig))
    .map {
      case res @ FcmSuccessResponse(name) =>
        println(s"Successful $name")
        res
      case res @ FcmErrorResponse(errorMessage) =>
        println(s"Send error $errorMessage")
        res
    }
    .runWith(Sink.seq)

Java

copysourceimport org.apache.pekko.stream.connectors.google.firebase.fcm.FcmSettings;
import org.apache.pekko.stream.connectors.google.firebase.fcm.v1.models.*;
import org.apache.pekko.stream.connectors.google.firebase.fcm.v1.javadsl.GoogleFcm;

CompletionStage<List<FcmResponse>> result1 =
    Source.single(notification)
        .via(GoogleFcm.send(fcmConfig))
        .map(
            res -> {
              if (res.isSuccess()) {
                FcmSuccessResponse response = (FcmSuccessResponse) res;
                System.out.println("Successful " + response.getName());
              } else {
                FcmErrorResponse response = (FcmErrorResponse) res;
                System.out.println("Send error " + response.getRawError());
              }
              return res;
            })
        .runWith(Sink.seq(), system);

With this type of send you can get responses from the server. These responses can be FcmSuccessResponse or FcmErrorResponse. You can choose what you want to do with this information, but keep in mind if you try to resend the failed messages you will need to use exponential backoff! (see Apache Pekko docs RestartFlow.onFailuresWithBackoff)

If you don’t care if the notification was sent successfully, you may use fireAndForget.

Scala

copysourceimport org.apache.pekko
import pekko.actor.ActorSystem
import pekko.stream.connectors.google.firebase.fcm.FcmSettings
import pekko.stream.connectors.google.firebase.fcm.v1.models._
import pekko.stream.connectors.google.firebase.fcm.v1.scaladsl.GoogleFcm

val fcmConfig = FcmSettings()
val notification = FcmNotification("Test", "This is a test notification!", Token("token"))
Source
  .single(notification)
  .runWith(GoogleFcm.fireAndForget(fcmConfig))

Java

copysourceimport org.apache.pekko.stream.connectors.google.firebase.fcm.FcmSettings;
import org.apache.pekko.stream.connectors.google.firebase.fcm.v1.models.*;
import org.apache.pekko.stream.connectors.google.firebase.fcm.v1.javadsl.GoogleFcm;

FcmSettings fcmConfig = FcmSettings.create();
FcmNotification notification =
    FcmNotification.basic("Test", "This is a test notification!", new Token("token"));
Source.single(notification).runWith(GoogleFcm.fireAndForget(fcmConfig), system);

With fire and forget you will just send messages and ignore all the errors.

To help the integration and error handling or logging, there is a variation of the flow where you can send data beside your notification.

Scala

copysourceimport org.apache.pekko
import pekko.actor.ActorSystem
import pekko.stream.connectors.google.firebase.fcm.FcmSettings
import pekko.stream.connectors.google.firebase.fcm.v1.models._
import pekko.stream.connectors.google.firebase.fcm.v1.scaladsl.GoogleFcm

val result2: Future[immutable.Seq[(FcmResponse, String)]] =
  Source
    .single((notification, "superData"))
    .via(GoogleFcm.sendWithPassThrough(fcmConfig))
    .runWith(Sink.seq)

Java

copysourceimport org.apache.pekko.stream.connectors.google.firebase.fcm.FcmSettings;
import org.apache.pekko.stream.connectors.google.firebase.fcm.v1.models.*;
import org.apache.pekko.stream.connectors.google.firebase.fcm.v1.javadsl.GoogleFcm;

CompletionStage<List<Pair<FcmResponse, String>>> result2 =
    Source.single(Pair.create(notification, "superData"))
        .via(GoogleFcm.sendWithPassThrough(fcmConfig))
        .runWith(Sink.seq(), system);

Here I send a simple string, but you could use any type.

Scala only

You can build any notification described in the original documentation. It can be done by hand, or using some builder method. If you build your notification from scratch with options (and not with the provided builders), worth to check isSendable before sending.

Scala

copysourceval buildedNotification = FcmNotification.empty
  .withTarget(Topic("testers"))
  .withBasicNotification("title", "body")
  // .withAndroidConfig(AndroidConfig(...))
  // .withApnsConfig(ApnsConfig(...))
  .withWebPushConfig(
    WebPushConfig(
      headers = Option(Map.empty),
      data = Option(Map.empty),
      notification =
        Option("{\"title\": \"web-title\", \"body\": \"web-body\", \"icon\": \"http://example.com/icon.png\"}")))
val sendable = buildedNotification.isSendable

There is a condition builder too.

Scala

copysourceimport Condition.{ Topic => CTopic }
val condition = Condition(CTopic("TopicA") && (CTopic("TopicB") || (CTopic("TopicC") && !CTopic("TopicD"))))
val conditioneddNotification = FcmNotification("Test", "This is a test notification!", condition)