You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ph...@apache.org on 2018/01/10 11:12:01 UTC
[19/19] nifi-minifi-cpp git commit: MINIFICPP-342: MQTT extension
MINIFICPP-342: MQTT extension
This closes #228.
Signed-off-by: Marc Parisi <ph...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/commit/a8703b5c
Tree: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/tree/a8703b5c
Diff: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/diff/a8703b5c
Branch: refs/heads/master
Commit: a8703b5c79380ae3a24ca3099572bf67b10e6f46
Parents: 0da444e
Author: Bin Qiu <be...@gmail.com>
Authored: Thu Jan 4 08:23:14 2018 -0800
Committer: Marc Parisi <ph...@apache.org>
Committed: Wed Jan 10 06:11:21 2018 -0500
----------------------------------------------------------------------
CMakeLists.txt | 7 +
Extensions.md | 1 +
LICENSE | 16 +
PROCESSORS.md | 62 +
README.md | 2 +
bootstrap.sh | 20 +-
extensions/mqtt/AbstractMQTTProcessor.cpp | 158 +
extensions/mqtt/AbstractMQTTProcessor.h | 154 +
extensions/mqtt/CMakeLists.txt | 75 +
extensions/mqtt/ConsumeMQTT.cpp | 110 +
extensions/mqtt/ConsumeMQTT.h | 122 +
extensions/mqtt/MQTTLoader.cpp | 29 +
extensions/mqtt/MQTTLoader.h | 72 +
extensions/mqtt/PublishMQTT.cpp | 106 +
extensions/mqtt/PublishMQTT.h | 142 +
libminifi/test/mqtt-tests/CMakeLists.txt | 39 +
thirdparty/paho.mqtt.c/.gitignore | 6 +
thirdparty/paho.mqtt.c/.gitreview | 5 +
thirdparty/paho.mqtt.c/.pydevproject | 5 +
thirdparty/paho.mqtt.c/CMakeLists.txt | 86 +
thirdparty/paho.mqtt.c/CONTRIBUTING.md | 66 +
thirdparty/paho.mqtt.c/Makefile | 283 ++
thirdparty/paho.mqtt.c/README.md | 215 ++
.../MQTTVersion/MQTTVersion.vcxproj | 166 +
.../MQTTVersion/MQTTVersion.vcxproj.filters | 36 +
.../MQTTVersion/MQTTVersion.vcxproj.user | 4 +
.../Windows Build/Paho C MQTT APIs.sln | 155 +
.../paho-mqtt3a/paho-mqtt3a.vcxproj | 202 ++
.../paho-mqtt3a/paho-mqtt3a.vcxproj.filters | 153 +
.../paho-mqtt3a/paho-mqtt3a.vcxproj.user | 3 +
.../paho-mqtt3as/paho-mqtt3as.vcxproj | 204 ++
.../paho-mqtt3as/paho-mqtt3as.vcxproj.filters | 147 +
.../paho-mqtt3as/paho-mqtt3as.vcxproj.user | 4 +
.../paho-mqtt3c/paho-mqtt3c.vcxproj | 172 +
.../paho-mqtt3c/paho-mqtt3c.vcxproj.filters | 79 +
.../paho-mqtt3c/paho-mqtt3c.vcxproj.user | 3 +
.../paho-mqtt3cs/paho-mqtt3cs.vcxproj | 204 ++
.../paho-mqtt3cs/paho-mqtt3cs.vcxproj.filters | 147 +
.../paho-mqtt3cs/paho-mqtt3cs.vcxproj.user | 4 +
.../Windows Build/stdoutsub/stdoutsub.vcxproj | 175 +
.../stdoutsub/stdoutsub.vcxproj.filters | 30 +
.../stdoutsub/stdoutsub.vcxproj.user | 3 +
.../Windows Build/stdoutsuba/stdoutsuba.vcxproj | 170 +
.../stdoutsuba/stdoutsuba.vcxproj.filters | 30 +
.../stdoutsuba/stdoutsuba.vcxproj.user | 3 +
.../Windows Build/test1/test1.vcxproj | 173 +
.../Windows Build/test1/test1.vcxproj.filters | 30 +
.../Windows Build/test2/test2.vcxproj | 177 +
.../Windows Build/test2/test2.vcxproj.filters | 45 +
.../Windows Build/test3/test3.vcxproj | 178 +
.../Windows Build/test3/test3.vcxproj.filters | 30 +
.../Windows Build/test4/test4.vcxproj | 173 +
.../Windows Build/test4/test4.vcxproj.filters | 30 +
.../Windows Build/test5/test5.vcxproj | 173 +
.../Windows Build/test5/test5.vcxproj.filters | 30 +
.../Windows Build/test9/test9.vcxproj | 170 +
.../Windows Build/test9/test9.vcxproj.filters | 30 +
thirdparty/paho.mqtt.c/about.html | 28 +
thirdparty/paho.mqtt.c/android/Android.mk | 140 +
thirdparty/paho.mqtt.c/appveyor.yml | 50 +
thirdparty/paho.mqtt.c/build.xml | 316 ++
thirdparty/paho.mqtt.c/cbuild.bat | 19 +
.../paho.mqtt.c/cmake/CPackDebConfig.cmake.in | 91 +
.../cmake/modules/CMakeDebHelper.cmake | 74 +
.../cmake/modules/CMakeDebHelperInstall.cmake | 17 +
.../cmake/toolchain.linux-arm11.cmake | 10 +
.../paho.mqtt.c/cmake/toolchain.win32.cmake | 15 +
.../paho.mqtt.c/cmake/toolchain.win64.cmake | 15 +
thirdparty/paho.mqtt.c/debian/CMakeLists.txt | 16 +
thirdparty/paho.mqtt.c/dist/Makefile | 11 +
thirdparty/paho.mqtt.c/dist/paho-c.spec | 78 +
thirdparty/paho.mqtt.c/doc/CMakeLists.txt | 40 +
thirdparty/paho.mqtt.c/doc/DoxyfileV3AsyncAPI | 1803 ++++++++++
.../paho.mqtt.c/doc/DoxyfileV3AsyncAPI.in | 1804 ++++++++++
thirdparty/paho.mqtt.c/doc/DoxyfileV3ClientAPI | 1803 ++++++++++
.../paho.mqtt.c/doc/DoxyfileV3ClientAPI.in | 1804 ++++++++++
.../paho.mqtt.c/doc/DoxyfileV3ClientInternal | 1851 ++++++++++
.../paho.mqtt.c/doc/DoxyfileV3ClientInternal.in | 1852 ++++++++++
thirdparty/paho.mqtt.c/doc/pahologo.png | Bin 0 -> 12197 bytes
thirdparty/paho.mqtt.c/edl-v10 | 15 +
thirdparty/paho.mqtt.c/epl-v10 | 70 +
thirdparty/paho.mqtt.c/notice.html | 108 +
thirdparty/paho.mqtt.c/src/CMakeLists.txt | 165 +
thirdparty/paho.mqtt.c/src/Clients.c | 55 +
thirdparty/paho.mqtt.c/src/Clients.h | 209 ++
thirdparty/paho.mqtt.c/src/Heap.c | 481 +++
thirdparty/paho.mqtt.c/src/Heap.h | 76 +
thirdparty/paho.mqtt.c/src/LinkedList.c | 500 +++
thirdparty/paho.mqtt.c/src/LinkedList.h | 105 +
thirdparty/paho.mqtt.c/src/Log.c | 572 ++++
thirdparty/paho.mqtt.c/src/Log.h | 85 +
thirdparty/paho.mqtt.c/src/MQTTAsync.c | 3227 ++++++++++++++++++
thirdparty/paho.mqtt.c/src/MQTTAsync.h | 1728 ++++++++++
thirdparty/paho.mqtt.c/src/MQTTClient.c | 2102 ++++++++++++
thirdparty/paho.mqtt.c/src/MQTTClient.h | 1396 ++++++++
.../paho.mqtt.c/src/MQTTClientPersistence.h | 254 ++
thirdparty/paho.mqtt.c/src/MQTTPacket.c | 755 ++++
thirdparty/paho.mqtt.c/src/MQTTPacket.h | 262 ++
thirdparty/paho.mqtt.c/src/MQTTPacketOut.c | 269 ++
thirdparty/paho.mqtt.c/src/MQTTPacketOut.h | 34 +
thirdparty/paho.mqtt.c/src/MQTTPersistence.c | 654 ++++
thirdparty/paho.mqtt.c/src/MQTTPersistence.h | 74 +
.../paho.mqtt.c/src/MQTTPersistenceDefault.c | 841 +++++
.../paho.mqtt.c/src/MQTTPersistenceDefault.h | 33 +
thirdparty/paho.mqtt.c/src/MQTTProtocol.h | 46 +
thirdparty/paho.mqtt.c/src/MQTTProtocolClient.c | 769 +++++
thirdparty/paho.mqtt.c/src/MQTTProtocolClient.h | 55 +
thirdparty/paho.mqtt.c/src/MQTTProtocolOut.c | 242 ++
thirdparty/paho.mqtt.c/src/MQTTProtocolOut.h | 46 +
thirdparty/paho.mqtt.c/src/MQTTVersion.c | 230 ++
thirdparty/paho.mqtt.c/src/Messages.c | 104 +
thirdparty/paho.mqtt.c/src/Messages.h | 24 +
thirdparty/paho.mqtt.c/src/OsWrapper.c | 28 +
thirdparty/paho.mqtt.c/src/OsWrapper.h | 42 +
thirdparty/paho.mqtt.c/src/SSLSocket.c | 917 +++++
thirdparty/paho.mqtt.c/src/SSLSocket.h | 51 +
thirdparty/paho.mqtt.c/src/Socket.c | 898 +++++
thirdparty/paho.mqtt.c/src/Socket.h | 143 +
thirdparty/paho.mqtt.c/src/SocketBuffer.c | 413 +++
thirdparty/paho.mqtt.c/src/SocketBuffer.h | 84 +
thirdparty/paho.mqtt.c/src/StackTrace.c | 207 ++
thirdparty/paho.mqtt.c/src/StackTrace.h | 71 +
thirdparty/paho.mqtt.c/src/Thread.c | 462 +++
thirdparty/paho.mqtt.c/src/Thread.h | 73 +
thirdparty/paho.mqtt.c/src/Tree.c | 724 ++++
thirdparty/paho.mqtt.c/src/Tree.h | 115 +
thirdparty/paho.mqtt.c/src/VersionInfo.h.in | 7 +
.../paho.mqtt.c/src/samples/CMakeLists.txt | 65 +
.../paho.mqtt.c/src/samples/MQTTAsync_publish.c | 154 +
.../src/samples/MQTTAsync_subscribe.c | 190 ++
.../src/samples/MQTTClient_publish.c | 60 +
.../src/samples/MQTTClient_publish_async.c | 96 +
.../src/samples/MQTTClient_subscribe.c | 95 +
thirdparty/paho.mqtt.c/src/samples/paho_c_pub.c | 379 ++
thirdparty/paho.mqtt.c/src/samples/paho_c_sub.c | 359 ++
.../paho.mqtt.c/src/samples/paho_cs_pub.c | 276 ++
.../paho.mqtt.c/src/samples/paho_cs_sub.c | 270 ++
thirdparty/paho.mqtt.c/src/utf-8.c | 230 ++
thirdparty/paho.mqtt.c/src/utf-8.h | 23 +
thirdparty/paho.mqtt.c/travis-build.sh | 15 +
thirdparty/paho.mqtt.c/travis-env-vars | 2 +
thirdparty/paho.mqtt.c/travis-install.sh | 18 +
thirdparty/paho.mqtt.c/travis-macos-vars | 2 +
143 files changed, 38998 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/CMakeLists.txt b/CMakeLists.txt
index e6f89a8..05d3425 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -172,6 +172,12 @@ if (ENABLE_ALL OR ENABLE_GPS)
createExtension(GPS-EXTENSION "GPS EXTENSIONS" "Enables LibGPS Functionality and the GetGPS processor." "extensions/gps" "${TEST_DIR}/gps-tests")
endif()
+## Create MQTT Extension
+option(ENABLE_MQTT "Enables the mqtt extension." OFF)
+if(ENABLE_ALL OR ENABLE_MQTT)
+ createExtension(MQTT-EXTENSIONS "MQTT EXTENSIONS" "This Enables MQTT functionality including PublishMQTT/ConsumeMQTT" "extensions/mqtt" "${TEST_DIR}/mqtt-tests" "TRUE" "thirdparty/paho.mqtt.c")
+endif()
+
option(ENABLE_PCAP "Enables the PCAP extension." OFF)
if(ENABLE_ALL OR ENABLE_PCAP)
createExtension(PCAP-EXTENSION "PCAP EXTENSIONS" "Enables libPCAP Functionality and the PacketCapture processor." "extensions/pcap" "${TEST_DIR}/pcap-tests")
@@ -246,6 +252,7 @@ set(CPACK_PACKAGE_FILE_NAME "${ASSEMBLY_BASE_NAME}")
set(CPACK_BINARY_TGZ, "ON")
set(CPACK_ARCHIVE_COMPONENT_INSTALL ON)
set(CPACK_RDKAFKA_COMPONENT_INSTALL ON)
+set(CPACK_MQTT_COMPONENT_INSTALL ON)
set(CPACK_COMPONENTS_ALL bin)
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/Extensions.md
----------------------------------------------------------------------
diff --git a/Extensions.md b/Extensions.md
index 24ef07e..3ccaf6c 100644
--- a/Extensions.md
+++ b/Extensions.md
@@ -27,6 +27,7 @@ Currently we support the following extension flags:
- `-DENABLE_PCAP=TRUE`
- `-DENABLE_GPS=TRUE`
- `-DENABLE_TENSORFLOW=TRUE`
+ - `-DENABLE_MQTT=TRUE`
For more information on these extensions, please visit [the Extension How-To on our wiki](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=74685143)
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/LICENSE
----------------------------------------------------------------------
diff --git a/LICENSE b/LICENSE
index b4a2c77..a270b73 100644
--- a/LICENSE
+++ b/LICENSE
@@ -1037,3 +1037,19 @@ SHALL THE COPYRIGHT HOLDERS OR ANYONE DISTRIBUTING THE SOFTWARE BE LIABLE
FOR ANY DAMAGES OR OTHER LIABILITY, WHETHER IN CONTRACT, TORT OR OTHERWISE,
ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
DEALINGS IN THE SOFTWARE.
+
+This product bundles 'paho mqtt' which is available under a Eclipse Software License.
+
+Eclipse Distribution License - v 1.0
+
+Copyright (c) 2007, Eclipse Foundation, Inc. and its licensors.
+
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:
+
+ Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.
+ Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.
+ Neither the name of the Eclipse Foundation, Inc. nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/PROCESSORS.md
----------------------------------------------------------------------
diff --git a/PROCESSORS.md b/PROCESSORS.md
index 09c7d24..6c4eedf 100644
--- a/PROCESSORS.md
+++ b/PROCESSORS.md
@@ -723,3 +723,65 @@ default values, and whether a property supports the NiFi Expression Language.
| success | Any FlowFile that is successfully sent to Kafka will be routed to this Relationship |
| failure | Any FlowFile that cannot be sent to Kafka will be routed to this Relationship |
+## PublishMQTT
+
+This Processor puts the contents of a FlowFile to a MQTT broker for a sepcified topic. The
+content of a FlowFile becomes the payload of the MQTT message.
+
+### Properties
+
+In the list below, the names of required properties appear in bold. Any other
+properties (not in bold) are considered optional. The table also indicates any
+default values, and whether a property supports the NiFi Expression Language.
+
+| Name | Default Value | Allowable Values | Description |
+| - | - | - | - |
+| **Broker URI** | | | The URI to use to connect to the MQTT broker |
+| **Topic** | | | The topic to publish the message to |
+| Session state | | | Whether to start afresh or resume previous flows |
+| Client ID | | | MQTT client ID to use |
+| Username | | | Username to use when connecting to the broker |
+| Password | | | Password to use when connecting to the broker |
+| Keep Alive Interval | | | Defines the maximum time interval between messages sent or received |
+| Connection Timeout | | | Maximum time interval the client will wait for the network connection to the MQTT server |
+| Quality of Service | | | The Quality of Service(QoS) to send the message with. Accepts three values '0', '1' and '2' |
+| Retain | | | Retain MQTT published record in broker |
+| Max Flow Segment Size | | Maximum flow content payload segment size for the MQTT record |
+
+### Relationships
+
+| Name | Description |
+| - | - |
+| success | Any FlowFile that is successfully sent to broker will be routed to this Relationship |
+| failure | Any FlowFile that cannot be sent to broker will be routed to this Relationship |
+
+## ConsumeMQTT
+
+This Processor gets the contents of a FlowFile from a MQTT broker for a sepcified topic. The
+the payload of the MQTT message becomes content of a FlowFile
+
+### Properties
+
+In the list below, the names of required properties appear in bold. Any other
+properties (not in bold) are considered optional. The table also indicates any
+default values, and whether a property supports the NiFi Expression Language.
+
+| Name | Default Value | Allowable Values | Description |
+| - | - | - | - |
+| **Broker URI** | | | The URI to use to connect to the MQTT broker |
+| **Topic** | | | The topic to publish the message to |
+| Session state | | | Whether to start afresh or resume previous flows |
+| Client ID | | | MQTT client ID to use |
+| Username | | | Username to use when connecting to the broker |
+| Password | | | Password to use when connecting to the broker |
+| Keep Alive Interval | | | Defines the maximum time interval between messages sent or received |
+| Connection Timeout | | | Maximum time interval the client will wait for the network connection to the MQTT server |
+| Quality of Service | | | The Quality of Service(QoS) to send the message with. Accepts three values '0', '1' and '2' |
+| Max Flow Segment Size | | Maximum flow content payload segment size for the MQTT record |
+
+### Relationships
+
+| Name | Description |
+| - | - |
+| success | Any FlowFile that is successfully sent to broker will be routed to this Relationship |
+
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/README.md
----------------------------------------------------------------------
diff --git a/README.md b/README.md
index e22e464..40f7c97 100644
--- a/README.md
+++ b/README.md
@@ -65,6 +65,8 @@ MiNiFi - C++ supports the following processors:
* [UnfocusArchiveEntry](PROCESSORS.md#unfocusarchiveentry)
* [ManipulateArchive](PROCESSORS.md#manipulatearchive)
* [PublishKafka](PROCESSORS.md#publishkafka)
+* [PublishMQTT](PROCESSORS.md#publishMQTT)
+* [ConsumeMQTT](PROCESSORS.md#consumeMQTT)
## Caveats
* 0.4.0 represents a non-GA release, APIs and interfaces are subject to change
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/bootstrap.sh
----------------------------------------------------------------------
diff --git a/bootstrap.sh b/bootstrap.sh
index 270a4a5..bc8b1de 100755
--- a/bootstrap.sh
+++ b/bootstrap.sh
@@ -209,6 +209,8 @@ add_dependency GPS_ENABLED "gpsd"
add_disabled_option KAFKA_ENABLED ${FALSE} "ENABLE_LIBRDKAFKA" "3.4.0"
+add_disabled_option MQTT_ENABLED ${FALSE} "ENABLE_MQTT"
+
#add_disabled_option BUSTACHE_ENABLED ${FALSE} "ENABLE_BUSTACHE"
#add_dependency BUSTACHE_ENABLED "boost"
@@ -360,10 +362,11 @@ show_supported_features() {
echo "I. GPS support .................$(print_feature_status GPS_ENABLED)"
echo "J. TensorFlow Support ..........$(print_feature_status TENSORFLOW_ENABLED)"
echo "K. Bustache Support ............$(print_feature_status BUSTACHE_ENABLED)"
- echo "L. Enable all extensions"
- echo "M. Portable Build ..............$(print_feature_status PORTABLE_BUILD)"
- echo "N. Build with Debug symbols ....$(print_feature_status DEBUG_SYMBOLS)"
- echo "O. Continue with these options"
+ echo "L. MQTT Support ................$(print_feature_status MQTT_ENABLED)"
+ echo "M. Enable all extensions"
+ echo "N. Portable Build ..............$(print_feature_status PORTABLE_BUILD)"
+ echo "O. Build with Debug symbols ....$(print_feature_status DEBUG_SYMBOLS)"
+ echo "P. Continue with these options"
echo "Q. Exit"
echo "* Extension cannot be installed due to"
echo -e " version of cmake or other software\r\n"
@@ -384,10 +387,11 @@ read_options(){
i) ToggleFeature GPS_ENABLED ;;
j) ToggleFeature TENSORFLOW_ENABLED ;;
k) ToggleFeature BUSTACHE_ENABLED ;;
- l) EnableAllFeatures ;;
- m) ToggleFeature PORTABLE_BUILD ;;
- n) ToggleFeature DEBUG_SYMBOLS ;;
- o) FEATURES_SELECTED="true" ;;
+ l) ToggleFeature MQTT_ENABLED ;;
+ m) EnableAllFeatures ;;
+ n) ToggleFeature PORTABLE_BUILD ;;
+ o) ToggleFeature DEBUG_SYMBOLS ;;
+ p) FEATURES_SELECTED="true" ;;
q) exit 0;;
*) echo -e "${RED}Please enter an option A-L...${NO_COLOR}" && sleep 2
esac
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/extensions/mqtt/AbstractMQTTProcessor.cpp
----------------------------------------------------------------------
diff --git a/extensions/mqtt/AbstractMQTTProcessor.cpp b/extensions/mqtt/AbstractMQTTProcessor.cpp
new file mode 100644
index 0000000..9ce052d
--- /dev/null
+++ b/extensions/mqtt/AbstractMQTTProcessor.cpp
@@ -0,0 +1,158 @@
+/**
+ * @file AbstractMQTTProcessor.cpp
+ * AbstractMQTTProcessor class implementation
+ *
+ * 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.
+ */
+#include "AbstractMQTTProcessor.h"
+#include <stdio.h>
+#include <memory>
+#include <string>
+#include "utils/TimeUtil.h"
+#include "utils/StringUtils.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace processors {
+
+core::Property AbstractMQTTProcessor::BrokerURL("Broker URI", "The URI to use to connect to the MQTT broker", "");
+core::Property AbstractMQTTProcessor::CleanSession("Session state", "Whether to start afresh or resume previous flows. See the allowable value descriptions for more details", "true");
+core::Property AbstractMQTTProcessor::ClientID("Client ID", "MQTT client ID to use", "");
+core::Property AbstractMQTTProcessor::UserName("Username", "Username to use when connecting to the broker", "");
+core::Property AbstractMQTTProcessor::PassWord("Password", "Password to use when connecting to the broker", "");
+core::Property AbstractMQTTProcessor::KeepLiveInterval("Keep Alive Interval", "Defines the maximum time interval between messages sent or received", "60 sec");
+core::Property AbstractMQTTProcessor::ConnectionTimeOut("Connection Timeout", "Maximum time interval the client will wait for the network connection to the MQTT server", "30 sec");
+core::Property AbstractMQTTProcessor::QOS("Quality of Service", "The Quality of Service(QoS) to send the message with. Accepts three values '0', '1' and '2'", "MQTT_QOS_0");
+core::Property AbstractMQTTProcessor::Topic("Topic", "The topic to publish the message to", "");
+core::Relationship AbstractMQTTProcessor::Success("success", "FlowFiles that are sent successfully to the destination are transferred to this relationship");
+core::Relationship AbstractMQTTProcessor::Failure("failure", "FlowFiles that failed to send to the destination are transferred to this relationship");
+
+void AbstractMQTTProcessor::initialize() {
+ // Set the supported properties
+ std::set<core::Property> properties;
+ properties.insert(BrokerURL);
+ properties.insert(CleanSession);
+ properties.insert(ClientID);
+ properties.insert(UserName);
+ properties.insert(PassWord);
+ properties.insert(KeepLiveInterval);
+ properties.insert(ConnectionTimeOut);
+ properties.insert(QOS);
+ properties.insert(Topic);
+ setSupportedProperties(properties);
+ // Set the supported relationships
+ std::set<core::Relationship> relationships;
+ relationships.insert(Success);
+ relationships.insert(Failure);
+ setSupportedRelationships(relationships);
+}
+
+void AbstractMQTTProcessor::onSchedule(core::ProcessContext *context, core::ProcessSessionFactory *sessionFactory) {
+ std::string value;
+ int64_t valInt;
+ value = "";
+ if (context->getProperty(BrokerURL.getName(), value) && !value.empty()) {
+ uri_ = value;
+ logger_->log_info("AbstractMQTTProcessor: BrokerURL [%s]", uri_);
+ }
+ value = "";
+ if (context->getProperty(ClientID.getName(), value) && !value.empty()) {
+ clientID_ = value;
+ logger_->log_info("AbstractMQTTProcessor: ClientID [%s]", clientID_);
+ }
+ value = "";
+ if (context->getProperty(Topic.getName(), value) && !value.empty()) {
+ topic_ = value;
+ logger_->log_info("AbstractMQTTProcessor: Topic [%s]", topic_);
+ }
+ value = "";
+ if (context->getProperty(UserName.getName(), value) && !value.empty()) {
+ userName_ = value;
+ logger_->log_info("AbstractMQTTProcessor: UserName [%s]", userName_);
+ }
+ value = "";
+ if (context->getProperty(PassWord.getName(), value) && !value.empty()) {
+ passWord_ = value;
+ logger_->log_info("AbstractMQTTProcessor: PassWord [%s]", passWord_);
+ }
+ value = "";
+ if (context->getProperty(CleanSession.getName(), value) && !value.empty() &&
+ org::apache::nifi::minifi::utils::StringUtils::StringToBool(value, cleanSession_)) {
+ logger_->log_info("AbstractMQTTProcessor: CleanSession [%d]", cleanSession_);
+ }
+ value = "";
+ if (context->getProperty(KeepLiveInterval.getName(), value) && !value.empty()) {
+ core::TimeUnit unit;
+ if (core::Property::StringToTime(value, valInt, unit) && core::Property::ConvertTimeUnitToMS(valInt, unit, valInt)) {
+ keepAliveInterval_ = valInt/1000;
+ logger_->log_info("AbstractMQTTProcessor: KeepLiveInterval [%ll]", keepAliveInterval_);
+ }
+ }
+ value = "";
+ if (context->getProperty(ConnectionTimeOut.getName(), value) && !value.empty()) {
+ core::TimeUnit unit;
+ if (core::Property::StringToTime(value, valInt, unit) && core::Property::ConvertTimeUnitToMS(valInt, unit, valInt)) {
+ connectionTimeOut_ = valInt/1000;
+ logger_->log_info("AbstractMQTTProcessor: ConnectionTimeOut [%ll]", connectionTimeOut_);
+ }
+ }
+ value = "";
+ if (context->getProperty(QOS.getName(), value) && !value.empty() && (value == MQTT_QOS_0 || value == MQTT_QOS_1 || MQTT_QOS_2) &&
+ core::Property::StringToInt(value, valInt)) {
+ qos_ = valInt;
+ logger_->log_info("AbstractMQTTProcessor: QOS [%ll]", qos_);
+ }
+ if (!client_) {
+ MQTTClient_create(&client_, uri_.c_str(), clientID_.c_str(), MQTTCLIENT_PERSISTENCE_NONE, NULL);
+ }
+ if (client_) {
+ MQTTClient_setCallbacks(client_, (void *) this, connectionLost, msgReceived, msgDelivered);
+ // call reconnect to bootstrap
+ this->reconnect();
+ }
+}
+
+bool AbstractMQTTProcessor::reconnect() {
+ if (!client_)
+ return false;
+ if (MQTTClient_isConnected(client_))
+ return true;
+ MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
+ conn_opts.keepAliveInterval = keepAliveInterval_;
+ conn_opts.cleansession = cleanSession_;
+ if (!userName_.empty()) {
+ conn_opts.username = userName_.c_str();
+ conn_opts.password = passWord_.c_str();
+ }
+ if (MQTTClient_connect(client_, &conn_opts) != MQTTCLIENT_SUCCESS) {
+ logger_->log_error("Failed to connect to MQTT broker %s", uri_);
+ return false;
+ }
+ if (isSubscriber_) {
+ MQTTClient_subscribe(client_, topic_.c_str(), qos_);
+ }
+ return true;
+}
+
+} /* namespace processors */
+} /* namespace minifi */
+} /* namespace nifi */
+} /* namespace apache */
+} /* namespace org */
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/extensions/mqtt/AbstractMQTTProcessor.h
----------------------------------------------------------------------
diff --git a/extensions/mqtt/AbstractMQTTProcessor.h b/extensions/mqtt/AbstractMQTTProcessor.h
new file mode 100644
index 0000000..7870c2b
--- /dev/null
+++ b/extensions/mqtt/AbstractMQTTProcessor.h
@@ -0,0 +1,154 @@
+/**
+ * @file AbstractMQTTProcessor.h
+ * AbstractMQTTProcessor class declaration
+ *
+ * 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.
+ */
+#ifndef __ABSTRACTMQTT_H__
+#define __ABSTRACTMQTT_H__
+
+#include "FlowFileRecord.h"
+#include "core/Processor.h"
+#include "core/ProcessSession.h"
+#include "core/Core.h"
+#include "core/Resource.h"
+#include "core/logging/LoggerConfiguration.h"
+#include "MQTTClient.h"
+
+#define MQTT_QOS_0 "0"
+#define MQTT_QOS_1 "1"
+#define MQTT_QOS_2 "2"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace processors {
+
+// AbstractMQTTProcessor Class
+class AbstractMQTTProcessor : public core::Processor {
+ public:
+ // Constructor
+ /*!
+ * Create a new processor
+ */
+ explicit AbstractMQTTProcessor(std::string name, uuid_t uuid = NULL)
+ : core::Processor(name, uuid),
+ logger_(logging::LoggerFactory<AbstractMQTTProcessor>::getLogger()) {
+ client_ = nullptr;
+ cleanSession_ = false;
+ keepAliveInterval_ = 60;
+ connectionTimeOut_ = 30;
+ qos_ = 0;
+ isSubscriber_ = false;
+ }
+ // Destructor
+ virtual ~AbstractMQTTProcessor() {
+ if (isSubscriber_) {
+ MQTTClient_unsubscribe(client_, topic_.c_str());
+ }
+ if (client_ && MQTTClient_isConnected(client_)) {
+ MQTTClient_disconnect(client_, connectionTimeOut_);
+ }
+ if (client_)
+ MQTTClient_destroy(&client_);
+ }
+ // Processor Name
+ static constexpr char const* ProcessorName = "AbstractMQTTProcessor";
+ // Supported Properties
+ static core::Property BrokerURL;
+ static core::Property ClientID;
+ static core::Property UserName;
+ static core::Property PassWord;
+ static core::Property CleanSession;
+ static core::Property KeepLiveInterval;
+ static core::Property ConnectionTimeOut;
+ static core::Property Topic;
+ static core::Property QOS;
+
+ // Supported Relationships
+ static core::Relationship Failure;
+ static core::Relationship Success;
+
+ public:
+ /**
+ * Function that's executed when the processor is scheduled.
+ * @param context process context.
+ * @param sessionFactory process session factory that is used when creating
+ * ProcessSession objects.
+ */
+ virtual void onSchedule(core::ProcessContext *context, core::ProcessSessionFactory *sessionFactory);
+ // OnTrigger method, implemented by NiFi AbstractMQTTProcessor
+ virtual void onTrigger(core::ProcessContext *context, core::ProcessSession *session) {
+ }
+ // OnTrigger method, implemented by NiFi AbstractMQTTProcessor
+ virtual void onTrigger(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) {
+ }
+ // Initialize, over write by NiFi AbstractMQTTProcessor
+ virtual void initialize(void);
+ // MQTT async callbacks
+ static void msgDelivered(void *context, MQTTClient_deliveryToken dt) {
+ AbstractMQTTProcessor *processor = (AbstractMQTTProcessor *) context;
+ processor->delivered_token_ = dt;
+ }
+ static int msgReceived(void *context, char *topicName, int topicLen, MQTTClient_message *message) {
+ AbstractMQTTProcessor *processor = (AbstractMQTTProcessor *) context;
+ if (processor->isSubscriber_) {
+ if (!processor->enqueueReceiveMQTTMsg(message))
+ MQTTClient_freeMessage(&message);
+ } else {
+ MQTTClient_freeMessage(&message);
+ }
+ MQTTClient_free(topicName);
+ return 1;
+ }
+ static void connectionLost(void *context, char *cause) {
+ AbstractMQTTProcessor *processor = (AbstractMQTTProcessor *) context;
+ processor->reconnect();
+ }
+ bool reconnect();
+ // enqueue receive MQTT message
+ virtual bool enqueueReceiveMQTTMsg(MQTTClient_message *message) {
+ return false;
+ }
+
+ protected:
+ MQTTClient client_;
+ MQTTClient_deliveryToken delivered_token_;
+ std::string uri_;
+ std::string topic_;
+ int64_t keepAliveInterval_;
+ int64_t connectionTimeOut_;
+ int64_t qos_;
+ bool cleanSession_;
+ std::string clientID_;
+ std::string userName_;
+ std::string passWord_;
+ bool isSubscriber_;
+
+ private:
+ std::shared_ptr<logging::Logger> logger_;
+};
+
+REGISTER_RESOURCE(AbstractMQTTProcessor);
+
+} /* namespace processors */
+} /* namespace minifi */
+} /* namespace nifi */
+} /* namespace apache */
+} /* namespace org */
+
+#endif
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/extensions/mqtt/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/extensions/mqtt/CMakeLists.txt b/extensions/mqtt/CMakeLists.txt
new file mode 100644
index 0000000..7e186ba
--- /dev/null
+++ b/extensions/mqtt/CMakeLists.txt
@@ -0,0 +1,75 @@
+#
+# 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.
+#
+
+set(CMAKE_EXE_LINKER_FLAGS "-Wl,--export-all-symbols")
+set(CMAKE_SHARED_LINKER_FLAGS "-Wl,--export-symbols")
+
+include_directories(../../libminifi/include ../../libminifi/include/core ../../thirdparty/spdlog-20170710/include ../../thirdparty/concurrentqueue ../../thirdparty/yaml-cpp-yaml-cpp-0.5.3/include ${CIVET_THIRDPARTY_ROOT}/include ../../thirdparty/jsoncpp/include ../../thirdparty/)
+
+include_directories(../../thirdparty/paho.mqtt.c/src)
+
+file(GLOB SOURCES "*.cpp")
+
+add_library(minifi-mqtt-extensions STATIC ${SOURCES})
+set_property(TARGET minifi-mqtt-extensions PROPERTY POSITION_INDEPENDENT_CODE ON)
+if(THREADS_HAVE_PTHREAD_ARG)
+ target_compile_options(PUBLIC minifi-mqtt-extensions "-pthread")
+endif()
+if(CMAKE_THREAD_LIBS_INIT)
+ target_link_libraries(minifi-mqtt-extensions "${CMAKE_THREAD_LIBS_INIT}")
+endif()
+
+
+# Include UUID
+find_package(UUID REQUIRED)
+target_link_libraries(minifi-mqtt-extensions ${LIBMINIFI} ${UUID_LIBRARIES} ${JSONCPP_LIB})
+add_dependencies(minifi-mqtt-extensions jsoncpp_project)
+find_package(OpenSSL REQUIRED)
+include_directories(${OPENSSL_INCLUDE_DIR})
+target_link_libraries(minifi-mqtt-extensions ${CMAKE_DL_LIBS} )
+if (MQTT_FOUND AND NOT BUILD_MQTT)
+ target_link_libraries(minifi-mqtt-extensions ${MQTT_LIBRARIES} )
+else()
+ target_link_libraries(minifi-mqtt-extensions paho-mqtt3a )
+ target_link_libraries(minifi-mqtt-extensions paho-mqtt3c )
+ target_link_libraries(minifi-mqtt-extensions paho-mqtt3as )
+ target_link_libraries(minifi-mqtt-extensions paho-mqtt3cs )
+endif()
+find_package(ZLIB REQUIRED)
+include_directories(${ZLIB_INCLUDE_DIRS})
+target_link_libraries (minifi-mqtt-extensions ${ZLIB_LIBRARIES})
+if (WIN32)
+ set_target_properties(minifi-mqtt-extensions PROPERTIES
+ LINK_FLAGS "/WHOLEMQTT"
+ )
+elseif (APPLE)
+ set_target_properties(minifi-mqtt-extensions PROPERTIES
+ LINK_FLAGS "-Wl,-all_load"
+ )
+else ()
+ set_target_properties(minifi-mqtt-extensions PROPERTIES
+ LINK_FLAGS "-Wl,--whole-mqtt"
+ )
+endif ()
+
+
+SET (MQTT-EXTENSIONS minifi-mqtt-extensions PARENT_SCOPE)
+
+register_extension(minifi-mqtt-extensions)
+
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/extensions/mqtt/ConsumeMQTT.cpp
----------------------------------------------------------------------
diff --git a/extensions/mqtt/ConsumeMQTT.cpp b/extensions/mqtt/ConsumeMQTT.cpp
new file mode 100644
index 0000000..12de6bf
--- /dev/null
+++ b/extensions/mqtt/ConsumeMQTT.cpp
@@ -0,0 +1,110 @@
+/**
+ * @file ConsumeMQTT.cpp
+ * ConsumeMQTT class implementation
+ *
+ * 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.
+ */
+#include "ConsumeMQTT.h"
+#include <stdio.h>
+#include <algorithm>
+#include <memory>
+#include <string>
+#include <map>
+#include <set>
+#include "utils/TimeUtil.h"
+#include "utils/StringUtils.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace processors {
+
+core::Property ConsumeMQTT::MaxQueueSize("Max Flow Segment Size", "Maximum flow content payload segment size for the MQTT record", "");
+
+void ConsumeMQTT::initialize() {
+ // Set the supported properties
+ std::set<core::Property> properties;
+ properties.insert(BrokerURL);
+ properties.insert(CleanSession);
+ properties.insert(ClientID);
+ properties.insert(UserName);
+ properties.insert(PassWord);
+ properties.insert(KeepLiveInterval);
+ properties.insert(ConnectionTimeOut);
+ properties.insert(QOS);
+ properties.insert(Topic);
+ properties.insert(MaxQueueSize);
+ setSupportedProperties(properties);
+ // Set the supported relationships
+ std::set<core::Relationship> relationships;
+ relationships.insert(Success);
+ setSupportedRelationships(relationships);
+}
+
+bool ConsumeMQTT::enqueueReceiveMQTTMsg(MQTTClient_message *message) {
+ if (queue_.size_approx() >= maxQueueSize_) {
+ logger_->log_debug("MQTT queue full");
+ return false;
+ } else {
+ queue_.enqueue(message);
+ logger_->log_debug("enqueue MQTT message length %d", message->payloadlen);
+ return true;
+ }
+}
+
+void ConsumeMQTT::onSchedule(core::ProcessContext *context, core::ProcessSessionFactory *sessionFactory) {
+ AbstractMQTTProcessor::onSchedule(context, sessionFactory);
+ std::string value;
+ int64_t valInt;
+ value = "";
+ if (context->getProperty(MaxQueueSize.getName(), value) && !value.empty() && core::Property::StringToInt(value, valInt)) {
+ maxQueueSize_ = valInt;
+ logger_->log_info("ConsumeMQTT: max queue size [%ll]", maxQueueSize_);
+ }
+}
+
+void ConsumeMQTT::onTrigger(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) {
+ // reconnect if necessary
+ reconnect();
+ std::deque<MQTTClient_message *> msg_queue;
+ getReceivedMQTTMsg(msg_queue);
+ while (!msg_queue.empty()) {
+ MQTTClient_message *message = msg_queue.front();
+ std::shared_ptr<core::FlowFile> processFlowFile = session->create();
+ ConsumeMQTT::WriteCallback callback(message);
+ session->write(processFlowFile, &callback);
+ if (callback.status_ < 0) {
+ logger_->log_error("ConsumeMQTT fail for the flow with UUID %s", processFlowFile->getUUIDStr());
+ session->remove(processFlowFile);
+ } else {
+ session->putAttribute(processFlowFile, MQTT_BROKER_ATTRIBUTE, uri_.c_str());
+ session->putAttribute(processFlowFile, MQTT_TOPIC_ATTRIBUTE, topic_.c_str());
+ logger_->log_debug("ConsumeMQTT processing success for the flow with UUID %s topic %s", processFlowFile->getUUIDStr(), topic_);
+ session->transfer(processFlowFile, Success);
+ }
+ MQTTClient_freeMessage(&message);
+ msg_queue.pop_front();
+ }
+}
+
+} /* namespace processors */
+} /* namespace minifi */
+} /* namespace nifi */
+} /* namespace apache */
+} /* namespace org */
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/extensions/mqtt/ConsumeMQTT.h
----------------------------------------------------------------------
diff --git a/extensions/mqtt/ConsumeMQTT.h b/extensions/mqtt/ConsumeMQTT.h
new file mode 100644
index 0000000..f5155fb
--- /dev/null
+++ b/extensions/mqtt/ConsumeMQTT.h
@@ -0,0 +1,122 @@
+/**
+ * @file ConsumeMQTT.h
+ * ConsumeMQTT class declaration
+ *
+ * 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.
+ */
+#ifndef __CONSUME_MQTT_H__
+#define __CONSUME_MQTT_H__
+
+#include <climits>
+#include <deque>
+#include "FlowFileRecord.h"
+#include "core/Processor.h"
+#include "core/ProcessSession.h"
+#include "core/Core.h"
+#include "core/Resource.h"
+#include "core/Property.h"
+#include "core/logging/LoggerConfiguration.h"
+#include "concurrentqueue.h"
+#include "MQTTClient.h"
+#include "AbstractMQTTProcessor.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace processors {
+
+#define MQTT_TOPIC_ATTRIBUTE "mqtt.topic"
+#define MQTT_BROKER_ATTRIBUTE "mqtt.broker"
+
+// ConsumeMQTT Class
+class ConsumeMQTT: public processors::AbstractMQTTProcessor {
+public:
+ // Constructor
+ /*!
+ * Create a new processor
+ */
+ explicit ConsumeMQTT(std::string name, uuid_t uuid = NULL)
+ : processors::AbstractMQTTProcessor(name, uuid), logger_(logging::LoggerFactory<ConsumeMQTT>::getLogger()) {
+ isSubscriber_ = true;
+ maxQueueSize_ = 100;
+ }
+ // Destructor
+ virtual ~ConsumeMQTT() {
+ MQTTClient_message *message;
+ while (queue_.try_dequeue(message)) {
+ MQTTClient_freeMessage(&message);
+ }
+ }
+ // Processor Name
+ static constexpr char const* ProcessorName = "ConsumeMQTT";
+ // Supported Properties
+ static core::Property MaxQueueSize;
+ // Nest Callback Class for write stream
+ class WriteCallback: public OutputStreamCallback {
+ public:
+ WriteCallback(MQTTClient_message *message) :
+ message_(message) {
+ status_ = 0;
+ }
+ MQTTClient_message *message_;
+ int64_t process(std::shared_ptr<io::BaseStream> stream) {
+ int64_t len = stream->write(reinterpret_cast<uint8_t*>(message_->payload), message_->payloadlen);
+ if (len < 0)
+ status_ = -1;
+ return len;
+ }
+ int status_;
+ };
+
+public:
+ /**
+ * Function that's executed when the processor is scheduled.
+ * @param context process context.
+ * @param sessionFactory process session factory that is used when creating
+ * ProcessSession objects.
+ */
+ void onSchedule(core::ProcessContext *context, core::ProcessSessionFactory *sessionFactory);
+ // OnTrigger method, implemented by NiFi ConsumeMQTT
+ virtual void onTrigger(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session);
+ // Initialize, over write by NiFi ConsumeMQTT
+ virtual void initialize(void);
+ virtual bool enqueueReceiveMQTTMsg(MQTTClient_message *message);
+
+protected:
+ void getReceivedMQTTMsg(std::deque<MQTTClient_message *> &msg_queue) {
+ MQTTClient_message *message;
+ while (queue_.try_dequeue(message)) {
+ msg_queue.push_back(message);
+ }
+ }
+
+private:
+ std::shared_ptr<logging::Logger> logger_;
+ std::mutex mutex_;
+ uint64_t maxQueueSize_;
+ moodycamel::ConcurrentQueue<MQTTClient_message *> queue_;
+};
+
+REGISTER_RESOURCE (ConsumeMQTT);
+
+} /* namespace processors */
+} /* namespace minifi */
+} /* namespace nifi */
+} /* namespace apache */
+} /* namespace org */
+
+#endif
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/extensions/mqtt/MQTTLoader.cpp
----------------------------------------------------------------------
diff --git a/extensions/mqtt/MQTTLoader.cpp b/extensions/mqtt/MQTTLoader.cpp
new file mode 100644
index 0000000..869d41f
--- /dev/null
+++ b/extensions/mqtt/MQTTLoader.cpp
@@ -0,0 +1,29 @@
+/**
+ *
+ * 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.
+ */
+#include "MQTTLoader.h"
+#include "core/FlowConfiguration.h"
+
+bool MQTTFactory::added = core::FlowConfiguration::add_static_func("createMQTTFactory");
+
+extern "C" {
+
+void *createMQTTFactory(void) {
+ return new MQTTFactory();
+}
+
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/extensions/mqtt/MQTTLoader.h
----------------------------------------------------------------------
diff --git a/extensions/mqtt/MQTTLoader.h b/extensions/mqtt/MQTTLoader.h
new file mode 100644
index 0000000..d337af5
--- /dev/null
+++ b/extensions/mqtt/MQTTLoader.h
@@ -0,0 +1,72 @@
+/**
+ *
+ * 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.
+ */
+#ifndef EXTENSION_MQTTLOADER_H
+#define EXTENSION_MQTTLOADER_H
+
+#include "PublishMQTT.h"
+#include "ConsumeMQTT.h"
+#include "core/ClassLoader.h"
+
+class __attribute__((visibility("default"))) MQTTFactory : public core::ObjectFactory {
+ public:
+ MQTTFactory() {
+
+ }
+
+ /**
+ * Gets the name of the object.
+ * @return class name of processor
+ */
+ virtual std::string getName() {
+ return "MQTTFactory";
+ }
+
+ virtual std::string getClassName() {
+ return "MQTTFactory";
+ }
+ /**
+ * Gets the class name for the object
+ * @return class name for the processor.
+ */
+ virtual std::vector<std::string> getClassNames() {
+ std::vector<std::string> class_names;
+ class_names.push_back("PublishMQTT");
+ class_names.push_back("ConsumeMQTT");
+ return class_names;
+ }
+
+ virtual std::unique_ptr<ObjectFactory> assign(const std::string &class_name) {
+ if (utils::StringUtils::equalsIgnoreCase(class_name, "PublishMQTT")) {
+ return std::unique_ptr<ObjectFactory>(new core::DefautObjectFactory<minifi::processors::PublishMQTT>());
+ }
+ else if (utils::StringUtils::equalsIgnoreCase(class_name, "ConsumeMQTT")) {
+ return std::unique_ptr<ObjectFactory>(new core::DefautObjectFactory<minifi::processors::ConsumeMQTT>());
+ }
+ else {
+ return nullptr;
+ }
+ }
+
+ static bool added;
+
+};
+
+extern "C" {
+void *createMQTTFactory(void);
+}
+#endif /* EXTENSION_MQTTLOADER_H */
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/extensions/mqtt/PublishMQTT.cpp
----------------------------------------------------------------------
diff --git a/extensions/mqtt/PublishMQTT.cpp b/extensions/mqtt/PublishMQTT.cpp
new file mode 100644
index 0000000..44ce298
--- /dev/null
+++ b/extensions/mqtt/PublishMQTT.cpp
@@ -0,0 +1,106 @@
+/**
+ * @file PublishMQTT.cpp
+ * PublishMQTT class implementation
+ *
+ * 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.
+ */
+#include "PublishMQTT.h"
+#include <stdio.h>
+#include <algorithm>
+#include <memory>
+#include <string>
+#include <map>
+#include <set>
+#include "utils/TimeUtil.h"
+#include "utils/StringUtils.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace processors {
+
+core::Property PublishMQTT::Retain("Retain", "Retain MQTT published record in broker", "false");
+core::Property PublishMQTT::MaxFlowSegSize("Max Flow Segment Size", "Maximum flow content payload segment size for the MQTT record", "");
+
+void PublishMQTT::initialize() {
+ // Set the supported properties
+ std::set<core::Property> properties;
+ properties.insert(BrokerURL);
+ properties.insert(CleanSession);
+ properties.insert(ClientID);
+ properties.insert(UserName);
+ properties.insert(PassWord);
+ properties.insert(KeepLiveInterval);
+ properties.insert(ConnectionTimeOut);
+ properties.insert(QOS);
+ properties.insert(Topic);
+ properties.insert(Retain);
+ properties.insert(MaxFlowSegSize);
+ setSupportedProperties(properties);
+ // Set the supported relationships
+ std::set<core::Relationship> relationships;
+ relationships.insert(Success);
+ relationships.insert(Failure);
+ setSupportedRelationships(relationships);
+}
+
+void PublishMQTT::onSchedule(core::ProcessContext *context, core::ProcessSessionFactory *sessionFactory) {
+ AbstractMQTTProcessor::onSchedule(context, sessionFactory);
+ std::string value;
+ int64_t valInt;
+ value = "";
+ if (context->getProperty(MaxFlowSegSize.getName(), value) && !value.empty() && core::Property::StringToInt(value, valInt)) {
+ max_seg_size_ = valInt;
+ logger_->log_info("PublishMQTT: max flow segment size [%ll]", max_seg_size_);
+ }
+ value = "";
+ if (context->getProperty(Retain.getName(), value) && !value.empty() && org::apache::nifi::minifi::utils::StringUtils::StringToBool(value, retain_)) {
+ logger_->log_info("PublishMQTT: Retain [%d]", retain_);
+ }
+}
+
+void PublishMQTT::onTrigger(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) {
+ std::shared_ptr<core::FlowFile> flowFile = session->get();
+
+ if (!flowFile) {
+ return;
+ }
+
+ if (!reconnect()) {
+ logger_->log_error("MQTT connect to %s failed", uri_);
+ session->transfer(flowFile, Failure);
+ return;
+ }
+
+ PublishMQTT::ReadCallback callback(flowFile->getSize(), max_seg_size_, topic_, client_, qos_, retain_, delivered_token_);
+ session->read(flowFile, &callback);
+ if (callback.status_ < 0) {
+ logger_->log_error("Failed to send flow to MQTT topic %s", topic_);
+ session->transfer(flowFile, Failure);
+ } else {
+ logger_->log_debug("Sent flow with length %d to MQTT topic %s", callback.read_size_, topic_);
+ session->transfer(flowFile, Success);
+ }
+}
+
+} /* namespace processors */
+} /* namespace minifi */
+} /* namespace nifi */
+} /* namespace apache */
+} /* namespace org */
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/extensions/mqtt/PublishMQTT.h
----------------------------------------------------------------------
diff --git a/extensions/mqtt/PublishMQTT.h b/extensions/mqtt/PublishMQTT.h
new file mode 100644
index 0000000..ed17bd4
--- /dev/null
+++ b/extensions/mqtt/PublishMQTT.h
@@ -0,0 +1,142 @@
+/**
+ * @file PublishMQTT.h
+ * PublishMQTT class declaration
+ *
+ * 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.
+ */
+#ifndef __PUBLISH_MQTT_H__
+#define __PUBLISH_MQTT_H__
+
+#include "FlowFileRecord.h"
+#include "core/Processor.h"
+#include "core/ProcessSession.h"
+#include "core/Core.h"
+#include "core/Resource.h"
+#include "core/Property.h"
+#include "core/logging/LoggerConfiguration.h"
+#include "MQTTClient.h"
+#include "AbstractMQTTProcessor.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace processors {
+
+// PublishMQTT Class
+class PublishMQTT: public processors::AbstractMQTTProcessor {
+public:
+ // Constructor
+ /*!
+ * Create a new processor
+ */
+ explicit PublishMQTT(std::string name, uuid_t uuid = NULL)
+ : processors::AbstractMQTTProcessor(name, uuid), logger_(logging::LoggerFactory<PublishMQTT>::getLogger()) {
+ retain_ = false;
+ max_seg_size_ = ULLONG_MAX;
+ }
+ // Destructor
+ virtual ~PublishMQTT() {
+ }
+ // Processor Name
+ static constexpr char const* ProcessorName = "PublishMQTT";
+ // Supported Properties
+ static core::Property Retain;
+ static core::Property MaxFlowSegSize;
+
+ // Nest Callback Class for read stream
+ class ReadCallback: public InputStreamCallback {
+ public:
+ ReadCallback(uint64_t flow_size, uint64_t max_seg_size, const std::string &key, MQTTClient client,
+ int qos, bool retain, MQTTClient_deliveryToken &token) :
+ flow_size_(flow_size), max_seg_size_(max_seg_size), key_(key), client_(client),
+ qos_(qos), retain_(retain), token_(token) {
+ status_ = 0;
+ read_size_ = 0;
+ }
+ ~ReadCallback() {
+ }
+ int64_t process(std::shared_ptr<io::BaseStream> stream) {
+ if (flow_size_ < max_seg_size_)
+ max_seg_size_ = flow_size_;
+ std::vector<unsigned char> buffer;
+ buffer.reserve(max_seg_size_);
+ read_size_ = 0;
+ status_ = 0;
+ while (read_size_ < flow_size_) {
+ int readRet = stream->read(&buffer[0], max_seg_size_);
+ if (readRet < 0) {
+ status_ = -1;
+ return read_size_;
+ }
+ if (readRet > 0) {
+ MQTTClient_message pubmsg = MQTTClient_message_initializer;
+ pubmsg.payload = &buffer[0];
+ pubmsg.payloadlen = readRet;
+ pubmsg.qos = qos_;
+ pubmsg.retained = retain_;
+ if (MQTTClient_publishMessage(client_, key_.c_str(), &pubmsg, &token_) != MQTTCLIENT_SUCCESS) {
+ status_ = -1;
+ return -1;
+ }
+ read_size_ += readRet;
+ } else {
+ break;
+ }
+ }
+ return read_size_;
+ }
+ uint64_t flow_size_;
+ uint64_t max_seg_size_;
+ std::string key_;
+ MQTTClient client_;;
+ int status_;
+ int read_size_;
+ int qos_;
+ int retain_;
+ MQTTClient_deliveryToken &token_;
+ };
+
+public:
+ /**
+ * Function that's executed when the processor is scheduled.
+ * @param context process context.
+ * @param sessionFactory process session factory that is used when creating
+ * ProcessSession objects.
+ */
+ void onSchedule(core::ProcessContext *context, core::ProcessSessionFactory *sessionFactory);
+ // OnTrigger method, implemented by NiFi PublishMQTT
+ virtual void onTrigger(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session);
+ // Initialize, over write by NiFi PublishMQTT
+ virtual void initialize(void);
+
+protected:
+
+private:
+ uint64_t max_seg_size_;
+ bool retain_;
+ std::shared_ptr<logging::Logger> logger_;
+};
+
+REGISTER_RESOURCE (PublishMQTT);
+
+} /* namespace processors */
+} /* namespace minifi */
+} /* namespace nifi */
+} /* namespace apache */
+} /* namespace org */
+
+#endif
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/libminifi/test/mqtt-tests/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/libminifi/test/mqtt-tests/CMakeLists.txt b/libminifi/test/mqtt-tests/CMakeLists.txt
new file mode 100644
index 0000000..1e465b1
--- /dev/null
+++ b/libminifi/test/mqtt-tests/CMakeLists.txt
@@ -0,0 +1,39 @@
+#
+# 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.
+#
+
+file(GLOB KAFKA_INTEGRATION_TESTS "*.cpp")
+
+SET(EXTENSIONS_TEST_COUNT 0)
+FOREACH(testfile ${KAFKA_INTEGRATION_TESTS})
+ get_filename_component(testfilename "${testfile}" NAME_WE)
+ add_executable("${testfilename}" "${testfile}")
+ target_include_directories(${testfilename} BEFORE PRIVATE "${CMAKE_SOURCE_DIR}/extensions/librdkafka")
+ target_include_directories(${testfilename} BEFORE PRIVATE "${CMAKE_SOURCE_DIR}/thirdparty/librdkafka-0.11.1/src")
+ target_include_directories(${testfilename} BEFORE PRIVATE "${CMAKE_SOURCE_DIR}/thirdparty/librdkafka-0.11.1/src-cpp")
+ createTests("${testfilename}")
+ target_link_libraries(${testfilename} ${CATCH_MAIN_LIB})
+ if (APPLE)
+ target_link_libraries (${testfilename} -Wl,-all_load minifi-rdkafka-extensions)
+ else ()
+ target_link_libraries (${testfilename} -Wl,--whole-archive minifi-rdkafka-extensions -Wl,--no-whole-archive)
+ endif ()
+ MATH(EXPR EXTENSIONS_TEST_COUNT "${EXTENSIONS_TEST_COUNT}+1")
+ add_test(NAME "${testfilename}" COMMAND "${testfilename}" WORKING_DIRECTORY ${TEST_DIR})
+ENDFOREACH()
+message("-- Finished building ${KAFKA-EXTENSIONS_TEST_COUNT} Lib Kafka related test file(s)...")
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/thirdparty/paho.mqtt.c/.gitignore
----------------------------------------------------------------------
diff --git a/thirdparty/paho.mqtt.c/.gitignore b/thirdparty/paho.mqtt.c/.gitignore
new file mode 100644
index 0000000..5b2b31f
--- /dev/null
+++ b/thirdparty/paho.mqtt.c/.gitignore
@@ -0,0 +1,6 @@
+/dep/
+/build/
+/build.paho/
+*.swp
+*.pyc
+/build.paho
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/thirdparty/paho.mqtt.c/.gitreview
----------------------------------------------------------------------
diff --git a/thirdparty/paho.mqtt.c/.gitreview b/thirdparty/paho.mqtt.c/.gitreview
new file mode 100644
index 0000000..2d16be7
--- /dev/null
+++ b/thirdparty/paho.mqtt.c/.gitreview
@@ -0,0 +1,5 @@
+[gerrit]
+host=git.eclipse.org
+port=29418
+project=paho/org.eclipse.paho.mqtt.c
+defaultbranch=develop
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/thirdparty/paho.mqtt.c/.pydevproject
----------------------------------------------------------------------
diff --git a/thirdparty/paho.mqtt.c/.pydevproject b/thirdparty/paho.mqtt.c/.pydevproject
new file mode 100644
index 0000000..40e9f40
--- /dev/null
+++ b/thirdparty/paho.mqtt.c/.pydevproject
@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="UTF-8" standalone="no"?>
+<?eclipse-pydev version="1.0"?><pydev_project>
+<pydev_property name="org.python.pydev.PYTHON_PROJECT_INTERPRETER">Default</pydev_property>
+<pydev_property name="org.python.pydev.PYTHON_PROJECT_VERSION">python 2.7</pydev_property>
+</pydev_project>
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/thirdparty/paho.mqtt.c/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/thirdparty/paho.mqtt.c/CMakeLists.txt b/thirdparty/paho.mqtt.c/CMakeLists.txt
new file mode 100644
index 0000000..35c3fc7
--- /dev/null
+++ b/thirdparty/paho.mqtt.c/CMakeLists.txt
@@ -0,0 +1,86 @@
+#*******************************************************************************
+# Copyright (c) 2015, 2017 logi.cals GmbH and others
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Eclipse Public License v1.0
+# and Eclipse Distribution License v1.0 which accompany this distribution.
+#
+# The Eclipse Public License is available at
+# http://www.eclipse.org/legal/epl-v10.html
+# and the Eclipse Distribution License is available at
+# http://www.eclipse.org/org/documents/edl-v10.php.
+#
+# Contributors:
+# Rainer Poisel - initial version
+# Genis Riera Perez - Add support for building debian package
+#*******************************************************************************/
+
+# Note: on OS X you should install XCode and the associated command-line tools
+
+CMAKE_MINIMUM_REQUIRED(VERSION 2.8.4)
+PROJECT("paho" C)
+MESSAGE(STATUS "CMake version: " ${CMAKE_VERSION})
+MESSAGE(STATUS "CMake system name: " ${CMAKE_SYSTEM_NAME})
+
+SET(CMAKE_SCRIPTS "${CMAKE_SOURCE_DIR}/cmake")
+SET(CMAKE_MODULE_PATH "${CMAKE_SOURCE_DIR}/cmake/modules")
+
+## build settings
+SET(PAHO_VERSION_MAJOR 1)
+SET(PAHO_VERSION_MINOR 2)
+SET(PAHO_VERSION_PATCH 0)
+SET(CLIENT_VERSION ${PAHO_VERSION_MAJOR}.${PAHO_VERSION_MINOR}.${PAHO_VERSION_PATCH})
+
+INCLUDE(GNUInstallDirs)
+
+STRING(TIMESTAMP BUILD_TIMESTAMP UTC)
+MESSAGE(STATUS "Timestamp is ${BUILD_TIMESTAMP}")
+
+IF(WIN32)
+ ADD_DEFINITIONS(-D_CRT_SECURE_NO_DEPRECATE -DWIN32_LEAN_AND_MEAN -MD)
+ELSEIF(${CMAKE_SYSTEM_NAME} STREQUAL "Darwin")
+ ADD_DEFINITIONS(-DOSX)
+ENDIF()
+
+## build options
+SET(PAHO_WITH_SSL TRUE CACHE BOOL "Flag that defines whether to build ssl-enabled binaries too. ")
+SET(PAHO_BUILD_STATIC FALSE CACHE BOOL "Build static library")
+SET(PAHO_BUILD_DOCUMENTATION FALSE CACHE BOOL "Create and install the HTML based API documentation (requires Doxygen)")
+SET(PAHO_BUILD_SAMPLES FALSE CACHE BOOL "Build sample programs")
+SET(PAHO_BUILD_DEB_PACKAGE FALSE CACHE BOOL "Build debian package")
+SET(PAHO_ENABLE_TESTING FALSE CACHE BOOL "Build tests and run")
+
+ADD_SUBDIRECTORY(src)
+IF(PAHO_BUILD_SAMPLES)
+ ADD_SUBDIRECTORY(src/samples)
+ENDIF()
+
+IF(PAHO_BUILD_DOCUMENTATION)
+ ADD_SUBDIRECTORY(doc)
+ENDIF()
+
+### packaging settings
+IF (WIN32)
+ SET(CPACK_GENERATOR "ZIP")
+ELSEIF(PAHO_BUILD_DEB_PACKAGE)
+ SET(CPACK_GENERATOR "DEB")
+ CONFIGURE_FILE(${CMAKE_SCRIPTS}/CPackDebConfig.cmake.in
+ ${CMAKE_BINARY_DIR}/CPackDebConfig.cmake @ONLY)
+ SET(CPACK_PROJECT_CONFIG_FILE ${CMAKE_BINARY_DIR}/CPackDebConfig.cmake)
+ ADD_SUBDIRECTORY(debian)
+ELSE()
+ SET(CPACK_GENERATOR "TGZ")
+ENDIF()
+
+SET(CPACK_PACKAGE_VERSION_MAJOR ${PAHO_VERSION_MAJOR})
+SET(CPACK_PACKAGE_VERSION_MINOR ${PAHO_VERSION_MINOR})
+SET(CPACK_PACKAGE_VERSION_PATCH ${PAHO_VERSION_PATCH})
+INCLUDE(CPack)
+
+IF(PAHO_ENABLE_TESTING)
+ ENABLE_TESTING()
+ INCLUDE_DIRECTORIES(test src)
+ ADD_SUBDIRECTORY(test)
+ELSE()
+ INCLUDE_DIRECTORIES(src)
+ENDIF()
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/thirdparty/paho.mqtt.c/CONTRIBUTING.md
----------------------------------------------------------------------
diff --git a/thirdparty/paho.mqtt.c/CONTRIBUTING.md b/thirdparty/paho.mqtt.c/CONTRIBUTING.md
new file mode 100644
index 0000000..1ebe9d1
--- /dev/null
+++ b/thirdparty/paho.mqtt.c/CONTRIBUTING.md
@@ -0,0 +1,66 @@
+# Contributing to Paho
+
+Thanks for your interest in this project!
+
+You can contribute bugfixes and new features by sending pull requests through GitHub.
+
+## Legal
+
+In order for your contribution to be accepted, it must comply with the Eclipse Foundation IP policy.
+
+Please read the [Eclipse Foundation policy on accepting contributions via Git](http://wiki.eclipse.org/Development_Resources/Contributing_via_Git).
+
+1. Sign the [Eclipse CLA](http://www.eclipse.org/legal/CLA.php)
+ 1. Register for an Eclipse Foundation User ID. You can register [here](https://dev.eclipse.org/site_login/createaccount.php).
+ 2. Log into the [Projects Portal](https://projects.eclipse.org/), and click on the '[Eclipse CLA](https://projects.eclipse.org/user/sign/cla)' link.
+2. Go to your [account settings](https://dev.eclipse.org/site_login/myaccount.php#open_tab_accountsettings) and add your GitHub username to your account.
+3. Make sure that you _sign-off_ your Git commits in the following format:
+ ``` Signed-off-by: John Smith <jo...@nowhere.com> ``` This is usually at the bottom of the commit message. You can automate this by adding the '-s' flag when you make the commits. e.g. ```git commit -s -m "Adding a cool feature"```
+4. Ensure that the email address that you make your commits with is the same one you used to sign up to the Eclipse Foundation website with.
+
+## Contributing a change
+
+1. [Fork the repository on GitHub](https://github.com/eclipse/paho.mqtt.c/fork)
+2. Clone the forked repository onto your computer: ``` git clone https://github.com/<your username>/paho.mqtt.c.git ```
+3. Create a new branch from the latest ```develop``` branch with ```git checkout -b YOUR_BRANCH_NAME origin/develop```
+4. Make your changes
+5. If developing a new feature, make sure to include JUnit tests.
+6. Ensure that all new and existing tests pass.
+7. Commit the changes into the branch: ``` git commit -s ``` Make sure that your commit message is meaningful and describes your changes correctly.
+8. If you have a lot of commits for the change, squash them into a single / few commits.
+9. Push the changes in your branch to your forked repository.
+10. Finally, go to [https://github.com/eclipse/paho.mqtt.c](https://github.com/eclipse/paho.mqtt.c) and create a pull request from your "YOUR_BRANCH_NAME" branch to the ```develop``` one to request review and merge of the commits in your pushed branch.
+
+
+What happens next depends on the content of the patch. If it is 100% authored
+by the contributor and is less than 1000 lines (and meets the needs of the
+project), then it can be pulled into the main repository. If not, more steps
+are required. These are detailed in the
+[legal process poster](http://www.eclipse.org/legal/EclipseLegalProcessPoster.pdf).
+
+
+
+## Developer resources:
+
+
+Information regarding source code management, builds, coding standards, and more.
+
+- [https://projects.eclipse.org/projects/iot.paho/developer](https://projects.eclipse.org/projects/iot.paho/developer)
+
+Contact:
+--------
+
+Contact the project developers via the project's development
+[mailing list](https://dev.eclipse.org/mailman/listinfo/paho-dev).
+
+Search for bugs:
+----------------
+
+This project uses GitHub Issues here: [github.com/eclipse/paho.mqtt.c/issues](https://github.com/eclipse/paho.mqtt.c/issues) to track ongoing development and issues.
+
+Create a new bug:
+-----------------
+
+Be sure to search for existing bugs before you create another one. Remember that contributions are always welcome!
+
+- [Create new Paho bug](https://github.com/eclipse/paho.mqtt.c/issues/new)
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a8703b5c/thirdparty/paho.mqtt.c/Makefile
----------------------------------------------------------------------
diff --git a/thirdparty/paho.mqtt.c/Makefile b/thirdparty/paho.mqtt.c/Makefile
new file mode 100755
index 0000000..63d745f
--- /dev/null
+++ b/thirdparty/paho.mqtt.c/Makefile
@@ -0,0 +1,283 @@
+#*******************************************************************************
+# Copyright (c) 2009, 2017 IBM Corp.
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Eclipse Public License v1.0
+# and Eclipse Distribution License v1.0 which accompany this distribution.
+#
+# The Eclipse Public License is available at
+# http://www.eclipse.org/legal/epl-v10.html
+# and the Eclipse Distribution License is available at
+# http://www.eclipse.org/org/documents/edl-v10.php.
+#
+# Contributors:
+# Ian Craggs - initial API and implementation and/or initial documentation
+# Allan Stockdill-Mander - SSL updates
+# Andy Piper - various fixes
+# Ian Craggs - OSX build
+# Rainer Poisel - support for multi-core builds and cross-compilation
+#*******************************************************************************/
+
+# Note: on OS X you should install XCode and the associated command-line tools
+
+SHELL = /bin/sh
+.PHONY: clean, mkdir, install, uninstall, html
+
+ifndef release.version
+ release.version = 1.2.0
+endif
+
+# determine current platform
+BUILD_TYPE ?= debug
+ifeq ($(OS),Windows_NT)
+ OSTYPE ?= $(OS)
+ MACHINETYPE ?= $(PROCESSOR_ARCHITECTURE)
+else
+ OSTYPE ?= $(shell uname -s)
+ MACHINETYPE ?= $(shell uname -m)
+ build.level = $(shell date)
+endif # OS
+ifeq ($(OSTYPE),linux)
+ OSTYPE = Linux
+endif
+
+# assume this is normally run in the main Paho directory
+ifndef srcdir
+ srcdir = src
+endif
+
+ifndef blddir
+ blddir = build/output
+endif
+
+ifndef blddir_work
+ blddir_work = build
+endif
+
+ifndef docdir
+ docdir = $(blddir)/doc
+endif
+
+ifndef docdir_work
+ docdir_work = $(blddir)/../doc
+endif
+
+ifndef prefix
+ prefix = /usr/local
+endif
+
+ifndef exec_prefix
+ exec_prefix = ${prefix}
+endif
+
+bindir = $(exec_prefix)/bin
+includedir = $(prefix)/include
+libdir = $(exec_prefix)/lib
+
+SOURCE_FILES = $(wildcard $(srcdir)/*.c)
+SOURCE_FILES_C = $(filter-out $(srcdir)/MQTTAsync.c $(srcdir)/MQTTVersion.c $(srcdir)/SSLSocket.c, $(SOURCE_FILES))
+SOURCE_FILES_CS = $(filter-out $(srcdir)/MQTTAsync.c $(srcdir)/MQTTVersion.c, $(SOURCE_FILES))
+SOURCE_FILES_A = $(filter-out $(srcdir)/MQTTClient.c $(srcdir)/MQTTVersion.c $(srcdir)/SSLSocket.c, $(SOURCE_FILES))
+SOURCE_FILES_AS = $(filter-out $(srcdir)/MQTTClient.c $(srcdir)/MQTTVersion.c, $(SOURCE_FILES))
+
+HEADERS = $(srcdir)/*.h
+HEADERS_C = $(filter-out $(srcdir)/MQTTAsync.h, $(HEADERS))
+HEADERS_A = $(HEADERS)
+
+SAMPLE_FILES_C = paho_cs_pub paho_cs_sub MQTTClient_publish MQTTClient_publish_async MQTTClient_subscribe
+SYNC_SAMPLES = ${addprefix ${blddir}/samples/,${SAMPLE_FILES_C}}
+
+SAMPLE_FILES_A = paho_c_pub paho_c_sub MQTTAsync_subscribe MQTTAsync_publish
+ASYNC_SAMPLES = ${addprefix ${blddir}/samples/,${SAMPLE_FILES_A}}
+
+TEST_FILES_C = test1 test2 sync_client_test test_mqtt4sync
+SYNC_TESTS = ${addprefix ${blddir}/test/,${TEST_FILES_C}}
+
+TEST_FILES_CS = test3
+SYNC_SSL_TESTS = ${addprefix ${blddir}/test/,${TEST_FILES_CS}}
+
+TEST_FILES_A = test4 test9 test_mqtt4async
+ASYNC_TESTS = ${addprefix ${blddir}/test/,${TEST_FILES_A}}
+
+TEST_FILES_AS = test5
+ASYNC_SSL_TESTS = ${addprefix ${blddir}/test/,${TEST_FILES_AS}}
+
+# The names of the four different libraries to be built
+MQTTLIB_C = paho-mqtt3c
+MQTTLIB_CS = paho-mqtt3cs
+MQTTLIB_A = paho-mqtt3a
+MQTTLIB_AS = paho-mqtt3as
+
+CC ?= gcc
+
+ifndef INSTALL
+INSTALL = install
+endif
+INSTALL_PROGRAM = $(INSTALL)
+INSTALL_DATA = $(INSTALL) -m 644
+DOXYGEN_COMMAND = doxygen
+
+MAJOR_VERSION = 1
+MINOR_VERSION = 0
+VERSION = ${MAJOR_VERSION}.${MINOR_VERSION}
+
+MQTTLIB_C_TARGET = ${blddir}/lib${MQTTLIB_C}.so.${VERSION}
+MQTTLIB_CS_TARGET = ${blddir}/lib${MQTTLIB_CS}.so.${VERSION}
+MQTTLIB_A_TARGET = ${blddir}/lib${MQTTLIB_A}.so.${VERSION}
+MQTTLIB_AS_TARGET = ${blddir}/lib${MQTTLIB_AS}.so.${VERSION}
+MQTTVERSION_TARGET = ${blddir}/MQTTVersion
+
+CCFLAGS_SO = -g -fPIC $(CFLAGS) -Os -Wall -fvisibility=hidden -I$(blddir_work)
+FLAGS_EXE = $(LDFLAGS) -I ${srcdir} -lpthread -L ${blddir}
+FLAGS_EXES = $(LDFLAGS) -I ${srcdir} ${START_GROUP} -lpthread -lssl -lcrypto ${END_GROUP} -L ${blddir}
+
+LDCONFIG ?= /sbin/ldconfig
+LDFLAGS_C = $(LDFLAGS) -shared -Wl,-init,$(MQTTCLIENT_INIT) -lpthread
+LDFLAGS_CS = $(LDFLAGS) -shared $(START_GROUP) -lpthread $(EXTRA_LIB) -lssl -lcrypto $(END_GROUP) -Wl,-init,$(MQTTCLIENT_INIT)
+LDFLAGS_A = $(LDFLAGS) -shared -Wl,-init,$(MQTTASYNC_INIT) -lpthread
+LDFLAGS_AS = $(LDFLAGS) -shared $(START_GROUP) -lpthread $(EXTRA_LIB) -lssl -lcrypto $(END_GROUP) -Wl,-init,$(MQTTASYNC_INIT)
+
+SED_COMMAND = sed \
+ -e "s/@CLIENT_VERSION@/${release.version}/g" \
+ -e "s/@BUILD_TIMESTAMP@/${build.level}/g"
+
+ifeq ($(OSTYPE),Linux)
+
+MQTTCLIENT_INIT = MQTTClient_init
+MQTTASYNC_INIT = MQTTAsync_init
+START_GROUP = -Wl,--start-group
+END_GROUP = -Wl,--end-group
+
+EXTRA_LIB = -ldl
+
+LDFLAGS_C += -Wl,-soname,lib$(MQTTLIB_C).so.${MAJOR_VERSION}
+LDFLAGS_CS += -Wl,-soname,lib$(MQTTLIB_CS).so.${MAJOR_VERSION} -Wl,-no-whole-archive
+LDFLAGS_A += -Wl,-soname,lib${MQTTLIB_A}.so.${MAJOR_VERSION}
+LDFLAGS_AS += -Wl,-soname,lib${MQTTLIB_AS}.so.${MAJOR_VERSION} -Wl,-no-whole-archive
+
+else ifeq ($(OSTYPE),Darwin)
+
+MQTTCLIENT_INIT = _MQTTClient_init
+MQTTASYNC_INIT = _MQTTAsync_init
+START_GROUP =
+END_GROUP =
+
+EXTRA_LIB = -ldl
+
+CCFLAGS_SO += -Wno-deprecated-declarations -DOSX -I /usr/local/opt/openssl/include
+LDFLAGS_C += -Wl,-install_name,lib$(MQTTLIB_C).so.${MAJOR_VERSION}
+LDFLAGS_CS += -Wl,-install_name,lib$(MQTTLIB_CS).so.${MAJOR_VERSION} -L /usr/local/opt/openssl/lib
+LDFLAGS_A += -Wl,-install_name,lib${MQTTLIB_A}.so.${MAJOR_VERSION}
+LDFLAGS_AS += -Wl,-install_name,lib${MQTTLIB_AS}.so.${MAJOR_VERSION} -L /usr/local/opt/openssl/lib
+FLAGS_EXE += -DOSX
+FLAGS_EXES += -L /usr/local/opt/openssl/lib
+
+endif
+
+all: build
+
+build: | mkdir ${MQTTLIB_C_TARGET} ${MQTTLIB_CS_TARGET} ${MQTTLIB_A_TARGET} ${MQTTLIB_AS_TARGET} ${MQTTVERSION_TARGET} ${SYNC_SAMPLES} ${ASYNC_SAMPLES} ${SYNC_TESTS} ${SYNC_SSL_TESTS} ${ASYNC_TESTS} ${ASYNC_SSL_TESTS}
+
+clean:
+ rm -rf ${blddir}/*
+ rm -rf ${blddir_work}/*
+
+mkdir:
+ -mkdir -p ${blddir}/samples
+ -mkdir -p ${blddir}/test
+ echo OSTYPE is $(OSTYPE)
+
+${SYNC_TESTS}: ${blddir}/test/%: ${srcdir}/../test/%.c $(MQTTLIB_C_TARGET)
+ ${CC} -DNOSTACKTRACE $(srcdir)/Thread.c -g -o $@ $< -l${MQTTLIB_C} ${FLAGS_EXE}
+
+${SYNC_SSL_TESTS}: ${blddir}/test/%: ${srcdir}/../test/%.c $(MQTTLIB_CS_TARGET)
+ ${CC} -g -o $@ $< -l${MQTTLIB_CS} ${FLAGS_EXES}
+
+${ASYNC_TESTS}: ${blddir}/test/%: ${srcdir}/../test/%.c $(MQTTLIB_CS_TARGET)
+ ${CC} -g -o $@ $< -l${MQTTLIB_A} ${FLAGS_EXE}
+
+${ASYNC_SSL_TESTS}: ${blddir}/test/%: ${srcdir}/../test/%.c $(MQTTLIB_CS_TARGET) $(MQTTLIB_AS_TARGET)
+ ${CC} -g -o $@ $< -l${MQTTLIB_AS} ${FLAGS_EXES}
+
+${SYNC_SAMPLES}: ${blddir}/samples/%: ${srcdir}/samples/%.c $(MQTTLIB_C_TARGET)
+ ${CC} -o $@ $< -l${MQTTLIB_C} ${FLAGS_EXE}
+
+${ASYNC_SAMPLES}: ${blddir}/samples/%: ${srcdir}/samples/%.c $(MQTTLIB_A_TARGET)
+ ${CC} -o $@ $< -l${MQTTLIB_A} ${FLAGS_EXE}
+
+$(blddir_work)/VersionInfo.h: $(srcdir)/VersionInfo.h.in
+ $(SED_COMMAND) $< > $@
+
+${MQTTLIB_C_TARGET}: ${SOURCE_FILES_C} ${HEADERS_C} $(blddir_work)/VersionInfo.h
+ ${CC} ${CCFLAGS_SO} -o $@ ${SOURCE_FILES_C} ${LDFLAGS_C}
+ -ln -s lib$(MQTTLIB_C).so.${VERSION} ${blddir}/lib$(MQTTLIB_C).so.${MAJOR_VERSION}
+ -ln -s lib$(MQTTLIB_C).so.${MAJOR_VERSION} ${blddir}/lib$(MQTTLIB_C).so
+
+${MQTTLIB_CS_TARGET}: ${SOURCE_FILES_CS} ${HEADERS_C} $(blddir_work)/VersionInfo.h
+ ${CC} ${CCFLAGS_SO} -o $@ ${SOURCE_FILES_CS} -DOPENSSL ${LDFLAGS_CS}
+ -ln -s lib$(MQTTLIB_CS).so.${VERSION} ${blddir}/lib$(MQTTLIB_CS).so.${MAJOR_VERSION}
+ -ln -s lib$(MQTTLIB_CS).so.${MAJOR_VERSION} ${blddir}/lib$(MQTTLIB_CS).so
+
+${MQTTLIB_A_TARGET}: ${SOURCE_FILES_A} ${HEADERS_A} $(blddir_work)/VersionInfo.h
+ ${CC} ${CCFLAGS_SO} -o $@ ${SOURCE_FILES_A} ${LDFLAGS_A}
+ -ln -s lib$(MQTTLIB_A).so.${VERSION} ${blddir}/lib$(MQTTLIB_A).so.${MAJOR_VERSION}
+ -ln -s lib$(MQTTLIB_A).so.${MAJOR_VERSION} ${blddir}/lib$(MQTTLIB_A).so
+
+${MQTTLIB_AS_TARGET}: ${SOURCE_FILES_AS} ${HEADERS_A} $(blddir_work)/VersionInfo.h
+ ${CC} ${CCFLAGS_SO} -o $@ ${SOURCE_FILES_AS} -DOPENSSL ${LDFLAGS_AS}
+ -ln -s lib$(MQTTLIB_AS).so.${VERSION} ${blddir}/lib$(MQTTLIB_AS).so.${MAJOR_VERSION}
+ -ln -s lib$(MQTTLIB_AS).so.${MAJOR_VERSION} ${blddir}/lib$(MQTTLIB_AS).so
+
+${MQTTVERSION_TARGET}: $(srcdir)/MQTTVersion.c $(srcdir)/MQTTAsync.h ${MQTTLIB_A_TARGET} $(MQTTLIB_CS_TARGET)
+ ${CC} ${FLAGS_EXE} -o $@ -l${MQTTLIB_A} $(srcdir)/MQTTVersion.c -ldl
+
+strip_options:
+ $(eval INSTALL_OPTS := -s)
+
+install-strip: build strip_options install
+
+install: build
+ $(INSTALL_DATA) ${INSTALL_OPTS} ${MQTTLIB_C_TARGET} $(DESTDIR)${libdir}
+ $(INSTALL_DATA) ${INSTALL_OPTS} ${MQTTLIB_CS_TARGET} $(DESTDIR)${libdir}
+ $(INSTALL_DATA) ${INSTALL_OPTS} ${MQTTLIB_A_TARGET} $(DESTDIR)${libdir}
+ $(INSTALL_DATA) ${INSTALL_OPTS} ${MQTTLIB_AS_TARGET} $(DESTDIR)${libdir}
+ $(INSTALL_PROGRAM) ${INSTALL_OPTS} ${MQTTVERSION_TARGET} $(DESTDIR)${bindir}
+ $(LDCONFIG) $(DESTDIR)${libdir}
+ ln -s lib$(MQTTLIB_C).so.${MAJOR_VERSION} $(DESTDIR)${libdir}/lib$(MQTTLIB_C).so
+ ln -s lib$(MQTTLIB_CS).so.${MAJOR_VERSION} $(DESTDIR)${libdir}/lib$(MQTTLIB_CS).so
+ ln -s lib$(MQTTLIB_A).so.${MAJOR_VERSION} $(DESTDIR)${libdir}/lib$(MQTTLIB_A).so
+ ln -s lib$(MQTTLIB_AS).so.${MAJOR_VERSION} $(DESTDIR)${libdir}/lib$(MQTTLIB_AS).so
+ $(INSTALL_DATA) ${srcdir}/MQTTAsync.h $(DESTDIR)${includedir}
+ $(INSTALL_DATA) ${srcdir}/MQTTClient.h $(DESTDIR)${includedir}
+ $(INSTALL_DATA) ${srcdir}/MQTTClientPersistence.h $(DESTDIR)${includedir}
+
+uninstall:
+ rm $(DESTDIR)${libdir}/lib$(MQTTLIB_C).so.${VERSION}
+ rm $(DESTDIR)${libdir}/lib$(MQTTLIB_CS).so.${VERSION}
+ rm $(DESTDIR)${libdir}/lib$(MQTTLIB_A).so.${VERSION}
+ rm $(DESTDIR)${libdir}/lib$(MQTTLIB_AS).so.${VERSION}
+ rm $(DESTDIR)${bindir}/MQTTVersion
+ $(LDCONFIG) $(DESTDIR)${libdir}
+ rm $(DESTDIR)${libdir}/lib$(MQTTLIB_C).so
+ rm $(DESTDIR)${libdir}/lib$(MQTTLIB_CS).so
+ rm $(DESTDIR)${libdir}/lib$(MQTTLIB_A).so
+ rm $(DESTDIR)${libdir}/lib$(MQTTLIB_AS).so
+ rm $(DESTDIR)${includedir}/MQTTAsync.h
+ rm $(DESTDIR)${includedir}/MQTTClient.h
+ rm $(DESTDIR)${includedir}/MQTTClientPersistence.h
+
+REGEX_DOXYGEN := \
+ 's;@PROJECT_SOURCE_DIR@/src/\?;;' \
+ 's;@PROJECT_SOURCE_DIR@;..;' \
+ 's;@CMAKE_CURRENT_BINARY_DIR@;../build/output;'
+SED_DOXYGEN := $(foreach sed_exp,$(REGEX_DOXYGEN),-e $(sed_exp))
+define process_doxygen
+ cd ${srcdir}; sed $(SED_DOXYGEN) ../doc/${1}.in > ../$(docdir_work)/${1}
+ cd ${srcdir}; $(DOXYGEN_COMMAND) ../$(docdir_work)/${1}
+endef
+html:
+ -mkdir -p $(docdir_work)
+ -mkdir -p ${docdir}
+ $(call process_doxygen,DoxyfileV3ClientAPI)
+ $(call process_doxygen,DoxyfileV3AsyncAPI)
+ $(call process_doxygen,DoxyfileV3ClientInternal)