HUAWEI Push Kit

.github/autolabeler.yml

@@ -26,6 +26,7 @@ dependency-change: "/project/Dependencies.scala"
 'p:google-fcm': ["/google-fcm"]
 'p:hbase': ["/hbase"]
 'p:hdfs': ["/hdfs"]
+'p:hms-push-kit': ["/hms-push-kit"]
 'p:influxdb': ["/influxdb"]
 'p:ironmq': ["/ironmq"]
 'p:jms': ["/jms"]

build.sbt

@@ -25,6 +25,7 @@ lazy val alpakka = project
     googleFcm,
     hbase,
     hdfs,
+    huaweiPushKit,
     influxdb,
     ironmq,
     jms,
@@ -235,6 +236,10 @@ lazy val hbase = alpakkaProject("hbase", "hbase", Dependencies.HBase, Test / for
 
 lazy val hdfs = alpakkaProject("hdfs", "hdfs", Dependencies.Hdfs, fatalWarnings := true)
 
+lazy val huaweiPushKit =
+  alpakkaProject("huawei-push-kit", "huawei.pushkit", Dependencies.HuaweiPushKit, fatalWarnings := true)
+    .disablePlugins(MimaPlugin)
+
 lazy val influxdb = alpakkaProject("influxdb", "influxdb", Dependencies.InfluxDB, fatalWarnings := false)
 
 lazy val ironmq = alpakkaProject(

docs/src/main/paradox/huawei-push-kit.md

@@ -0,0 +1,86 @@
+# HUAWEI Push Kit
+
+@@@ note { title="HUAWEI Push Kit" }
+
+HUAWEI Push Kit is a messaging service provided for you. It establishes a messaging channel from the cloud to devices. By integrating Push Kit, you can send messages to your apps on users' devices in real time.
+
+@@@
+
+The Alpakka HUAWEI Push Kit connector provides a way to send notifications with [HUAWEI Push Kit](https://developer.huawei.com/consumer/en/hms/huawei-pushkit).
+
+@@project-info{ projectId="huawei-push-kit" }
+
+## Artifacts
+
+@@dependency [sbt,Maven,Gradle] {
+group=com.lightbend.akka
+artifact=akka-stream-alpakka-huawei-push-kit_$scala.binary.version$
+version=$project.version$
+symbol2=AkkaVersion
+value2=$akka.version$
+group2=com.typesafe.akka
+artifact2=akka-stream_$scala.binary.version$
+version2=AkkaVersion
+symbol3=AkkaHttpVersion
+value3=$akka-http.version$
+group3=com.typesafe.akka
+artifact3=akka-http_$scala.binary.version$
+version3=AkkaHttpVersion
+group4=com.typesafe.akka
+artifact4=akka-http-spray-json_$scala.binary.version$
+version4=AkkaHttpVersion
+}
+
+The table below shows direct dependencies of this module and the second tab shows all libraries it depends on transitively.
+
+@@dependencies { projectId="huawei-push-kit" }
+
+## Settings
+
+Prepare your credentials for access to HUAWEI Push Kit.
+
+Scala
+: @@snip [snip](/huawei-push-kit/src/test/scala/docs/scaladsl/PushKitExamples.scala) { #init-credentials }
+
+Java
+: @@snip [snip](/huawei-push-kit/src/test/java/docs/javadsl/PushKitExamples.java) { #init-credentials }
+
+The last two parameters in [HmsSettings](akka.stream.alpakka.huawei.pushkit.HmsSettings) are the predefined values.
+You can send test notifications [(so called validate only).](https://developer.huawei.com/consumer/en/doc/development/HMSCore-References-V5/https-send-api-0000001050986197-V5)
+And you can set the number of maximum concurrent connections.
+
+## Sending notifications
+
+To send a notification message create your notification object, and send it!
+
+Scala
+: @@snip [snip](/huawei-push-kit/src/test/scala/docs/scaladsl/PushKitExamples.scala) { #imports #asFlow-send }
+
+Java
+: @@snip [snip](/huawei-push-kit/src/test/java/docs/javadsl/PushKitExamples.java) { #imports #asFlow-send }
+
+With this type of send you can get responses from the server.
+These responses can be @scaladoc[PushKitResponse](akka.stream.alpakka.huawei.pushkit.PushKitResponse) or @scaladoc[ErrorResponse](akka.stream.alpakka.huawei.pushkit.ErrorResponse).
+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 [Akka docs `RestartFlow.onFailuresWithBackoff`](https://doc.akka.io/docs/akka/current/stream/operators/RestartFlow/onFailuresWithBackoff.html))
+
+If you don't care if the notification was sent successfully, you may use `fireAndForget`.
+
+Scala
+: @@snip [snip](/huawei-push-kit/src/test/scala/docs/scaladsl/PushKitExamples.scala) { #imports #simple-send }
+
+Java
+: @@snip [snip](/huawei-push-kit/src/test/java/docs/javadsl/PushKitExamples.java) { #imports #simple-send }
+
+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 only
+
+You can build notification described in the original documentation.
+It can be done by hand, or using some builder method.
+Example is condition builder.
+
+Scala
+: @@snip [snip](/huawei-push-kit/src/test/scala/docs/scaladsl/PushKitExamples.scala) { #condition-builder }

docs/src/main/paradox/index.md

@@ -40,6 +40,7 @@ The [Alpakka project](https://doc.akka.io/docs/alpakka/current/) is an open sour
 * [gRPC](external/grpc.md)
 * [Hadoop Distributed File System](hdfs.md)
 * [HBase](hbase.md)
+* [HUAWEI Push Kit](huawei-push-kit.md)
 * [HTTP](external/http.md)
 * [IBM Bluemix Cloud Object storage](bluemix-cos.md)
 * [IBM DB2 Event Store](external/db2-event-store.md)

huawei-push-kit/src/main/resources/reference.conf

@@ -0,0 +1,14 @@
+alpakka.huawei.pushkit {
+  # An address of a proxy that will be used for all connections using HTTP CONNECT tunnel.
+  # forward-proxy {
+  #   host = "proxy"
+  #   port = 8080
+  #   credentials {
+  #     username = "username"
+  #     password = "password"
+  #   }
+  #   trustPem {
+  #     pemPath = "/path/to/file.pem"
+  #   }
+  # }
+}

huawei-push-kit/src/main/scala/akka/stream/alpakka/huawei/pushkit/ForwardProxyHttpsContext.scala

@@ -0,0 +1,58 @@
+/*
+ * Copyright (C) 2016-2020 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.stream.alpakka.huawei.pushkit
+
+import akka.actor.ActorSystem
+import akka.http.scaladsl.{Http, HttpsConnectionContext}
+
+import java.io.FileInputStream
+import java.security.KeyStore
+import java.security.cert.{CertificateFactory, X509Certificate}
+import javax.net.ssl.{SSLContext, TrustManagerFactory}
+
+private[pushkit] object ForwardProxyHttpsContext {
+
+  val SSL = "SSL"
+  val X509 = "X509"
+
+  implicit class ForwardProxyHttpsContext(forwardProxy: ForwardProxy) {
+
+    def httpsContext(system: ActorSystem) = {
+      forwardProxy.trustPem match {
+        case Some(trustPem) => createHttpsContext(trustPem)
+        case None => defaultHttpsContext(system)
+      }
+    }
+  }
+
+  private def defaultHttpsContext(implicit system: ActorSystem) = {
+    Http().createDefaultClientHttpsContext()
+  }
+
+  private def createHttpsContext(trustPem: ForwardProxyTrustPem) = {
+    val certificate = x509Certificate(trustPem)
+    val sslContext = SSLContext.getInstance(SSL)
+
+    val alias = certificate.getIssuerDN.getName
+    val trustStore = KeyStore.getInstance(KeyStore.getDefaultType)
+    trustStore.load(null, null)
+    trustStore.setCertificateEntry(alias, certificate)
+
+    val tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm)
+    tmf.init(trustStore)
+    val trustManagers = tmf.getTrustManagers
+    sslContext.init(null, trustManagers, null)
+    new HttpsConnectionContext(sslContext)
+  }
+
+  private def x509Certificate(trustPem: ForwardProxyTrustPem) = {
+    val stream = new FileInputStream(trustPem.pemPath)
+    var result: X509Certificate = null
+    try result = CertificateFactory.getInstance(X509).generateCertificate(stream).asInstanceOf[X509Certificate]
+    finally if (stream != null) stream.close()
+    result
+  }
+
+}

huawei-push-kit/src/main/scala/akka/stream/alpakka/huawei/pushkit/ForwardProxyPoolSettings.scala

@@ -0,0 +1,32 @@
+/*
+ * Copyright (C) 2016-2020 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.stream.alpakka.huawei.pushkit
+
+import akka.actor.ActorSystem
+import akka.http.scaladsl.ClientTransport
+import akka.http.scaladsl.model.headers.BasicHttpCredentials
+import akka.http.scaladsl.settings.{ClientConnectionSettings, ConnectionPoolSettings}
+
+import java.net.InetSocketAddress
+
+private[pushkit] object ForwardProxyPoolSettings {
+
+  implicit class ForwardProxyPoolSettings(forwardProxy: ForwardProxy) {
+
+    def poolSettings(system: ActorSystem) = {
+      val address = InetSocketAddress.createUnresolved(forwardProxy.host, forwardProxy.port)
+      val transport = forwardProxy.credentials.fold(ClientTransport.httpsProxy(address))(
+        c => ClientTransport.httpsProxy(address, BasicHttpCredentials(c.username, c.password))
+      )
+
+      ConnectionPoolSettings(system)
+        .withConnectionSettings(
+          ClientConnectionSettings(system)
+            .withTransport(transport)
+        )
+    }
+  }
+
+}