This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 9766836 [IOTDB-1319] Trigger module: sink utility (#3038)
9766836 is described below
commit 9766836b4db82fe56c041c32fcd837017de22114
Author: Steve Yurong Su <[email protected]>
AuthorDate: Thu Apr 22 11:30:28 2021 +0800
[IOTDB-1319] Trigger module: sink utility (#3038)
---
docs/UserGuide/Advanced-Features/Triggers.md | 278 +++++++++++++++++++++
docs/zh/UserGuide/Advanced-Features/Triggers.md | 278 +++++++++++++++++++++
example/pom.xml | 1 +
example/trigger/pom.xml | 124 +++++++++
.../org/apache/iotdb/trigger/TriggerExample.java | 127 ++++++++++
server/pom.xml | 4 +
.../resources/conf/iotdb-engine.properties | 6 +-
.../Evaluator.java => sink/api/Configuration.java} | 14 +-
.../api/Evaluator.java => sink/api/Event.java} | 14 +-
.../api/Evaluator.java => sink/api/Handler.java} | 16 +-
.../exception/SinkException.java} | 15 +-
.../db/sink/local/LocalIoTDBConfiguration.java | 51 ++++
.../local/LocalIoTDBEvent.java} | 23 +-
.../iotdb/db/sink/local/LocalIoTDBHandler.java | 104 ++++++++
.../iotdb/db/sink/mqtt/MQTTConfiguration.java | 74 ++++++
.../mqtt/MQTTEvent.java} | 48 ++--
.../org/apache/iotdb/db/sink/mqtt/MQTTHandler.java | 85 +++++++
.../iotdb/db/utils/windowing/api/Evaluator.java | 3 +-
.../windowing/runtime/WindowEvaluationTask.java | 2 +-
.../apache/iotdb/db/sink/LocalIoTDBSinkTest.java | 217 ++++++++++++++++
.../org/apache/iotdb/db/sink/MQTTSinkTest.java | 204 +++++++++++++++
21 files changed, 1621 insertions(+), 67 deletions(-)
diff --git a/docs/UserGuide/Advanced-Features/Triggers.md
b/docs/UserGuide/Advanced-Features/Triggers.md
index a4f4df8..2c63cbb 100644
--- a/docs/UserGuide/Advanced-Features/Triggers.md
+++ b/docs/UserGuide/Advanced-Features/Triggers.md
@@ -489,6 +489,284 @@ The maximum number of window evaluation tasks that can be
pending for execution.
+### Sink Utility
+
+The sink utility provides the ability for triggers to connect to external
systems.
+
+It provides a programming paradigm. Each sink utility contains a `Handler` for
processing data sending, a `Configuration` for configuring `Handler`, and an
`Event` for describing the sending data.
+
+
+
+#### LocalIoTDBSink
+
+`LocalIoTDBSink` is used to insert data points to the local sequence.
+
+Before writing data, it is not required that the time series have been created.
+
+**Note**, in the scenario used for triggers, the listening time series and the
written target time series should not be in the same storage group.
+
+Example:
+
+```java
+final String device = "root.alerting";
+final String[] measurements = new String[] {"local"};
+final TSDataType[] dataTypes = new TSDataType[] {TSDataType.DOUBLE};
+
+LocalIoTDBHandler localIoTDBHandler = new LocalIoTDBHandler();
+localIoTDBHandler.open(new LocalIoTDBConfiguration(device, measurements,
dataTypes));
+
+// insert 100 data points
+for (int i = 0; i < 100; ++i) {
+ final long timestamp = i;
+ final double value = i;
+ localIoTDBHandler.onEvent(new LocalIoTDBEvent(timestamp, value));
+}
+```
+
+Note that when you need to insert data points to a time series of type `TEXT`,
you need to use `org.apache.iotdb.tsfile.utils.Binary`:
+
+```java
+// insert 100 data points
+for (int i = 0; i < 100; ++i) {
+ final long timestamp = i;
+ final String value = "" + i;
+ localIoTDBHandler.onEvent(new LocalIoTDBEvent(timestamp,
Binary.valueOf(value)));
+}
+```
+
+
+
+#### MQTTSink
+
+In triggers, you can use `MQTTSink` to send data points to other IoTDB
instances.
+
+Before sending data, it is not required that the time series have been created.
+
+Example:
+
+```java
+final String host = "127.0.0.1";
+final int port = 1883;
+final String username = "root";
+final String password = "root";
+final PartialPath device = new PartialPath("root.alerting");
+final String[] measurements = new String[] {"remote"};
+
+MQTTHandler mqttHandler = new MQTTHandler();
+mqttHandler.open(new MQTTConfiguration(host, port, username, password, device,
measurements));
+
+final String topic = "test";
+final QoS qos = QoS.EXACTLY_ONCE;
+final boolean retain = false;
+// send 100 data points
+for (int i = 0; i < 100; ++i) {
+ final long timestamp = i;
+ final double value = i;
+ mqttHandler.onEvent(new MQTTEvent(topic, qos, retain, timestamp, value));
+}
+```
+
+
+
+## Maven Project Example
+
+If you use [Maven](http://search.maven.org/), you can refer to our sample
project **trigger-example**.
+
+You can find it
[here](https://github.com/apache/iotdb/tree/master/example/trigger).
+
+It shows:
+
+* How to use Maven to manage your trigger project
+* How to listen to data changes based on the user programming interface
+* How to use the windowing utility
+* How to use the sink utility
+
+```java
+package org.apache.iotdb.trigger;
+
+import org.apache.iotdb.db.engine.trigger.api.Trigger;
+import org.apache.iotdb.db.engine.trigger.api.TriggerAttributes;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.sink.mqtt.MQTTConfiguration;
+import org.apache.iotdb.db.sink.mqtt.MQTTEvent;
+import org.apache.iotdb.db.sink.mqtt.MQTTHandler;
+import org.apache.iotdb.db.sink.local.LocalIoTDBConfiguration;
+import org.apache.iotdb.db.sink.local.LocalIoTDBEvent;
+import org.apache.iotdb.db.sink.local.LocalIoTDBHandler;
+import
org.apache.iotdb.db.utils.windowing.configuration.SlidingSizeWindowConfiguration;
+import
org.apache.iotdb.db.utils.windowing.handler.SlidingSizeWindowEvaluationHandler;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+
+import org.fusesource.mqtt.client.QoS;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TriggerExample implements Trigger {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(TriggerExample.class);
+
+ private static final String TARGET_DEVICE = "root.alerting";
+
+ private final LocalIoTDBHandler localIoTDBHandler = new LocalIoTDBHandler();
+ private final MQTTHandler mqttHandler = new MQTTHandler();
+
+ private SlidingSizeWindowEvaluationHandler windowEvaluationHandler;
+
+ @Override
+ public void onCreate(TriggerAttributes attributes) throws Exception {
+ LOGGER.info("onCreate(TriggerAttributes attributes)");
+
+ double lo = attributes.getDouble("lo");
+ double hi = attributes.getDouble("hi");
+
+ openSinkHandlers();
+
+ windowEvaluationHandler =
+ new SlidingSizeWindowEvaluationHandler(
+ new SlidingSizeWindowConfiguration(TSDataType.DOUBLE, 5, 5),
+ window -> {
+ double avg = 0;
+ for (int i = 0; i < window.size(); ++i) {
+ avg += window.getDouble(i);
+ }
+ avg /= window.size();
+
+ if (avg < lo || hi < avg) {
+ localIoTDBHandler.onEvent(new
LocalIoTDBEvent(window.getTime(0), avg));
+ mqttHandler.onEvent(
+ new MQTTEvent("test", QoS.EXACTLY_ONCE, false,
window.getTime(0), avg));
+ }
+ });
+ }
+
+ @Override
+ public void onDrop() throws Exception {
+ LOGGER.info("onDrop()");
+ closeSinkHandlers();
+ }
+
+ @Override
+ public void onStart() throws Exception {
+ LOGGER.info("onStart()");
+ openSinkHandlers();
+ }
+
+ @Override
+ public void onStop() throws Exception {
+ LOGGER.info("onStop()");
+ closeSinkHandlers();
+ }
+
+ @Override
+ public Double fire(long timestamp, Double value) {
+ windowEvaluationHandler.collect(timestamp, value);
+ return value;
+ }
+
+ @Override
+ public double[] fire(long[] timestamps, double[] values) {
+ for (int i = 0; i < timestamps.length; ++i) {
+ windowEvaluationHandler.collect(timestamps[i], values[i]);
+ }
+ return values;
+ }
+
+ private void openSinkHandlers() throws Exception {
+ localIoTDBHandler.open(
+ new LocalIoTDBConfiguration(
+ TARGET_DEVICE, new String[] {"local"}, new TSDataType[]
{TSDataType.DOUBLE}));
+ mqttHandler.open(
+ new MQTTConfiguration(
+ "127.0.0.1",
+ 1883,
+ "root",
+ "root",
+ new PartialPath(TARGET_DEVICE),
+ new String[] {"remote"}));
+ }
+
+ private void closeSinkHandlers() throws Exception {
+ localIoTDBHandler.close();
+ mqttHandler.close();
+ }
+}
+```
+
+You can try this trigger by following the steps below:
+
+* Enable MQTT service by modifying `iotdb-engine.properties`
+
+ ``` properties
+ # whether to enable the mqtt service.
+ enable_mqtt_service=true
+ ```
+
+* Start the IoTDB server
+
+* Create time series via cli
+
+ ``` sql
+ CREATE TIMESERIES root.sg1.d1.s1 WITH DATATYPE=DOUBLE, ENCODING=PLAIN;
+ ```
+
+* Place the JAR (`trigger-example-0.12.0-SNAPSHOT.jar`) of **trigger-example**
in the directory `iotdb-server-0.12.0-SNAPSHOT/ext/trigger` (or in a
subdirectory of `iotdb-server-0.12.0-SNAPSHOT/ext/trigger`)
+
+ > You can specify the root path to load the trigger JAR package by modifying
the `trigger_root_dir` in the configuration file.
+
+* Use the SQL statement to register the trigger, assuming that the name given
to the trigger is `window-avg-alerter`
+
+* Use the `CREATE TRIGGER` statement to register the trigger via cli
+
+ ```sql
+ CREATE TRIGGER window-avg-alerter
+ AFTER INSERT
+ ON root.sg1.d1.s1
+ AS "org.apache.iotdb.trigger.TriggerExample"
+ WITH (
+ "lo" = "0",
+ "hi" = "10.0"
+ )
+ ```
+
+* Use cli to insert test data
+
+ ``` sql
+ INSERT INTO root.sg1.d1(timestamp, s1) VALUES (1, 0);
+ INSERT INTO root.sg1.d1(timestamp, s1) VALUES (2, 2);
+ INSERT INTO root.sg1.d1(timestamp, s1) VALUES (3, 4);
+ INSERT INTO root.sg1.d1(timestamp, s1) VALUES (4, 6);
+ INSERT INTO root.sg1.d1(timestamp, s1) VALUES (5, 8);
+
+ INSERT INTO root.sg1.d1(timestamp, s1) VALUES (6, 10);
+ INSERT INTO root.sg1.d1(timestamp, s1) VALUES (7, 12);
+ INSERT INTO root.sg1.d1(timestamp, s1) VALUES (8, 14);
+ INSERT INTO root.sg1.d1(timestamp, s1) VALUES (9, 16);
+ INSERT INTO root.sg1.d1(timestamp, s1) VALUES (10, 18);
+ ```
+
+* Use cli to query data to verify the behavior of the trigger
+
+ ``` sql
+ SELECT * FROM root.alerting;
+ ```
+
+* Under normal circumstances, the following results should be shown
+
+ ``` sql
+ IoTDB> SELECT * FROM root.alerting;
+ +-----------------------------+--------------------+-------------------+
+ | Time|root.alerting.remote|root.alerting.local|
+ +-----------------------------+--------------------+-------------------+
+ |1970-01-01T08:00:00.006+08:00| 14.0| 14.0|
+ +-----------------------------+--------------------+-------------------+
+ Total line number = 1
+ It costs 0.006s
+ ```
+
+That's all, please enjoy it :D
+
+
+
## Important Notes
* The trigger is implemented based on the reflection mechanism. Triggers can
be dynamically registered and dropped without restarting the server.
diff --git a/docs/zh/UserGuide/Advanced-Features/Triggers.md
b/docs/zh/UserGuide/Advanced-Features/Triggers.md
index 3c4ac28..b73ab9e 100644
--- a/docs/zh/UserGuide/Advanced-Features/Triggers.md
+++ b/docs/zh/UserGuide/Advanced-Features/Triggers.md
@@ -506,6 +506,284 @@ SlidingTimeWindowEvaluationHandler handler =
+### Sink工具类
+
+Sink工具类为触发器提供了连接外部系统的能力。
+
+它提供了一套编程范式。每一个Sink工具都包含一个用于处理数据发送的`Handler`、一个用于配置`Handler`的`Configuration`，还有一个用于描述发送数据的`Event`。
+
+
+
+#### LocalIoTDBSink
+
+`LocalIoTDBSink`用于向本地序列写入数据点。
+
+在写入数据前，不要求时间序列已被创建。
+
+**注意**，在触发器场景中，侦听的时间序列和写入的目标时间序列不要在同一个存储组下。
+
+使用示例：
+
+```java
+final String device = "root.alerting";
+final String[] measurements = new String[] {"local"};
+final TSDataType[] dataTypes = new TSDataType[] {TSDataType.DOUBLE};
+
+LocalIoTDBHandler localIoTDBHandler = new LocalIoTDBHandler();
+localIoTDBHandler.open(new LocalIoTDBConfiguration(device, measurements,
dataTypes));
+
+// insert 100 data points
+for (int i = 0; i < 100; ++i) {
+ final long timestamp = i;
+ final double value = i;
+ localIoTDBHandler.onEvent(new LocalIoTDBEvent(timestamp, value));
+}
+```
+
+注意，当您需要向某个`TEXT`类型的序列写入数据时，您需要借助`org.apache.iotdb.tsfile.utils.Binary`：
+
+```java
+// insert 100 data points
+for (int i = 0; i < 100; ++i) {
+ final long timestamp = i;
+ final String value = "" + i;
+ localIoTDBHandler.onEvent(new LocalIoTDBEvent(timestamp,
Binary.valueOf(value)));
+}
+```
+
+
+
+#### MQTTSink
+
+触发器可以使用`MQTTSink`向其他的 IoTDB 实例发送数据点。
+
+在发送数据前，不要求时间序列已被创建。
+
+使用示例：
+
+```java
+final String host = "127.0.0.1";
+final int port = 1883;
+final String username = "root";
+final String password = "root";
+final PartialPath device = new PartialPath("root.alerting");
+final String[] measurements = new String[] {"remote"};
+
+MQTTHandler mqttHandler = new MQTTHandler();
+mqttHandler.open(new MQTTConfiguration(host, port, username, password, device,
measurements));
+
+final String topic = "test";
+final QoS qos = QoS.EXACTLY_ONCE;
+final boolean retain = false;
+// send 100 data points
+for (int i = 0; i < 100; ++i) {
+ final long timestamp = i;
+ final double value = i;
+ mqttHandler.onEvent(new MQTTEvent(topic, qos, retain, timestamp, value));
+}
+```
+
+
+
+## 完整的Maven示例项目
+
+如果您使用[Maven](http://search.maven.org/)，可以参考我们编写的示例项目**trigger-example**。
+
+您可以在[这里](https://github.com/apache/iotdb/tree/master/example/trigger)找到它。
+
+它展示了：
+
+* 如何使用Maven管理您的trigger项目
+* 如何基于触发器的用户编程接口实现数据侦听
+* 如何使用窗口工具类
+* 如何使用Sink工具类
+
+```java
+package org.apache.iotdb.trigger;
+
+import org.apache.iotdb.db.engine.trigger.api.Trigger;
+import org.apache.iotdb.db.engine.trigger.api.TriggerAttributes;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.sink.mqtt.MQTTConfiguration;
+import org.apache.iotdb.db.sink.mqtt.MQTTEvent;
+import org.apache.iotdb.db.sink.mqtt.MQTTHandler;
+import org.apache.iotdb.db.sink.local.LocalIoTDBConfiguration;
+import org.apache.iotdb.db.sink.local.LocalIoTDBEvent;
+import org.apache.iotdb.db.sink.local.LocalIoTDBHandler;
+import
org.apache.iotdb.db.utils.windowing.configuration.SlidingSizeWindowConfiguration;
+import
org.apache.iotdb.db.utils.windowing.handler.SlidingSizeWindowEvaluationHandler;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+
+import org.fusesource.mqtt.client.QoS;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TriggerExample implements Trigger {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(TriggerExample.class);
+
+ private static final String TARGET_DEVICE = "root.alerting";
+
+ private final LocalIoTDBHandler localIoTDBHandler = new LocalIoTDBHandler();
+ private final MQTTHandler mqttHandler = new MQTTHandler();
+
+ private SlidingSizeWindowEvaluationHandler windowEvaluationHandler;
+
+ @Override
+ public void onCreate(TriggerAttributes attributes) throws Exception {
+ LOGGER.info("onCreate(TriggerAttributes attributes)");
+
+ double lo = attributes.getDouble("lo");
+ double hi = attributes.getDouble("hi");
+
+ openSinkHandlers();
+
+ windowEvaluationHandler =
+ new SlidingSizeWindowEvaluationHandler(
+ new SlidingSizeWindowConfiguration(TSDataType.DOUBLE, 5, 5),
+ window -> {
+ double avg = 0;
+ for (int i = 0; i < window.size(); ++i) {
+ avg += window.getDouble(i);
+ }
+ avg /= window.size();
+
+ if (avg < lo || hi < avg) {
+ localIoTDBHandler.onEvent(new
LocalIoTDBEvent(window.getTime(0), avg));
+ mqttHandler.onEvent(
+ new MQTTEvent("test", QoS.EXACTLY_ONCE, false,
window.getTime(0), avg));
+ }
+ });
+ }
+
+ @Override
+ public void onDrop() throws Exception {
+ LOGGER.info("onDrop()");
+ closeSinkHandlers();
+ }
+
+ @Override
+ public void onStart() throws Exception {
+ LOGGER.info("onStart()");
+ openSinkHandlers();
+ }
+
+ @Override
+ public void onStop() throws Exception {
+ LOGGER.info("onStop()");
+ closeSinkHandlers();
+ }
+
+ @Override
+ public Double fire(long timestamp, Double value) {
+ windowEvaluationHandler.collect(timestamp, value);
+ return value;
+ }
+
+ @Override
+ public double[] fire(long[] timestamps, double[] values) {
+ for (int i = 0; i < timestamps.length; ++i) {
+ windowEvaluationHandler.collect(timestamps[i], values[i]);
+ }
+ return values;
+ }
+
+ private void openSinkHandlers() throws Exception {
+ localIoTDBHandler.open(
+ new LocalIoTDBConfiguration(
+ TARGET_DEVICE, new String[] {"local"}, new TSDataType[]
{TSDataType.DOUBLE}));
+ mqttHandler.open(
+ new MQTTConfiguration(
+ "127.0.0.1",
+ 1883,
+ "root",
+ "root",
+ new PartialPath(TARGET_DEVICE),
+ new String[] {"remote"}));
+ }
+
+ private void closeSinkHandlers() throws Exception {
+ localIoTDBHandler.close();
+ mqttHandler.close();
+ }
+}
+```
+
+您可以按照下面的步骤试用这个触发器：
+
+* 在`iotdb-engine.properties`中启用MQTT服务
+
+ ``` properties
+ # whether to enable the mqtt service.
+ enable_mqtt_service=true
+ ```
+
+* 启动 IoTDB 服务器
+
+* 通过 cli 创建时间序列
+
+ ``` sql
+ CREATE TIMESERIES root.sg1.d1.s1 WITH DATATYPE=DOUBLE, ENCODING=PLAIN;
+ ```
+
+* 将 **trigger-example** 中打包好的JAR（`trigger-example-0.12.0-SNAPSHOT.jar`）放置到目录
`iotdb-server-0.12.0-SNAPSHOT/ext/trigger`
（也可以是`iotdb-server-0.12.0-SNAPSHOT/ext/trigger`的子目录）下
+
+ > 您可以通过修改配置文件中的`trigger_root_dir`来指定加载触发器JAR包的根路径。
+
+* 使用SQL语句注册该触发器，假定赋予该触发器的名字为`window-avg-alerter`
+
+* 使用`CREATE TRIGGER`语句注册该触发器
+
+ ```sql
+ CREATE TRIGGER window-avg-alerter
+ AFTER INSERT
+ ON root.sg1.d1.s1
+ AS "org.apache.iotdb.trigger.TriggerExample"
+ WITH (
+ "lo" = "0",
+ "hi" = "10.0"
+ )
+ ```
+
+* 使用 cli 插入测试数据
+
+ ``` sql
+ INSERT INTO root.sg1.d1(timestamp, s1) VALUES (1, 0);
+ INSERT INTO root.sg1.d1(timestamp, s1) VALUES (2, 2);
+ INSERT INTO root.sg1.d1(timestamp, s1) VALUES (3, 4);
+ INSERT INTO root.sg1.d1(timestamp, s1) VALUES (4, 6);
+ INSERT INTO root.sg1.d1(timestamp, s1) VALUES (5, 8);
+
+ INSERT INTO root.sg1.d1(timestamp, s1) VALUES (6, 10);
+ INSERT INTO root.sg1.d1(timestamp, s1) VALUES (7, 12);
+ INSERT INTO root.sg1.d1(timestamp, s1) VALUES (8, 14);
+ INSERT INTO root.sg1.d1(timestamp, s1) VALUES (9, 16);
+ INSERT INTO root.sg1.d1(timestamp, s1) VALUES (10, 18);
+ ```
+
+* 使用 cli 查询数据以验证触发器的行为
+
+ ``` sql
+ SELECT * FROM root.alerting;
+ ```
+
+* 正常情况下，得到如下结果
+
+ ``` sql
+ IoTDB> SELECT * FROM root.alerting;
+ +-----------------------------+--------------------+-------------------+
+ | Time|root.alerting.remote|root.alerting.local|
+ +-----------------------------+--------------------+-------------------+
+ |1970-01-01T08:00:00.006+08:00| 14.0| 14.0|
+ +-----------------------------+--------------------+-------------------+
+ Total line number = 1
+ It costs 0.006s
+ ```
+
+以上就是基本的使用方法，希望您能喜欢 :D
+
+
+
## 重要注意事项
* 触发器是通过反射技术动态装载的，因此您在装载过程中无需启停服务器。
diff --git a/example/pom.xml b/example/pom.xml
index 8db4d4a..588fc7c 100644
--- a/example/pom.xml
+++ b/example/pom.xml
@@ -44,6 +44,7 @@
<module>mqtt</module>
<module>pulsar</module>
<module>udf</module>
+ <module>trigger</module>
</modules>
<build>
<pluginManagement>
diff --git a/example/trigger/pom.xml b/example/trigger/pom.xml
new file mode 100644
index 0000000..8701de2
--- /dev/null
+++ b/example/trigger/pom.xml
@@ -0,0 +1,124 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-examples</artifactId>
+ <version>0.12.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+ <artifactId>trigger-example</artifactId>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-server</artifactId>
+ <version>0.12.0-SNAPSHOT</version>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.3</version>
+ <configuration>
+ <source>8</source>
+ <target>8</target>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <version>0.13</version>
+ <configuration>
+ <consoleOutput>false</consoleOutput>
+ </configuration>
+ <executions>
+ <execution>
+ <id>license-check</id>
+ <phase>verify</phase>
+ <goals>
+ <goal>check</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>com.diffplug.spotless</groupId>
+ <artifactId>spotless-maven-plugin</artifactId>
+ <version>2.4.2</version>
+ <configuration>
+ <java>
+ <googleJavaFormat>
+ <version>1.7</version>
+ <style>GOOGLE</style>
+ </googleJavaFormat>
+ <importOrder>
+ <order>org.apache.iotdb,,javax,java,\#</order>
+ </importOrder>
+ <removeUnusedImports/>
+ </java>
+ </configuration>
+ <executions>
+ <execution>
+ <id>spotless-check</id>
+ <phase>validate</phase>
+ <goals>
+ <goal>check</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ <profiles>
+ <profile>
+ <id>get-jar-with-dependencies</id>
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <version>3.1.0</version>
+ <configuration>
+ <descriptorRefs>
+
<descriptorRef>jar-with-dependencies</descriptorRef>
+ </descriptorRefs>
+ </configuration>
+ <executions>
+ <execution>
+ <id>make-assembly</id>
+ <!-- this is used for inheritance merges -->
+ <phase>package</phase>
+ <!-- bind to the packaging phase -->
+ <goals>
+ <goal>single</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
+</project>
diff --git
a/example/trigger/src/main/java/org/apache/iotdb/trigger/TriggerExample.java
b/example/trigger/src/main/java/org/apache/iotdb/trigger/TriggerExample.java
new file mode 100644
index 0000000..55d96aa
--- /dev/null
+++ b/example/trigger/src/main/java/org/apache/iotdb/trigger/TriggerExample.java
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.trigger;
+
+import org.apache.iotdb.db.engine.trigger.api.Trigger;
+import org.apache.iotdb.db.engine.trigger.api.TriggerAttributes;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.sink.local.LocalIoTDBConfiguration;
+import org.apache.iotdb.db.sink.local.LocalIoTDBEvent;
+import org.apache.iotdb.db.sink.local.LocalIoTDBHandler;
+import org.apache.iotdb.db.sink.mqtt.MQTTConfiguration;
+import org.apache.iotdb.db.sink.mqtt.MQTTEvent;
+import org.apache.iotdb.db.sink.mqtt.MQTTHandler;
+import
org.apache.iotdb.db.utils.windowing.configuration.SlidingSizeWindowConfiguration;
+import
org.apache.iotdb.db.utils.windowing.handler.SlidingSizeWindowEvaluationHandler;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+
+import org.fusesource.mqtt.client.QoS;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TriggerExample implements Trigger {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(TriggerExample.class);
+
+ private static final String TARGET_DEVICE = "root.alerting";
+
+ private final LocalIoTDBHandler localIoTDBHandler = new LocalIoTDBHandler();
+ private final MQTTHandler mqttHandler = new MQTTHandler();
+
+ private SlidingSizeWindowEvaluationHandler windowEvaluationHandler;
+
+ @Override
+ public void onCreate(TriggerAttributes attributes) throws Exception {
+ LOGGER.info("onCreate(TriggerAttributes attributes)");
+
+ double lo = attributes.getDouble("lo");
+ double hi = attributes.getDouble("hi");
+
+ openSinkHandlers();
+
+ windowEvaluationHandler =
+ new SlidingSizeWindowEvaluationHandler(
+ new SlidingSizeWindowConfiguration(TSDataType.DOUBLE, 5, 5),
+ window -> {
+ double avg = 0;
+ for (int i = 0; i < window.size(); ++i) {
+ avg += window.getDouble(i);
+ }
+ avg /= window.size();
+
+ if (avg < lo || hi < avg) {
+ localIoTDBHandler.onEvent(new
LocalIoTDBEvent(window.getTime(0), avg));
+ mqttHandler.onEvent(
+ new MQTTEvent("test", QoS.EXACTLY_ONCE, false,
window.getTime(0), avg));
+ }
+ });
+ }
+
+ @Override
+ public void onDrop() throws Exception {
+ LOGGER.info("onDrop()");
+ closeSinkHandlers();
+ }
+
+ @Override
+ public void onStart() throws Exception {
+ LOGGER.info("onStart()");
+ openSinkHandlers();
+ }
+
+ @Override
+ public void onStop() throws Exception {
+ LOGGER.info("onStop()");
+ closeSinkHandlers();
+ }
+
+ @Override
+ public Double fire(long timestamp, Double value) {
+ windowEvaluationHandler.collect(timestamp, value);
+ return value;
+ }
+
+ @Override
+ public double[] fire(long[] timestamps, double[] values) {
+ for (int i = 0; i < timestamps.length; ++i) {
+ windowEvaluationHandler.collect(timestamps[i], values[i]);
+ }
+ return values;
+ }
+
+ private void openSinkHandlers() throws Exception {
+ localIoTDBHandler.open(
+ new LocalIoTDBConfiguration(
+ TARGET_DEVICE, new String[] {"local"}, new TSDataType[]
{TSDataType.DOUBLE}));
+ mqttHandler.open(
+ new MQTTConfiguration(
+ "127.0.0.1",
+ 1883,
+ "root",
+ "root",
+ new PartialPath(TARGET_DEVICE),
+ new String[] {"remote"}));
+ }
+
+ private void closeSinkHandlers() throws Exception {
+ localIoTDBHandler.close();
+ mqttHandler.close();
+ }
+}
diff --git a/server/pom.xml b/server/pom.xml
index fecd1d9..c30b7d0 100644
--- a/server/pom.xml
+++ b/server/pom.xml
@@ -203,6 +203,10 @@
<artifactId>netty-buffer</artifactId>
<version>4.1.27.Final</version>
</dependency>
+ <dependency>
+ <groupId>org.fusesource.mqtt-client</groupId>
+ <artifactId>mqtt-client</artifactId>
+ </dependency>
</dependencies>
<build>
<plugins>
diff --git a/server/src/assembly/resources/conf/iotdb-engine.properties
b/server/src/assembly/resources/conf/iotdb-engine.properties
index 3881a0b..f909b39 100644
--- a/server/src/assembly/resources/conf/iotdb-engine.properties
+++ b/server/src/assembly/resources/conf/iotdb-engine.properties
@@ -643,7 +643,7 @@ timestamp_precision=ms
# The size of log buffer for every trigger management operation plan. If the
size of a trigger
# management operation plan is larger than this parameter, the trigger
management operation plan
# will be rejected by TriggerManager.
-tlog_buffer_size=1048576
+# tlog_buffer_size=1048576
# Uncomment the following field to configure the trigger root directory.
# For Window platform
@@ -655,11 +655,11 @@ tlog_buffer_size=1048576
# trigger_root_dir=ext/trigger
# How many threads can be used for evaluating sliding windows. When <= 0, use
CPU core number.
-concurrent_window_evaluation_thread=0
+# concurrent_window_evaluation_thread=0
# Max number of window evaluation tasks that can be pending for execution.
When <= 0, the value is
# 64 by default.
-max_pending_window_evaluation_tasks = 64
+# max_pending_window_evaluation_tasks = 64
####################
### Index Configuration
diff --git
a/server/src/main/java/org/apache/iotdb/db/utils/windowing/api/Evaluator.java
b/server/src/main/java/org/apache/iotdb/db/sink/api/Configuration.java
similarity index 74%
copy from
server/src/main/java/org/apache/iotdb/db/utils/windowing/api/Evaluator.java
copy to server/src/main/java/org/apache/iotdb/db/sink/api/Configuration.java
index 7e2047a..f24994f 100644
---
a/server/src/main/java/org/apache/iotdb/db/utils/windowing/api/Evaluator.java
+++ b/server/src/main/java/org/apache/iotdb/db/sink/api/Configuration.java
@@ -17,16 +17,6 @@
* under the License.
*/
-package org.apache.iotdb.db.utils.windowing.api;
+package org.apache.iotdb.db.sink.api;
-import java.util.concurrent.RejectedExecutionException;
-
-@FunctionalInterface
-public interface Evaluator {
-
- void evaluate(Window window);
-
- default void onRejection(Window window) {
- throw new RejectedExecutionException();
- }
-}
+public interface Configuration {}
diff --git
a/server/src/main/java/org/apache/iotdb/db/utils/windowing/api/Evaluator.java
b/server/src/main/java/org/apache/iotdb/db/sink/api/Event.java
similarity index 74%
copy from
server/src/main/java/org/apache/iotdb/db/utils/windowing/api/Evaluator.java
copy to server/src/main/java/org/apache/iotdb/db/sink/api/Event.java
index 7e2047a..a4a70e0 100644
---
a/server/src/main/java/org/apache/iotdb/db/utils/windowing/api/Evaluator.java
+++ b/server/src/main/java/org/apache/iotdb/db/sink/api/Event.java
@@ -17,16 +17,6 @@
* under the License.
*/
-package org.apache.iotdb.db.utils.windowing.api;
+package org.apache.iotdb.db.sink.api;
-import java.util.concurrent.RejectedExecutionException;
-
-@FunctionalInterface
-public interface Evaluator {
-
- void evaluate(Window window);
-
- default void onRejection(Window window) {
- throw new RejectedExecutionException();
- }
-}
+public interface Event {}
diff --git
a/server/src/main/java/org/apache/iotdb/db/utils/windowing/api/Evaluator.java
b/server/src/main/java/org/apache/iotdb/db/sink/api/Handler.java
similarity index 69%
copy from
server/src/main/java/org/apache/iotdb/db/utils/windowing/api/Evaluator.java
copy to server/src/main/java/org/apache/iotdb/db/sink/api/Handler.java
index 7e2047a..1827b0c 100644
---
a/server/src/main/java/org/apache/iotdb/db/utils/windowing/api/Evaluator.java
+++ b/server/src/main/java/org/apache/iotdb/db/sink/api/Handler.java
@@ -17,16 +17,16 @@
* under the License.
*/
-package org.apache.iotdb.db.utils.windowing.api;
+package org.apache.iotdb.db.sink.api;
-import java.util.concurrent.RejectedExecutionException;
+public interface Handler<C extends Configuration, E extends Event> {
-@FunctionalInterface
-public interface Evaluator {
+ @SuppressWarnings("squid:S112")
+ default void open(C configuration) throws Exception {}
- void evaluate(Window window);
+ @SuppressWarnings("squid:S112")
+ default void close() throws Exception {}
- default void onRejection(Window window) {
- throw new RejectedExecutionException();
- }
+ @SuppressWarnings("squid:S112")
+ void onEvent(E event) throws Exception;
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/utils/windowing/api/Evaluator.java
b/server/src/main/java/org/apache/iotdb/db/sink/exception/SinkException.java
similarity index 74%
copy from
server/src/main/java/org/apache/iotdb/db/utils/windowing/api/Evaluator.java
copy to
server/src/main/java/org/apache/iotdb/db/sink/exception/SinkException.java
index 7e2047a..66ef627 100644
---
a/server/src/main/java/org/apache/iotdb/db/utils/windowing/api/Evaluator.java
+++ b/server/src/main/java/org/apache/iotdb/db/sink/exception/SinkException.java
@@ -17,16 +17,15 @@
* under the License.
*/
-package org.apache.iotdb.db.utils.windowing.api;
+package org.apache.iotdb.db.sink.exception;
-import java.util.concurrent.RejectedExecutionException;
+public class SinkException extends Exception {
-@FunctionalInterface
-public interface Evaluator {
-
- void evaluate(Window window);
+ public SinkException(String message) {
+ super(message);
+ }
- default void onRejection(Window window) {
- throw new RejectedExecutionException();
+ public SinkException(String message, Throwable cause) {
+ super(message, cause);
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/sink/local/LocalIoTDBConfiguration.java
b/server/src/main/java/org/apache/iotdb/db/sink/local/LocalIoTDBConfiguration.java
new file mode 100644
index 0000000..576a47e
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/sink/local/LocalIoTDBConfiguration.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.sink.local;
+
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.sink.api.Configuration;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+
+public class LocalIoTDBConfiguration implements Configuration {
+
+ private final PartialPath device;
+ private final String[] measurements;
+ private final TSDataType[] dataTypes;
+
+ public LocalIoTDBConfiguration(String device, String[] measurements,
TSDataType[] dataTypes)
+ throws IllegalPathException {
+ this.device = new PartialPath(device);
+ this.measurements = measurements;
+ this.dataTypes = dataTypes;
+ }
+
+ public PartialPath getDevice() {
+ return device;
+ }
+
+ public String[] getMeasurements() {
+ return measurements;
+ }
+
+ public TSDataType[] getDataTypes() {
+ return dataTypes;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/utils/windowing/api/Evaluator.java
b/server/src/main/java/org/apache/iotdb/db/sink/local/LocalIoTDBEvent.java
similarity index 65%
copy from
server/src/main/java/org/apache/iotdb/db/utils/windowing/api/Evaluator.java
copy to server/src/main/java/org/apache/iotdb/db/sink/local/LocalIoTDBEvent.java
index 7e2047a..62bfd68 100644
---
a/server/src/main/java/org/apache/iotdb/db/utils/windowing/api/Evaluator.java
+++ b/server/src/main/java/org/apache/iotdb/db/sink/local/LocalIoTDBEvent.java
@@ -17,16 +17,25 @@
* under the License.
*/
-package org.apache.iotdb.db.utils.windowing.api;
+package org.apache.iotdb.db.sink.local;
-import java.util.concurrent.RejectedExecutionException;
+import org.apache.iotdb.db.sink.api.Event;
-@FunctionalInterface
-public interface Evaluator {
+public class LocalIoTDBEvent implements Event {
- void evaluate(Window window);
+ private final long timestamp;
+ private final Object[] values;
- default void onRejection(Window window) {
- throw new RejectedExecutionException();
+ public LocalIoTDBEvent(long timestamp, Object... values) {
+ this.timestamp = timestamp;
+ this.values = values;
+ }
+
+ public long getTimestamp() {
+ return timestamp;
+ }
+
+ public Object[] getValues() {
+ return values;
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/sink/local/LocalIoTDBHandler.java
b/server/src/main/java/org/apache/iotdb/db/sink/local/LocalIoTDBHandler.java
new file mode 100644
index 0000000..69a6537
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/sink/local/LocalIoTDBHandler.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.sink.local;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.qp.executor.IPlanExecutor;
+import org.apache.iotdb.db.qp.executor.PlanExecutor;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
+import org.apache.iotdb.db.service.IoTDB;
+import org.apache.iotdb.db.sink.api.Handler;
+import org.apache.iotdb.db.sink.exception.SinkException;
+import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+
+import java.util.Collections;
+
+import static
org.apache.iotdb.db.utils.EncodingInferenceUtils.getDefaultEncoding;
+
+public class LocalIoTDBHandler implements Handler<LocalIoTDBConfiguration,
LocalIoTDBEvent> {
+
+ private IPlanExecutor executor;
+
+ private PartialPath device;
+ private String[] measurements;
+ private TSDataType[] dataTypes;
+
+ @Override
+ public void open(LocalIoTDBConfiguration configuration) throws Exception {
+ executor = new PlanExecutor();
+
+ device = configuration.getDevice();
+ measurements = configuration.getMeasurements();
+ dataTypes = configuration.getDataTypes();
+
+ createOrCheckTimeseries();
+ }
+
+ private void createOrCheckTimeseries() throws MetadataException,
SinkException {
+ for (int i = 0; i < measurements.length; ++i) {
+ String measurement = measurements[i];
+ TSDataType dataType = dataTypes[i];
+
+ PartialPath path = new PartialPath(device.getFullPath(), measurement);
+ if (!IoTDB.metaManager.isPathExist(path)) {
+ IoTDB.metaManager.createTimeseries(
+ path,
+ dataType,
+ getDefaultEncoding(dataType),
+ TSFileDescriptor.getInstance().getConfig().getCompressor(),
+ Collections.emptyMap());
+ } else {
+ if (!IoTDB.metaManager.getSeriesSchema(device,
measurement).getType().equals(dataType)) {
+ throw new SinkException(
+ String.format("The data type of %s you provided was not
correct.", path));
+ }
+ }
+ }
+ }
+
+ @Override
+ public void onEvent(LocalIoTDBEvent event)
+ throws QueryProcessException, StorageEngineException,
StorageGroupNotSetException {
+ InsertRowPlan plan = new InsertRowPlan();
+ plan.setNeedInferType(false);
+ plan.setDeviceId(device);
+ plan.setMeasurements(measurements);
+ plan.setDataTypes(dataTypes);
+ plan.setTime(event.getTimestamp());
+ plan.setValues(event.getValues());
+ executeNonQuery(plan);
+ }
+
+ private void executeNonQuery(PhysicalPlan plan)
+ throws QueryProcessException, StorageGroupNotSetException,
StorageEngineException {
+ if (IoTDBDescriptor.getInstance().getConfig().isReadOnly()) {
+ throw new QueryProcessException(
+ "Current system mode is read-only, non-query operation is not
supported.");
+ }
+ executor.processNonQuery(plan);
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/sink/mqtt/MQTTConfiguration.java
b/server/src/main/java/org/apache/iotdb/db/sink/mqtt/MQTTConfiguration.java
new file mode 100644
index 0000000..7e89b81
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/sink/mqtt/MQTTConfiguration.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.sink.mqtt;
+
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.sink.api.Configuration;
+
+public class MQTTConfiguration implements Configuration {
+
+ private final String host;
+ private final int port;
+
+ private final String username;
+ private final String password;
+
+ private final PartialPath device;
+ private final String[] measurements;
+
+ public MQTTConfiguration(
+ String host,
+ int port,
+ String username,
+ String password,
+ PartialPath device,
+ String[] measurements) {
+ this.host = host;
+ this.port = port;
+ this.username = username;
+ this.password = password;
+ this.device = device;
+ this.measurements = measurements;
+ }
+
+ public String getHost() {
+ return host;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ public String getUsername() {
+ return username;
+ }
+
+ public String getPassword() {
+ return password;
+ }
+
+ public String[] getMeasurements() {
+ return measurements;
+ }
+
+ public PartialPath getDevice() {
+ return device;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/utils/windowing/runtime/WindowEvaluationTask.java
b/server/src/main/java/org/apache/iotdb/db/sink/mqtt/MQTTEvent.java
similarity index 51%
copy from
server/src/main/java/org/apache/iotdb/db/utils/windowing/runtime/WindowEvaluationTask.java
copy to server/src/main/java/org/apache/iotdb/db/sink/mqtt/MQTTEvent.java
index 794b794..1c876ec 100644
---
a/server/src/main/java/org/apache/iotdb/db/utils/windowing/runtime/WindowEvaluationTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/sink/mqtt/MQTTEvent.java
@@ -17,28 +17,46 @@
* under the License.
*/
-package org.apache.iotdb.db.utils.windowing.runtime;
+package org.apache.iotdb.db.sink.mqtt;
-import org.apache.iotdb.db.concurrent.WrappedRunnable;
-import org.apache.iotdb.db.utils.windowing.api.Evaluator;
-import org.apache.iotdb.db.utils.windowing.api.Window;
+import org.apache.iotdb.db.sink.api.Event;
-public class WindowEvaluationTask extends WrappedRunnable {
+import org.fusesource.mqtt.client.QoS;
- private final Evaluator evaluator;
- private final Window window;
+public class MQTTEvent implements Event {
- public WindowEvaluationTask(Evaluator evaluator, Window window) {
- this.evaluator = evaluator;
- this.window = window;
+ private final String topic;
+ private final QoS qos;
+ private final boolean retain;
+
+ private final long timestamp;
+ private final Object[] values;
+
+ public MQTTEvent(String topic, QoS qos, boolean retain, long timestamp,
Object... values) {
+ this.topic = topic;
+ this.qos = qos;
+ this.retain = retain;
+ this.timestamp = timestamp;
+ this.values = values;
+ }
+
+ public String getTopic() {
+ return topic;
+ }
+
+ public QoS getQoS() {
+ return qos;
+ }
+
+ public boolean retain() {
+ return retain;
}
- @Override
- public void runMayThrow() {
- evaluator.evaluate(window);
+ public long getTimestamp() {
+ return timestamp;
}
- public void onRejection() {
- evaluator.onRejection(window);
+ public Object[] getValues() {
+ return values;
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/sink/mqtt/MQTTHandler.java
b/server/src/main/java/org/apache/iotdb/db/sink/mqtt/MQTTHandler.java
new file mode 100644
index 0000000..342a3e1
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/sink/mqtt/MQTTHandler.java
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.sink.mqtt;
+
+import org.apache.iotdb.db.sink.api.Handler;
+import org.apache.iotdb.db.sink.exception.SinkException;
+import org.apache.iotdb.tsfile.utils.Binary;
+
+import org.fusesource.mqtt.client.BlockingConnection;
+import org.fusesource.mqtt.client.MQTT;
+
+public class MQTTHandler implements Handler<MQTTConfiguration, MQTTEvent> {
+
+ private BlockingConnection connection;
+
+ private String payloadFormatter;
+
+ @Override
+ public void open(MQTTConfiguration configuration) throws Exception {
+ MQTT mqtt = new MQTT();
+ mqtt.setHost(configuration.getHost(), configuration.getPort());
+ mqtt.setUserName(configuration.getUsername());
+ mqtt.setPassword(configuration.getPassword());
+ connection = mqtt.blockingConnection();
+ connection.connect();
+
+ payloadFormatter = generatePayloadFormatter(configuration);
+ }
+
+ private static String generatePayloadFormatter(MQTTConfiguration
configuration)
+ throws SinkException {
+ return String.format(
+ "{\"device\":\"%s\",\"measurements\":[%s]%s",
+ configuration.getDevice(),
+ arrayToJson(configuration.getMeasurements()),
+ ",\"timestamp\":%d,\"values\":[%s]}");
+ }
+
+ private static String arrayToJson(Object[] array) throws SinkException {
+ if (array.length <= 0) {
+ throw new SinkException("The number of measurements should be
positive.");
+ }
+
+ StringBuilder sb = new StringBuilder(objectToJson(array[0]));
+ for (int i = 1; i < array.length; ++i) {
+ sb.append(',').append(objectToJson(array[i]));
+ }
+ return sb.toString();
+ }
+
+ private static String objectToJson(Object object) {
+ return (object instanceof String || object instanceof Binary)
+ ? ('\"' + object.toString() + '\"')
+ : object.toString();
+ }
+
+ @Override
+ public void close() throws Exception {
+ connection.disconnect();
+ }
+
+ @Override
+ public void onEvent(MQTTEvent event) throws Exception {
+ String payload =
+ String.format(payloadFormatter, event.getTimestamp(),
arrayToJson(event.getValues()));
+ connection.publish(event.getTopic(), payload.getBytes(), event.getQoS(),
event.retain());
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/utils/windowing/api/Evaluator.java
b/server/src/main/java/org/apache/iotdb/db/utils/windowing/api/Evaluator.java
index 7e2047a..9937b4b 100644
---
a/server/src/main/java/org/apache/iotdb/db/utils/windowing/api/Evaluator.java
+++
b/server/src/main/java/org/apache/iotdb/db/utils/windowing/api/Evaluator.java
@@ -24,7 +24,8 @@ import java.util.concurrent.RejectedExecutionException;
@FunctionalInterface
public interface Evaluator {
- void evaluate(Window window);
+ @SuppressWarnings("squid:S112")
+ void evaluate(Window window) throws Exception;
default void onRejection(Window window) {
throw new RejectedExecutionException();
diff --git
a/server/src/main/java/org/apache/iotdb/db/utils/windowing/runtime/WindowEvaluationTask.java
b/server/src/main/java/org/apache/iotdb/db/utils/windowing/runtime/WindowEvaluationTask.java
index 794b794..053296b 100644
---
a/server/src/main/java/org/apache/iotdb/db/utils/windowing/runtime/WindowEvaluationTask.java
+++
b/server/src/main/java/org/apache/iotdb/db/utils/windowing/runtime/WindowEvaluationTask.java
@@ -34,7 +34,7 @@ public class WindowEvaluationTask extends WrappedRunnable {
}
@Override
- public void runMayThrow() {
+ public void runMayThrow() throws Exception {
evaluator.evaluate(window);
}
diff --git
a/server/src/test/java/org/apache/iotdb/db/sink/LocalIoTDBSinkTest.java
b/server/src/test/java/org/apache/iotdb/db/sink/LocalIoTDBSinkTest.java
new file mode 100644
index 0000000..adcd434
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/sink/LocalIoTDBSinkTest.java
@@ -0,0 +1,217 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.sink;
+
+import org.apache.iotdb.db.sink.local.LocalIoTDBConfiguration;
+import org.apache.iotdb.db.sink.local.LocalIoTDBEvent;
+import org.apache.iotdb.db.sink.local.LocalIoTDBHandler;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.jdbc.Config;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.utils.Binary;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.sql.Types;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+public class LocalIoTDBSinkTest {
+
+ @Before
+ public void setUp() throws Exception {
+ EnvironmentUtils.envSetUp();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ EnvironmentUtils.cleanEnv();
+ }
+
+ @Test
+ public void onEventUsingSingleSensorHandler() throws Exception {
+ LocalIoTDBHandler localIoTDBHandler = new LocalIoTDBHandler();
+ localIoTDBHandler.open(
+ new LocalIoTDBConfiguration(
+ "root.sg1.d1", new String[] {"s1"}, new TSDataType[]
{TSDataType.INT32}));
+
+ for (int i = 0; i < 10000; ++i) {
+ localIoTDBHandler.onEvent(new LocalIoTDBEvent(i, i));
+ }
+
+ localIoTDBHandler.close();
+
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ Assert.assertTrue(statement.execute("select * from root"));
+
+ try (ResultSet resultSet = statement.getResultSet()) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+
+ checkHeader(
+ resultSetMetaData,
+ "Time,root.sg1.d1.s1,",
+ new int[] {
+ Types.TIMESTAMP, Types.INTEGER,
+ });
+
+ int count = 0;
+ while (resultSet.next()) {
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ assertEquals(count, Double.parseDouble(resultSet.getString(i)),
0.0);
+ }
+ count++;
+ }
+ Assert.assertEquals(10000, count);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void onEventUsingMultiSensorsHandler() throws Exception {
+ LocalIoTDBHandler localIoTDBHandler = new LocalIoTDBHandler();
+ localIoTDBHandler.open(
+ new LocalIoTDBConfiguration(
+ "root.sg1.d1",
+ new String[] {"s1", "s2", "s3", "s4", "s5", "s6"},
+ new TSDataType[] {
+ TSDataType.INT32,
+ TSDataType.INT64,
+ TSDataType.FLOAT,
+ TSDataType.DOUBLE,
+ TSDataType.BOOLEAN,
+ TSDataType.TEXT
+ }));
+
+ for (int i = 0; i < 10000; ++i) {
+ localIoTDBHandler.onEvent(
+ new LocalIoTDBEvent(
+ i,
+ i,
+ (long) i,
+ (float) i,
+ (double) i,
+ i % 2 == 0,
+ Binary.valueOf(String.valueOf(i))));
+ }
+
+ localIoTDBHandler.close();
+
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ Assert.assertTrue(statement.execute("select * from root"));
+
+ try (ResultSet resultSet = statement.getResultSet()) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+
+ checkHeader(
+ resultSetMetaData,
+ "Time,root.sg1.d1.s1,root.sg1.d1.s2,root.sg1.d1.s3,"
+ + "root.sg1.d1.s4,root.sg1.d1.s5,root.sg1.d1.s6,",
+ new int[] {
+ Types.TIMESTAMP,
+ Types.INTEGER,
+ Types.BIGINT,
+ Types.FLOAT,
+ Types.DOUBLE,
+ Types.BOOLEAN,
+ Types.VARCHAR,
+ });
+
+ int count = 0;
+ while (resultSet.next()) {
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ try {
+ assertEquals(count, Double.parseDouble(resultSet.getString(i)),
0.0);
+ } catch (NumberFormatException e) {
+ assertEquals(count % 2 == 0,
Boolean.parseBoolean(resultSet.getString(i)));
+ }
+ }
+ count++;
+ }
+ Assert.assertEquals(10000, count);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ private void checkHeader(
+ ResultSetMetaData resultSetMetaData, String expectedHeaderStrings, int[]
expectedTypes)
+ throws SQLException {
+ String[] expectedHeaders = expectedHeaderStrings.split(",");
+ Map<String, Integer> expectedHeaderToTypeIndexMap = new HashMap<>();
+ for (int i = 0; i < expectedHeaders.length; ++i) {
+ expectedHeaderToTypeIndexMap.put(expectedHeaders[i], i);
+ }
+
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ Integer typeIndex =
expectedHeaderToTypeIndexMap.get(resultSetMetaData.getColumnName(i));
+ Assert.assertNotNull(typeIndex);
+ Assert.assertEquals(expectedTypes[typeIndex],
resultSetMetaData.getColumnType(i));
+ }
+ }
+
+ @Test(expected = ClassCastException.class)
+ public void onEventWithWrongType1() throws Exception {
+ LocalIoTDBHandler localIoTDBHandler = new LocalIoTDBHandler();
+ localIoTDBHandler.open(
+ new LocalIoTDBConfiguration(
+ "root.sg1.d1", new String[] {"s1"}, new TSDataType[]
{TSDataType.INT32}));
+
+ localIoTDBHandler.onEvent(new LocalIoTDBEvent(0,
Binary.valueOf(String.valueOf(0))));
+
+ localIoTDBHandler.close();
+ }
+
+ @Test(expected = ClassCastException.class)
+ public void onEventWithWrongType2() throws Exception {
+ LocalIoTDBHandler localIoTDBHandler = new LocalIoTDBHandler();
+ localIoTDBHandler.open(
+ new LocalIoTDBConfiguration(
+ "root.sg1.d1", new String[] {"s1"}, new TSDataType[]
{TSDataType.TEXT}));
+
+ localIoTDBHandler.onEvent(new LocalIoTDBEvent(0, String.valueOf(0)));
+
+ localIoTDBHandler.close();
+ }
+}
diff --git a/server/src/test/java/org/apache/iotdb/db/sink/MQTTSinkTest.java
b/server/src/test/java/org/apache/iotdb/db/sink/MQTTSinkTest.java
new file mode 100644
index 0000000..8e7b92e
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/sink/MQTTSinkTest.java
@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.sink;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.sink.mqtt.MQTTConfiguration;
+import org.apache.iotdb.db.sink.mqtt.MQTTEvent;
+import org.apache.iotdb.db.sink.mqtt.MQTTHandler;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.jdbc.Config;
+
+import org.fusesource.mqtt.client.QoS;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.sql.Types;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+@SuppressWarnings("squid:S2925")
+public class MQTTSinkTest {
+
+ @Before
+ public void setUp() throws Exception {
+ IoTDBDescriptor.getInstance().getConfig().setEnableMQTTService(true);
+ EnvironmentUtils.envSetUp();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ EnvironmentUtils.cleanEnv();
+ }
+
+ @Test
+ public void onEventUsingSingleSensorHandler() throws Exception {
+ MQTTHandler mqttHandler = new MQTTHandler();
+ mqttHandler.open(
+ new MQTTConfiguration(
+ "127.0.0.1",
+ 1883,
+ "root",
+ "root",
+ new PartialPath("root.sg1.d1"),
+ new String[] {"s1"}));
+
+ for (int i = 0; i < 10000; ++i) {
+ mqttHandler.onEvent(new MQTTEvent("test", QoS.EXACTLY_ONCE, false, i,
i));
+ }
+
+ mqttHandler.close();
+
+ Thread.sleep(1000);
+
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ Assert.assertTrue(statement.execute("select * from root"));
+
+ try (ResultSet resultSet = statement.getResultSet()) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+
+ checkHeader(
+ resultSetMetaData,
+ "Time,root.sg1.d1.s1,",
+ new int[] {
+ Types.TIMESTAMP, Types.FLOAT,
+ });
+
+ int count = 0;
+ while (resultSet.next()) {
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ assertEquals(count, Double.parseDouble(resultSet.getString(i)),
0.0);
+ }
+ count++;
+ }
+ Assert.assertEquals(10000, count);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void onEventUsingMultiSensorsHandler() throws Exception {
+ MQTTHandler mqttHandler = new MQTTHandler();
+ mqttHandler.open(
+ new MQTTConfiguration(
+ "127.0.0.1",
+ 1883,
+ "root",
+ "root",
+ new PartialPath("root.sg1.d1"),
+ new String[] {"s1", "s2", "s3", "s4", "s5", "s6"}));
+
+ for (int i = 0; i < 10000; ++i) {
+ mqttHandler.onEvent(
+ new MQTTEvent(
+ "test",
+ QoS.EXACTLY_ONCE,
+ false,
+ i,
+ i,
+ (long) i,
+ (float) i,
+ (double) i,
+ i % 2 == 0,
+ String.valueOf(i)));
+ }
+
+ mqttHandler.close();
+
+ Thread.sleep(1000);
+
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ Assert.assertTrue(statement.execute("select * from root"));
+
+ try (ResultSet resultSet = statement.getResultSet()) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+
+ checkHeader(
+ resultSetMetaData,
+ "Time,root.sg1.d1.s1,root.sg1.d1.s2,root.sg1.d1.s3,"
+ + "root.sg1.d1.s4,root.sg1.d1.s5,root.sg1.d1.s6,",
+ new int[] {
+ Types.TIMESTAMP,
+ Types.FLOAT,
+ Types.FLOAT,
+ Types.FLOAT,
+ Types.FLOAT,
+ Types.BOOLEAN,
+ Types.FLOAT,
+ });
+
+ int count = 0;
+ while (resultSet.next()) {
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ try {
+ assertEquals(count, Double.parseDouble(resultSet.getString(i)),
0.0);
+ } catch (NumberFormatException e) {
+ assertEquals(count % 2 == 0,
Boolean.parseBoolean(resultSet.getString(i)));
+ }
+ }
+ count++;
+ }
+ Assert.assertEquals(10000, count);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ private void checkHeader(
+ ResultSetMetaData resultSetMetaData, String expectedHeaderStrings, int[]
expectedTypes)
+ throws SQLException {
+ String[] expectedHeaders = expectedHeaderStrings.split(",");
+ Map<String, Integer> expectedHeaderToTypeIndexMap = new HashMap<>();
+ for (int i = 0; i < expectedHeaders.length; ++i) {
+ expectedHeaderToTypeIndexMap.put(expectedHeaders[i], i);
+ }
+
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ Integer typeIndex =
expectedHeaderToTypeIndexMap.get(resultSetMetaData.getColumnName(i));
+ Assert.assertNotNull(typeIndex);
+ Assert.assertEquals(expectedTypes[typeIndex],
resultSetMetaData.getColumnType(i));
+ }
+ }
+}
