[IOTDB-6236] Pipe: CountPointProcessor example

example/pipe-count-point-processor/pom.xml

@@ -0,0 +1,66 @@
+<?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>1.3.1-SNAPSHOT</version>
+    </parent>
+    <artifactId>pipe-count-point-processor-example</artifactId>
+    <name>IoTDB: Example: Pipe: Count Point Processor</name>
+    <properties>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.iotdb</groupId>
+            <artifactId>pipe-api</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.iotdb</groupId>
+            <artifactId>iotdb-server</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.iotdb</groupId>
+            <artifactId>common-api</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-jar-plugin</artifactId>
+                <version>3.2.0</version>
+                <configuration/>
+            </plugin>
+        </plugins>
+    </build>
+</project>

example/pipe-count-point-processor/src/main/java/org/apache/iotdb/CountPointProcessor.java

@@ -0,0 +1,83 @@
+/*
+ * 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;
+
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
+import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
+import org.apache.iotdb.pipe.api.PipeProcessor;
+import org.apache.iotdb.pipe.api.collector.EventCollector;
+import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+import org.apache.iotdb.pipe.api.event.Event;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
+import org.apache.iotdb.tsfile.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class CountPointProcessor implements PipeProcessor {
+  private static final String AGGREGATE_SERIES_KEY = "aggregate-series";
+  private static AtomicLong writePointCount = new AtomicLong(0);
+
+  private PartialPath aggregateSeries;
+
+  @Override
+  public void validate(PipeParameterValidator validator) throws Exception {
+    validator.validateRequiredAttribute(AGGREGATE_SERIES_KEY);
+  }
+
+  @Override
+  public void customize(PipeParameters parameters, PipeProcessorRuntimeConfiguration configuration)
+      throws Exception {
+    this.aggregateSeries = new PartialPath(parameters.getString(AGGREGATE_SERIES_KEY));
+  }
+
+  @Override
+  public void process(TabletInsertionEvent tabletInsertionEvent, EventCollector eventCollector)
+      throws Exception {
+    tabletInsertionEvent.processTablet(
+        (tablet, rowCollector) -> {
+          writePointCount.addAndGet(tablet.rowSize);
+        });
+  }
+
+  @Override
+  public void process(Event event, EventCollector eventCollector) throws Exception {
+    if (event instanceof PipeHeartbeatEvent) {
+      Tablet tablet =
+          new Tablet(
+              aggregateSeries.getDevice(),
+              Collections.singletonList(
+                  new MeasurementSchema(aggregateSeries.getMeasurement(), TSDataType.INT64)),
+              1);
+      tablet.rowSize = 1;
+      tablet.addTimestamp(0, System.currentTimeMillis());
+      tablet.addValue(aggregateSeries.getMeasurement(), 0, writePointCount.get());
+      eventCollector.collect(new PipeRawTabletInsertionEvent(tablet, false, null, null, false));
+    }
+  }
+
+  @Override
+  public void close() throws Exception {}
+}

example/opc-ua-sink/pom.xml → example/pipe-opc-ua-sink/pom.xml

@@ -19,18 +19,15 @@
     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">
-
+<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">
     <parent>
         <groupId>org.apache.iotdb</groupId>
         <artifactId>iotdb-examples</artifactId>
         <version>1.3.1-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
-    <artifactId>opc-ua-sink-example</artifactId>
-    <name>IoTDB: Example: OPCUA Sink</name>
+    <artifactId>pipe-opc-ua-sink-example</artifactId>
+    <name>IoTDB: Example: Pipe: OPCUA Sink</name>
     <dependencies>
         <dependency>
             <groupId>org.eclipse.milo</groupId>

example/pom.xml

@@ -39,14 +39,15 @@
         <module>flink</module>
         <module>mqtt</module>
         <module>mqtt-customize</module>
-        <module>opc-ua-sink</module>
+        <module>pipe-opc-ua-sink</module>
         <module>pulsar</module>
         <module>udf</module>
         <module>trigger</module>
         <module>rabbitmq</module>
         <module>rest-java-example</module>
         <module>flink-sql</module>
         <module>schema</module>
+        <module>pipe-count-point-processor</module>
     </modules>
     <build>
         <pluginManagement>