コンテンツにスキップ
machbase-neo
MQTT API
MQTTの検索

MQTTの検索

MQTTでデータベースクエリを実行するには、db/queryトピックにリクエストを送信します。サーバーは、db/replyトピック、またはリクエストのreplyフィールドで指定したトピックに結果を返します。

クエリJSON

パラメーター既定値説明
qn/a実行するSQLクエリ文字列
p省略可能。SQLプレースホルダーに渡すバインドパラメーターです。
- ?の位置パラメーター（JSON配列）：["name", 1234] Since v8.0.75
- :nameの名前付きパラメーター（JSONオブジェクト）：{"name": "wave.sin", "n": 5} Since v8.7.0
db省略可能。複数データベース環境で使用する対象データベース名です。 Since v8.7.0
replydb/replyクエリ結果を受信するトピック
formatjson結果形式：json、csv、box
timeformatns時刻の単位：s、ms、us、ns
tzUTCタイムゾーン：UTC、Local、地域指定
compress圧縮なし圧縮方式：gzip
rownumfalse行番号を含めるかどうか：true、false

format=jsonで使用できる追加パラメーター Since v8.0.12

以下のオプションは、format=jsonの場合にのみ使用できます。

パラメーター既定値説明
transposefalse行配列の代わりに、列配列（cols）を生成します。
rowsFlattenfalseJSONオブジェクトのrowsフィールドの配列次元を1つ減らします。
rowsArrayfalse各レコードをオブジェクトとする配列のJSONを生成します。

format=csvで使用できるパラメーター

パラメーター既定値説明
headerskipを指定するとヘッダーを含めません。
precision-1浮動小数点数の桁数：-1は丸めなし、0は整数

基本例では、クライアントがdb/reply/#を購読し、replyにdb/reply/my_queryを指定してdb/queryトピックにクエリを発行します。これにより、複数のメッセージから自身への応答を識別できます。バインドプレースホルダーを使用する場合は、pフィールドに?用のJSON配列を指定します。

{
    "q": "select name,time,value from example where name = ? limit ?",
    "p": ["wave.sin", 5],
    "format": "csv",
    "reply": "db/reply/my_query"
}

または、:nameの名前付きパラメーター用のJSONオブジェクトを指定します。 Since v8.7.0

{
    "q": "select name,time,value from example where name = :name limit :n",
    "p": {"name": "wave.sin", "n": 5},
    "format": "csv",
    "reply": "db/reply/my_query"
}
MQTTでクエリを実行して応答を受信する例（MQTTX.appを使用）

MQTTでクエリを実行して応答を受信する例（MQTTX.appを使用）

クライアントの例

JSHアプリ

Since v8.5.0

この例では、応答トピックを購読し、SQLクエリを発行して、MQTTで結果を受信する手順を説明します。

  1. 応答トピックの購読
    まず、db/reply/my_queryのような結果を受信するトピックを購読します。

  2. SQLクエリの発行
    SQLクエリ（q）、結果形式（format）、応答トピック（reply）を含むメッセージを、db/queryトピックに発行します。

  3. 応答の受信と処理
    サーバーはクエリを処理し、指定した応答トピックに結果を送信します。クライアントはメッセージを受信して結果を出力します。

JSHのサンプルコード：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

const process = require("process");
const mqtt = require("mqtt");

const topicReply = "db/reply/my_query";
const topicQuery = "db/query";
const queryRequest = {
    q: `select name,time,value from example limit 5`,
    format: 'csv',
    reply: topicReply,
};

var client = new mqtt.Client({
    servers: ["tcp://127.0.0.1:5653"],
    keepAlive: 10,
});
client.on('open', () => {
    console.println('---- subscribe:', topicReply);
    client.subscribe(topicReply, {qos:0})
});
client.on('error', (err) => {
    console.println('MQTT ERROR:', err.message);
});
client.on('close', () => {
    console.println('---- disconnected');
});
client.on('subscribed', (topic, reason) => {
    console.println('---- publish:', topicQuery);
    client.publish(topicQuery, JSON.stringify(queryRequest));
});
client.on('message', (msg) => {
    console.println('---- reply')
    console.println(msg.payload);
    client.unsubscribe(msg.topic);
});
client.on('unsubscribed', (topic, reason) => {
    console.println('---- unsubscribed:', topic, 'reason:', reason);
    setTimeout(()=>{
        client.close();
    }, 500)
});

実行と結果出力：

/work > ./mqtt_query.js
---- subscribe: db/reply/my_query ----
---- publish: db/query ----
---- reply ----
name,time,value
my-car,1782260468085501458,1.2345
my-car,1782260474814668541,1.35795
my-car,1782260474827077041,1.4814
my-car,1782260474839257291,1.60485

---- unsubscribed: db/reply/my_query reason: 0 ----
---- disconnected ----

Node.js クライアント 

npm install mqtt --save
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

const mqtt = require("mqtt");

const client = mqtt.connect("mqtt://127.0.0.1:5653", {
    clean: true,
    connectTimeout: 3000,
    autoUseTopicAlias: true,
    protocolVersion: 5,
});

const sqlText = "SELECT time,value FROM example "+
    "where name = 'neo_cpu.percent' limit 3";

client.on("connect", () => {
    client.subscribe("db/reply/#", (err) => {
        if (!err) {
            const req = {
                q: sqlText,
                format: "box",
                precision: 1,
                timeformat: "15:04:05",
            };
            client.publish("db/query", JSON.stringify(req));
        }
    });
});

client.on("message", (topic, message) => {
    console.log(message.toString());
    client.end();
});
$ node main.js

+----------+-------+
| TIME     | VALUE |
+----------+-------+
| 05:46:19 | 69.4  |
| 05:46:22 | 26.4  |
| 05:46:25 | 42.8  |
+----------+-------+

Go クライアント

応答の構造定義

type Result struct {
	Success bool       `json:"success"`
	Reason  string     `json:"reason"`
	Elapse  string     `json:"elapse"`
	Data    ResultData `json:"data"`
}

type ResultData struct {
	Columns []string `json:"columns"`
	Types   []string `json:"types"`
	Rows    [][]any  `json:"rows"`
}

db/replyの購読

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

client.Subscribe("db/reply", 1, func(_ paho.Client, msg paho.Message) {
    buff := msg.Payload()
    result := Result{}
    if err := json.Unmarshal(buff, &result); err != nil {
        panic(err)
    }
    if !result.Success {
        fmt.Println("RECV: query failed:", result.Reason)
        return
    }
    if len(result.Data.Rows) == 0 {
        fmt.Println("Empty result")
        return
    }
    for i, rec := range result.Data.Rows {
        // 各レコードに必要な処理を実行します。
        name := rec[0].(string)
        ts := time.Unix(0, int64(rec[1].(float64)))
        value := float64(rec[2].(float64))
        fmt.Println(i+1, name, ts, value)
    }
})

‘db/query’への発行

jsonStr := `{ "q": "select * from EXAMPLE order by time desc limit 5" }`
client.Publish("db/query", 1, false, []byte(jsonStr))
最終更新日
MQTT v3.1のデータ書き込み