Alpakka AWS SQS Connectorを試してみた

Alpakka 0.3で AWS SQS Connector が追加されたので試してみました。

<blockquote class="twitter-tweet" data-lang="ja"><p lang="en" dir="ltr">Released <a href="https://twitter.com/hashtag/Alpakka?src=hash">#Alpakka</a> 0.3. AWS SQS, Cassandra, FTP. Great stuff, thanks all contributors. <a href="https://t.co/ObyGvhamWu">https://t.co/ObyGvhamWu</a></p>&mdash; Akka Team (@akkateam) <a href="https://twitter.com/akkateam/status/804647813808263168">2016年12月2日</a></blockquote>

## Alpakka
[Github akka/alpakka](https://github.com/akka/alpakka)
> This project provides a home to Akka Streams connectors to various technologies, protocols or libraries.

Akka Streams コネクターが提供されています。
Akka Streams に関しては[こちら](http://www.slideshare.net/ktoso/2014-akkastreamstokyojapanese)や[こちら](http://www.slideshare.net/kazukinegoro5/akka-streams-100-scalamatsuri)をご覧ください。

## 前準備
- AWSのアカウントを作成し`Access key ID`と`Secret access key`を取得
- SQSでキューを作成し、URLを取得

## 実装
こちらの[リファレンス](http://developer.lightbend.com/docs/alpakka/0.3/sqs.html)を参考（ほぼそのまま）にコードを書きます。 `config`は適宜読み替えてください。
メッセージが送信されると本文を出力し、そのまま削除するようにしています。

```scala:build.sbt
name := "alpakka-sqs-sample"

version := "1.0"

scalaVersion := "2.12.0"

libraryDependencies += "com.lightbend.akka" %% "akka-stream-alpakka-sqs" % "0.3"
```

```scala:AlpakkaSQSSample.scala
import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.alpakka.sqs.scaladsl.SqsSource
import com.amazonaws.auth.BasicAWSCredentials
import com.amazonaws.services.sqs.AmazonSQSAsyncClient
import com.amazonaws.services.sqs.model.DeleteMessageRequest
import com.typesafe.config.ConfigFactory

import scala.concurrent.duration.{ FiniteDuration, _ }

object AlpakkaSQSSample {
  def main(args: Array[String]): Unit = {
    val config = ConfigFactory.load()

    val credentials = new BasicAWSCredentials(
      config.getString("aws.accessKey"), config.getString("aws.secretKey")
    )
    implicit val sqsClient: AmazonSQSAsyncClient =
      new AmazonSQSAsyncClient(credentials).withEndpoint(config.getString("aws.sqs.endpoint"))

    implicit val system = ActorSystem()
    implicit val mat = ActorMaterializer()

    val queue = config.getString("aws.sqs.url")

    SqsSource(queue)
      .runForeach((message) => {
        println(message.getBody)
        sqsClient.deleteMessage(
          new DeleteMessageRequest(queue, message.getReceiptHandle)
        )
      })
  }
}
```

ロングポーリングタイムアウトやバッファーサイズなどは`SqsSourceSettings(longPollingDuration: FiniteDuration, maxBufferSize: Int, maxBatchSize: Int)`で設定可能です。（`SqsSource.apply`の第二引数）
デフォルトの設定は[こちら](https://github.com/akka/alpakka/blob/master/sqs/src/main/scala/akka/stream/alpakka/sqs/SqsSourceStage.scala#L18)です。

## 実行
```bash
$ sbt
> run
```

メッセージを送信すると以下のように本文が表示されます。

```bash
[info] Running ***.AlpakkaSQSSample 
message-1
message-2
```

## 最後に
Akka Streamsがさらに身近な存在になりました。
[リファレンス](http://developer.lightbend.com/docs/alpakka/0.3/sqs.html)通りに[ElasticMQ](https://github.com/adamw/elasticmq)を利用すれば、SQSを利用しなくてもすぐ試すことができます。
ちなみに、FIFOキューだとちょっと気になる挙動をしていたので[Akka User List](https://groups.google.com/forum/?hl=ja#!topic/akka-user/PSnmaSkTDkQ)で質問してみました。

## 参考
- [Github](https://github.com/akka/alpakka)
- [Reference](http://developer.lightbend.com/docs/alpakka/0.3/)