DynamoDB Streamのデータをsocket.ioを使ってブラウザに流す

## やること

- TwitterのストリームデータをKinesisを経由してDynamoDBに保存します。
- DynamoDB Streamにデータを流し、ブラウザでストリームデータを可視化する

## 構成図
![構成図 (1).png](https://qiita-image-store.s3.amazonaws.com/0/65478/a1a66bb7-e067-9196-753b-1c697d00dec5.png)


## Twitter Stream → Kinesis → DynamoDB
[AWS Lambda Reference Architecture: Real-time Stream Processing](https://github.com/awslabs/lambda-refarch-streamprocessing)
awslabsのアカウントにそのままやりたいことがありました。動かすとKinesisにデータを流して、キャプチャの通りDynamoDBにデータを保存してくれます。

<img width="735" alt="スクリーンショット 2016-03-28 22.05.39.png" src="https://qiita-image-store.s3.amazonaws.com/0/65478/392eb168-6334-2173-97d3-8fd73f8034f1.png">

## DynamoDB → DynamoDB Stream → Lambda → SNS http Notification
### DynamoDB Streamの有効化
キャプチャの箇所からDynamoDB Streamを有効化します
<img width="404" alt="スクリーンショット 2016-03-28 22.08.44.png" src="https://qiita-image-store.s3.amazonaws.com/0/65478/845f8f70-dea3-d9cd-5a30-4c7131e156e3.png">

トリガーを有効化して、DynamoDB Streamから流れてきたデータを処理するLambdaファンクションを指定します
<img width="557" alt="スクリーンショット 2016-03-28 22.09.29.png" src="https://qiita-image-store.s3.amazonaws.com/0/65478/15baf356-a330-c717-2652-e15c1fcc0f55.png">

## DynamoDBトリガーLambdaファンクション
以下のLambdaファンクションでDynamoDB Streamのデータを処理します。
受け取ったデータをSNSにpublishします。

```js:dynamodbTrigger
var aws = require('aws-sdk');

var sns = new aws.SNS();

exports.handler = function(event, context) {
    event.Records.forEach(function(record) {
        console.log('DynamoDB Record: %j', record.dynamodb);
        sns.publish({
        Message: JSON.stringify(record.dynamodb),
        Subject: 'message title ',
        TopicArn: '<SNS TopicのARN>'
        }, function(err, data) {
            if (err) {
                console.log(err.stack);
                context.done(err, 'Errors!');  
                return;
            }

            console.log(data);

            // Notify Lambda that we are finished
            context.done(null, 'push sent');  
        });
    });
};
```
## SNS http Notification → Socket.io → ブラウザ
SNS http Notificationを設定します。そしてデータをSocket.ioで受け取った後、ブラウザにデータを流します。
<img width="465" alt="スクリーンショット 2016-03-28 22.18.22.png" src="https://qiita-image-store.s3.amazonaws.com/0/65478/f21dc1bd-ee19-9ca6-b83a-dcfed7c2fabf.png">

### Socket.ioサーバ
[[Node.js]Amazon SNSでHTTPを使って通知を受け取る[aws-sdk-js]](http://dev.classmethod.jp/cloud/amazonsns-http/)
を参考にSNS http NotificationのエンドポイントとなるSocket.ioサーバを立てます
ソースコードは以下のとおり

```js:app.js
var http = require('http');
var url = require("url");
var async = require('async');
var AWS = require('aws-sdk');
 
AWS.config.loadFromPath('./config/aws.json');
var sns = new AWS.SNS();
 
var httpServer = http.createServer(handler);
var io = require('socket.io').listen(httpServer);
var fs = require('fs');
//asyncを使って順番に処理を実行
async.series([
    //HTTPサーバ起動
    function (callback) {
        httpServer.listen(3000);
        callback(null, 1);
    },
    //subscrive開始
    function (callback) {
        initSubscriber(callback);
        callback(null, 2);
    }
], function (err, results) {
    if (err) {
        throw err;
    }
});
 
function handler(req, res) {
    var path = url.parse(req.url).pathname;
 
    if (path === "/httpsns") {
        var body = '';
        req.on('data', function (data) {
            body += data;
        });
        req.on('end', function () {
            res.writeHead(200, {
                'Content-Type':'text/html'
            });
 
            var obj = JSON.parse(body);
            if (obj.Type === "SubscriptionConfirmation") {
                sns.confirmSubscription({ TopicArn:obj.TopicArn, Token:obj.Token}, function (err, data) {
                    console.log("confirmSubscription");
                });
            } else if (obj.Type === "Notification" && obj.Message !== undefined) {
                console.log(obj.Subject + ":" + obj.Message);
    			io.sockets.emit('msg', obj.Message);
            }
            res.end('OK');
        });
    } else {
        res.writeHead(200, {'Content-Type':'text/html'});
        res.end(fs.readFileSync('index.html'));
    }
}
 
function initSubscriber(callback) {
    var args = {
        TopicArn:"<SNSトピックのARN>",
        Protocol:'http',
        Endpoint:"http://<host>:3000/httpsns"
    };
 
    sns.subscribe(args, function (err, data) {
        console.log("subscribe start.");        
    });
}
```

以下のコマンドでサーバを起動します

```
$ node app.js
```

### Socket.ioクライアント
```html:index.html
<!DOCTYPE html>
<meta charset="UTF-8">
<title>chat</title>
<script src="/socket.io/socket.io.js"></script>
<script src="//ajax.googleapis.com/ajax/libs/jquery/1.10.2/jquery.min.js"></script>
<script>
$(function() {
  var socket = io.connect();

  socket.on('msg', function(data) {
        console.log(data);
    $('div').prepend(data + '<br>');

  });
});
</script>
<h2>ツイート情報</h2></h2><div></div>
```

## 動作確認
http://＜host＞:3000 にブラウザでアクセスすると以下のとおりデータがリアルタイムに流れてきます。
[![DynamoDB Streamのデータをsocket.io経由でブラウザに流す](https://i.ytimg.com/vi/F4QWf0I7Jfs/hqdefault.jpg)](https://www.youtube.com/watch?v=F4QWf0I7Jfs)