add SNS connector with publish sink #204

build.sbt

@@ -1,7 +1,7 @@
 lazy val alpakka = project
   .in(file("."))
   .enablePlugins(PublishUnidoc)
-  .aggregate(amqp, cassandra, dynamodb, files, ftp, hbase, jms, mqtt, s3, simpleCodecs, sqs, sse, awslambda)
+  .aggregate(amqp, cassandra, dynamodb, files, ftp, hbase, jms, mqtt, s3, simpleCodecs, sns, sqs, sse, awslambda)
 
 lazy val amqp = project
   .enablePlugins(AutomateHeaderPlugin)
@@ -77,6 +77,13 @@ lazy val simpleCodecs = project
     name := "akka-stream-alpakka-simple-codecs"
   )
 
+lazy val sns = project
+  .enablePlugins(AutomateHeaderPlugin)
+  .settings(
+    name := "akka-stream-alpakka-sns",
+    Dependencies.Sns
+  )
+
 lazy val sqs = project
   .in(file("sqs"))
   .enablePlugins(AutomateHeaderPlugin)

docs/src/main/paradox/connectors.md

@@ -4,6 +4,7 @@
 
 * [AMQP Connector](amqp.md)
 * [AWS DynamoDB Connector](dynamodb.md)
+* [AWS SNS Connector](sns.md)
 * [AWS SQS Connector](sqs.md)
 * [AWS Lambda Connector](awslambda.md)
 * [Cassandra Connector](cassandra.md)

docs/src/main/paradox/sns.md

@@ -0,0 +1,93 @@
+# AWS SNS Connector
+
+The AWS SNS connector provides an Akka Stream Flow and Sink for push notifications through AWS SNS.
+
+For more information about AWS SNS please visit the [official documentation](https://aws.amazon.com/documentation/sns/).
+
+## Artifacts
+
+sbt
+:   @@@vars
+    ```scala
+    libraryDependencies += "com.lightbend.akka" %% "akka-stream-alpakka-sns" % "$version$"
+    ```
+    @@@
+
+Maven
+:   @@@vars
+    ```xml
+    <dependency>
+      <groupId>com.lightbend.akka</groupId>
+      <artifactId>akka-stream-alpakka-sns_$scala.binaryVersion$</artifactId>
+      <version>$version$</version>
+    </dependency>
+    ```
+    @@@
+
+Gradle
+:   @@@vars
+    ```gradle
+    dependencies {
+      compile group: "com.lightbend.akka", name: "akka-stream-alpakka-sns_$scala.binaryVersion$", version: "$version$"
+    }
+    ```
+    @@@
+
+## Usage
+
+Sources provided by this connector need a prepared `AmazonSNSAsyncClient` to publish messages to a topic.
+
+Scala
+: @@snip (../../../../sns/src/test/scala/akka/stream/alpakka/sns/scaladsl/Examples.scala) { #init-client }
+
+Java
+: @@snip (../../../../sns/src/test/java/akka/stream/alpakka/sns/javadsl/Examples.java) { #init-client }
+
+We will also need an @scaladoc[ActorSystem](akka.actor.ActorSystem) and an @scaladoc[ActorMaterializer](akka.stream.ActorMaterializer).
+
+Scala
+: @@snip (../../../../sns/src/test/scala/akka/stream/alpakka/sns/scaladsl/Examples.scala) { #init-system }
+
+Java
+: @@snip (../../../../sns/src/test/java/akka/stream/alpakka/sns/javadsl/Examples.java) { #init-system }
+
+This is all preparation that we are going to need.
+
+### Publish messages to a SNS topic
+
+Now we can publish a String message to any SNS topic where we have access to by providing the topic ARN to the
+@scaladoc[SnsPublishFlow](akka.stream.alpakka.sns.scaladsl.SnsPublishFlow$) or
+@scaladoc[SnsPublishSink](akka.stream.alpakka.sns.scaladsl.SnsPublishSink$) factory.
+
+### SnsPublishFlow
+
+Scala
+: @@snip (../../../../sns/src/test/scala/akka/stream/alpakka/sns/scaladsl/Examples.scala) { #use-flow }
+
+Java
+: @@snip (../../../../sns/src/test/java/akka/stream/alpakka/sns/javadsl/Examples.java) { #use-flow }
+
+As you can see, this would publish the messages from the source to the specified AWS SNS topic.
+After a message has been successfully published, a
+[PublishResult](http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/sns/model/PublishResult.html)
+will be pushed downstream.
+
+### SnsPublishSink
+
+Scala
+: @@snip (../../../../sns/src/test/scala/akka/stream/alpakka/sns/scaladsl/Examples.scala) { #use-sink }
+
+Java
+: @@snip (../../../../sns/src/test/java/akka/stream/alpakka/sns/javadsl/Examples.java) { #use-sink }
+
+As you can see, this would publish the messages from the source to the specified AWS SNS topic.
+
+### Running the example code
+
+The code in this guide is part of runnable tests of this project. You are welcome to edit the code and run it in sbt.
+
+Scala
+:   ```
+    sbt
+    > sns/test
+    ```

project/Dependencies.scala

@@ -97,6 +97,13 @@ object Dependencies {
     )
   )
 
+  val Sns = Seq(
+    libraryDependencies ++= Seq(
+      "com.amazonaws" % "aws-java-sdk-sns" % "1.11.95",        // ApacheV2
+      "org.mockito"   % "mockito-core"     % "2.7.11"    % Test // MIT
+    )
+  )
+
   val Sqs = Seq(
     libraryDependencies ++= Seq(
       "com.amazonaws" % "aws-java-sdk-sqs" % "1.11.76",        // ApacheV2

sns/src/main/scala/akka/stream/alpakka/sns/SnsPublishFlowStage.scala

@@ -0,0 +1,59 @@
+/*
+ * Copyright (C) 2016 Lightbend Inc. <http://www.lightbend.com>
+ */
+package akka.stream.alpakka.sns
+
+import akka.stream._
+import akka.stream.stage._
+import com.amazonaws.handlers.AsyncHandler
+import com.amazonaws.services.sns.AmazonSNSAsync
+import com.amazonaws.services.sns.model.{PublishRequest, PublishResult}
+
+final class SnsPublishFlowStage(topicArn: String, snsClient: AmazonSNSAsync)
+    extends GraphStage[FlowShape[String, PublishResult]] {
+
+  private val in = Inlet[String]("SnsPublishFlow.in")
+  private val out = Outlet[PublishResult]("SnsPublishFlow.out")
+
+  override def shape: FlowShape[String, PublishResult] = FlowShape.of(in, out)
+
+  override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = {
+    new GraphStageLogic(shape) with InHandler with OutHandler with StageLogging {
+
+      private val failureCallback = getAsyncCallback[Throwable](handleFailure)
+      private val successCallback = getAsyncCallback[PublishResult](handleSuccess)
+
+      private def handleFailure(ex: Throwable): Unit =
+        failStage(ex)
+
+      private def handleSuccess(result: PublishResult): Unit = {
+        log.debug("Published SNS message: {}", result.getMessageId)
+
+        if (isAvailable(out)) {
+          if (!hasBeenPulled(in)) tryPull(in)
+          push(out, result)
+        }
+      }
+
+      override def onPush(): Unit = {
+        val request = new PublishRequest().withTopicArn(topicArn).withMessage(grab(in))
+
+        snsClient.publishAsync(request,
+          new AsyncHandler[PublishRequest, PublishResult] {
+          override def onError(exception: Exception): Unit =
+            failureCallback.invoke(exception)
+
+          override def onSuccess(request: PublishRequest, result: PublishResult): Unit =
+            successCallback.invoke(result)
+        })
+      }
+
+      override def onPull(): Unit = {
+        if (isClosed(in)) completeStage()
+        if (!hasBeenPulled(in)) tryPull(in)
+      }
+
+      setHandlers(in, out, this)
+    }
+  }
+}

sns/src/main/scala/akka/stream/alpakka/sns/SnsPublishSinkStage.scala

@@ -0,0 +1,75 @@
+/*
+ * Copyright (C) 2016 Lightbend Inc. <http://www.lightbend.com>
+ */
+package akka.stream.alpakka.sns
+
+import akka.Done
+import akka.stream.stage._
+import akka.stream.{Attributes, Inlet, SinkShape}
+import com.amazonaws.handlers.AsyncHandler
+import com.amazonaws.services.sns.AmazonSNSAsync
+import com.amazonaws.services.sns.model.{PublishRequest, PublishResult}
+
+import scala.concurrent.{Future, Promise}
+
+final class SnsPublishSinkStage(topicArn: String, snsClient: AmazonSNSAsync)
+    extends GraphStageWithMaterializedValue[SinkShape[String], Future[Done]] {
+
+  private val in = Inlet[String]("SnsPublishSink.in")
+
+  override def shape: SinkShape[String] = SinkShape.of(in)
+
+  override def createLogicAndMaterializedValue(inheritedAttributes: Attributes): (GraphStageLogic, Future[Done]) = {
+    val completed = Promise[Done]()
+
+    val logic = new GraphStageLogic(shape) with InHandler with StageLogging {
+
+      private def handleFailure(ex: Throwable): Unit = {
+        failStage(ex)
+        completed.tryFailure(ex)
+      }
+
+      private def handleSuccess(result: PublishResult): Unit = {
+        log.debug("Published SNS message: {}", result.getMessageId)
+        completed.trySuccess(Done)
+
+        if (!hasBeenPulled(in)) tryPull(in)
+      }
+
+      private val failureCallback = getAsyncCallback[Throwable](handleFailure)
+      private val successCallback = getAsyncCallback[PublishResult](handleSuccess)
+
+      override def onPush(): Unit = {
+        val request = new PublishRequest().withTopicArn(topicArn).withMessage(grab(in))
+
+        snsClient.publishAsync(request,
+          new AsyncHandler[PublishRequest, PublishResult] {
+          override def onError(exception: Exception): Unit =
+            failureCallback.invoke(exception)
+
+          override def onSuccess(request: PublishRequest, result: PublishResult): Unit =
+            successCallback.invoke(result)
+        })
+      }
+
+      override def onUpstreamFinish(): Unit = {
+        completeStage()
+        completed.trySuccess(Done)
+      }
+
+      override def onUpstreamFailure(ex: Throwable): Unit = {
+        failStage(ex)
+        completed.tryFailure(ex)
+      }
+
+      override def preStart(): Unit = {
+        setKeepGoing(true)
+        pull(in)
+      }
+
+      setHandler(in, this)
+    }
+
+    (logic, completed.future)
+  }
+}

sns/src/main/scala/akka/stream/alpakka/sns/javadsl/SnsPublishFlow.scala

@@ -0,0 +1,20 @@
+/*
+ * Copyright (C) 2016 Lightbend Inc. <http://www.lightbend.com>
+ */
+package akka.stream.alpakka.sns.javadsl
+
+import akka.NotUsed
+import akka.stream.alpakka.sns.SnsPublishFlowStage
+import akka.stream.javadsl.Flow
+import com.amazonaws.services.sns.AmazonSNSAsync
+import com.amazonaws.services.sns.model.PublishResult
+
+object SnsPublishFlow {
+
+  /**
+   * Java API: creates a [[SnsPublishFlowStage]] for a SNS topic using an [[AmazonSNSAsync]]
+   */
+  def create(topicArn: String, snsClient: AmazonSNSAsync): Flow[String, PublishResult, NotUsed] =
+    Flow.fromGraph(new SnsPublishFlowStage(topicArn, snsClient))
+
+}

sns/src/main/scala/akka/stream/alpakka/sns/javadsl/SnsPublishSink.scala

@@ -0,0 +1,26 @@
+/*
+ * Copyright (C) 2016 Lightbend Inc. <http://www.lightbend.com>
+ */
+package akka.stream.alpakka.sns.javadsl
+
+import java.util.concurrent.CompletionStage
+
+import akka.Done
+import akka.stream.alpakka.sns.SnsPublishSinkStage
+import akka.stream.javadsl.{Sink => JSink}
+import akka.stream.scaladsl.Sink
+import com.amazonaws.services.sns.AmazonSNSAsync
+
+import scala.compat.java8.FutureConverters._
+
+object SnsPublishSink {
+
+  /**
+   * Java API: creates a [[SnsPublishSinkStage]] for a SNS topic using an [[AmazonSNSAsync]]
+   */
+  def create(topicArn: String, snsClient: AmazonSNSAsync): JSink[String, CompletionStage[Done]] = {
+    val sink = Sink.fromGraph(new SnsPublishSinkStage(topicArn, snsClient))
+    sink.mapMaterializedValue(_.toJava).asJava
+  }
+
+}

sns/src/main/scala/akka/stream/alpakka/sns/scaladsl/SnsPublishFlow.scala

@@ -0,0 +1,20 @@
+/*
+ * Copyright (C) 2016 Lightbend Inc. <http://www.lightbend.com>
+ */
+package akka.stream.alpakka.sns.scaladsl
+
+import akka.NotUsed
+import akka.stream.alpakka.sns.SnsPublishFlowStage
+import akka.stream.scaladsl.Flow
+import com.amazonaws.services.sns.AmazonSNSAsync
+import com.amazonaws.services.sns.model.PublishResult
+
+object SnsPublishFlow {
+
+  /**
+   * Scala API: creates a [[SnsPublishFlowStage]] for a SNS topic using an [[AmazonSNSAsync]]
+   */
+  def apply(topicArn: String)(implicit snsClient: AmazonSNSAsync): Flow[String, PublishResult, NotUsed] =
+    Flow.fromGraph(new SnsPublishFlowStage(topicArn, snsClient))
+
+}

sns/src/main/scala/akka/stream/alpakka/sns/scaladsl/SnsPublishSink.scala

@@ -0,0 +1,21 @@
+/*
+ * Copyright (C) 2016 Lightbend Inc. <http://www.lightbend.com>
+ */
+package akka.stream.alpakka.sns.scaladsl
+
+import akka.Done
+import akka.stream.alpakka.sns.SnsPublishSinkStage
+import akka.stream.scaladsl.Sink
+import com.amazonaws.services.sns.AmazonSNSAsync
+
+import scala.concurrent.Future
+
+object SnsPublishSink {
+
+  /**
+   * Scala API: creates a [[SnsPublishSinkStage]] for a SNS topic using an [[AmazonSNSAsync]]
+   */
+  def apply(topicArn: String)(implicit snsClient: AmazonSNSAsync): Sink[String, Future[Done]] =
+    Sink.fromGraph(new SnsPublishSinkStage(topicArn, snsClient))
+
+}