SQS flows + Embedded ElasticMQ #239

kciesielski · 2017-03-27T20:57:53Z

This PR introduces quit a lot of changes to the SQS connector, however it keeps the public Scala API backward compatible. Highlights:

Introducing SqsFlow.create() and SqsAckFlow.create() for creating SQS publishers as flows.
Removing SqsSinkStage and SqsAckSinkStage as these stages can be build from flows.
BREAKING changing the Java API to make it return a Java CompletableStage instead of Scala Future.
Updating documentation, using current AWS SDK instead of deprecated functions.
Using embedded ElasticMQ for tests, no need to run external queue via docker.

The design of flow stages is based on similar stages in akka-stream-kafka

kciesielski · 2017-03-27T20:58:48Z

@s12v Could you take a look at this PR?

s12v · 2017-03-27T22:05:15Z

sqs/src/main/scala/akka/stream/alpakka/sqs/SqsAckFlowStage.scala

+                  new AsyncHandler[DeleteMessageRequest, DeleteMessageResult] {
+
+                  override def onError(exception: Exception): Unit = {
+                    r.failure(exception)


why it's failure here and tryFailure in the other handler?

Should be failure in both, fixing.

s12v · 2017-03-27T22:07:09Z

sqs/src/main/scala/akka/stream/alpakka/sqs/SqsAckFlowStage.scala

+
+                  override def onError(exception: Exception): Unit = {
+                    r.failure(exception)
+                    onComplete()


I thought it's better to not directly call methods in GraphStageLogic, and always invoke callback

That's right, I'll fix this to use callbacks.

s12v · 2017-03-27T22:08:43Z

sqs/src/main/scala/akka/stream/alpakka/sqs/SqsAckFlowStage.scala

+              case RequeueWithDelay(delaySeconds) =>
+                sqsClient
+                  .sendMessageAsync(new SendMessageRequest(queueUrl, message.getBody).withDelaySeconds(delaySeconds),
+                    new AsyncHandler[SendMessageRequest, SendMessageResult] {


this object always does exactly the same, would it be better to create it once and reuse?

It isn't, it closes over the promise.

johanandren

Added some feedback.

johanandren · 2017-03-28T08:09:02Z

sqs/src/main/scala/akka/stream/alpakka/sqs/SqsAckFlowStage.scala

+
+    val logic = new GraphStageLogic(shape) with StageLogging {
+      private var inFlight = new AtomicInteger(0)
+      @volatile private var inIsClosed = false


Put all interaction in methods called from async callbacks instead of these thread primitives and make use of the guarantees streams gives you.

Of course, fixing to callbacks.

johanandren · 2017-03-28T08:09:10Z

sqs/src/main/scala/akka/stream/alpakka/sqs/SqsAckFlowStage.scala

+import scala.concurrent.{Future, Promise}
+import scala.util.{Failure, Success, Try}
+
+class SqsAckFlowStage(queueUrl: String, sqsClient: AmazonSQSAsync)


final and maybe private[sqs] as well. The factories are the public API.

Right, fixing.

johanandren · 2017-03-28T08:10:20Z

sqs/src/main/scala/akka/stream/alpakka/sqs/SqsAckFlowStage.scala

+
+      setHandler(out,
+        new OutHandler {
+        override def onPull() =


Strange indentation/linebreaks.

It's automatically managed by scalafmt

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

sqs/src/main/scala/akka/stream/alpakka/sqs/SqsAckFlowStage.scala

+
+            def onComplete(): Unit =
+              if (inFlight.decrementAndGet() == 0 && inIsClosed)
+                checkForCompletionCB.invoke(())


Exact duplicate of the other onComplete?

That's right, removing.

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

sqs/src/main/scala/akka/stream/alpakka/sqs/SqsAckFlowStage.scala

+              case RequeueWithDelay(delaySeconds) =>
+                sqsClient
+                  .sendMessageAsync(new SendMessageRequest(queueUrl, message.getBody).withDelaySeconds(delaySeconds),
+                    new AsyncHandler[SendMessageRequest, SendMessageResult] {


It isn't, it closes over the promise.

johanandren · 2017-03-28T08:13:06Z

sqs/src/main/scala/akka/stream/alpakka/sqs/SqsAckFlowStage.scala

+                checkForCompletionCB.invoke(())
+
+            val (message, action) = grab(in)
+            val r = Promise[AmazonWebServiceResult[ResponseMetadata]]


responsePromise or something like that instead of r. Single char names are ok in a small/few line scope, not in a larger scope like this.

Thanks, renaming.

johanandren · 2017-03-28T08:13:54Z

sqs/src/main/scala/akka/stream/alpakka/sqs/SqsAckFlowStage.scala

+
+  override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = {
+
+    val logic = new GraphStageLogic(shape) with StageLogging {


No need to capture it in the logic val, just return it directly.

Thanks, fixing.

johanandren · 2017-03-28T08:15:39Z

sqs/src/main/scala/akka/stream/alpakka/sqs/javadsl/SqsAckSink.scala

   */
  def create(queueUrl: String,
             settings: SqsAckSinkSettings,
-             sqsClient: AmazonSQSAsync): Sink[MessageActionPair, Future[Done]] =
-    Sink.fromGraph(new SqsAckSinkStage(queueUrl, settings, sqsClient))
+             sqsClient: AmazonSQSAsync): Sink[MessageActionPair, CompletionStage[Done]] =


johanandren · 2017-03-28T08:17:02Z

sqs/src/main/scala/akka/stream/alpakka/sqs/javadsl/package.scala

+package object javadsl {
+  def identityFunction[T] = new Function[T, T] {
+    override def apply(param: T): T = param
+  }


Isn't used anywhere so can be removed? (If it was used a val would be better, to avoid the allocation of a new function on every use)

Sure, I'll remove it.

dpfeiffer · 2017-03-28T09:42:29Z

sqs/src/test/java/akka/stream/alpakka/sqs/javadsl/BaseSqsTest.java

+    private InetSocketAddress sqsAddress;
+    private int sqsPort = 0;
+    protected String sqsEndpoint;
+    protected AWSCredentialsProvider credentialsProvider = new AWSCredentialsProvider() {


If I remember correctly there should be a StaticAWSCredentialsProvider.

Thanks, I'll use it.

- Mainly managing local state via callbacks + some minor changes - Additionally, use dedicated result types to represent data returned from flow (AWS response metadata + original msg body)

kciesielski · 2017-04-12T14:37:05Z

I had to rebase this PR against master, there were merge conflicts. All review remarks are addressed, let me know if I overlooked something. Thanks a lot for reviewing @dpfeiffer @johanandren @s12v

johanandren · 2017-04-13T08:27:31Z

LGTM, thanks @kciesielski !

s12v reviewed Mar 27, 2017

View reviewed changes

johanandren reviewed Mar 28, 2017

View reviewed changes

dpfeiffer reviewed Mar 28, 2017

View reviewed changes

s12v mentioned this pull request Apr 11, 2017

SQS: Flow versions of SqsAckSink and SqsSink #255

Closed

kciesielski force-pushed the sqs-flow-elasticmq branch from 64332c4 to 9c55423 Compare April 12, 2017 10:05

kciesielski added 6 commits April 12, 2017 16:36

Add a flow stage and use ElasticMQ

f7e0fd3

Use flow-based stage for ACKs

db2a5ae

Use AmazonSQSAsync instead of AmazonSQSAsyncClient

f78d2f8

Adjusts docs and tests

139d13f

Fixes after code review

20ea9c8

- Mainly managing local state via callbacks + some minor changes - Additionally, use dedicated result types to represent data returned from flow (AWS response metadata + original msg body)

Use AWSStaticCredentialsProvided

ed7ad2d

kciesielski force-pushed the sqs-flow-elasticmq branch from 9c55423 to ed7ad2d Compare April 12, 2017 14:36

johanandren merged commit 9ad7b4a into akka:master Apr 13, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

SQS flows + Embedded ElasticMQ #239

SQS flows + Embedded ElasticMQ #239

kciesielski commented Mar 27, 2017 •

edited

Loading

kciesielski commented Mar 27, 2017

s12v Mar 27, 2017

kciesielski Apr 12, 2017

s12v Mar 27, 2017

kciesielski Apr 12, 2017

s12v Mar 27, 2017

johanandren Mar 28, 2017

johanandren left a comment

johanandren Mar 28, 2017

kciesielski Apr 12, 2017

johanandren Mar 28, 2017

kciesielski Apr 12, 2017

johanandren Mar 28, 2017

kciesielski Apr 12, 2017

johanandren Mar 28, 2017

kciesielski Apr 12, 2017

johanandren Mar 28, 2017

johanandren Mar 28, 2017

kciesielski Apr 12, 2017

johanandren Mar 28, 2017

kciesielski Apr 12, 2017

johanandren Mar 28, 2017

johanandren Mar 28, 2017

kciesielski Apr 12, 2017

dpfeiffer Mar 28, 2017

kciesielski Apr 12, 2017

kciesielski commented Apr 12, 2017

johanandren commented Apr 13, 2017


		override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = {

		val logic = new GraphStageLogic(shape) with StageLogging {

SQS flows + Embedded ElasticMQ #239

SQS flows + Embedded ElasticMQ #239

Conversation

kciesielski commented Mar 27, 2017 • edited Loading

kciesielski commented Mar 27, 2017

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

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

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

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

kciesielski commented Apr 12, 2017

johanandren commented Apr 13, 2017

kciesielski commented Mar 27, 2017 •

edited

Loading