This is an automated email from the ASF dual-hosted git repository.
rong 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 c643667761 [IOTDB-3251] ForwardTrigger: support MQTT/HTTP data
forwarding (#5870)
c643667761 is described below
commit c64366776113cccb67374b4b86d65935bc4a07e2
Author: 刘威 <[email protected]>
AuthorDate: Sun May 29 15:56:10 2022 +0800
[IOTDB-3251] ForwardTrigger: support MQTT/HTTP data forwarding (#5870)
Co-authored-by: gongning <[email protected]>
---
.../org/apache/iotdb/trigger/AlertingExample.java | 5 +-
.../org/apache/iotdb/trigger/TriggerExample.java | 4 +-
integration/pom.xml | 6 +
.../db/engine/trigger/example/Accumulator.java | 13 +-
.../iotdb/db/engine/trigger/example/Counter.java | 13 +-
.../db/integration/IoTDBTriggerForwardIT.java | 305 +++++++++++++++++++++
.../resources/conf/iotdb-engine.properties | 18 +-
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 65 +++++
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 31 +++
.../iotdb/db/engine/trigger/api/Trigger.java | 37 +--
.../db/engine/trigger/builtin/ForwardTrigger.java | 229 ++++++++++++++++
.../db/engine/trigger/executor/TriggerEngine.java | 14 +-
.../engine/trigger/executor/TriggerExecutor.java | 43 +--
.../iotdb/db/engine/trigger/sink/api/Event.java | 8 +-
.../iotdb/db/engine/trigger/sink/api/Handler.java | 5 +
.../engine/trigger/sink/forward/ForwardEvent.java | 70 +++++
.../HTTPForwardConfiguration.java} | 31 ++-
.../{api/Event.java => http/HTTPForwardEvent.java} | 12 +-
.../trigger/sink/http/HTTPForwardHandler.java | 133 +++++++++
.../sink/mqtt/MQTTForwardConfiguration.java | 137 +++++++++
.../{api/Event.java => mqtt/MQTTForwardEvent.java} | 12 +-
.../trigger/sink/mqtt/MQTTForwardHandler.java | 97 +++++++
.../db/engine/trigger/utils/BatchHandlerQueue.java | 150 ++++++++++
.../engine/trigger/utils/HTTPConnectionPool.java | 49 ++++
.../trigger/utils/MQTTConnectionFactory.java | 115 ++++++++
.../engine/trigger/utils/MQTTConnectionPool.java | 79 ++++++
.../db/protocol/mqtt/JSONPayloadFormatter.java | 30 +-
.../metadata/idtable/trigger_example/Counter.java | 13 +-
28 files changed, 1644 insertions(+), 80 deletions(-)
diff --git
a/example/trigger/src/main/java/org/apache/iotdb/trigger/AlertingExample.java
b/example/trigger/src/main/java/org/apache/iotdb/trigger/AlertingExample.java
index 500caf2b92..48975a1dd8 100644
---
a/example/trigger/src/main/java/org/apache/iotdb/trigger/AlertingExample.java
+++
b/example/trigger/src/main/java/org/apache/iotdb/trigger/AlertingExample.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.trigger;
+import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.engine.trigger.api.Trigger;
import org.apache.iotdb.db.engine.trigger.api.TriggerAttributes;
import
org.apache.iotdb.db.engine.trigger.sink.alertmanager.AlertManagerConfiguration;
@@ -71,7 +72,7 @@ public class AlertingExample implements Trigger {
}
@Override
- public Double fire(long timestamp, Double value) throws Exception {
+ public Double fire(long timestamp, Double value, PartialPath path) throws
Exception {
if (value > 100.0) {
labels.put("value", String.valueOf(value));
labels.put("severity", "critical");
@@ -88,7 +89,7 @@ public class AlertingExample implements Trigger {
}
@Override
- public double[] fire(long[] timestamps, double[] values) throws Exception {
+ public double[] fire(long[] timestamps, double[] values, PartialPath path)
throws Exception {
for (double value : values) {
if (value > 100.0) {
labels.put("value", String.valueOf(value));
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
index c6565172d5..7e1a6155a7 100644
--- a/example/trigger/src/main/java/org/apache/iotdb/trigger/TriggerExample.java
+++ b/example/trigger/src/main/java/org/apache/iotdb/trigger/TriggerExample.java
@@ -99,14 +99,14 @@ public class TriggerExample implements Trigger {
}
@Override
- public Double fire(long timestamp, Double value) throws Exception {
+ public Double fire(long timestamp, Double value, PartialPath path) throws
Exception {
tryOpenSinksFirstOnFire();
windowEvaluationHandler.collect(timestamp, value);
return value;
}
@Override
- public double[] fire(long[] timestamps, double[] values) throws Exception {
+ public double[] fire(long[] timestamps, double[] values, PartialPath path)
throws Exception {
tryOpenSinksFirstOnFire();
for (int i = 0; i < timestamps.length; ++i) {
windowEvaluationHandler.collect(timestamps[i], values[i]);
diff --git a/integration/pom.xml b/integration/pom.xml
index 08537c32a6..2e572d63fc 100644
--- a/integration/pom.xml
+++ b/integration/pom.xml
@@ -60,6 +60,12 @@
<groupId>junit</groupId>
<artifactId>junit</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <version>${awaitility.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<dependencyManagement>
<dependencies>
diff --git
a/integration/src/test/java/org/apache/iotdb/db/engine/trigger/example/Accumulator.java
b/integration/src/test/java/org/apache/iotdb/db/engine/trigger/example/Accumulator.java
index dc2bcb1b07..c2d662c463 100644
---
a/integration/src/test/java/org/apache/iotdb/db/engine/trigger/example/Accumulator.java
+++
b/integration/src/test/java/org/apache/iotdb/db/engine/trigger/example/Accumulator.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.engine.trigger.example;
+import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.engine.trigger.api.Trigger;
import org.apache.iotdb.db.engine.trigger.api.TriggerAttributes;
import org.apache.iotdb.tsfile.exception.NotImplementedException;
@@ -48,36 +49,36 @@ public class Accumulator implements Trigger {
}
@Override
- public Integer fire(long timestamp, Integer value) {
+ public Integer fire(long timestamp, Integer value, PartialPath path) {
accumulator += value;
return value;
}
@Override
- public Long fire(long timestamp, Long value) {
+ public Long fire(long timestamp, Long value, PartialPath path) {
accumulator += value;
return value;
}
@Override
- public Float fire(long timestamp, Float value) {
+ public Float fire(long timestamp, Float value, PartialPath path) {
accumulator += value;
return value;
}
@Override
- public Double fire(long timestamp, Double value) {
+ public Double fire(long timestamp, Double value, PartialPath path) {
accumulator += value;
return value;
}
@Override
- public Boolean fire(long timestamp, Boolean value) {
+ public Boolean fire(long timestamp, Boolean value, PartialPath path) {
throw new NotImplementedException();
}
@Override
- public Binary fire(long timestamp, Binary value) {
+ public Binary fire(long timestamp, Binary value, PartialPath path) {
throw new NotImplementedException();
}
diff --git
a/integration/src/test/java/org/apache/iotdb/db/engine/trigger/example/Counter.java
b/integration/src/test/java/org/apache/iotdb/db/engine/trigger/example/Counter.java
index 6281e8c7b6..7f52620c20 100644
---
a/integration/src/test/java/org/apache/iotdb/db/engine/trigger/example/Counter.java
+++
b/integration/src/test/java/org/apache/iotdb/db/engine/trigger/example/Counter.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.engine.trigger.example;
+import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.engine.trigger.api.Trigger;
import org.apache.iotdb.db.engine.trigger.api.TriggerAttributes;
import org.apache.iotdb.tsfile.utils.Binary;
@@ -47,37 +48,37 @@ public class Counter implements Trigger {
}
@Override
- public Integer fire(long timestamp, Integer value) {
+ public Integer fire(long timestamp, Integer value, PartialPath path) {
++counter;
return value;
}
@Override
- public Long fire(long timestamp, Long value) {
+ public Long fire(long timestamp, Long value, PartialPath path) {
++counter;
return value;
}
@Override
- public Float fire(long timestamp, Float value) {
+ public Float fire(long timestamp, Float value, PartialPath path) {
++counter;
return value;
}
@Override
- public Double fire(long timestamp, Double value) {
+ public Double fire(long timestamp, Double value, PartialPath path) {
++counter;
return value;
}
@Override
- public Boolean fire(long timestamp, Boolean value) {
+ public Boolean fire(long timestamp, Boolean value, PartialPath path) {
++counter;
return value;
}
@Override
- public Binary fire(long timestamp, Binary value) {
+ public Binary fire(long timestamp, Binary value, PartialPath path) {
++counter;
return value;
}
diff --git
a/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBTriggerForwardIT.java
b/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBTriggerForwardIT.java
new file mode 100644
index 0000000000..3473752a66
--- /dev/null
+++
b/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBTriggerForwardIT.java
@@ -0,0 +1,305 @@
+/*
+ * 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.integration;
+
+import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.engine.trigger.sink.forward.ForwardEvent;
+import org.apache.iotdb.db.service.IoTDB;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.jdbc.Config;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+
+import com.google.common.collect.Lists;
+import com.google.gson.Gson;
+import com.google.gson.JsonArray;
+import com.sun.net.httpserver.HttpServer;
+import io.moquette.BrokerConstants;
+import io.moquette.broker.Server;
+import io.moquette.broker.config.IConfig;
+import io.moquette.broker.config.MemoryConfig;
+import io.moquette.interception.AbstractInterceptHandler;
+import io.moquette.interception.InterceptHandler;
+import io.moquette.interception.messages.InterceptPublishMessage;
+import org.apache.http.HttpStatus;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.Assert.fail;
+
+public class IoTDBTriggerForwardIT {
+ private volatile long count = 0;
+ private volatile Exception exception = null;
+
+ private HttpServer httpServer;
+ private Server mqttServer;
+
+ private final Gson gson = new Gson();
+ private final AtomicLong resultCount = new AtomicLong(0);
+
+ private final Thread dataGenerator =
+ new Thread() {
+ @Override
+ public void run() {
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
"root");
+ Statement statement = connection.createStatement()) {
+
+ do {
+ ++count;
+ statement.execute(
+ String.format(
+ "insert into
root.vehicle.a.b.c.d1(timestamp,s1,s2,s3,s4,s5,s6)
values(%d,%d,%d,%d,%d,%s,'%d')",
+ count, count, count, count, count, count % 2 == 0 ?
"true" : "false", count));
+ } while (!isInterrupted());
+ } catch (Exception e) {
+ exception = e;
+ }
+ }
+ };
+
+ @Before
+ public void setUp() throws Exception {
+ EnvironmentUtils.envSetUp();
+ createTimeseries();
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ EnvironmentUtils.cleanEnv();
+ }
+
+ @Test
+ public void testForwardHTTPTrigger() throws InterruptedException {
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ startHTTPService();
+ statement.execute(
+ "create trigger trigger_forward_http_before before insert on
root.vehicle.a.b.c.d1.s1 "
+ + "as
'org.apache.iotdb.db.engine.trigger.builtin.ForwardTrigger' "
+ + "with ('protocol' = 'http', 'endpoint' =
'http://127.0.0.1:8080/')");
+ statement.execute(
+ "create trigger trigger_forward_http_after after insert on
root.vehicle.a.b.c.d1.s2 "
+ + "as
'org.apache.iotdb.db.engine.trigger.builtin.ForwardTrigger' "
+ + "with ('protocol' = 'http', 'endpoint' =
'http://127.0.0.1:8080/')");
+ startDataGenerator();
+ waitCountIncreaseBy(500);
+ stopDataGenerator();
+ // ensure no exception occurs when inserting data
+ if (exception != null) {
+ fail(exception.getMessage());
+ }
+
+ await().atMost(1, TimeUnit.MINUTES).until(() -> 2 * count ==
resultCount.get());
+ if (exception != null) {
+ fail(exception.getMessage());
+ }
+ } catch (Exception e) {
+ fail(e.getMessage());
+ } finally {
+ if (httpServer != null) {
+ httpServer.stop(0);
+ }
+ stopDataGenerator();
+ }
+ }
+
+ @Test
+ public void testForwardMQTTTrigger() throws InterruptedException {
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ startMQTTService();
+ statement.execute(
+ "create trigger trigger_forward_mqtt_before before insert on
root.vehicle.a.b.c.d1.s3 "
+ + "as
'org.apache.iotdb.db.engine.trigger.builtin.ForwardTrigger' "
+ + "with ('protocol' = 'mqtt', 'host' = '127.0.0.1', 'port' =
'1884',"
+ + " 'username' = 'root', 'password' = 'root', 'topic' =
'mqtt-test')");
+ statement.execute(
+ "create trigger trigger_forward_mqtt_after after insert on
root.vehicle.a.b.c.d1.s4 "
+ + "as
'org.apache.iotdb.db.engine.trigger.builtin.ForwardTrigger' "
+ + "with ('protocol' = 'mqtt', 'host' = '127.0.0.1', 'port' =
'1884',"
+ + " 'username' = 'root', 'password' = 'root', 'topic' =
'mqtt-test')");
+ startDataGenerator();
+ waitCountIncreaseBy(500);
+ stopDataGenerator();
+ // ensure no exception occurs when inserting data
+ if (exception != null) {
+ fail(exception.getMessage());
+ }
+
+ await().atMost(1, TimeUnit.MINUTES).until(() -> 2 * count ==
resultCount.get());
+ if (exception != null) {
+ fail(exception.getMessage());
+ }
+ } catch (SQLException | InterruptedException | IOException e) {
+ fail(e.getMessage());
+ } finally {
+ stopDataGenerator();
+ if (mqttServer != null) {
+ mqttServer.stopServer();
+ }
+ }
+ }
+
+ private void createTimeseries() throws MetadataException {
+ IoTDB.schemaProcessor.createTimeseries(
+ new PartialPath("root.vehicle.a.b.c.d1.s1"),
+ TSDataType.INT32,
+ TSEncoding.PLAIN,
+ CompressionType.UNCOMPRESSED,
+ null);
+ IoTDB.schemaProcessor.createTimeseries(
+ new PartialPath("root.vehicle.a.b.c.d1.s2"),
+ TSDataType.INT64,
+ TSEncoding.PLAIN,
+ CompressionType.UNCOMPRESSED,
+ null);
+ IoTDB.schemaProcessor.createTimeseries(
+ new PartialPath("root.vehicle.a.b.c.d1.s3"),
+ TSDataType.FLOAT,
+ TSEncoding.PLAIN,
+ CompressionType.UNCOMPRESSED,
+ null);
+ IoTDB.schemaProcessor.createTimeseries(
+ new PartialPath("root.vehicle.a.b.c.d1.s4"),
+ TSDataType.DOUBLE,
+ TSEncoding.PLAIN,
+ CompressionType.UNCOMPRESSED,
+ null);
+ IoTDB.schemaProcessor.createTimeseries(
+ new PartialPath("root.vehicle.a.b.c.d1.s5"),
+ TSDataType.BOOLEAN,
+ TSEncoding.PLAIN,
+ CompressionType.UNCOMPRESSED,
+ null);
+ IoTDB.schemaProcessor.createTimeseries(
+ new PartialPath("root.vehicle.a.b.c.d1.s6"),
+ TSDataType.TEXT,
+ TSEncoding.PLAIN,
+ CompressionType.UNCOMPRESSED,
+ null);
+ }
+
+ private void startDataGenerator() {
+ dataGenerator.start();
+ }
+
+ private void stopDataGenerator() throws InterruptedException {
+ if (!dataGenerator.isInterrupted()) {
+ dataGenerator.interrupt();
+ }
+ dataGenerator.join();
+ }
+
+ private void waitCountIncreaseBy(final long increment) throws
InterruptedException {
+ final long previous = count;
+ while (count - previous < increment) {
+ Thread.sleep(100);
+ }
+ }
+
+ private void startHTTPService() throws IOException {
+ httpServer = HttpServer.create(new InetSocketAddress(8080), 0);
+ httpServer.createContext(
+ "/",
+ exchange -> {
+ String entity = "";
+ try {
+ InputStream in = exchange.getRequestBody();
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ byte[] b = new byte[1024 * 8];
+ int len;
+ while ((len = in.read(b)) != -1) {
+ out.write(b, 0, len);
+ }
+ entity = out.toString();
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+
+ if (!checkPayload(entity)) {
+ exception = new IOException("HTTP forward payload error");
+ }
+ JsonArray receiveData = gson.fromJson(entity, JsonArray.class);
+ resultCount.addAndGet(receiveData.size());
+
+ exchange.sendResponseHeaders(HttpStatus.SC_OK, -1);
+ });
+ httpServer.start();
+ }
+
+ private void startMQTTService() throws IOException {
+ Properties properties = new Properties();
+ properties.setProperty(BrokerConstants.HOST_PROPERTY_NAME, "0.0.0.0");
+ properties.setProperty(BrokerConstants.PORT_PROPERTY_NAME, "1884");
+
properties.setProperty(BrokerConstants.BROKER_INTERCEPTOR_THREAD_POOL_SIZE,
"1");
+ IConfig config = new MemoryConfig(properties);
+
+ List<InterceptHandler> handlers = Lists.newArrayList(new
ForwardTestHandler());
+
+ mqttServer = new Server();
+ mqttServer.startServer(config, handlers);
+ }
+
+ private class ForwardTestHandler extends AbstractInterceptHandler {
+ @Override
+ public String getID() {
+ return "forward-test-handler";
+ }
+
+ @Override
+ public void onPublish(InterceptPublishMessage msg) {
+ String payload = msg.getPayload().toString(StandardCharsets.UTF_8);
+ if (!checkPayload(payload)) {
+ exception = new IOException("MQTT forward payload error");
+ }
+ JsonArray receiveData = gson.fromJson(payload, JsonArray.class);
+ resultCount.addAndGet(receiveData.size());
+ }
+ }
+
+ private boolean checkPayload(String payload) {
+ return payload.matches(ForwardEvent.PAYLOADS_FORMATTER_REGEX);
+ }
+}
diff --git a/server/src/assembly/resources/conf/iotdb-engine.properties
b/server/src/assembly/resources/conf/iotdb-engine.properties
index d551acf77d..97c4335e49 100644
--- a/server/src/assembly/resources/conf/iotdb-engine.properties
+++ b/server/src/assembly/resources/conf/iotdb-engine.properties
@@ -1037,4 +1037,20 @@ timestamp_precision=ms
# The cache size for schema page in one schema file
# A bigger cache makes it faster but costs more space and more volatile when
evicts item from cache
# Datatype: int
-# page_cache_in_schema_file=1024
\ No newline at end of file
+# page_cache_in_schema_file=1024
+
+####################
+### Trigger Forward
+####################
+# Number of queues per forwarding trigger
+trigger_forward_max_queue_number=8
+# The length of one of the queues per forwarding trigger
+trigger_forward_max_size_per_queue=2000
+# Trigger forwarding data size per batch
+trigger_forward_batch_size=50
+# Trigger HTTP forward pool size
+trigger_forward_http_pool_size=200
+# Trigger HTTP forward pool max connection for per route
+trigger_forward_http_pool_max_per_route=20
+# Trigger MQTT forward pool size
+trigger_forward_mqtt_pool_size=4
\ No newline at end of file
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 4d3be15291..9087355e00 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -920,6 +920,23 @@ public class IoTDBConfig {
/** Cache expire time of user and role */
private int authorCacheExpireTime = 30;
+ /** Number of queues per forwarding trigger */
+ private int triggerForwardMaxQueueNumber = 8;
+ /** The length of one of the queues per forwarding trigger */
+ private int triggerForwardMaxSizePerQueue = 2000;
+
+ /** Trigger forwarding data size per batch */
+ private int triggerForwardBatchSize = 50;
+
+ /** Trigger HTTP forward pool size */
+ private int triggerForwardHTTPPoolSize = 200;
+
+ /** Trigger HTTP forward pool max connection for per route */
+ private int triggerForwardHTTPPOOLMaxPerRoute = 20;
+
+ /** Trigger MQTT forward pool size */
+ private int triggerForwardMQTTPoolSize = 4;
+
IoTDBConfig() {}
public float getUdfMemoryBudgetInMB() {
@@ -2863,4 +2880,52 @@ public class IoTDBConfig {
public void setAuthorCacheExpireTime(int authorCacheExpireTime) {
this.authorCacheExpireTime = authorCacheExpireTime;
}
+
+ public int getTriggerForwardMaxQueueNumber() {
+ return triggerForwardMaxQueueNumber;
+ }
+
+ public void setTriggerForwardMaxQueueNumber(int
triggerForwardMaxQueueNumber) {
+ this.triggerForwardMaxQueueNumber = triggerForwardMaxQueueNumber;
+ }
+
+ public int getTriggerForwardMaxSizePerQueue() {
+ return triggerForwardMaxSizePerQueue;
+ }
+
+ public void setTriggerForwardMaxSizePerQueue(int
triggerForwardMaxSizePerQueue) {
+ this.triggerForwardMaxSizePerQueue = triggerForwardMaxSizePerQueue;
+ }
+
+ public int getTriggerForwardBatchSize() {
+ return triggerForwardBatchSize;
+ }
+
+ public void setTriggerForwardBatchSize(int triggerForwardBatchSize) {
+ this.triggerForwardBatchSize = triggerForwardBatchSize;
+ }
+
+ public int getTriggerForwardHTTPPoolSize() {
+ return triggerForwardHTTPPoolSize;
+ }
+
+ public void setTriggerForwardHTTPPoolSize(int triggerForwardHTTPPoolSize) {
+ this.triggerForwardHTTPPoolSize = triggerForwardHTTPPoolSize;
+ }
+
+ public int getTriggerForwardHTTPPOOLMaxPerRoute() {
+ return triggerForwardHTTPPOOLMaxPerRoute;
+ }
+
+ public void setTriggerForwardHTTPPOOLMaxPerRoute(int
triggerForwardHTTPPOOLMaxPerRoute) {
+ this.triggerForwardHTTPPOOLMaxPerRoute = triggerForwardHTTPPOOLMaxPerRoute;
+ }
+
+ public int getTriggerForwardMQTTPoolSize() {
+ return triggerForwardMQTTPoolSize;
+ }
+
+ public void setTriggerForwardMQTTPoolSize(int triggerForwardMQTTPoolSize) {
+ this.triggerForwardMQTTPoolSize = triggerForwardMQTTPoolSize;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index f93f78331d..9db6ea1774 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -1514,6 +1514,37 @@ public class IoTDBDescriptor {
if (tlogBufferSize > 0) {
conf.setTlogBufferSize(tlogBufferSize);
}
+
+ conf.setTriggerForwardMaxQueueNumber(
+ Integer.parseInt(
+ properties.getProperty(
+ "trigger_forward_max_queue_number",
+ Integer.toString(conf.getTriggerForwardMaxQueueNumber()))));
+ conf.setTriggerForwardMaxSizePerQueue(
+ Integer.parseInt(
+ properties.getProperty(
+ "trigger_forward_max_size_per_queue",
+ Integer.toString(conf.getTriggerForwardMaxSizePerQueue()))));
+ conf.setTriggerForwardBatchSize(
+ Integer.parseInt(
+ properties.getProperty(
+ "trigger_forward_batch_size",
+ Integer.toString(conf.getTriggerForwardBatchSize()))));
+ conf.setTriggerForwardHTTPPoolSize(
+ Integer.parseInt(
+ properties.getProperty(
+ "trigger_forward_http_pool_size",
+ Integer.toString(conf.getTriggerForwardHTTPPoolSize()))));
+ conf.setTriggerForwardHTTPPOOLMaxPerRoute(
+ Integer.parseInt(
+ properties.getProperty(
+ "trigger_forward_http_pool_max_per_route",
+
Integer.toString(conf.getTriggerForwardHTTPPOOLMaxPerRoute()))));
+ conf.setTriggerForwardMQTTPoolSize(
+ Integer.parseInt(
+ properties.getProperty(
+ "trigger_forward_mqtt_pool_size",
+ Integer.toString(conf.getTriggerForwardMQTTPoolSize()))));
}
private void loadCQProps(Properties properties) {
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/api/Trigger.java
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/api/Trigger.java
index 9a2b9f16b4..48f9c227f3 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/trigger/api/Trigger.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/trigger/api/Trigger.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.engine.trigger.api;
+import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.tsfile.utils.Binary;
/** User Guide: docs/UserGuide/Operation Manual/Triggers.md */
@@ -37,79 +38,79 @@ public interface Trigger {
default void onStop() throws Exception {}
@SuppressWarnings("squid:S112")
- default Integer fire(long timestamp, Integer value) throws Exception {
+ default Integer fire(long timestamp, Integer value, PartialPath path) throws
Exception {
return value;
}
- default int[] fire(long[] timestamps, int[] values) throws Exception {
+ default int[] fire(long[] timestamps, int[] values, PartialPath path) throws
Exception {
int size = timestamps.length;
for (int i = 0; i < size; ++i) {
- fire(timestamps[i], values[i]);
+ fire(timestamps[i], values[i], path);
}
return values;
}
@SuppressWarnings("squid:S112")
- default Long fire(long timestamp, Long value) throws Exception {
+ default Long fire(long timestamp, Long value, PartialPath path) throws
Exception {
return value;
}
- default long[] fire(long[] timestamps, long[] values) throws Exception {
+ default long[] fire(long[] timestamps, long[] values, PartialPath path)
throws Exception {
int size = timestamps.length;
for (int i = 0; i < size; ++i) {
- fire(timestamps[i], values[i]);
+ fire(timestamps[i], values[i], path);
}
return values;
}
@SuppressWarnings("squid:S112")
- default Float fire(long timestamp, Float value) throws Exception {
+ default Float fire(long timestamp, Float value, PartialPath path) throws
Exception {
return value;
}
- default float[] fire(long[] timestamps, float[] values) throws Exception {
+ default float[] fire(long[] timestamps, float[] values, PartialPath path)
throws Exception {
int size = timestamps.length;
for (int i = 0; i < size; ++i) {
- fire(timestamps[i], values[i]);
+ fire(timestamps[i], values[i], path);
}
return values;
}
@SuppressWarnings("squid:S112")
- default Double fire(long timestamp, Double value) throws Exception {
+ default Double fire(long timestamp, Double value, PartialPath path) throws
Exception {
return value;
}
- default double[] fire(long[] timestamps, double[] values) throws Exception {
+ default double[] fire(long[] timestamps, double[] values, PartialPath path)
throws Exception {
int size = timestamps.length;
for (int i = 0; i < size; ++i) {
- fire(timestamps[i], values[i]);
+ fire(timestamps[i], values[i], path);
}
return values;
}
@SuppressWarnings("squid:S112")
- default Boolean fire(long timestamp, Boolean value) throws Exception {
+ default Boolean fire(long timestamp, Boolean value, PartialPath path) throws
Exception {
return value;
}
- default boolean[] fire(long[] timestamps, boolean[] values) throws Exception
{
+ default boolean[] fire(long[] timestamps, boolean[] values, PartialPath
path) throws Exception {
int size = timestamps.length;
for (int i = 0; i < size; ++i) {
- fire(timestamps[i], values[i]);
+ fire(timestamps[i], values[i], path);
}
return values;
}
@SuppressWarnings("squid:S112")
- default Binary fire(long timestamp, Binary value) throws Exception {
+ default Binary fire(long timestamp, Binary value, PartialPath path) throws
Exception {
return value;
}
- default Binary[] fire(long[] timestamps, Binary[] values) throws Exception {
+ default Binary[] fire(long[] timestamps, Binary[] values, PartialPath path)
throws Exception {
int size = timestamps.length;
for (int i = 0; i < size; ++i) {
- fire(timestamps[i], values[i]);
+ fire(timestamps[i], values[i], path);
}
return values;
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/builtin/ForwardTrigger.java
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/builtin/ForwardTrigger.java
new file mode 100644
index 0000000000..0a22cf4d7e
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/builtin/ForwardTrigger.java
@@ -0,0 +1,229 @@
+/*
+ * 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.engine.trigger.builtin;
+
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.engine.trigger.api.Trigger;
+import org.apache.iotdb.db.engine.trigger.api.TriggerAttributes;
+import org.apache.iotdb.db.engine.trigger.sink.api.Configuration;
+import org.apache.iotdb.db.engine.trigger.sink.api.Event;
+import org.apache.iotdb.db.engine.trigger.sink.api.Handler;
+import org.apache.iotdb.db.engine.trigger.sink.exception.SinkException;
+import org.apache.iotdb.db.engine.trigger.sink.http.HTTPForwardConfiguration;
+import org.apache.iotdb.db.engine.trigger.sink.http.HTTPForwardEvent;
+import org.apache.iotdb.db.engine.trigger.sink.http.HTTPForwardHandler;
+import org.apache.iotdb.db.engine.trigger.sink.mqtt.MQTTForwardConfiguration;
+import org.apache.iotdb.db.engine.trigger.sink.mqtt.MQTTForwardEvent;
+import org.apache.iotdb.db.engine.trigger.sink.mqtt.MQTTForwardHandler;
+import org.apache.iotdb.db.engine.trigger.utils.BatchHandlerQueue;
+import org.apache.iotdb.db.exception.TriggerExecutionException;
+import org.apache.iotdb.tsfile.utils.Binary;
+
+import java.util.HashMap;
+
+public class ForwardTrigger implements Trigger {
+
+ private static final String PROTOCOL_HTTP = "http";
+ private static final String PROTOCOL_MQTT = "mqtt";
+
+ private Handler forwardHandler;
+ private Configuration forwardConfig;
+ private BatchHandlerQueue<Event> queue;
+ private final HashMap<String, String> labels = new HashMap<>();
+ private String protocol;
+
+ @Override
+ public void onCreate(TriggerAttributes attributes) throws Exception {
+ protocol = attributes.getStringOrDefault("protocol",
PROTOCOL_HTTP).toLowerCase();
+ int queueNumber = attributes.getIntOrDefault("queueNumber", 8);
+ int queueSize = attributes.getIntOrDefault("queueSize", 2000);
+ int batchSize = attributes.getIntOrDefault("batchSize", 50);
+
+ switch (protocol) {
+ case PROTOCOL_HTTP:
+ forwardConfig = createHTTPConfiguration(attributes);
+ forwardHandler = new HTTPForwardHandler();
+ break;
+ case PROTOCOL_MQTT:
+ forwardConfig = createMQTTConfiguration(attributes);
+ forwardHandler = new MQTTForwardHandler();
+ break;
+ default:
+ throw new TriggerExecutionException("Forward protocol doesn't
support.");
+ }
+ queue = new BatchHandlerQueue<>(queueNumber, queueSize, batchSize,
forwardHandler);
+ forwardHandler.open(forwardConfig);
+ }
+
+ private HTTPForwardConfiguration createHTTPConfiguration(TriggerAttributes
attributes)
+ throws SinkException {
+ String endpoint = attributes.getString("endpoint");
+ boolean stopIfException =
attributes.getBooleanOrDefault("stopIfException", false);
+ HTTPForwardConfiguration forwardConfig =
+ new HTTPForwardConfiguration(endpoint, stopIfException);
+ forwardConfig.checkConfig();
+ return forwardConfig;
+ }
+
+ private MQTTForwardConfiguration createMQTTConfiguration(TriggerAttributes
attributes)
+ throws SinkException {
+ String host = attributes.getString("host");
+ int port = attributes.getInt("port");
+ String username = attributes.getString("username");
+ String password = attributes.getString("password");
+ String topic = attributes.getString("topic");
+ long reconnectDelay = attributes.getLongOrDefault("reconnectDelay", 10L);
+ long connectAttemptsMax =
attributes.getLongOrDefault("connectAttemptsMax", 3L);
+ String qos = attributes.getStringOrDefault("qos", "exactly_once");
+ int poolSize = attributes.getIntOrDefault("poolSize", 4);
+ boolean retain = attributes.getBooleanOrDefault("retain", false);
+ boolean stopIfException =
attributes.getBooleanOrDefault("stopIfException", false);
+
+ MQTTForwardConfiguration forwardConfig =
+ new MQTTForwardConfiguration(
+ host,
+ port,
+ username,
+ password,
+ topic,
+ reconnectDelay,
+ connectAttemptsMax,
+ qos,
+ retain,
+ poolSize,
+ stopIfException);
+ forwardConfig.checkConfig();
+ return forwardConfig;
+ }
+
+ @Override
+ public void onDrop() throws Exception {
+ forwardHandler.close();
+ }
+
+ @Override
+ public void onStart() throws Exception {
+ forwardHandler.open(forwardConfig);
+ }
+
+ @Override
+ public void onStop() throws Exception {
+ forwardHandler.close();
+ }
+
+ private void offerEventToQueue(long timestamp, Object value, PartialPath
path) throws Exception {
+ Event event;
+ switch (protocol) {
+ case PROTOCOL_HTTP:
+ event = new HTTPForwardEvent(timestamp, value, path);
+ break;
+ case PROTOCOL_MQTT:
+ event = new MQTTForwardEvent(timestamp, value, path);
+ break;
+ default:
+ throw new TriggerExecutionException("Forward protocol doesn't
support.");
+ }
+ queue.offer(event);
+ }
+
+ @Override
+ public Integer fire(long timestamp, Integer value, PartialPath path) throws
Exception {
+ offerEventToQueue(timestamp, value, path);
+ return value;
+ }
+
+ @Override
+ public int[] fire(long[] timestamps, int[] values, PartialPath path) throws
Exception {
+ for (int i = 0; i < timestamps.length; i++) {
+ offerEventToQueue(timestamps[i], values[i], path);
+ }
+ return values;
+ }
+
+ @Override
+ public Long fire(long timestamp, Long value, PartialPath path) throws
Exception {
+ offerEventToQueue(timestamp, value, path);
+ return value;
+ }
+
+ @Override
+ public long[] fire(long[] timestamps, long[] values, PartialPath path)
throws Exception {
+ for (int i = 0; i < timestamps.length; i++) {
+ offerEventToQueue(timestamps[i], values[i], path);
+ }
+ return values;
+ }
+
+ @Override
+ public Float fire(long timestamp, Float value, PartialPath path) throws
Exception {
+ offerEventToQueue(timestamp, value, path);
+ return value;
+ }
+
+ @Override
+ public float[] fire(long[] timestamps, float[] values, PartialPath path)
throws Exception {
+ for (int i = 0; i < timestamps.length; i++) {
+ offerEventToQueue(timestamps[i], values[i], path);
+ }
+ return values;
+ }
+
+ @Override
+ public Double fire(long timestamp, Double value, PartialPath path) throws
Exception {
+ offerEventToQueue(timestamp, value, path);
+ return value;
+ }
+
+ @Override
+ public double[] fire(long[] timestamps, double[] values, PartialPath path)
throws Exception {
+ for (int i = 0; i < timestamps.length; i++) {
+ offerEventToQueue(timestamps[i], values[i], path);
+ }
+ return values;
+ }
+
+ @Override
+ public Boolean fire(long timestamp, Boolean value, PartialPath path) throws
Exception {
+ offerEventToQueue(timestamp, value, path);
+ return value;
+ }
+
+ @Override
+ public boolean[] fire(long[] timestamps, boolean[] values, PartialPath path)
throws Exception {
+ for (int i = 0; i < timestamps.length; i++) {
+ offerEventToQueue(timestamps[i], values[i], path);
+ }
+ return values;
+ }
+
+ @Override
+ public Binary fire(long timestamp, Binary value, PartialPath path) throws
Exception {
+ offerEventToQueue(timestamp, value, path);
+ return value;
+ }
+
+ @Override
+ public Binary[] fire(long[] timestamps, Binary[] values, PartialPath path)
throws Exception {
+ for (int i = 0; i < timestamps.length; i++) {
+ offerEventToQueue(timestamps[i], values[i], path);
+ }
+ return values;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/executor/TriggerEngine.java
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/executor/TriggerEngine.java
index fcf17622ce..07605f6d07 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/executor/TriggerEngine.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/executor/TriggerEngine.java
@@ -56,7 +56,12 @@ public class TriggerEngine {
continue;
}
for (TriggerExecutor executor : mNode.getUpperTriggerExecutorList()) {
- executor.fireIfActivated(event, timestamp, values[i],
mNode.getSchema().getType());
+ executor.fireIfActivated(
+ event,
+ timestamp,
+ values[i],
+ mNode.getSchema().getType(),
+ insertRowPlan.getPaths().get(i));
}
}
}
@@ -83,7 +88,12 @@ public class TriggerEngine {
continue;
}
for (TriggerExecutor executor : mNode.getUpperTriggerExecutorList()) {
- executor.fireIfActivated(event, timestamps, columns[i],
mNode.getSchema().getType());
+ executor.fireIfActivated(
+ event,
+ timestamps,
+ columns[i],
+ mNode.getSchema().getType(),
+ insertTabletPlan.getPaths().get(i));
}
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/executor/TriggerExecutor.java
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/executor/TriggerExecutor.java
index ea26758559..732b4c0a62 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/executor/TriggerExecutor.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/executor/TriggerExecutor.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.engine.trigger.executor;
+import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.engine.trigger.api.Trigger;
import org.apache.iotdb.db.engine.trigger.api.TriggerAttributes;
@@ -136,36 +137,37 @@ public class TriggerExecutor {
}
public void fireIfActivated(
- TriggerEvent event, long timestamp, Object value, TSDataType
seriesDataType)
+ TriggerEvent event, long timestamp, Object value, TSDataType
seriesDataType, PartialPath path)
throws TriggerExecutionException {
if (!registrationInformation.isStopped() &&
event.equals(registrationInformation.getEvent())) {
- fire(timestamp, value, seriesDataType);
+ fire(timestamp, value, seriesDataType, path);
}
}
- private synchronized void fire(long timestamp, Object value, TSDataType
seriesDataType)
+ private synchronized void fire(
+ long timestamp, Object value, TSDataType seriesDataType, PartialPath
path)
throws TriggerExecutionException {
Thread.currentThread().setContextClassLoader(classLoader);
try {
switch (seriesDataType) {
case INT32:
- trigger.fire(timestamp, (Integer) value);
+ trigger.fire(timestamp, (Integer) value, path);
break;
case INT64:
- trigger.fire(timestamp, (Long) value);
+ trigger.fire(timestamp, (Long) value, path);
break;
case FLOAT:
- trigger.fire(timestamp, (Float) value);
+ trigger.fire(timestamp, (Float) value, path);
break;
case DOUBLE:
- trigger.fire(timestamp, (Double) value);
+ trigger.fire(timestamp, (Double) value, path);
break;
case BOOLEAN:
- trigger.fire(timestamp, (Boolean) value);
+ trigger.fire(timestamp, (Boolean) value, path);
break;
case TEXT:
- trigger.fire(timestamp, (Binary) value);
+ trigger.fire(timestamp, (Binary) value, path);
break;
default:
throw new TriggerExecutionException("Unsupported series data type.");
@@ -178,36 +180,41 @@ public class TriggerExecutor {
}
public void fireIfActivated(
- TriggerEvent event, long[] timestamps, Object values, TSDataType
seriesDataType)
+ TriggerEvent event,
+ long[] timestamps,
+ Object values,
+ TSDataType seriesDataType,
+ PartialPath path)
throws TriggerExecutionException {
if (!registrationInformation.isStopped() &&
event.equals(registrationInformation.getEvent())) {
- fire(timestamps, values, seriesDataType);
+ fire(timestamps, values, seriesDataType, path);
}
}
- private synchronized void fire(long[] timestamps, Object values, TSDataType
seriesDataType)
+ private synchronized void fire(
+ long[] timestamps, Object values, TSDataType seriesDataType, PartialPath
path)
throws TriggerExecutionException {
Thread.currentThread().setContextClassLoader(classLoader);
try {
switch (seriesDataType) {
case INT32:
- trigger.fire(timestamps, (int[]) values);
+ trigger.fire(timestamps, (int[]) values, path);
break;
case INT64:
- trigger.fire(timestamps, (long[]) values);
+ trigger.fire(timestamps, (long[]) values, path);
break;
case FLOAT:
- trigger.fire(timestamps, (float[]) values);
+ trigger.fire(timestamps, (float[]) values, path);
break;
case DOUBLE:
- trigger.fire(timestamps, (double[]) values);
+ trigger.fire(timestamps, (double[]) values, path);
break;
case BOOLEAN:
- trigger.fire(timestamps, (boolean[]) values);
+ trigger.fire(timestamps, (boolean[]) values, path);
break;
case TEXT:
- trigger.fire(timestamps, (Binary[]) values);
+ trigger.fire(timestamps, (Binary[]) values, path);
break;
default:
throw new TriggerExecutionException("Unsupported series data type.");
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/api/Event.java
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/api/Event.java
index 3b3b94acbc..c46e4dd187 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/api/Event.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/api/Event.java
@@ -19,4 +19,10 @@
package org.apache.iotdb.db.engine.trigger.sink.api;
-public interface Event {}
+import org.apache.iotdb.commons.path.PartialPath;
+
+public interface Event {
+ default PartialPath getFullPath() {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/api/Handler.java
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/api/Handler.java
index dbdf62061e..5f98d2005a 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/api/Handler.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/api/Handler.java
@@ -19,6 +19,8 @@
package org.apache.iotdb.db.engine.trigger.sink.api;
+import java.util.List;
+
public interface Handler<C extends Configuration, E extends Event> {
@SuppressWarnings("squid:S112")
@@ -29,4 +31,7 @@ public interface Handler<C extends Configuration, E extends
Event> {
@SuppressWarnings("squid:S112")
void onEvent(E event) throws Exception;
+
+ @SuppressWarnings("squid:S112")
+ default void onEvent(List<E> events) throws Exception {}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/forward/ForwardEvent.java
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/forward/ForwardEvent.java
new file mode 100644
index 0000000000..c38886ee40
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/forward/ForwardEvent.java
@@ -0,0 +1,70 @@
+/*
+ * 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.engine.trigger.sink.forward;
+
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.engine.trigger.sink.api.Event;
+import org.apache.iotdb.tsfile.utils.Binary;
+
+public class ForwardEvent implements Event {
+ private final long timestamp;
+ private final Object value;
+ private final PartialPath fullPath;
+
+ private static final String PAYLOAD_FORMATTER =
+
"{\"device\":\"%s\",\"measurement\":\"%s\",\"timestamp\":%d,\"value\":%s}";
+
+ public static final String PAYLOADS_FORMATTER_REGEX =
+
"\\[(\\{\"device\":\".*\",\"measurement\":\".*\",\"timestamp\":\\d*,\"value\":.*},)*"
+ +
"(\\{\"device\":\".*\",\"measurement\":\".*\",\"timestamp\":\\d*,\"value\":.*})]";
+
+ public ForwardEvent(long timestamp, Object value, PartialPath fullPath) {
+ this.timestamp = timestamp;
+ this.value = value;
+ this.fullPath = fullPath;
+ }
+
+ public String toJsonString() {
+ return String.format(
+ PAYLOAD_FORMATTER,
+ fullPath.getDevice(),
+ fullPath.getMeasurement(),
+ timestamp,
+ objectToJson(value));
+ }
+
+ private static String objectToJson(Object object) {
+ return (object instanceof String || object instanceof Binary)
+ ? ('\"' + object.toString() + '\"')
+ : object.toString();
+ }
+
+ public long getTimestamp() {
+ return timestamp;
+ }
+
+ public Object getValue() {
+ return value;
+ }
+
+ public PartialPath getFullPath() {
+ return fullPath;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/api/Handler.java
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/http/HTTPForwardConfiguration.java
similarity index 50%
copy from
server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/api/Handler.java
copy to
server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/http/HTTPForwardConfiguration.java
index dbdf62061e..530b784615 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/api/Handler.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/http/HTTPForwardConfiguration.java
@@ -17,16 +17,31 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.trigger.sink.api;
+package org.apache.iotdb.db.engine.trigger.sink.http;
-public interface Handler<C extends Configuration, E extends Event> {
+import org.apache.iotdb.db.engine.trigger.sink.api.Configuration;
+import org.apache.iotdb.db.engine.trigger.sink.exception.SinkException;
- @SuppressWarnings("squid:S112")
- default void open(C configuration) throws Exception {}
+public class HTTPForwardConfiguration implements Configuration {
+ private final String endpoint;
+ private final boolean stopIfException;
- @SuppressWarnings("squid:S112")
- default void close() throws Exception {}
+ public HTTPForwardConfiguration(String endpoint, boolean stopIfException) {
+ this.endpoint = endpoint;
+ this.stopIfException = stopIfException;
+ }
- @SuppressWarnings("squid:S112")
- void onEvent(E event) throws Exception;
+ public void checkConfig() throws SinkException {
+ if (endpoint == null || endpoint.isEmpty()) {
+ throw new SinkException("HTTP config item error");
+ }
+ }
+
+ public boolean isStopIfException() {
+ return stopIfException;
+ }
+
+ public String getEndpoint() {
+ return endpoint;
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/api/Event.java
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/http/HTTPForwardEvent.java
similarity index 69%
copy from
server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/api/Event.java
copy to
server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/http/HTTPForwardEvent.java
index 3b3b94acbc..bfb407a566 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/api/Event.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/http/HTTPForwardEvent.java
@@ -17,6 +17,14 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.trigger.sink.api;
+package org.apache.iotdb.db.engine.trigger.sink.http;
-public interface Event {}
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.engine.trigger.sink.forward.ForwardEvent;
+
+public class HTTPForwardEvent extends ForwardEvent {
+
+ public HTTPForwardEvent(long timestamp, Object value, PartialPath fullPath) {
+ super(timestamp, value, fullPath);
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/http/HTTPForwardHandler.java
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/http/HTTPForwardHandler.java
new file mode 100644
index 0000000000..1c091653ad
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/http/HTTPForwardHandler.java
@@ -0,0 +1,133 @@
+/*
+ * 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.engine.trigger.sink.http;
+
+import org.apache.iotdb.db.engine.trigger.sink.api.Handler;
+import org.apache.iotdb.db.engine.trigger.sink.exception.SinkException;
+import org.apache.iotdb.db.engine.trigger.utils.HTTPConnectionPool;
+
+import org.apache.http.HttpStatus;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.List;
+
+public class HTTPForwardHandler implements Handler<HTTPForwardConfiguration,
HTTPForwardEvent> {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(HTTPForwardHandler.class);
+
+ private static CloseableHttpClient client;
+ private static int referenceCount;
+
+ private HttpPost request;
+ private HTTPForwardConfiguration config;
+
+ private static synchronized void closeClient() throws IOException {
+ if (--referenceCount == 0) {
+ client.close();
+ }
+ }
+
+ private static synchronized void openClient() {
+ if (referenceCount++ == 0) {
+ PoolingHttpClientConnectionManager connectionManager =
HTTPConnectionPool.getInstance();
+ client =
HttpClients.custom().setConnectionManager(connectionManager).build();
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ closeClient();
+ }
+
+ @Override
+ public void open(HTTPForwardConfiguration config) {
+ this.config = config;
+ if (this.request == null) {
+ this.request = new HttpPost(config.getEndpoint());
+ request.setHeader("Accept", "application/json");
+ request.setHeader("Content-type", "application/json");
+ }
+
+ openClient();
+ }
+
+ @Override
+ public void onEvent(HTTPForwardEvent event) throws SinkException {
+ CloseableHttpResponse response = null;
+ try {
+ request.setEntity(new StringEntity("[" + event.toJsonString() + "]"));
+ response = client.execute(request);
+ if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) {
+ throw new SinkException(response.getStatusLine().toString());
+ }
+ } catch (Exception e) {
+ if (config.isStopIfException()) {
+ throw new SinkException("HTTP Forward Exception", e);
+ }
+ LOGGER.error("HTTP Forward Exception", e);
+ } finally {
+ try {
+ if (null != response) {
+ response.close();
+ }
+ } catch (IOException e) {
+ LOGGER.error("Connection Close Exception", e);
+ }
+ }
+ }
+
+ @Override
+ public void onEvent(List<HTTPForwardEvent> events) throws SinkException {
+ CloseableHttpResponse response = null;
+ try {
+ StringBuilder sb = new StringBuilder().append("[");
+ for (HTTPForwardEvent event : events) {
+ sb.append(event.toJsonString()).append(", ");
+ }
+ sb.replace(sb.lastIndexOf(", "), sb.length(), "").append("]");
+ request.setEntity(new StringEntity(sb.toString()));
+ response = client.execute(request);
+ if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) {
+ throw new SinkException(response.getStatusLine().toString());
+ }
+ } catch (Exception e) {
+ if (config.isStopIfException()) {
+ throw new SinkException("HTTP Forward Exception", e);
+ }
+ LOGGER.error("HTTP Forward Exception", e);
+ } finally {
+ try {
+ if (null != response) {
+ response.close();
+ }
+ } catch (IOException e) {
+ LOGGER.error("Connection Close Exception", e);
+ }
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/mqtt/MQTTForwardConfiguration.java
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/mqtt/MQTTForwardConfiguration.java
new file mode 100644
index 0000000000..820ac925fe
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/mqtt/MQTTForwardConfiguration.java
@@ -0,0 +1,137 @@
+/*
+ * 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.engine.trigger.sink.mqtt;
+
+import org.apache.iotdb.db.engine.trigger.sink.api.Configuration;
+import org.apache.iotdb.db.engine.trigger.sink.exception.SinkException;
+
+import org.fusesource.mqtt.client.QoS;
+
+public class MQTTForwardConfiguration implements Configuration {
+ private final String host;
+ private final int port;
+ private final String username;
+ private final String password;
+ private final String topic;
+ private final long reconnectDelay;
+ private final long connectAttemptsMax;
+ private final QoS qos;
+ private final boolean retain;
+ private final int poolSize;
+ private final boolean stopIfException;
+
+ public MQTTForwardConfiguration(
+ String host,
+ int port,
+ String username,
+ String password,
+ String topic,
+ long reconnectDelay,
+ long connectAttemptsMax,
+ String qos,
+ boolean retain,
+ int poolSize,
+ boolean stopIfException)
+ throws SinkException {
+ this.host = host;
+ this.port = port;
+ this.username = username;
+ this.password = password;
+ this.topic = topic;
+ this.reconnectDelay = reconnectDelay;
+ this.connectAttemptsMax = connectAttemptsMax;
+ this.qos = parseQoS(qos);
+ this.retain = retain;
+ this.poolSize = poolSize;
+ this.stopIfException = stopIfException;
+ }
+
+ private static QoS parseQoS(String qos) throws SinkException {
+ switch (qos.toLowerCase()) {
+ case "exactly_once":
+ return QoS.EXACTLY_ONCE;
+ case "at_least_once":
+ return QoS.AT_LEAST_ONCE;
+ case "at_most_once":
+ return QoS.AT_MOST_ONCE;
+ default:
+ throw new SinkException("Unable to identify QoS config");
+ }
+ }
+
+ public void checkConfig() throws SinkException {
+ if (host == null
+ || host.isEmpty()
+ || port < 0
+ || port > 65535
+ || username == null
+ || username.isEmpty()
+ || password == null
+ || password.isEmpty()
+ || topic == null
+ || topic.isEmpty()) {
+ throw new SinkException("MQTT config item error");
+ }
+ }
+
+ public String getHost() {
+ return host;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ public String getUsername() {
+ return username;
+ }
+
+ public String getPassword() {
+ return password;
+ }
+
+ public String getTopic() {
+ return topic;
+ }
+
+ public long getReconnectDelay() {
+ return reconnectDelay;
+ }
+
+ public long getConnectAttemptsMax() {
+ return connectAttemptsMax;
+ }
+
+ public QoS getQos() {
+ return qos;
+ }
+
+ public boolean isRetain() {
+ return retain;
+ }
+
+ public int getPoolSize() {
+ return poolSize;
+ }
+
+ public boolean isStopIfException() {
+ return stopIfException;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/api/Event.java
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/mqtt/MQTTForwardEvent.java
similarity index 69%
copy from
server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/api/Event.java
copy to
server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/mqtt/MQTTForwardEvent.java
index 3b3b94acbc..bd99824ab1 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/api/Event.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/mqtt/MQTTForwardEvent.java
@@ -17,6 +17,14 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.trigger.sink.api;
+package org.apache.iotdb.db.engine.trigger.sink.mqtt;
-public interface Event {}
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.engine.trigger.sink.forward.ForwardEvent;
+
+public class MQTTForwardEvent extends ForwardEvent {
+
+ public MQTTForwardEvent(long timestamp, Object value, PartialPath fullPath) {
+ super(timestamp, value, fullPath);
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/mqtt/MQTTForwardHandler.java
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/mqtt/MQTTForwardHandler.java
new file mode 100644
index 0000000000..175a9be5b9
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/sink/mqtt/MQTTForwardHandler.java
@@ -0,0 +1,97 @@
+/*
+ * 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.engine.trigger.sink.mqtt;
+
+import org.apache.iotdb.db.engine.trigger.sink.api.Handler;
+import org.apache.iotdb.db.engine.trigger.sink.exception.SinkException;
+import org.apache.iotdb.db.engine.trigger.utils.MQTTConnectionFactory;
+import org.apache.iotdb.db.engine.trigger.utils.MQTTConnectionPool;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+public class MQTTForwardHandler implements Handler<MQTTForwardConfiguration,
MQTTForwardEvent> {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(MQTTForwardHandler.class);
+
+ private MQTTConnectionPool connectionPool;
+ private MQTTForwardConfiguration config;
+
+ @Override
+ public void open(MQTTForwardConfiguration config) throws Exception {
+ this.config = config;
+ MQTTConnectionFactory factory =
+ new MQTTConnectionFactory(
+ config.getHost(),
+ config.getPort(),
+ config.getUsername(),
+ config.getPassword(),
+ config.getConnectAttemptsMax(),
+ config.getReconnectDelay());
+ connectionPool =
+ MQTTConnectionPool.getInstance(
+ config.getHost(),
+ config.getPort(),
+ config.getUsername(),
+ factory,
+ config.getPoolSize());
+ }
+
+ @Override
+ public void close() throws Exception {
+ connectionPool.clearAndClose();
+ }
+
+ @Override
+ public void onEvent(MQTTForwardEvent event) throws SinkException {
+ try {
+ connectionPool.publish(
+ config.getTopic(),
+ ("[" + event.toJsonString() + "]").getBytes(),
+ config.getQos(),
+ config.isRetain());
+ } catch (Exception e) {
+ if (config.isStopIfException()) {
+ throw new SinkException("MQTT Forward Exception", e);
+ }
+ LOGGER.error("MQTT Forward Exception", e);
+ }
+ }
+
+ @Override
+ public void onEvent(List<MQTTForwardEvent> events) throws SinkException {
+ StringBuilder sb = new StringBuilder().append("[");
+ for (MQTTForwardEvent event : events) {
+ sb.append(event.toJsonString()).append(", ");
+ }
+ sb.replace(sb.lastIndexOf(", "), sb.length(), "").append("]");
+ try {
+ connectionPool.publish(
+ config.getTopic(), sb.toString().getBytes(), config.getQos(),
config.isRetain());
+ } catch (Exception e) {
+ if (config.isStopIfException()) {
+ throw new SinkException("MQTT Forward Exception", e);
+ }
+ LOGGER.error("MQTT Forward Exception", e);
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/utils/BatchHandlerQueue.java
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/utils/BatchHandlerQueue.java
new file mode 100644
index 0000000000..bf5d478e8a
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/utils/BatchHandlerQueue.java
@@ -0,0 +1,150 @@
+/*
+ * 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.engine.trigger.utils;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.trigger.sink.api.Event;
+import org.apache.iotdb.db.engine.trigger.sink.api.Handler;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Each Trigger instantiate a ForwardQueue
+ *
+ * @param <T> Subclass of Event
+ */
+public class BatchHandlerQueue<T extends Event> {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(BatchHandlerQueue.class);
+
+ private final int queueNumber;
+ private final int queueSize;
+ private final int batchSize;
+ private final AtomicInteger atomicCount = new AtomicInteger();
+
+ private final ArrayBlockingQueue<T>[] queues;
+
+ private final Handler handler;
+
+ public BatchHandlerQueue(int queueNumber, int queueSize, int batchSize,
Handler handler) {
+ this.queueNumber =
+ Math.min(
+ queueNumber,
+
IoTDBDescriptor.getInstance().getConfig().getTriggerForwardMaxQueueNumber());
+ this.queueSize =
+ Math.min(
+ queueSize,
+
IoTDBDescriptor.getInstance().getConfig().getTriggerForwardMaxSizePerQueue());
+ this.batchSize =
+ Math.min(batchSize,
IoTDBDescriptor.getInstance().getConfig().getTriggerForwardBatchSize());
+ this.handler = handler;
+ queues = new ArrayBlockingQueue[this.queueNumber];
+ for (int i = 0; i < queueNumber; i++) {
+ queues[i] = new ArrayBlockingQueue<>(this.queueSize);
+ Thread t =
+ new ForwardQueueConsumer(
+ handler.getClass().getSimpleName()
+ + "-"
+ + BatchHandlerQueue.class.getSimpleName()
+ + "-"
+ + i,
+ queues[i]);
+ t.setDaemon(true);
+ t.start();
+ }
+ }
+
+ private int getQueueID(int hashCode) {
+ return (hashCode & 0x7FFFFFFF) % queues.length;
+ }
+
+ public boolean offer(T event) {
+ // Group by device or polling
+ if (event.getFullPath() != null) {
+ return
queues[getQueueID(event.getFullPath().getDevice().hashCode())].offer(event);
+ } else {
+ return queues[getQueueID(atomicCount.incrementAndGet())].offer(event);
+ }
+ }
+
+ public void put(T event) throws InterruptedException {
+ // Group by device or polling
+ if (event.getFullPath() != null) {
+
queues[getQueueID(event.getFullPath().getDevice().hashCode())].put(event);
+ } else {
+ queues[getQueueID(atomicCount.incrementAndGet())].put(event);
+ }
+ }
+
+ private void handle(ArrayList<T> events) throws Exception {
+ handler.onEvent(events);
+ }
+
+ class ForwardQueueConsumer extends Thread {
+
+ ArrayBlockingQueue<T> queue;
+
+ public ForwardQueueConsumer(String name, ArrayBlockingQueue<T> queue) {
+ super(name);
+ this.queue = queue;
+ }
+
+ public void run() {
+ final long maxWaitMillis = 500;
+ final ArrayList<T> list = new ArrayList<>();
+ long startMillis = System.currentTimeMillis();
+ long restMillis = maxWaitMillis;
+ while (true) {
+ try {
+ T obj;
+ if (list.isEmpty()) {
+ obj = queue.take();
+ } else {
+ obj = queue.poll(restMillis, TimeUnit.MILLISECONDS);
+ }
+ if (obj != null) {
+ list.add(obj);
+ queue.drainTo(list, batchSize - list.size());
+ if (list.size() < batchSize) {
+ long waitMillis = System.currentTimeMillis() - startMillis;
+ if (waitMillis < maxWaitMillis) {
+ restMillis = maxWaitMillis - waitMillis;
+ continue;
+ }
+ }
+ }
+ handle(list);
+ list.clear();
+ startMillis = System.currentTimeMillis();
+ } catch (InterruptedException e) {
+ break;
+ } catch (Throwable t) {
+ LOGGER.error("ForwardTaskQueue consumer error", t);
+ }
+ }
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/utils/HTTPConnectionPool.java
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/utils/HTTPConnectionPool.java
new file mode 100644
index 0000000000..0a6503287d
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/utils/HTTPConnectionPool.java
@@ -0,0 +1,49 @@
+/*
+ * 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.engine.trigger.utils;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+
+import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
+
+public class HTTPConnectionPool {
+
+ private static volatile PoolingHttpClientConnectionManager
clientConnectionManager;
+
+ private HTTPConnectionPool() {}
+
+ public static PoolingHttpClientConnectionManager getInstance() {
+ if (clientConnectionManager == null) {
+ synchronized (HTTPConnectionPool.class) {
+ if (clientConnectionManager == null) {
+ clientConnectionManager = new PoolingHttpClientConnectionManager();
+ // Set the max number of connections
+ clientConnectionManager.setMaxTotal(
+
IoTDBDescriptor.getInstance().getConfig().getTriggerForwardHTTPPoolSize());
+ // Set the maximum number of connections per host and the specified
number of connections
+ // per website, which will not affect the access of other websites
+ clientConnectionManager.setDefaultMaxPerRoute(
+
IoTDBDescriptor.getInstance().getConfig().getTriggerForwardHTTPPOOLMaxPerRoute());
+ }
+ }
+ }
+ return clientConnectionManager;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/utils/MQTTConnectionFactory.java
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/utils/MQTTConnectionFactory.java
new file mode 100644
index 0000000000..4de66a12fa
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/utils/MQTTConnectionFactory.java
@@ -0,0 +1,115 @@
+/*
+ * 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.engine.trigger.utils;
+
+import org.apache.iotdb.db.engine.trigger.sink.exception.SinkException;
+
+import org.apache.commons.pool2.BasePooledObjectFactory;
+import org.apache.commons.pool2.PooledObject;
+import org.apache.commons.pool2.impl.DefaultPooledObject;
+import org.fusesource.mqtt.client.BlockingConnection;
+import org.fusesource.mqtt.client.MQTT;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class MQTTConnectionFactory extends
BasePooledObjectFactory<BlockingConnection> {
+ private final String host;
+ private final int port;
+ private final String username;
+ private final String password;
+ private final long connectAttemptsMax;
+ private final long reconnectDelay;
+
+ private static final AtomicInteger atomicCount = new AtomicInteger();
+ private static final String CLIENT_NAME = "MQTTClient";
+
+ public MQTTConnectionFactory(
+ String host,
+ int port,
+ String username,
+ String password,
+ long connectAttemptsMax,
+ long reconnectDelay) {
+ this.host = host;
+ this.port = port;
+ this.username = username;
+ this.password = password;
+ this.connectAttemptsMax = connectAttemptsMax;
+ this.reconnectDelay = reconnectDelay;
+ }
+
+ @Override
+ public BlockingConnection create() throws Exception {
+ MQTT mqtt = new MQTT();
+ mqtt.setClientId(CLIENT_NAME + atomicCount.incrementAndGet());
+ mqtt.setHost(host, port);
+ mqtt.setUserName(username);
+ mqtt.setPassword(password);
+ mqtt.setConnectAttemptsMax(connectAttemptsMax);
+ mqtt.setReconnectDelay(reconnectDelay);
+
+ BlockingConnection connection = mqtt.blockingConnection();
+ try {
+ connection.connect();
+ } catch (Exception e) {
+ if (connection != null) {
+ if (connection.isConnected()) {
+ connection.disconnect();
+ }
+ connection.kill();
+ }
+ throw new SinkException("MQTT Connection activate error", e);
+ }
+ return connection;
+ }
+
+ @Override
+ public PooledObject<BlockingConnection> wrap(BlockingConnection
blockingConnection) {
+ return new DefaultPooledObject<>(blockingConnection);
+ }
+
+ @Override
+ public boolean validateObject(PooledObject<BlockingConnection> p) {
+ if (p == null) {
+ return false;
+ }
+ BlockingConnection connection = p.getObject();
+ return connection != null && connection.isConnected();
+ }
+
+ @Override
+ public void destroyObject(PooledObject<BlockingConnection> p) throws
Exception {
+ if (p == null) {
+ return;
+ }
+ BlockingConnection connection = p.getObject();
+ try {
+ if (connection != null) {
+ if (connection.isConnected()) {
+ connection.disconnect();
+ }
+ connection.kill();
+ }
+ } catch (Exception e) {
+ throw new SinkException("MQTT connection destroy error", e);
+ }
+ super.destroyObject(p);
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/utils/MQTTConnectionPool.java
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/utils/MQTTConnectionPool.java
new file mode 100644
index 0000000000..fbc775a1a3
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/utils/MQTTConnectionPool.java
@@ -0,0 +1,79 @@
+/*
+ * 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.engine.trigger.utils;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+
+import org.apache.commons.pool2.impl.GenericObjectPool;
+import org.fusesource.mqtt.client.BlockingConnection;
+import org.fusesource.mqtt.client.QoS;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class MQTTConnectionPool extends GenericObjectPool<BlockingConnection> {
+
+ // Each host:port,username corresponds to a singleton instance
+ private static final Map<String, MQTTConnectionPool>
MQTT_CONNECTION_POOL_MAP =
+ new ConcurrentHashMap<>();
+ private final AtomicInteger referenceCount = new AtomicInteger(0);
+
+ public static MQTTConnectionPool getInstance(
+ String host, int port, String username, MQTTConnectionFactory factory,
int size)
+ throws Exception {
+ String key = host + ":" + port + "," + username;
+ MQTTConnectionPool connectionPool =
+ MQTT_CONNECTION_POOL_MAP.computeIfAbsent(key, k -> new
MQTTConnectionPool(factory, size));
+ if (connectionPool.referenceCount.getAndIncrement() == 0) {
+ connectionPool.preparePool();
+ }
+ return connectionPool;
+ }
+
+ private MQTTConnectionPool(MQTTConnectionFactory factory, int size) {
+ super(factory);
+ setMaxTotal(
+ Math.min(size,
IoTDBDescriptor.getInstance().getConfig().getTriggerForwardMQTTPoolSize()));
+ setMinIdle(1);
+ }
+
+ public void connect() throws Exception {
+ BlockingConnection connection = borrowObject();
+ if (!connection.isConnected()) {
+ connection.connect();
+ }
+ returnObject(connection);
+ }
+
+ public void clearAndClose() {
+ clear();
+ if (referenceCount.decrementAndGet() == 0) {
+ close();
+ }
+ }
+
+ public void publish(final String topic, final byte[] payload, final QoS qos,
final boolean retain)
+ throws Exception {
+ BlockingConnection connection = this.borrowObject();
+ connection.publish(topic, payload, qos, retain);
+ returnObject(connection);
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/mqtt/JSONPayloadFormatter.java
b/server/src/main/java/org/apache/iotdb/db/protocol/mqtt/JSONPayloadFormatter.java
index 2436fcac0d..3493252501 100644
---
a/server/src/main/java/org/apache/iotdb/db/protocol/mqtt/JSONPayloadFormatter.java
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/mqtt/JSONPayloadFormatter.java
@@ -20,8 +20,11 @@ package org.apache.iotdb.db.protocol.mqtt;
import com.google.common.collect.Lists;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParseException;
+import com.google.gson.JsonSyntaxException;
import com.google.gson.reflect.TypeToken;
import io.netty.buffer.ByteBuf;
@@ -50,13 +53,28 @@ public class JSONPayloadFormatter implements
PayloadFormatter {
return null;
}
String txt = payload.toString(StandardCharsets.UTF_8);
- JsonObject jsonObject = GSON.fromJson(txt, JsonObject.class);
+ try {
+ JsonObject jsonObject = GSON.fromJson(txt, JsonObject.class);
- if (jsonObject.get(JSON_KEY_TIMESTAMP) != null) {
- return formatJson(jsonObject);
- }
- if (jsonObject.get(JSON_KEY_TIMESTAMPS) != null) {
- return formatBatchJson(jsonObject);
+ if (jsonObject.get(JSON_KEY_TIMESTAMP) != null) {
+ return formatJson(jsonObject);
+ }
+ if (jsonObject.get(JSON_KEY_TIMESTAMPS) != null) {
+ return formatBatchJson(jsonObject);
+ }
+ } catch (JsonSyntaxException e) {
+ JsonArray jsonArray = GSON.fromJson(txt, JsonArray.class);
+ List<Message> messages = new ArrayList<>();
+ for (JsonElement jsonElement : jsonArray) {
+ JsonObject jsonObject = jsonElement.getAsJsonObject();
+ if (jsonObject.get(JSON_KEY_TIMESTAMP) != null) {
+ messages.addAll(formatJson(jsonObject));
+ }
+ if (jsonObject.get(JSON_KEY_TIMESTAMPS) != null) {
+ messages.addAll(formatBatchJson(jsonObject));
+ }
+ }
+ return messages;
}
throw new JsonParseException("payload is invalidate");
}
diff --git
a/server/src/test/java/org/apache/iotdb/db/metadata/idtable/trigger_example/Counter.java
b/server/src/test/java/org/apache/iotdb/db/metadata/idtable/trigger_example/Counter.java
index 4db46caa94..3c86592de5 100644
---
a/server/src/test/java/org/apache/iotdb/db/metadata/idtable/trigger_example/Counter.java
+++
b/server/src/test/java/org/apache/iotdb/db/metadata/idtable/trigger_example/Counter.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.metadata.idtable.trigger_example;
+import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.engine.trigger.api.Trigger;
import org.apache.iotdb.db.engine.trigger.api.TriggerAttributes;
import org.apache.iotdb.tsfile.utils.Binary;
@@ -47,37 +48,37 @@ public class Counter implements Trigger {
}
@Override
- public Integer fire(long timestamp, Integer value) {
+ public Integer fire(long timestamp, Integer value, PartialPath path) {
++counter;
return value;
}
@Override
- public Long fire(long timestamp, Long value) {
+ public Long fire(long timestamp, Long value, PartialPath path) {
++counter;
return value;
}
@Override
- public Float fire(long timestamp, Float value) {
+ public Float fire(long timestamp, Float value, PartialPath path) {
++counter;
return value;
}
@Override
- public Double fire(long timestamp, Double value) {
+ public Double fire(long timestamp, Double value, PartialPath path) {
++counter;
return value;
}
@Override
- public Boolean fire(long timestamp, Boolean value) {
+ public Boolean fire(long timestamp, Boolean value, PartialPath path) {
++counter;
return value;
}
@Override
- public Binary fire(long timestamp, Binary value) {
+ public Binary fire(long timestamp, Binary value, PartialPath path) {
++counter;
return value;
}
