add SNS connector with publish sink #204 #205

fgrutsch · 2017-02-25T12:20:08Z

Includes a new connector called sns
Includes a sink to publish messages to an AWS SNS topic

johanandren · 2017-03-01T08:27:30Z

johanandren

Some comments added. Looks promising!

johanandren · 2017-03-01T08:28:56Z

build.sbt

@@ -77,6 +77,14 @@ lazy val simpleCodecs = project
    name := "akka-stream-alpakka-simple-codecs"
  )

+lazy val sns = project
+  .in(file("sns"))


As the folder is the same name as the project/module, you don't have to specify .in(file(...)) it should be magically correct just by calling project (see module above for example)

Alright, I will try it out.

johanandren · 2017-03-01T08:30:09Z

docs/src/main/paradox/sns.md

@@ -0,0 +1,77 @@
+# AWS SNS Connector
+
+The AWS SNS connector provides an Akka Stream sink for AWS SNS.


Would be good with just a tad more about what SNS is, "provides an Akka Stream sink for push notifications through AWS SNS", so that a reader who doesn't know doesn't have to go to the official documentation directly.

Good point, I will add some more information.

johanandren · 2017-03-01T09:02:35Z

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

+
+import scala.concurrent.{Future, Promise}
+
+final class SnsPublishSinkStage(topicArn: String, snsClient: AmazonSNSAsync)


Is the AmazonSNSAsync thread safe? A stage can be materialized any number of times and all of them will share the same snsClient. If it cannot be shared a factory lambda is better (() => AmazonSNSAsync), that way every materialization can have its own client (which it will be responsible to manage/close etc in sync with the stage lifecycle).

Another option might be to take a AmazonSNSAsyncClientBuilder and .build() that each materialization (if the builder is immutable that is).

An additional thought: It could be better to make this a flow, emitting the values downstream as they have been ack:ed from AWS, the factory for the simple use case as sink could then be achieved through composition (SNSPublishFlow.to(Sink.ignore)).

Yes the AmazonSNSAsync is ThreadSafe regarding to the AWS docs. That's why I think it is fine to share a single AmazonSNSAsync among multiple materialization stages.

Also regarding your last comment about not making the AmazonSNSAsync implicit, I try to be consistent with other connectors, like the SQS one. There we also pass the AmazonSQSClient implicitly. I think it will be easier for people who use the connectors if the initialization of the GraphStages work the same way.

Good point to change the Sink to a Flow. I thought about keeping the Sink and provide an additional Flow GraphStage which will return the messageId of the published message, so clients can do further processing with the returned messageId. What do you think?

Ah, Hadn't noticed that how many connectors that do that. Thats a valid argument to keep it implicit.

Thanks for verifying that the client is threadsafe.

Maybe even just let it emit PublishResult downstream and then both String => String, and String => MessageId can be achieved through combinations with other stages.

Updated the SnsPublishFlow to emit a PublishResult.

This means we could delete the sink stage and express it by using the flow in the factory method instead:

val flow: Flow[String, PublishResult, NotUsed] val sink: Sink[String, Future[Done]] = flow.toMat(Sink.ignore)(Keep.right)

johanandren · 2017-03-01T09:03:49Z

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

+      }
+
+      def handleFailure(ex: Throwable): Unit = {
+        log.error(ex, "AmazonSNSAsync failure: {}", ex.getMessage)


No need to log the failure since we fail the stage and the materialized value with it. You could wrap it with a en exception to decorate with extra information from the stage though.

Alright I will remove it then.

johanandren · 2017-03-01T09:10:48Z

sns/src/test/scala/akka/stream/alpakka/sns/SnsPublishSinkSpec.scala

+              publishRequest,
+              new PublishResult().withMessageId("message-id")
+            )
+          new CompletableFuture()


This future is never completed though, shouldn't that cause the matVal to never complete, is there a bug around waiting for ack on publishing?

I will further investigate it.

fgrutsch · 2017-03-10T07:54:20Z

@johanandren I updated the PR regarding your comments. Furthemrore, I also added a SnsPublishFlow what I mentioned in one of my replies of your comments.

johanandren · 2017-03-12T16:41:00Z

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

+        val request = new PublishRequest().withTopicArn(topicArn).withMessage(grab(in))
+
+        snsClient.publishAsync(request,
+          new AsyncHandler[PublishRequest, PublishResult] {


Not necessarily anything to act on but, as this gets the values passed in on success/failure, it could be a single instance of AsyncHandler that you reuse rather than a new instance for every message.

johanandren · 2017-03-12T16:42:25Z

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

+      }
+
+      override def onPull(): Unit = {
+        if (isClosed(in)) completeStage()


There could be an ongoing publish that did not yet complete here, even if in was closed, you need to keep track of that and emit that last publish result when it comes, and then complete the stage.

johanandren · 2017-03-12T16:43:51Z

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

+
+import scala.compat.java8.FutureConverters._
+
+object SnsPublishSink {


Maybe this should then instead be SnsPublisher and have both .sink and .flow in the same place?

…osing

fgrutsch · 2017-03-14T21:41:11Z

@johanandren Updated regarding your comments. Removed the SnsPublishSink and provide a SnsPublisher with .flow and .sink factory methods. Also implemented a way to keep track of in flight messages for proper stage closing.

johanandren

LGTM, thanks @fg-devs !

patriknw

looking good, but some questions around the push/pull logic

patriknw · 2017-03-23T11:01:24Z

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

+        log.debug("Published SNS message: {}", result.getMessageId)
+        inFlight -= 1
+        if (isAvailable(out)) {
+          if (!hasBeenPulled(in)) tryPull(in)


why pull here? shouldn't that only be done from onPull?

Wouldn't this limit the Flow to process elements only one by one? Otherwise we would always have to wait for the output port to pull?

Thats why I also made inFlight as an Int and not a Boolean.

patriknw · 2017-03-23T11:03:42Z

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

+      private def handleSuccess(result: PublishResult): Unit = {
+        log.debug("Published SNS message: {}", result.getMessageId)
+        inFlight -= 1
+        if (isAvailable(out)) {


when will it not be available? should this be !isClosed
we are just dropping the message if it's not available so I want to be sure that we don't do that accidentally

patriknw · 2017-03-23T11:05:37Z

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

+      }
+
+      override def onPush(): Unit = {
+        inFlight += 1


can inFlight be > 1? perhaps use boolean if it can only be 0 or 1

patriknw · 2017-03-23T11:06:39Z

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

+
+      override def onPull(): Unit = {
+        if (isClosed(in) && inFlight == 0) completeStage()
+        if (!hasBeenPulled(in)) tryPull(in)


patriknw · 2017-03-23T11:07:43Z

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

+import com.amazonaws.services.sns.AmazonSNSAsync
+import com.amazonaws.services.sns.model.{PublishRequest, PublishResult}
+
+final class SnsPublishFlowStage(topicArn: String, snsClient: AmazonSNSAsync)


Internal API?
private[akka]

patriknw · 2017-03-23T11:08:24Z

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

+object SnsPublisher {
+
+  /**
+   * Java API: creates a [[Flow]] from a [[SnsPublishFlowStage]] for a SNS topic using an [[AmazonSNSAsync]]


users should not care about the implementation SnsPublishFlowStage

ktoso · 2017-04-07T14:56:19Z

Hi there @fg-devs, are you going to have some time to address the small comments made?
I'd hope so, would be awesome to include this in the next release :)

fgrutsch · 2017-04-08T06:14:28Z

@ktoso Hey, yes I will do the changes today.

# Conflicts: # build.sbt # docs/src/main/paradox/connectors.md

fgrutsch · 2017-04-08T17:13:48Z

@ktoso @patriknw Update the PR regarding your comments. The build for Scala 2.12 failed (there is already an open ticket)

# Conflicts: # build.sbt

johanandren · 2017-04-13T08:37:15Z

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

+import com.amazonaws.services.sns.AmazonSNSAsync
+import com.amazonaws.services.sns.model.{PublishRequest, PublishResult}
+
+private[akka] final class SnsPublishFlowStage(topicArn: String, snsClient: AmazonSNSAsync)


Does the client have a lifecycle, is is thread safe, should this be a client factory instead (() => AmazonSNSAsync) and have its lifecycle managed by the stage (`client.close() when the stage stops)?

NVM, I already asked that previously but github hid all comments and I had forgot.

johanandren

LGTM!

add SNS connector with publish sink akka#204

4280680

johanandren reviewed Mar 1, 2017

View reviewed changes

add SnsPublishFlow, update documentation and PR improvements

87da633

Fabian Grutsch added 2 commits March 11, 2017 19:02

emit SNS PublishResult in SnsPublishFlow

44cc23b

update SnsPublishFlow documentation

38b6b12

johanandren reviewed Mar 12, 2017

View reviewed changes

Fabian Grutsch added 4 commits March 14, 2017 22:36

provide SnsPublisher with sink and flow factory methods

3b81cff

remove SnsPublishSink and track inFlight messages for proper stage cl…

65c4edf

…osing

update documentation

02ba3ed

update code formatting

d11f49c

johanandren approved these changes Mar 15, 2017

View reviewed changes

patriknw self-assigned this Mar 20, 2017

patriknw reviewed Mar 23, 2017

View reviewed changes

Fabian Grutsch added 2 commits March 23, 2017 20:38

make SnsPublishFlowStage internal API

2cfdb15

update SnsPublisher documentation

2b36d4c

Fabian Grutsch added 2 commits April 8, 2017 18:19

Merge branch 'master' into aws_sns_sink

f744d00

# Conflicts: # build.sbt # docs/src/main/paradox/connectors.md

update push/pull logic and file headers

588d731

Merge branch 'master' into aws_sns_sink

8af3ebf

# Conflicts: # build.sbt

johanandren reviewed Apr 13, 2017

View reviewed changes

johanandren approved these changes Apr 13, 2017

View reviewed changes

johanandren merged commit d2b0363 into akka:master Apr 13, 2017

		@@ -0,0 +1,77 @@
		# AWS SNS Connector

		The AWS SNS connector provides an Akka Stream sink for AWS SNS.


		import scala.concurrent.{Future, Promise}

		final class SnsPublishSinkStage(topicArn: String, snsClient: AmazonSNSAsync)


		import scala.compat.java8.FutureConverters._

		object SnsPublishSink {

add SNS connector with publish sink #204 #205

add SNS connector with publish sink #204 #205

Conversation

fgrutsch commented Feb 25, 2017

johanandren commented Mar 1, 2017

johanandren left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fgrutsch Mar 4, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

johanandren Mar 12, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fgrutsch commented Mar 10, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fgrutsch commented Mar 14, 2017

johanandren left a comment

Choose a reason for hiding this comment

patriknw left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ktoso commented Apr 7, 2017

fgrutsch commented Apr 8, 2017

fgrutsch commented Apr 8, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

johanandren left a comment

Choose a reason for hiding this comment

fgrutsch Mar 4, 2017 •

edited

Loading

johanandren Mar 12, 2017 •

edited

Loading