MQTTでPub/Subしてみる

# MQTTって何？
MQ Telemetry Transport。メッセージキュープロトコルの１つ。
標準化されたメッセージキュープロトコルには、
AMQP(Advanced Message Queuing Protocol)、MQTT、STOMP(Streaming Text Oriented Messaging Protocol)などがありますが、MQTTの特徴を挙げると

* 軽い：プロトコル電文仕様が軽量で且つシンプル
* Pub/Sub型：データ配信モデルに特化、Queueモデルはサポートしない。そのためJob Queueモデルを利用した負荷分散には向かない
* QoS(Quality of Service)の指定ができる：クライアントとBroker間の通信品質を３つのQoSとしてパラメータ指定可能
 * QoS0：最高１回。最も軽い。メッセージが確実に届く保証はない、メッセージ配布に失敗しても再送をしない
 * QoS1：最低１回。必ずメッセージ配布するが、重複する可能性がある
 * QoS2：正確に１回。最も重い。必ずメッセージを配布して、重複も発生しない
* ネットワークが不安定な場所で動作する：→ QoS1とQoS2のBrokerとクライアント間の一定のメッセージ到達保証で担保の模様。HeatBeat時間を設定することでもブツブツ切れる端末もカバー

# 試してみる
ここでは、簡単なサンプルとして、JavaプロセスからブラウザへPub/Subを試してみます

## 環境
* OS：Windows7 32bit

## broker実装
* Mosquittoでも良かったんですが、Apache Apollo を利用します。 MQTTのだけではなくSTOMP/AMQP1.0もサポートしており、ちょっと試したかったので。
* http://activemq.apache.org/apollo/

## Publisher側コード
* Java ＋ MQTTクライアントとしてPaho（http://www.eclipse.org/paho/）を利用
* ここでは、/testtopic へ 「This is Test Message」をPublishしています

```java:TestPublish.java
// ClientIDは接続単位でユニークにする必要があるのでここではUUID利用
String clientId = "JavaClient" + UUID.randomUUID().toString();
MqttClient sampleClient = new MqttClient("tcp://localhost:61613", clientId, new MemoryPersistence());
MqttConnectOptions connOpts = new MqttConnectOptions();
connOpts.setUserName("admin");
connOpts.setPassword("password".toCharArray());
connOpts.setCleanSession(true);
sampleClient.connect(connOpts);
// Publishするメッセージ
String messageContent = "This is Test Message";
MqttMessage message = new MqttMessage(messageContent.getBytes());
// QoSは数値で指定
message.setQos(1);
sampleClient.publish("/testtopic", message);
sampleClient.disconnect();
```

## Subscriber側（Browser側JavaScript）コード
* ブラウザでMQTTを使う場合は、MQTT over WebSocketとなる。
* MQTTクライアントJavascriptライブラリとして同じくPahoを利用。メッセージをCallbackで受けてConsoleへ出力

```js
var clientId = "JavaScriptClient-"+(Math.floor(Math.random() * 100000));
client = new Paho.MQTT.Client('localhost', Number(61623), clientId);

// Callback設定
client.onConnectionLost = onConnectionLost;
client.onMessageArrived = onMessageArrived;

// 接続
client.connect({
	onSuccess:onConnect,
    onFailure:onFailure,
	userName:'admin', 
    password:'password' 
});

function onConnect() {
  // subscribeするTopics設定
  client.subscribe("/testtopic");
}

function onFailure(failure) {
  console.log("failure:"+failure.errorMessage);
}

// コネクション切断Callback
function onConnectionLost(responseObject) {
  if (responseObject.errorCode !== 0) {
    console.log("onConnectionLost:"+responseObject.errorMessage);
  }
}

// messeage到達Callback
function onMessageArrived(message) {
  console.log("onMessageArrived:"+message.payloadString);
}
```

## 実行してみる

* まずBrokerを実行しておく
``` 
C:\apps\ApolloBroker\mybroker\bin\apollo-broker.cmd run
```
## Subscriber

* Pub/SubではSubscriberをbrokerへ接続しておく必要があるので、前述のJavaScriptコードを実行しBrokerへ接続しておきます。
* ここではChromeとFirefoxの2プロセスから接続してみましょう
* 接続後Apolloの管理コンソール (http://127.0.0.1:61680/) から下図のようにConnector > ws で2つの接続が確認できます。
![foo.png](https://qiita-image-store.s3.amazonaws.com/0/82993/33d9a95b-672c-3fc8-6fa8-bc7fcf684d2e.png)
* また、Topicsを見るとSubscriber(Consumer)も2接続あるのも確認できます。
![Consumer.png](https://qiita-image-store.s3.amazonaws.com/0/82993/33d4e225-8e5c-5bd9-9f4a-b19e53a67b56.png)
* JavaプログラムでPublishを実行するとSubscriber側のブラウザのコンソールへメッセージが送信されます
![callback.png](https://qiita-image-store.s3.amazonaws.com/0/82993/89e3bdfa-4a57-aa96-431b-b43e9af9d4ff.png)

# 簡単なサンプル実行でしたが。。
今回全てローカル環境での簡単なサンプル実行でしたが、MQTTを利用することで、簡単に異プラットフォーム・異言語間で通信できることが確認できました。
MQTTクライアントはNode.jsやC#やPython・Androidと多数の言語・プラットフォームでサポートされているので、これらのプロセス・デバイス間で簡単にPub/Sub可能となりなかなか面白いですね。
Facebook MessengerのPushでも利用されていたり、Push基盤のプロトコルとしても利用できそうです。
後々、他のBroker実装やQoSの電文保証も検証してみようと思います。
AWSでもマネージドサービスとしてMQTTサービス（AWS IoT Message Broker）を展開するようなのでこちらも試してみたい！

# 参考情報

* MQTT V3.1 仕様 http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html
* MQTT Broker実装
 * Mosquitto http://mosquitto.org/
 * Apache Apollo   http://activemq.apache.org/apollo/
 * RabbitMQ http://www.rabbitmq.com/mqtt.html
* MQTT Client実装
 * Paho - (Android, C/C++, Java, JavaScript, Python, Go, .Net)  http://www.eclipse.org/paho/ 
 * M2Mqtt - MQTT Client Library for .Net.  https://www.nuget.org/packages/M2Mqtt/
 * Moscapsule - MQTT Client for iOS written in Swift. https://github.com/flightonary/Moscapsule