Amazon SQS互換のローカル実行環境ElasticMQの構築

Amazon SQSと互換があるElasticMQをインストールし、ローカルにあるLambdaのソースコードと連携するローカル実行環境を構築します

softwaremill/elasticmq

https://github.com/softwaremill/elasticmq

# ElasticMQのインストール

ElasticMQのインストール方法には複数ありますが、今回はDockerを使います。
特に、Docker環境としては、QNAPのNASにインストールします。理由は単にQNAP NASを利用しているから。

QNAPには、Docker環境として、ContainerStationがあります。
ElasticMQのDockerファイルは、リポジトリに対して、elasticmqの入力で検索にひっかかります。

https://hub.docker.com/r/softwaremill/elasticmq-native/

Dockerに馴染みのある方は、以下がわかりやすいかと思います。

　```docker run -p 9324:9324 -p 9325:9325 softwaremill/elasticmq-native```

これでもよいのですが、インストールされるホスト名がlocalhostとなってしまい、かっこ悪いので、設定ファイルを変更します。

（以降はQNAPでの操作です）
そこで、エントリポイントには以下を指定します。

```
/sbin/tini -- /opt/docker/bin/elasticmq-native-server -Dconfig.file=/opt2/elasticmq.conf -Dlogback.configurationFile=/opt/logback.xml
```

Configファイルは、「/opt2/elasticmq.conf」を参照するようにしています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/261826/44db2c06-dd90-b6d6-bda0-e2451de78fb1.png)

そして、/opt2をNAS上のフォルダに割り当てるようにします。
例：
ホストからのボリューム：/Container/elasticmq/opt2
マウントポイント：/opt2

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/261826/520fea1c-ac24-efe9-214a-1ed3f0acf989.png)

あとは、/Container/elasticmq/opt2の下に、elasticmq.confを作成すればよいです。QNAPのアプリである「Text Editor」を使うのがよいかと思います。

```/Container/elasticmq/opt2/elasticmq.conf
include classpath("application.conf")

# What is the outside visible address of this ElasticMQ node
# Used to create the queue URL (may be different from bind address!)
node-address {
  protocol = http
  host = 【外部からアクセスされる際のホスト名】
  port = 9324
  context-path = ""
}
```

ネットワークは、「NAT」にして、コンテナ側のポート番号9324と9325はそのままホスト側に割り当てます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/261826/e46873c0-d247-9593-6ea6-362c9a9f6c82.png)

以上で設定完了です。
インストール完了後、以下にアクセスすると、管理Webページが表示されます。

http://【外部からアクセスされる際のホスト名】:9325

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/261826/e8cda773-e7bb-2d32-6d13-e4f2a81fff67.png)

# Node.jsコード・AWS CLIの例

●キューの作成

（通常のキュー）

```sqs_create_queue.js
const SQS_ENDPOINT = "http://【外部からアクセスされる際のホスト名】:9324";
const SQS_QUEUE_NAME = "test";

const AWS = require('aws-sdk');
AWS.config.update({
//	region: 'ap-northeast-1'
	region: 'elasticmq'
});
const sqs = new AWS.SQS({
	apiVersion: '2012-11-05',
	endpoint: SQS_ENDPOINT
});

const params = {
	QueueName: SQS_QUEUE_NAME,
	Attributes: {
//		FifoQueue: "true",
	},
};

sqs.createQueue(params, (err, data) => {
  if (err) {
    console.log("Error", err);
    return;
  }
  console.log("Success", JSON.stringify(data, null, '\t'));
});
```

(FIFOキュー)

```sqs_create_queue.js
const SQS_ENDPOINT = "http://【外部からアクセスされる際のホスト名】:9324";
const SQS_QUEUE_NAME = "test.fifo";

const AWS = require('aws-sdk');
AWS.config.update({
//	region: 'ap-northeast-1'
	region: 'elasticmq'
});
const sqs = new AWS.SQS({
	apiVersion: '2012-11-05',
	endpoint: SQS_ENDPOINT
});

const params = {
	QueueName: SQS_QUEUE_NAME,
	Attributes: {
		FifoQueue: "true",
	},
};

sqs.createQueue(params, (err, data) => {
  if (err) {
    console.log("Error", err);
    return;
  }
  console.log("Success", JSON.stringify(data, null, '\t'));
});
```

```
> aws sqs create-queue --queue-name test --endpoint-url http://【外部からアクセスされる際のホスト名】:9324
```

●キューのリスト取得

```js:sqs_list_queue.js
const SQS_ENDPOINT = "http://【外部からアクセスされる際のホスト名】:9324";

const AWS = require('aws-sdk');
AWS.config.update({
//	region: 'ap-northeast-1'
	region: 'elasticmq'
});
const sqs = new AWS.SQS({
	apiVersion: '2012-11-05',
	endpoint: SQS_ENDPOINT
});

const params = {
};

sqs.listQueues(params, (err, data) => {
  if (err) {
    console.log("Error", err);
    return;
  }
  console.log("Success", JSON.stringify(data, null, '\t'));
});
```

```
> aws sqs list-queues --endpoint-url http://【外部からアクセスされる際のホスト名】:9324
```

●キューの削除

```js:sqs_delete_queue.js
const SQS_ENDPOINT = "http://【外部からアクセスされる際のホスト名】:9324";
const SQS_QUEUE_URL = "http://【外部からアクセスされる際のホスト名】:9324/000000000000/test";

const AWS = require('aws-sdk');
AWS.config.update({
//	region: 'ap-northeast-1'
	region: 'elasticmq'
});
const sqs = new AWS.SQS({
	apiVersion: '2012-11-05',
	endpoint: SQS_ENDPOINT
});

const params = {
	QueueUrl: SQS_QUEUE_URL
};

sqs.deleteQueue(params, (err, data) => {
  if (err) {
    console.log("Error", err);
    return;
  }
  console.log("Success", data);
});
```

```
> aws sqs delete-queue --queue-url http://【外部からアクセスされる際のホスト名】:9324/000000000000/test --endpoint-url http://【外部からアクセスされる際のホスト名】:9324
```

●メッセージの送信

```js:sqs_send_message.js
const SQS_ENDPOINT = "http://【外部からアクセスされる際のホスト名】:9324";
const SQS_QUEUE_URL = "http://【外部からアクセスされる際のホスト名】:9324/000000000000/test";

const AWS = require('aws-sdk');
AWS.config.update({
//	region: 'ap-northeast-1'
	region: 'elasticmq'
});
const sqs = new AWS.SQS({
	apiVersion: '2012-11-05',
	endpoint: SQS_ENDPOINT
});

const params = {
  QueueUrl: SQS_QUEUE_URL,
  MessageBody: "This is test message.",
  MessageAttributes: {
  },
};

sqs.sendMessage(params, (err, data) => {
  if (err) {
    console.log("Error", err);
    return;
  }
  console.log("Success", data);
});
```

```
> aws sqs send-message --queue-url http://【外部からアクセスされる際のホスト名】:9324/000000000000/test --message-body "This is test message." --endpoint-url http://【外部からアクセスされる際のホスト名】:9324
```

●メッセージの受信

```js:sqs_receive_message.js
const SQS_ENDPOINT = "http://【外部からアクセスされる際のホスト名】:9324";
const SQS_QUEUE_URL = "http://【外部からアクセスされる際のホスト名】:9324/000000000000/test";

const AWS = require('aws-sdk');
AWS.config.update({
//	region: 'ap-northeast-1'
	region: 'elasticmq'
});

const sqs = new AWS.SQS({
	apiVersion: '2012-11-05',
	endpoint: SQS_ENDPOINT
});

const params = {
 QueueUrl: SQS_QUEUE_URL,
 MaxNumberOfMessages: 10,
 MessageAttributeNames: [
    "All"
 ],
 VisibilityTimeout: 30,
 WaitTimeSeconds: 20
};

sqs.receiveMessage(params, (err, data) => {
  if (err) {
    console.log("Receive Error", err);
    return;
  }
  console.log(JSON.stringify(data, null, "\t"));
});
```

```
> aws sqs receive-message --queue-url http://【外部からアクセスされる際のホスト名】:9324/000000000000/test --endpoint-url http://【外部からアクセスされる際のホスト名】:9324
```

●メッセージの削除

```js:sqs_delete_message.js
const SQS_ENDPOINT = "http://【外部からアクセスされる際のホスト名】:9324";
const SQS_QUEUE_URL = "http://【外部からアクセスされる際のホスト名】:9324/000000000000/test";
const SQS_RECEIPT_HANDLE = "【送信した後に取得されるMessageId】";

const AWS = require('aws-sdk');
AWS.config.update({
//	region: 'ap-northeast-1'
	region: 'elasticmq'
});
const sqs = new AWS.SQS({
	apiVersion: '2012-11-05',
	endpoint: SQS_ENDPOINT
});

const params = {
	QueueUrl: SQS_QUEUE_URL,
	ReceiptHandle: SQS_RECEIPT_HANDLE
};

sqs.deleteMessage(params, (err, data) => {
  if (err) {
    console.log("Error", err);
    return;
  }
  console.log("Success", data);
});
```

```
> aws sqs delete-message --queue-url http://【外部からアクセスされる際のホスト名】:9324/000000000000/test --receipt-handle 【送信した後に取得されるMessageId】--endpoint-url http://【外部からアクセスされる際のホスト名】:9324
```

# メッセージ受信時にLambdaと連携する

面倒なので、簡単に設定できるようにしておきました。
使い方だけ。

以下に実装済みです。

poruruba/amplify_template

https://github.com/poruruba/amplify_template

ZIPダウンロード後展開します。

```
> unzip amplify_template-master.zip
> cd amplify_template-master
> npm install
> npm install aws-sdk --save-dev
> vi .env
```

下記は、以降で作成するLambdaで共通で使われます。

```.env
ROUTING_SQS_ENDPOINT=http://【外部からアクセスされる際のホスト名】:9324
```

Lambdaのロジックは以下のように作成します。

```
> cd api/controllers
> mkdir test-sqs
> cd test-sqs
> vi sqs.json
```

```json:sqs.json
[
  {
    "enable": true,
    "QueueUrl": "http://【外部からアクセスされる際のホスト名】:9324/000000000000/【キュー名】"
  }
]
```

Lambdaのロジックを実装します。以下例です。

```
> vi index.js
```

```js:api/controllers/test-sqs/index.js
'use strict';

exports.handler = async (event, context, callback) => {
  console.log(JSON.stringify(event, null, '\t'));
  return {};
};
```

これで完了です。
以下のように実行することで、ローカル実行環境が起動します。

```
> node app.js
```

sqs_send_message.js で送信すると、受信されてLambdaのロジック(index.js)が起動するかと思います。

内部で以下のような仲介処理を入れることで実現しています。

```js:api/controllers/routing_sqs.js
'use strict';

const THIS_BASE_PATH = process.env.THIS_BASE_PATH;
const CONTROLLERS_BASE = THIS_BASE_PATH + '/api/controllers/';

//const SQS_REGION = "ap-northeast-1";
const SQS_REGION = "elasticmq";
const SQS_ENDPOINT = process.env.ROUTING_SQS_ENDPOINT;

const SQS_TARGET_FNAME = "sqs.json";
const DEFAULT_HANDLER = "handler";

const AWS = require('aws-sdk');
AWS.config.update({
	region: SQS_REGION
});
const sqs = new AWS.SQS({
	apiVersion: '2012-11-05',
	endpoint: SQS_ENDPOINT
});

const fs = require('fs');

function parse_sqs() {
  // sqs.jsonの検索
  const folders = fs.readdirSync(CONTROLLERS_BASE);
  folders.forEach(folder => {
    try {
      const stats_dir = fs.statSync(CONTROLLERS_BASE + folder);
      if (!stats_dir.isDirectory())
        return;
    
      const fname = CONTROLLERS_BASE + folder + "/" + SQS_TARGET_FNAME;
      if (!fs.existsSync(fname))
        return;
      const stats_file = fs.statSync(fname);
      if (!stats_file.isFile())
        return;

      // pollerの登録
      const defs = JSON.parse(fs.readFileSync(fname).toString());
      parse_sqs_json(defs, CONTROLLERS_BASE + folder, folder);
    } catch (error) {
      console.log(error);
    }
  });
}

function parse_sqs_json(defs, folder, folder_name) {
  defs.forEach(item => {
    if (!item.enable )
      return;
      
    const handler = item.handler || DEFAULT_HANDLER;
    const proc = require(folder)[handler];
    const params = {
      QueueUrl: item.QueueUrl,
      AttributeNames: item.AttributeNames,
      MaxNumberOfMessages: (item.MaxNumberOfMessages === undefined) ? 1 : item.MaxNumberOfMessages,
      MessageAttributeNames: item.MessageAttributeNames,
      VisibilityTimeout: (item.VisibilityTimeout === undefined) ? 30 : item.VisibilityTimeout,
      WaitTimeSeconds: (item.WaitTimeSeconds === undefined) ? 20: item.WaitTimeSeconds
    };

    continuousReceive(params, proc);

    console.log("sqs(" + item.QueueUrl + ") " + handler + ' ' + folder_name);
  });
}

async function continuousReceive(params, func){
	while(true){
		try{
			await new Promise((resolve, reject) =>{
				sqs.receiveMessage(params, async (err, data) => {
				  if (err) {
				    return reject(err);
				  }
				  if( !data.Messages ){
				  	return resolve('message empty');
				  }

          try{
			  		const event = {
			  			Records: data.Messages.map(item => {
                return {
                  messageId: item.MessageId,
                  receiptHandle: item.ReceiptHandle,
                  body :item.Body,
                  attributes: item.Attributes,
                  messageAttributes: item.MessageAttributes,
                  md5OfBody: item.MD5OfBody,
                  md5OfMessageAttributes: item.MD5OfMessageAttributes,
                  awsRegion: SQS_REGION
                }
              })
			  		};

            const success_return = (msg) => {
              if( data.Messages ){
                const deleteParams = {
                  QueueUrl: params.QueueUrl,
                  Entries: data.Messages.map(item => { return { Id: item.MessageId, ReceiptHandle: item.ReceiptHandle } } )
                };
                sqs.deleteMessageBatch(deleteParams, (err, data) => {
                  if (err) {
                    return reject(err);
                  } else {
                    return resolve();
                  }
                });
              }
            };
            const context = {
              succeed: success_return,
              fail: (error) => resolve(error)
            };
            const task = func(event, context, (error, response) => {
              if (error) return resolve(error);
              return success_return(response);
            });
            if (task instanceof Promise || (task && typeof task.then === 'function')) {
              return task.then(ret => {
                if (ret)
                  return success_return(ret);
              })
              .catch(err => resolve(err));
            }else{
              if( task !== undefined )
                success_return(task);
            }
          }catch(error){
            console.log(error);
          }
				});
			});
		}catch(error){
			console.log(error);
			return;
		}
	}
}

module.exports = parse_sqs();
```

以上