diff --git a/CMakeLists.txt b/CMakeLists.txt
index f394bd2..2a4dc77 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -480,6 +480,18 @@
     createExtension(SFTP "SFTP EXTENSIONS" "This enables SFTP support" "extensions/sftp" "extensions/sftp/tests")
 endif()
 
+## Openwsman Extesions
+option(ENABLE_OPENWSMAN "Enables the Openwsman extensions." OFF)
+if (ENABLE_OPENWSMAN AND NOT DISABLE_CIVET AND NOT DISABLE_CURL)
+	include(BundledLibXml2)
+	use_bundled_libxml2(${CMAKE_CURRENT_SOURCE_DIR} ${CMAKE_CURRENT_BINARY_DIR})
+	list(APPEND CMAKE_MODULE_PATH "${CMAKE_SOURCE_DIR}/cmake/libxml2/dummy")
+
+	include(BundledOpenWSMAN)
+	use_bundled_openwsman(${CMAKE_CURRENT_SOURCE_DIR} ${CMAKE_CURRENT_BINARY_DIR})
+
+    createExtension(OPENWSMAN-EXTENSIONS "OPENWSMAN EXTENSIONS" "This enables Openwsman support" "extensions/openwsman")
+endif()
 
 ## NOW WE CAN ADD LIBRARIES AND EXTENSIONS TO MAIN
 add_subdirectory(main)
diff --git a/bootstrap.sh b/bootstrap.sh
index 9c96146..ff38708 100755
--- a/bootstrap.sh
+++ b/bootstrap.sh
@@ -305,6 +305,8 @@
 add_disabled_option SQL_ENABLED ${FALSE} "ENABLE_SQL"
 set_incompatible_with SQL_ENABLED SQLITE_ENABLED
 
+add_disabled_option OPENWSMAN_ENABLED ${FALSE} "ENABLE_OPENWSMAN"
+
 # Since the following extensions have limitations on
 add_disabled_option BUSTACHE_ENABLED ${FALSE} "ENABLE_BUSTACHE" "2.6" ${TRUE}
 add_dependency BUSTACHE_ENABLED "boost"
diff --git a/bstrp_functions.sh b/bstrp_functions.sh
index a13be2e..4ec336c 100755
--- a/bstrp_functions.sh
+++ b/bstrp_functions.sh
@@ -364,6 +364,7 @@
   echo "T. OpenCV Support ..............$(print_feature_status OPENCV_ENABLED)"
   echo "U. OPC-UA Support...............$(print_feature_status OPC_ENABLED)"
   echo "W. SQL Support..................$(print_feature_status SQL_ENABLED)"
+  echo "X. Openwsman Support ...........$(print_feature_status OPENWSMAN_ENABLED)"
   echo "****************************************"
   echo "            Build Options."
   echo "****************************************"
@@ -385,7 +386,7 @@
 
 read_feature_options(){
   local choice
-  read -p "Enter choice [ A - W or 1-6 ] " choice
+  read -p "Enter choice [ A - X or 1-6 ] " choice
   choice=$(echo ${choice} | tr '[:upper:]' '[:lower:]')
   case $choice in
     a) ToggleFeature ROCKSDB_ENABLED ;;
@@ -412,7 +413,8 @@
 	s) ToggleFeature SFTP_ENABLED ;;
     t) ToggleFeature OPENCV_ENABLED ;;
     u) ToggleFeature OPC_ENABLED ;;
-    w) ToggleFeature SQL_ENABLED ;;	
+    w) ToggleFeature SQL_ENABLED ;;
+    x) ToggleFeature OPENWSMAN_ENABLED ;;
     1) ToggleFeature TESTS_DISABLED ;;
     2) EnableAllFeatures ;;
     3) ToggleFeature JNI_ENABLED;;
@@ -430,7 +432,7 @@
       fi
       ;;
     q) exit 0;;
-    *) echo -e "${RED}Please enter an option A-W or 1-6...${NO_COLOR}" && sleep 2
+    *) echo -e "${RED}Please enter an option A-X or 1-6...${NO_COLOR}" && sleep 2
   esac
 }
 
diff --git a/cmake/BundledBZip2.cmake b/cmake/BundledBZip2.cmake
index 7ad99a6..9c3fb5b 100644
--- a/cmake/BundledBZip2.cmake
+++ b/cmake/BundledBZip2.cmake
@@ -39,7 +39,7 @@
     # Build project
     ExternalProject_Add(
             bzip2-external
-            URL https://sourceware.org/pub/bzip2/bzip2-1.0.8.tar.gz
+            URL https://sourceware.org/pub/bzip2/bzip2-1.0.8.tar.gz http://deb.debian.org/debian/pool/main/b/bzip2/bzip2_1.0.8.orig.tar.gz
             URL_HASH "SHA256=ab5a03176ee106d3f0fa90e381da478ddae405918153cca248e682cd0c4a2269"
             SOURCE_DIR "${BINARY_DIR}/thirdparty/bzip2-src"
             LIST_SEPARATOR % # This is needed for passing semicolon-separated lists
diff --git a/cmake/BundledLibXml2.cmake b/cmake/BundledLibXml2.cmake
new file mode 100644
index 0000000..61ad939
--- /dev/null
+++ b/cmake/BundledLibXml2.cmake
@@ -0,0 +1,94 @@
+#
+# 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.
+#
+
+function(use_bundled_libxml2 SOURCE_DIR BINARY_DIR)
+    message("Using bundled libxml2")
+
+    # Define patch step
+    if (WIN32)
+        set(PC "${Patch_EXECUTABLE}" -p1 -i "${SOURCE_DIR}/thirdparty/libxml2/libxml2-win.patch")
+    endif()
+
+    # Define byproducts
+    if (WIN32)
+        set(BYPRODUCT "lib/xml2.lib")
+    else()
+        set(BYPRODUCT "lib/libxml2.a")
+    endif()
+
+    # Set build options
+    if (WIN32)
+        set(LIBXML2_CMAKE_ARGS ${PASSTHROUGH_CMAKE_ARGS}
+                "-DCMAKE_INSTALL_PREFIX=${BINARY_DIR}/thirdparty/libxml2-install")
+    endif()
+
+    # Build project
+    set(LIBXML2_URL ftp://xmlsoft.org/libxml2/libxml2-2.9.10.tar.gz https://ftp.osuosl.org/pub/blfs/conglomeration/libxml2/libxml2-2.9.10.tar.gz)
+    set(LIBXML2_URL_HASH "SHA256=aafee193ffb8fe0c82d4afef6ef91972cbaf5feea100edc2f262750611b4be1f")
+
+    if (WIN32)
+        ExternalProject_Add(
+                libxml2-external
+                URL ${LIBXML2_URL}
+                URL_HASH ${LIBXML2_URL_HASH}
+                SOURCE_DIR "${BINARY_DIR}/thirdparty/libxml2-src"
+                LIST_SEPARATOR % # This is needed for passing semicolon-separated lists
+                CMAKE_ARGS ${LIBXML2_CMAKE_ARGS}
+                PATCH_COMMAND ${PC}
+                BUILD_BYPRODUCTS "${BINARY_DIR}/thirdparty/libxml2-install/${BYPRODUCT}"
+                EXCLUDE_FROM_ALL TRUE
+        )
+    else()
+        ExternalProject_Add(
+                libxml2-external
+                URL ${LIBXML2_URL}
+                URL_HASH ${LIBXML2_URL_HASH}
+                BUILD_IN_SOURCE true
+                SOURCE_DIR "${BINARY_DIR}/thirdparty/libxml2-src"
+                BUILD_COMMAND make
+                CMAKE_COMMAND ""
+                UPDATE_COMMAND ""
+                INSTALL_COMMAND make install
+                BUILD_BYPRODUCTS "${BINARY_DIR}/thirdparty/libxml2-install/${BYPRODUCT}"
+                CONFIGURE_COMMAND ""
+                PATCH_COMMAND ./configure --enable-shared=no --enable-static=yes --with-iconv=no --with-zlib=no --with-lzma=no --with-python=no --with-ftp=no --with-http=no --prefix=${BINARY_DIR}/thirdparty/libxml2-install
+                STEP_TARGETS build
+                EXCLUDE_FROM_ALL TRUE
+        )
+    endif()
+
+    # Set variables
+    set(LIBXML2_FOUND "YES" CACHE STRING "" FORCE)
+    set(LIBXML2_INCLUDE_DIR "${BINARY_DIR}/thirdparty/libxml2-install/include/libxml2" CACHE STRING "" FORCE)
+    set(LIBXML2_INCLUDE_DIRS "${LIBXML2_INCLUDE_DIR}" CACHE STRING "" FORCE)
+    set(LIBXML2_LIBRARY "${BINARY_DIR}/thirdparty/libxml2-install/${BYPRODUCT}" CACHE STRING "" FORCE)
+    set(LIBXML2_LIBRARIES "${LIBXML2_LIBRARY}" CACHE STRING "" FORCE)
+
+    # Set exported variables for FindPackage.cmake
+    set(PASSTHROUGH_VARIABLES ${PASSTHROUGH_VARIABLES} "-DEXPORTED_LIBXML2_INCLUDE_DIR=${LIBXML2_INCLUDE_DIR}" CACHE STRING "" FORCE)
+    set(PASSTHROUGH_VARIABLES ${PASSTHROUGH_VARIABLES} "-DEXPORTED_LIBXML2_LIBRARY=${LIBXML2_LIBRARY}" CACHE STRING "" FORCE)
+
+    # Create imported targets
+    file(MAKE_DIRECTORY ${LIBXML2_INCLUDE_DIR})
+
+    add_library(LibXml2::LibXml2 STATIC IMPORTED)
+    set_target_properties(LibXml2::LibXml2 PROPERTIES IMPORTED_LOCATION "${LIBXML2_LIBRARY}")
+    add_dependencies(LibXml2::LibXml2 libxml2-external)
+    set_property(TARGET LibXml2::LibXml2 APPEND PROPERTY INTERFACE_INCLUDE_DIRECTORIES "${LIBXML2_INCLUDE_DIR}")
+endfunction(use_bundled_libxml2)
diff --git a/cmake/BundledOpenWSMAN.cmake b/cmake/BundledOpenWSMAN.cmake
new file mode 100644
index 0000000..0392768
--- /dev/null
+++ b/cmake/BundledOpenWSMAN.cmake
@@ -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.
+
+function(use_bundled_openwsman SOURCE_DIR BINARY_DIR)
+    message("Using bundled openwsman")
+
+    # Define patch step
+    set(PC "${Patch_EXECUTABLE}" -p1 -i "${SOURCE_DIR}/thirdparty/openwsman/openwsman.patch")
+
+    # Define byproducts
+    if (APPLE)
+        set(PREFIX "lib/lib")
+        set(POSTFIX ".a")
+    elseif(WIN32)
+        message(FATAL_ERROR "OpenWSMAN Windows build is not supported")
+    else()
+        if("${CMAKE_SIZEOF_VOID_P}" EQUAL "8")
+            set(PREFIX "lib64/lib")
+        else()
+            set(PREFIX "lib/lib")
+        endif()
+        set(POSTFIX ".a")
+    endif()
+
+    set(BYPRODUCTS
+            "${PREFIX}wsman${POSTFIX}"
+            "${PREFIX}wsman_client${POSTFIX}"
+            "${PREFIX}wsman_curl_client_transport${POSTFIX}"
+            )
+
+    FOREACH(BYPRODUCT ${BYPRODUCTS})
+        LIST(APPEND OPENWSMAN_LIBRARIES_LIST "${BINARY_DIR}/thirdparty/openwsman-install/${BYPRODUCT}")
+    ENDFOREACH(BYPRODUCT)
+
+    # Set build options
+    set(OPENWSMAN_CMAKE_ARGS
+            ${PASSTHROUGH_CMAKE_ARGS}
+            -DCMAKE_BUILD_TYPE=${CMAKE_BUILD_TYPE}
+            "-DCMAKE_INSTALL_PREFIX=${BINARY_DIR}/thirdparty/openwsman-install"
+            -DBUILD_PYTHON=NO
+            -DBUILD_PYTHON3=NO
+            -DBUILD_LIBCIM=NO
+            -DBUILD_EXAMPLES=NO
+            -DBUILD_BINDINGS=NO
+            -DBUILD_RUBY=NO
+            -DBUILD_PERL=NO
+            -DBUILD_JAVA=NO
+            -DBUILD_CSHARP=NO
+            -DBUILD_CUNIT_TESTS=NO
+            -DDISABLE_PLUGINS=YES
+            -DUSE_PAM=NO
+            -DBUILD_TESTS=NO
+            -DDISABLE_SERVER=YES
+            -DBUILD_SHARED_LIBS=NO)
+
+    append_third_party_passthrough_args(OPENWSMAN_CMAKE_ARGS "${OPENWSMAN_CMAKE_ARGS}")
+
+    # Build project
+    ExternalProject_Add(
+            openwsman-external
+            URL "https://github.com/Openwsman/openwsman/archive/v2.6.11.tar.gz"
+            URL_HASH "SHA256=895eaaae62925f9416766ea3e71a5368210e6cfe13b23e4e0422fa0e75c2541c"
+            SOURCE_DIR "${BINARY_DIR}/thirdparty/openwsman-src"
+            LIST_SEPARATOR % # This is needed for passing semicolon-separated lists
+            CMAKE_ARGS ${OPENWSMAN_CMAKE_ARGS}
+            PATCH_COMMAND ${PC}
+            BUILD_BYPRODUCTS "${OPENWSMAN_LIBRARIES_LIST}"
+            EXCLUDE_FROM_ALL TRUE
+    )
+
+    # Set dependencies
+    add_dependencies(openwsman-external LibXml2::LibXml2 OpenSSL::SSL OpenSSL::Crypto CURL::libcurl)
+
+    # Set variables
+    set(OPENWSMAN_FOUND "YES" CACHE STRING "" FORCE)
+    set(OPENWSMAN_INCLUDE_DIR "${BINARY_DIR}/thirdparty/openwsman-src/include" CACHE STRING "" FORCE)
+    set(OPENWSMAN_LIBRARIES "${OPENWSMAN_LIBRARIES_LIST}" CACHE STRING "" FORCE)
+
+    # Create imported targets
+    file(MAKE_DIRECTORY ${OPENWSMAN_INCLUDE_DIR})
+
+    add_library(OpenWSMAN::libwsman_curl_client_transport STATIC IMPORTED)
+    set_target_properties(OpenWSMAN::libwsman_curl_client_transport PROPERTIES IMPORTED_LOCATION "${BINARY_DIR}/thirdparty/openwsman-install/${PREFIX}wsman_curl_client_transport${POSTFIX}")
+    add_dependencies(OpenWSMAN::libwsman_curl_client_transport openwsman-external)
+    set_property(TARGET OpenWSMAN::libwsman_curl_client_transport APPEND PROPERTY INTERFACE_LINK_LIBRARIES LibXml2::LibXml2 OpenSSL::SSL OpenSSL::Crypto CURL::libcurl)
+    set_property(TARGET OpenWSMAN::libwsman_curl_client_transport APPEND PROPERTY INTERFACE_INCLUDE_DIRECTORIES ${OPENWSMAN_INCLUDE_DIR})
+
+    add_library(OpenWSMAN::libwsman_client STATIC IMPORTED)
+    set_target_properties(OpenWSMAN::libwsman_client PROPERTIES IMPORTED_LOCATION "${BINARY_DIR}/thirdparty/openwsman-install/${PREFIX}wsman_client${POSTFIX}")
+    add_dependencies(OpenWSMAN::libwsman_client openwsman-external)
+    set_property(TARGET OpenWSMAN::libwsman_client APPEND PROPERTY INTERFACE_LINK_LIBRARIES LibXml2::LibXml2)
+    set_property(TARGET OpenWSMAN::libwsman_client APPEND PROPERTY INTERFACE_LINK_LIBRARIES OpenWSMAN::libwsman_curl_client_transport)
+    set_property(TARGET OpenWSMAN::libwsman_client APPEND PROPERTY INTERFACE_INCLUDE_DIRECTORIES ${OPENWSMAN_INCLUDE_DIR})
+
+    add_library(OpenWSMAN::libwsman STATIC IMPORTED)
+    set_target_properties(OpenWSMAN::libwsman PROPERTIES IMPORTED_LOCATION "${BINARY_DIR}/thirdparty/openwsman-install/${PREFIX}wsman${POSTFIX}")
+    add_dependencies(OpenWSMAN::libwsman openwsman-external)
+    set_property(TARGET OpenWSMAN::libwsman APPEND PROPERTY INTERFACE_LINK_LIBRARIES LibXml2::LibXml2)
+    set_property(TARGET OpenWSMAN::libwsman APPEND PROPERTY INTERFACE_LINK_LIBRARIES OpenWSMAN::libwsman_client)
+    set_property(TARGET OpenWSMAN::libwsman APPEND PROPERTY INTERFACE_INCLUDE_DIRECTORIES ${OPENWSMAN_INCLUDE_DIR})
+endfunction(use_bundled_openwsman)
diff --git a/cmake/libxml2/dummy/FindLibXml2.cmake b/cmake/libxml2/dummy/FindLibXml2.cmake
new file mode 100644
index 0000000..9c2ce0d
--- /dev/null
+++ b/cmake/libxml2/dummy/FindLibXml2.cmake
@@ -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.
+
+if(NOT LIBXML2_FOUND)
+    set(LIBXML2_FOUND "YES" CACHE STRING "" FORCE)
+    set(LIBXML2_INCLUDE_DIR "${EXPORTED_LIBXML2_INCLUDE_DIR}" CACHE STRING "" FORCE)
+    set(LIBXML2_INCLUDE_DIRS "${EXPORTED_LIBXML2_INCLUDE_DIR}" CACHE STRING "" FORCE)
+    set(LIBXML2_LIBRARIES "${EXPORTED_LIBXML2_LIBRARY}" CACHE STRING "" FORCE)
+endif()
+
+if(NOT TARGET LibXml2::LibXml2)
+    add_library(LibXml2::LibXml2 STATIC IMPORTED)
+    set_target_properties(LibXml2::LibXml2 PROPERTIES IMPORTED_LOCATION "${LIBXML2_LIBRARIES}")
+    set_property(TARGET LibXml2::LibXml2 APPEND PROPERTY INTERFACE_INCLUDE_DIRECTORIES "${LIBXML2_INCLUDE_DIR}")
+endif()
diff --git a/extensions/openwsman/CMakeLists.txt b/extensions/openwsman/CMakeLists.txt
new file mode 100644
index 0000000..a9470d3
--- /dev/null
+++ b/extensions/openwsman/CMakeLists.txt
@@ -0,0 +1,32 @@
+#
+# 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(${CMAKE_SOURCE_DIR}/extensions/ExtensionHeader.txt)
+
+file(GLOB SOURCES "processors/*.cpp")
+
+add_library(minifi-openwsman STATIC ${SOURCES})
+
+set_property(TARGET minifi-openwsman PROPERTY POSITION_INDEPENDENT_CODE ON)
+
+target_link_libraries(minifi-openwsman ${LIBMINIFI} Threads::Threads)
+target_link_libraries(minifi-openwsman OpenWSMAN::libwsman CIVETWEB::civetweb-cpp CIVETWEB::c-library LibXml2::LibXml2)
+
+SET (OPENWSMAN-EXTENSION minifi-openwsman PARENT_SCOPE)
+register_extension(minifi-openwsman)
diff --git a/extensions/openwsman/processors/SourceInitiatedSubscriptionListener.cpp b/extensions/openwsman/processors/SourceInitiatedSubscriptionListener.cpp
new file mode 100644
index 0000000..d285b95
--- /dev/null
+++ b/extensions/openwsman/processors/SourceInitiatedSubscriptionListener.cpp
@@ -0,0 +1,922 @@
+/**
+ *
+ * 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 "SourceInitiatedSubscriptionListener.h"
+
+#include <memory>
+#include <algorithm>
+#include <cctype>
+#include <cstdint>
+#include <cstring>
+#include <iostream>
+#include <iterator>
+#include <limits>
+#include <unordered_map>
+#include <set>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include <openssl/x509.h>
+extern "C" {
+#include "wsman-api.h"
+#include "wsman-xml-api.h"
+#include "wsman-xml-serialize.h"
+#include "wsman-xml-serializer.h"
+#include "wsman-soap.h"
+#include "wsman-soap-envelope.h"
+}
+
+#include "utils/ByteArrayCallback.h"
+#include "core/FlowFile.h"
+#include "core/logging/Logger.h"
+#include "core/ProcessContext.h"
+#include "core/Relationship.h"
+#include "io/DataStream.h"
+#include "io/StreamFactory.h"
+#include "ResourceClaim.h"
+#include "utils/StringUtils.h"
+#include "utils/ScopeGuard.h"
+#include "utils/file/FileUtils.h"
+
+#define XML_NS_CUSTOM_SUBSCRIPTION "http://schemas.microsoft.com/wbem/wsman/1/subscription"
+#define XML_NS_CUSTOM_AUTHENTICATION "http://schemas.microsoft.com/wbem/wsman/1/authentication"
+#define XML_NS_CUSTOM_POLICY "http://schemas.xmlsoap.org/ws/2002/12/policy"
+#define XML_NS_CUSTOM_MACHINEID "http://schemas.microsoft.com/wbem/wsman/1/machineid"
+#define WSMAN_CUSTOM_ACTION_ACK "http://schemas.dmtf.org/wbem/wsman/1/wsman/Ack"
+#define WSMAN_CUSTOM_ACTION_HEARTBEAT "http://schemas.dmtf.org/wbem/wsman/1/wsman/Heartbeat"
+#define WSMAN_CUSTOM_ACTION_EVENTS "http://schemas.dmtf.org/wbem/wsman/1/wsman/Events"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace processors {
+core::Property SourceInitiatedSubscriptionListener::ListenHostname(
+    core::PropertyBuilder::createProperty("Listen Hostname")->withDescription("The hostname or IP of this machine that will be advertised to event sources to connect to. "
+                                                                              "It must be contained as a Subject Alternative Name in the server certificate, "
+                                                                              "otherwise source machines will refuse to connect.")
+        ->isRequired(true)->build());
+core::Property SourceInitiatedSubscriptionListener::ListenPort(
+    core::PropertyBuilder::createProperty("Listen Port")->withDescription("The port to listen on.")
+        ->isRequired(true)->withDefaultValue<int64_t>(5986, core::StandardValidators::LISTEN_PORT_VALIDATOR())->build());
+core::Property SourceInitiatedSubscriptionListener::SubscriptionManagerPath(
+    core::PropertyBuilder::createProperty("Subscription Manager Path")->withDescription("The URI path that will be used for the WEC Subscription Manager endpoint.")
+        ->isRequired(true)->withDefaultValue("/wsman/SubscriptionManager/WEC")->build());
+core::Property SourceInitiatedSubscriptionListener::SubscriptionsBasePath(
+    core::PropertyBuilder::createProperty("Subscriptions Base Path")->withDescription("The URI path that will be used as the base for endpoints serving individual subscriptions.")
+        ->isRequired(true)->withDefaultValue("/wsman/subscriptions")->build());
+core::Property SourceInitiatedSubscriptionListener::SSLCertificate(
+    core::PropertyBuilder::createProperty("SSL Certificate")->withDescription("File containing PEM-formatted file including TLS/SSL certificate and key. "
+                                                                              "The root CA of the certificate must be the CA set in SSL Certificate Authority.")
+        ->isRequired(true)->build());
+core::Property SourceInitiatedSubscriptionListener::SSLCertificateAuthority(
+    core::PropertyBuilder::createProperty("SSL Certificate Authority")->withDescription("File containing the PEM-formatted CA that is the root CA for both this server's certificate "
+                                                                                        "and the event source clients' certificates.")
+        ->isRequired(true)->build());
+core::Property SourceInitiatedSubscriptionListener::SSLVerifyPeer(
+    core::PropertyBuilder::createProperty("SSL Verify Peer")->withDescription("Whether or not to verify the client's certificate")
+        ->isRequired(false)->withDefaultValue<bool>(true)->build());
+core::Property SourceInitiatedSubscriptionListener::XPathXmlQuery(
+    core::PropertyBuilder::createProperty("XPath XML Query")->withDescription("An XPath Query in structured XML format conforming to the Query Schema described in "
+                                                                              "https://docs.microsoft.com/en-gb/windows/win32/wes/queryschema-schema, "
+                                                                              "see an example here: https://docs.microsoft.com/en-gb/windows/win32/wes/consuming-events")
+        ->isRequired(true)
+        ->withDefaultValue("<QueryList>\n"
+                           "  <Query Id=\"0\">\n"
+                           "    <Select Path=\"Application\">*</Select>\n"
+                           "  </Query>\n"
+                           "</QueryList>\n")->build());
+core::Property SourceInitiatedSubscriptionListener::InitialExistingEventsStrategy(
+    core::PropertyBuilder::createProperty("Initial Existing Events Strategy")->withDescription("Defines the behaviour of the Processor when a new event source connects.\n"
+    "None: will not request existing events\n"
+    "All: will request all existing events matching the query")
+        ->isRequired(true)->withAllowableValues<std::string>({INITIAL_EXISTING_EVENTS_STRATEGY_NONE, INITIAL_EXISTING_EVENTS_STRATEGY_ALL})
+        ->withDefaultValue(INITIAL_EXISTING_EVENTS_STRATEGY_NONE)->build());
+core::Property SourceInitiatedSubscriptionListener::SubscriptionExpirationInterval(
+    core::PropertyBuilder::createProperty("Subscription Expiration Interval")->withDescription("The interval while a subscription is valid without renewal. "
+    "Because in a source-initiated subscription, the collector can not cancel the subscription, "
+    "setting this too large could cause unnecessary load on the source machine. "
+    "Setting this too small causes frequent reenumeration and resubscription which is ineffective.")
+        ->isRequired(true)->withDefaultValue<core::TimePeriodValue>("10 min")->build());
+core::Property SourceInitiatedSubscriptionListener::HeartbeatInterval(
+    core::PropertyBuilder::createProperty("Heartbeat Interval")->withDescription("The processor will ask the sources to send heartbeats with this interval.")
+        ->isRequired(true)->withDefaultValue<core::TimePeriodValue>("30 sec")->build());
+core::Property SourceInitiatedSubscriptionListener::MaxElements(
+    core::PropertyBuilder::createProperty("Max Elements")->withDescription("The maximum number of events a source will batch together and send at once.")
+        ->isRequired(true)->withDefaultValue<uint32_t>(20U)->build());
+core::Property SourceInitiatedSubscriptionListener::MaxLatency(
+    core::PropertyBuilder::createProperty("Max Latency")->withDescription("The maximum time a source will wait to send new events.")
+        ->isRequired(true)->withDefaultValue<core::TimePeriodValue>("10 sec")->build());
+core::Property SourceInitiatedSubscriptionListener::ConnectionRetryInterval(
+    core::PropertyBuilder::createProperty("Connection Retry Interval")->withDescription("The interval with which a source will try to reconnect to the server.")
+        ->withDefaultValue<core::TimePeriodValue>("10 sec")->build());
+core::Property SourceInitiatedSubscriptionListener::ConnectionRetryCount(
+    core::PropertyBuilder::createProperty("Connection Retry Count")->withDescription("The number of connection retries after which a source will consider the subscription expired.")
+        ->withDefaultValue<uint32_t>(5U)->build());
+
+core::Relationship SourceInitiatedSubscriptionListener::Success("success", "All Events are routed to success");
+
+constexpr char const* SourceInitiatedSubscriptionListener::ATTRIBUTE_WEF_REMOTE_MACHINEID;
+constexpr char const* SourceInitiatedSubscriptionListener::ATTRIBUTE_WEF_REMOTE_IP;
+
+constexpr char const* SourceInitiatedSubscriptionListener::INITIAL_EXISTING_EVENTS_STRATEGY_NONE;
+constexpr char const* SourceInitiatedSubscriptionListener::INITIAL_EXISTING_EVENTS_STRATEGY_ALL;
+
+constexpr char const* SourceInitiatedSubscriptionListener::ProcessorName;
+
+SourceInitiatedSubscriptionListener::SourceInitiatedSubscriptionListener(std::string name, utils::Identifier uuid)
+    : Processor(name, uuid)
+    , logger_(logging::LoggerFactory<SourceInitiatedSubscriptionListener>::getLogger())
+    , session_factory_(nullptr)
+    , listen_port_(0U)
+    , subscription_expiration_interval_(0)
+    , heartbeat_interval_(0)
+    , max_elements_(0U)
+    , max_latency_(0)
+    , connection_retry_interval_(0)
+    , connection_retry_count_(0U) {
+}
+
+SourceInitiatedSubscriptionListener::Handler::Handler(SourceInitiatedSubscriptionListener& processor)
+    : processor_(processor) {
+}
+
+SourceInitiatedSubscriptionListener::SubscriberData::SubscriberData()
+    : bookmark_(nullptr)
+    , subscription_(nullptr) {
+}
+
+SourceInitiatedSubscriptionListener::SubscriberData::~SubscriberData() {
+  clearSubscription();
+  clearBookmark();
+}
+
+void SourceInitiatedSubscriptionListener::SubscriberData::setSubscription(const std::string& subscription_version,
+    WsXmlDocH subscription,
+    const std::string& subscription_endpoint,
+    const std::string& subscription_identifier) {
+  clearSubscription();
+  subscription_version_ = subscription_version;
+  subscription_ = subscription;
+  subscription_endpoint_ = subscription_endpoint;
+  subscription_identifier_ = subscription_identifier;
+}
+
+void SourceInitiatedSubscriptionListener::SubscriberData::clearSubscription() {
+  subscription_version_.clear();
+  if (subscription_ != nullptr) {
+    ws_xml_destroy_doc(subscription_);
+  }
+  subscription_ = nullptr;
+}
+
+void SourceInitiatedSubscriptionListener::SubscriberData::setBookmark(WsXmlDocH bookmark) {
+  clearBookmark();
+  bookmark_ = bookmark;
+}
+
+void SourceInitiatedSubscriptionListener::SubscriberData::clearBookmark() {
+  if (bookmark_ != nullptr) {
+    ws_xml_destroy_doc(bookmark_);
+  }
+  bookmark_ = nullptr;
+}
+
+bool SourceInitiatedSubscriptionListener::persistState() const {
+  std::unordered_map<std::string, std::string> state_map;
+  size_t i = 0U;
+  for (const auto& subscriber : subscribers_) {
+    char* xml_buf = nullptr;
+    int xml_buf_size = 0;
+    ws_xml_dump_memory_enc(subscriber.second.bookmark_, &xml_buf, &xml_buf_size, "UTF-8");
+    state_map.emplace("subscriber." + std::to_string(i) + ".machineid", subscriber.first);
+    state_map.emplace("subscriber." + std::to_string(i) + ".bookmark", std::string(xml_buf, xml_buf_size));
+    i++;
+    ws_xml_free_memory(xml_buf);
+  }
+  return state_manager_->set(state_map);
+}
+
+bool SourceInitiatedSubscriptionListener::loadState() {
+  std::unordered_map<std::string, std::string> state_map;
+  if (!state_manager_->get(state_map)) {
+    return false;
+  }
+
+  for (size_t i = 0U;; i++) {
+    std::string machineId;
+    try {
+      machineId = state_map.at("subscriber." + std::to_string(i) + ".machineid");
+    } catch (...) {
+      break;
+    }
+
+    std::string bookmark;
+    try {
+      bookmark = state_map.at("subscriber." + std::to_string(i) + ".bookmark");
+    } catch (...) {
+      logger_->log_error("Bookmark for subscriber \"%s\" is missing, skipping", machineId);
+      continue;
+    }
+
+    WsXmlDocH doc = ws_xml_read_memory(bookmark.data(), bookmark.size(), "UTF-8", 0);
+    if (doc == nullptr) {
+      logger_->log_error("Failed to parse saved bookmark for subscriber \"%s\", skipping", machineId);
+      continue;
+    }
+    subscribers_[machineId].setBookmark(doc);
+  }
+
+  return true;
+}
+
+std::string SourceInitiatedSubscriptionListener::Handler::millisecondsToXsdDuration(int64_t milliseconds) {
+  char buf[1024];
+  snprintf(buf, sizeof(buf), "PT%lld.%03lldS", milliseconds / 1000, milliseconds % 1000);
+  return buf;
+}
+
+bool SourceInitiatedSubscriptionListener::Handler::handlePost(CivetServer* server, struct mg_connection* conn) {
+  const struct mg_request_info* req_info = mg_get_request_info(conn);
+  if (req_info == nullptr) {
+    processor_.logger_->log_error("Failed to get request info");
+    return false;
+  }
+
+  const char* endpoint = req_info->local_uri;
+  if (endpoint == nullptr) {
+    processor_.logger_->log_error("Failed to get called endpoint (local_uri)");
+    return false;
+  }
+  processor_.logger_->log_trace("Endpoint \"%s\" has been called", endpoint);
+
+  for (int i = 0; i < req_info->num_headers; i++) {
+    processor_.logger_->log_trace("Received header \"%s: %s\"", req_info->http_headers[i].name, req_info->http_headers[i].value);
+  }
+
+  const char* content_type = mg_get_header(conn, "Content-Type");
+  if (content_type == nullptr) {
+    processor_.logger_->log_error("Content-Type header missing");
+    return false;
+  }
+
+  std::string charset;
+  const char* charset_begin = strstr(content_type, "charset=");
+  if (charset_begin == nullptr) {
+    processor_.logger_->log_warn("charset missing from Content-Type header, assuming UTF-8");
+    charset = "UTF-8";
+  } else {
+    charset_begin += strlen("charset=");
+    const char* charset_end = strchr(charset_begin, ';');
+    if (charset_end == nullptr) {
+        charset = std::string(charset_begin);
+    } else {
+        charset = std::string(charset_begin, charset_end - charset_begin);
+    }
+  }
+  processor_.logger_->log_trace("charset is \"%s\"", charset.c_str());
+
+  std::vector<uint8_t> raw_data;
+  {
+    std::array<uint8_t, 16384U> buf;
+    int read_bytes;
+    while ((read_bytes = mg_read(conn, buf.data(), buf.size())) > 0) {
+      size_t orig_size = raw_data.size();
+      raw_data.resize(orig_size + read_bytes);
+      memcpy(raw_data.data() + orig_size, buf.data(), read_bytes);
+    }
+  }
+
+  if (raw_data.empty()) {
+    processor_.logger_->log_error("POST body is empty");
+    return false;
+  }
+
+  WsXmlDocH doc = ws_xml_read_memory(reinterpret_cast<char*>(raw_data.data()), raw_data.size(), charset.c_str(), 0);
+
+  if (doc == nullptr) {
+    processor_.logger_->log_error("Failed to parse POST body as XML");
+    return false;
+  }
+
+  {
+    WsXmlNodeH node = ws_xml_get_doc_root(doc);
+    char* xml_buf = nullptr;
+    int xml_buf_size = 0;
+    ws_xml_dump_memory_node_tree_enc(node, &xml_buf, &xml_buf_size, "UTF-8");
+    if (xml_buf != nullptr) {
+        logging::LOG_TRACE(processor_.logger_) << "Received request: \"" << std::string(xml_buf, xml_buf_size) << "\"";
+        ws_xml_free_memory(xml_buf);
+    }
+  }
+
+  if (endpoint == processor_.subscription_manager_path_) {
+    return this->handleSubscriptionManager(conn, endpoint, doc);
+  } else if (strncmp(endpoint, processor_.subscriptions_base_path_.c_str(), processor_.subscriptions_base_path_.length()) == 0) {
+    return this->handleSubscriptions(conn, endpoint, doc);
+  } else {
+    ws_xml_destroy_doc(doc);
+    return false;
+  }
+}
+
+std::string SourceInitiatedSubscriptionListener::Handler::getSoapAction(WsXmlDocH doc) {
+  WsXmlNodeH header = ws_xml_get_soap_header(doc);
+  if (header == nullptr) {
+    return "";
+  }
+  WsXmlNodeH action_node = ws_xml_get_child(header, 0 /*index*/, XML_NS_ADDRESSING, WSA_ACTION);
+  if (action_node == nullptr) {
+    return "";
+  }
+  char* text = ws_xml_get_node_text(action_node);
+  if (text == nullptr) {
+    return "";
+  }
+
+  return text;
+}
+
+std::string SourceInitiatedSubscriptionListener::Handler::getMachineId(WsXmlDocH doc) {
+  WsXmlNodeH header = ws_xml_get_soap_header(doc);
+  if (header == nullptr) {
+    return "";
+  }
+  WsXmlNodeH machineid_node = ws_xml_get_child(header, 0 /*index*/, XML_NS_CUSTOM_MACHINEID, "MachineID");
+  if (machineid_node == nullptr) {
+    return "";
+  }
+  char* text = ws_xml_get_node_text(machineid_node);
+  if (text == nullptr) {
+    return "";
+  }
+
+  return text;
+}
+
+bool SourceInitiatedSubscriptionListener::Handler::isAckRequested(WsXmlDocH doc) {
+  WsXmlNodeH header = ws_xml_get_soap_header(doc);
+  if (header == nullptr) {
+    return false;
+  }
+  WsXmlNodeH ack_requested_node = ws_xml_get_child(header, 0 /*index*/, XML_NS_WS_MAN, WSM_ACKREQUESTED);
+  return ack_requested_node != nullptr;
+}
+
+void SourceInitiatedSubscriptionListener::Handler::sendResponse(struct mg_connection* conn, const std::string& machineId, const std::string& remoteIp, char* xml_buf, size_t xml_buf_size) {
+  logging::LOG_TRACE(processor_.logger_) << "Sending response to " << machineId << " (" << remoteIp << "): \"" << std::string(xml_buf, xml_buf_size) << "\"";
+
+  mg_printf(conn, "HTTP/1.1 200 OK\r\n");
+  mg_printf(conn, "Content-Type: application/soap+xml;charset=UTF-8\r\n");
+  mg_printf(conn, "Authorization: %s\r\n", WSMAN_SECURITY_PROFILE_HTTPS_MUTUAL);
+  mg_printf(conn, "Content-Length: %zu\r\n", xml_buf_size);
+  mg_printf(conn, "\r\n");
+  mg_printf(conn, "%.*s", static_cast<int>(xml_buf_size), xml_buf);
+}
+
+bool SourceInitiatedSubscriptionListener::Handler::handleSubscriptionManager(struct mg_connection* conn, const std::string& endpoint, WsXmlDocH request) {
+  utils::ScopeGuard request_guard([&]() {
+      ws_xml_destroy_doc(request);
+  });
+
+  auto action = getSoapAction(request);
+  auto machine_id = getMachineId(request);
+  const struct mg_request_info* req_info = mg_get_request_info(conn);
+  std::string remote_ip = req_info->remote_addr;
+  if (action != ENUM_ACTION_ENUMERATE) {
+    processor_.logger_->log_error("%s called by %s (%s) with unknown Action \"%s\"", endpoint.c_str(), machine_id.c_str(), remote_ip.c_str(), action.c_str());
+    return false;  // TODO(bakaid): generate fault if possible
+  }
+
+  // Create reponse envelope from request
+  WsXmlDocH response = wsman_create_response_envelope(request, nullptr);
+  utils::ScopeGuard response_guard([&]() {
+    ws_xml_destroy_doc(response);
+  });
+
+  // Header
+  WsXmlNodeH response_header = ws_xml_get_soap_header(response);
+  // Header/MessageID
+  utils::Identifier msg_id = utils::IdGenerator::getIdGenerator()->generate();
+  ws_xml_add_child_format(response_header, XML_NS_ADDRESSING, WSA_MESSAGE_ID, "uuid:%s", msg_id.to_string().c_str());
+
+  // Body
+  WsXmlNodeH response_body = ws_xml_get_soap_body(response);
+  // Body/EnumerationResponse
+  WsXmlNodeH enumeration_response = ws_xml_add_child(response_body, XML_NS_ENUMERATION, WSENUM_ENUMERATE_RESP, nullptr);
+  // Body/EnumerationResponse/EnumerationContext
+  ws_xml_add_child(enumeration_response, XML_NS_ENUMERATION, WSENUM_ENUMERATION_CONTEXT, nullptr);
+  // Body/EnumerationResponse/Items
+  WsXmlNodeH enumeration_items = ws_xml_add_child(enumeration_response, XML_NS_WS_MAN, WSENUM_ITEMS, nullptr);
+  // Body/EnumerationResponse/EndOfSequence
+  ws_xml_add_child(enumeration_response, XML_NS_WS_MAN, WSENUM_END_OF_SEQUENCE, nullptr);
+
+  // Body/EnumerationResponse/Items/Subscription
+  WsXmlNodeH subscription = ws_xml_add_child(enumeration_items, nullptr, "Subscription", nullptr);
+  ws_xml_set_ns(subscription, XML_NS_CUSTOM_SUBSCRIPTION, "m");
+
+  // Body/EnumerationResponse/Items/Subscription/Version
+  std::lock_guard<std::mutex> lock(processor_.mutex_);
+  auto it = processor_.subscribers_.find(machine_id);
+
+  std::string subscription_version;
+  if (it != processor_.subscribers_.end() && it->second.subscription_ != nullptr) {
+    subscription_version = it->second.subscription_version_;
+  } else {
+    utils::Identifier id = utils::IdGenerator::getIdGenerator()->generate();
+    subscription_version = id.to_string();
+  }
+  ws_xml_add_child_format(subscription, XML_NS_CUSTOM_SUBSCRIPTION, "Version", "uuid:%s", subscription_version.c_str());
+
+  // Body/EnumerationResponse/Items/Subscription/Envelope
+  std::string subscription_identifier;
+  std::string subscription_endpoint;
+  if (it != processor_.subscribers_.end() && it->second.subscription_ != nullptr) {
+    WsXmlNodeH subscription_node = ws_xml_get_doc_root(it->second.subscription_);
+    ws_xml_copy_node(subscription_node, subscription);
+  } else {
+    WsXmlDocH subscription_doc = ws_xml_create_envelope();
+
+    // Header
+    WsXmlNodeH header = ws_xml_get_soap_header(subscription_doc);
+    WsXmlNodeH node;
+
+    // Header/Action
+    node = ws_xml_add_child(header, XML_NS_ADDRESSING, WSA_ACTION, EVT_ACTION_SUBSCRIBE);
+    ws_xml_add_node_attr(node, XML_NS_SOAP_1_2, SOAP_MUST_UNDERSTAND, "true");
+
+    // Header/MessageID
+    utils::Identifier msg_id = utils::IdGenerator::getIdGenerator()->generate();
+    ws_xml_add_child_format(header, XML_NS_ADDRESSING, WSA_MESSAGE_ID, "uuid:%s", msg_id.to_string().c_str());
+
+    // Header/To
+    node = ws_xml_add_child(header, XML_NS_ADDRESSING, WSA_TO, WSA_TO_ANONYMOUS);
+    ws_xml_add_node_attr(node, XML_NS_SOAP_1_2, SOAP_MUST_UNDERSTAND, "true");
+
+    // Header/ResourceURI
+    node = ws_xml_add_child(header, XML_NS_WS_MAN, WSM_RESOURCE_URI, "http://schemas.microsoft.com/wbem/wsman/1/windows/EventLog");
+    ws_xml_add_node_attr(node, XML_NS_SOAP_1_2, SOAP_MUST_UNDERSTAND, "true");
+
+    // Header/ReplyTo
+    node = ws_xml_add_child(header, XML_NS_ADDRESSING, WSA_REPLY_TO, nullptr);
+    node = ws_xml_add_child(node, XML_NS_ADDRESSING, WSA_ADDRESS, WSA_TO_ANONYMOUS);
+    ws_xml_add_node_attr(node, XML_NS_SOAP_1_2, SOAP_MUST_UNDERSTAND, "true");
+
+    // Header/OptionSet
+    WsXmlNodeH option_set = ws_xml_add_child(header, XML_NS_WS_MAN, WSM_OPTION_SET, nullptr);
+    ws_xml_ns_add(option_set, XML_NS_SCHEMA_INSTANCE, XML_NS_SCHEMA_INSTANCE_PREFIX);
+
+    // Header/OptionSet/Option (CDATA)
+    node = ws_xml_add_child(option_set, XML_NS_WS_MAN, WSM_OPTION, nullptr);
+    ws_xml_add_node_attr(node, nullptr, WSM_NAME, "CDATA");
+    ws_xml_add_node_attr(node, XML_NS_SCHEMA_INSTANCE, XML_SCHEMA_NIL, "true");
+
+    // Header/OptionSet/Option (IgnoreChannelError)
+    node = ws_xml_add_child(option_set, XML_NS_WS_MAN, WSM_OPTION, nullptr);
+    ws_xml_add_node_attr(node, nullptr, WSM_NAME, "IgnoreChannelError");
+    ws_xml_add_node_attr(node, XML_NS_SCHEMA_INSTANCE, XML_SCHEMA_NIL, "true");
+
+    // Body
+    WsXmlNodeH body = ws_xml_get_soap_body(subscription_doc);
+    WsXmlNodeH subscribe_node = ws_xml_add_child(body, XML_NS_EVENTING, WSEVENT_SUBSCRIBE, nullptr);
+
+    // Body/Delivery
+    {
+      utils::Identifier id = utils::IdGenerator::getIdGenerator()->generate();
+      subscription_identifier = id.to_string();
+    }
+    {
+      utils::Identifier id = utils::IdGenerator::getIdGenerator()->generate();
+      subscription_endpoint = processor_.subscriptions_base_path_ + "/" + id.to_string();
+    }
+
+    WsXmlNodeH delivery_node = ws_xml_add_child(subscribe_node, XML_NS_EVENTING, WSEVENT_DELIVERY, nullptr);
+    ws_xml_add_node_attr(delivery_node, nullptr, WSEVENT_DELIVERY_MODE, WSEVENT_DELIVERY_MODE_EVENTS);
+
+    // Body/Delivery/Heartbeats
+    ws_xml_add_child(delivery_node, XML_NS_WS_MAN, WSM_HEARTBEATS, millisecondsToXsdDuration(processor_.heartbeat_interval_).c_str());
+
+    // Body/Delivery/ConnectionRetry
+    auto connection_retry_node = ws_xml_add_child(delivery_node, XML_NS_WS_MAN, WSM_CONNECTIONRETRY, millisecondsToXsdDuration(processor_.connection_retry_interval_).c_str());
+    ws_xml_add_node_attr(connection_retry_node, nullptr, "Total", std::to_string(processor_.connection_retry_count_).c_str());
+
+    // Body/Delivery/NotifyTo and Body/EndTo are the same, so we will use this lambda to recreate the same tree
+    auto apply_endpoint_nodes = [&](WsXmlNodeH target_node) {
+      // ${target_node}/NotifyTo/Address
+      ws_xml_add_child_format(target_node, XML_NS_ADDRESSING, WSA_ADDRESS, "https://%s:%hu%s",
+                              processor_.listen_hostname_.c_str(),
+                              processor_.listen_port_,
+                              subscription_endpoint.c_str());
+      // ${target_node}/ReferenceProperties
+      node = ws_xml_add_child(target_node, XML_NS_ADDRESSING, WSA_REFERENCE_PROPERTIES, nullptr);
+      // ${target_node}/ReferenceProperties/Identifier
+      ws_xml_add_child_format(node, XML_NS_EVENTING, WSEVENT_IDENTIFIER, "%s", subscription_identifier.c_str());
+      // ${target_node}/Policy
+      WsXmlNodeH policy = ws_xml_add_child(target_node, nullptr, "Policy", nullptr);
+      ws_xml_set_ns(policy, XML_NS_CUSTOM_POLICY, "c");
+      ws_xml_ns_add(policy, XML_NS_CUSTOM_AUTHENTICATION, "auth");
+      // ${target_node}/Policy/ExactlyOne
+      WsXmlNodeH exactly_one = ws_xml_add_child(policy, XML_NS_CUSTOM_POLICY, "ExactlyOne", nullptr);
+      // ${target_node}/Policy/ExactlyOne/All
+      WsXmlNodeH all = ws_xml_add_child(exactly_one, XML_NS_CUSTOM_POLICY, "All", nullptr);
+      // ${target_node}/Policy/ExactlyOne/All/Authentication
+      WsXmlNodeH authentication = ws_xml_add_child(all, XML_NS_CUSTOM_AUTHENTICATION, "Authentication", nullptr);
+      ws_xml_add_node_attr(authentication, nullptr, "Profile", WSMAN_SECURITY_PROFILE_HTTPS_MUTUAL);
+      // ${target_node}/Policy/ExactlyOne/All/Authentication/ClientCertificate
+      WsXmlNodeH client_certificate = ws_xml_add_child(authentication, XML_NS_CUSTOM_AUTHENTICATION, "ClientCertificate", nullptr);
+      // ${target_node}/Policy/ExactlyOne/All/Authentication/ClientCertificate/Thumbprint
+      WsXmlNodeH thumbprint = ws_xml_add_child_format(client_certificate, XML_NS_CUSTOM_AUTHENTICATION, "Thumbprint", "%s", processor_.ssl_ca_cert_thumbprint_.c_str());
+      ws_xml_add_node_attr(thumbprint, nullptr, "Role", "issuer");
+    };
+
+    // Body/Delivery/NotifyTo
+    WsXmlNodeH notifyto_node = ws_xml_add_child(delivery_node, XML_NS_EVENTING, WSEVENT_NOTIFY_TO, nullptr);
+    apply_endpoint_nodes(notifyto_node);
+
+    // Body/EndTo
+    WsXmlNodeH endto_node = ws_xml_add_child(subscribe_node, XML_NS_EVENTING, WSEVENT_ENDTO, nullptr);
+    apply_endpoint_nodes(endto_node);
+
+    // Body/MaxElements
+    ws_xml_add_child(delivery_node, XML_NS_WS_MAN, WSM_MAX_ELEMENTS, std::to_string(processor_.max_elements_).c_str());
+    // Body/MaxTime
+    ws_xml_add_child(delivery_node, XML_NS_WS_MAN, WSENUM_MAX_TIME, millisecondsToXsdDuration(processor_.max_latency_).c_str());
+
+    // Body/Expires
+    ws_xml_add_child(subscribe_node, XML_NS_EVENTING, WSEVENT_EXPIRES, millisecondsToXsdDuration(processor_.subscription_expiration_interval_).c_str());
+
+    // Body/Filter
+    WsXmlNodeH filter_node = ws_xml_add_child(subscribe_node, XML_NS_WS_MAN, WSM_FILTER, processor_.xpath_xml_query_.c_str());
+    // ws_xml_add_node_attr(filter_node, nullptr, "Dialect", "http://schemas.microsoft.com/win/2004/08/events/eventquery");
+
+    // Body/Bookmark
+    if (it != processor_.subscribers_.end() && it->second.bookmark_ != nullptr) {
+      WsXmlNodeH bookmark_node = ws_xml_get_doc_root(it->second.bookmark_);
+      ws_xml_copy_node(bookmark_node, subscribe_node);
+    } else if (processor_.initial_existing_events_strategy_ == INITIAL_EXISTING_EVENTS_STRATEGY_ALL) {
+      ws_xml_add_child(subscribe_node, XML_NS_WS_MAN, WSM_BOOKMARK, "http://schemas.dmtf.org/wbem/wsman/1/wsman/bookmark/earliest");
+    }
+
+    // Body/SendBookmarks
+    ws_xml_add_child(subscribe_node, XML_NS_WS_MAN, WSM_SENDBOOKMARKS, nullptr);
+
+    // Copy the whole Subscription
+    WsXmlNodeH subscription_node = ws_xml_get_doc_root(subscription_doc);
+    ws_xml_copy_node(subscription_node, subscription);
+
+    // Save subscription
+    if (it == processor_.subscribers_.end()) {
+      it = processor_.subscribers_.emplace(machine_id, SubscriberData()).first;
+    }
+    it->second.setSubscription(subscription_version, subscription_doc, subscription_endpoint, subscription_identifier);
+  }
+
+  // Send response
+  char* xml_buf = nullptr;
+  int xml_buf_size = 0;
+  ws_xml_dump_memory_enc(response, &xml_buf, &xml_buf_size, "UTF-8");
+
+  sendResponse(conn, machine_id, req_info->remote_addr, xml_buf, xml_buf_size);
+
+  ws_xml_free_memory(xml_buf);
+
+  return true;
+}
+
+SourceInitiatedSubscriptionListener::Handler::WriteCallback::WriteCallback(char* text)
+    : text_(text) {
+}
+
+int64_t SourceInitiatedSubscriptionListener::Handler::WriteCallback::process(std::shared_ptr<io::BaseStream> stream) {
+  return stream->write(reinterpret_cast<uint8_t*>(text_), strlen(text_));
+}
+
+int SourceInitiatedSubscriptionListener::Handler::enumerateEventCallback(WsXmlNodeH node, void* data) {
+  if (data == nullptr) {
+    return 1;
+  }
+
+  std::shared_ptr<core::ProcessSession> session;
+  std::shared_ptr<logging::Logger> logger;
+  std::string machine_id;
+  std::string remote_ip;
+  std::tie(session, logger, machine_id, remote_ip) = *static_cast<std::tuple<std::shared_ptr<core::ProcessSession>, std::shared_ptr<logging::Logger>, std::string, std::string>*>(data);
+
+  char* text = ws_xml_get_node_text(node);
+  if (text == nullptr) {
+      logger->log_error("Failed to get text for node");
+      return 1;
+  }
+
+  try {
+    logger->log_trace("Found Event");
+    auto flow_file = std::static_pointer_cast<FlowFileRecord>(session->create());
+    if (flow_file == nullptr) {
+      logger->log_error("Failed to create FlowFile");
+      return 1;
+    }
+
+    WriteCallback callback(text);
+    session->write(flow_file, &callback);
+
+    session->putAttribute(flow_file, FlowAttributeKey(MIME_TYPE), "application/xml");
+    flow_file->addAttribute(ATTRIBUTE_WEF_REMOTE_MACHINEID, machine_id);
+    flow_file->addAttribute(ATTRIBUTE_WEF_REMOTE_IP, remote_ip);
+
+    session->transfer(flow_file, SourceInitiatedSubscriptionListener::Success);
+    session->commit();
+  } catch (const std::exception& e) {
+    logger->log_error("Caught exception while processing Events: %s", e.what());
+    return 1;
+  } catch (...) {
+    logger->log_error("Caught exception while processing Events");
+    return 1;
+  }
+
+  return 0;
+}
+
+bool SourceInitiatedSubscriptionListener::Handler::handleSubscriptions(struct mg_connection* conn, const std::string& endpoint, WsXmlDocH request) {
+  utils::ScopeGuard guard([&]() {
+      ws_xml_destroy_doc(request);
+  });
+  auto action = getSoapAction(request);
+  auto machine_id = getMachineId(request);
+  const struct mg_request_info* req_info = mg_get_request_info(conn);
+  std::string remote_ip = req_info->remote_addr;
+  if (action == EVT_ACTION_SUBEND) {
+    std::lock_guard<std::mutex> lock(processor_.mutex_);
+    auto it = processor_.subscribers_.find(machine_id);
+    if (it != processor_.subscribers_.end()) {
+        processor_.subscribers_.erase(it);
+    }
+    // TODO(bakaid): make sure whether we really have to clean the bookmark as well (based on the fault)
+  } else if (action == WSMAN_CUSTOM_ACTION_HEARTBEAT) {
+    processor_.logger_->log_debug("Received Heartbeat on %s from %s (%s)", endpoint.c_str(), machine_id.c_str(), remote_ip.c_str());
+  } else if (action == WSMAN_CUSTOM_ACTION_EVENTS) {
+    processor_.logger_->log_debug("Received Events on %s from %s (%s)", endpoint.c_str(), machine_id.c_str(), remote_ip.c_str());
+    // Body
+    WsXmlNodeH body = ws_xml_get_soap_body(request);
+    if (body == nullptr) {
+      processor_.logger_->log_error("Received malformed Events request on %s from %s (%s), SOAP Body missing", endpoint.c_str(), machine_id.c_str(), remote_ip.c_str());
+      return false;
+    }
+    // Body/Events
+    WsXmlNodeH events_node = ws_xml_get_child(body, 0 /*index*/, XML_NS_WS_MAN, WSM_EVENTS);
+    if (events_node == nullptr) {
+      processor_.logger_->log_error("Received malformed Events request on %s from %s (%s), Events missing", endpoint.c_str(), machine_id.c_str(), remote_ip.c_str());
+      return false;
+    }
+    const struct mg_request_info* req_info = mg_get_request_info(conn);
+    // Enumare Body/Events/Event nodes
+    auto session = processor_.session_factory_->createSession();
+    std::tuple<std::shared_ptr<core::ProcessSession>, std::shared_ptr<logging::Logger>, std::string, std::string> callback_args =
+        std::forward_as_tuple(session, processor_.logger_, machine_id, remote_ip);
+    int ret = ws_xml_enum_children(events_node, &SourceInitiatedSubscriptionListener::Handler::enumerateEventCallback, &callback_args, 0 /*bRecursive*/);
+    if (ret != 0) {
+      processor_.logger_->log_error("Failed to parse events on %s from %s (%s), rolling back session", endpoint.c_str(), machine_id.c_str(), remote_ip.c_str());
+      session->rollback();
+    }
+    // Header
+    WsXmlNodeH header = ws_xml_get_soap_header(request);
+    // Header/Bookmark
+    WsXmlNodeH bookmark_node = ws_xml_get_child(header, 0 /*index*/, XML_NS_WS_MAN, WSM_BOOKMARK);
+    if (ret == 0 && bookmark_node != nullptr) {
+      WsXmlDocH bookmark_doc = ws_xml_create_doc(XML_NS_WS_MAN, WSM_BOOKMARK);
+      WsXmlNodeH temp = ws_xml_get_doc_root(bookmark_doc);
+      ws_xml_duplicate_children(temp, bookmark_node);
+
+      std::lock_guard<std::mutex> lock(processor_.mutex_);
+      auto it = processor_.subscribers_.find(machine_id);
+      if (it != processor_.subscribers_.end()) {
+        it = processor_.subscribers_.emplace(machine_id, SubscriberData()).first;
+      }
+      it->second.setBookmark(bookmark_doc);
+      // Bookmark changed, invalidate stored subscription
+      it->second.clearSubscription();
+
+      // Persist state
+      processor_.persistState();
+
+      char* xml_buf = nullptr;
+      int xml_buf_size = 0;
+      ws_xml_dump_memory_enc(bookmark_doc, &xml_buf, &xml_buf_size, "UTF-8");
+      processor_.logger_->log_debug("Saved new bookmark for %s: \"%.*s\"", machine_id.c_str(), xml_buf_size, xml_buf);
+      ws_xml_free_memory(xml_buf);
+    }
+  } else {
+    processor_.logger_->log_error("%s called by %s (%s) with unknown Action \"%s\"", endpoint.c_str(), machine_id.c_str(), remote_ip.c_str(), action.c_str());
+    return false;  // TODO(bakaid): generate fault if possible
+  }
+
+  if (isAckRequested(request)) {
+    // Assemble ACK
+    WsXmlDocH ack = wsman_create_response_envelope(request, WSMAN_CUSTOM_ACTION_ACK);
+    // Header
+    WsXmlNodeH ack_header = ws_xml_get_soap_header(ack);
+
+    // Header/MessageID
+    utils::Identifier msg_id = utils::IdGenerator::getIdGenerator()->generate();
+    ws_xml_add_child_format(ack_header, XML_NS_ADDRESSING, WSA_MESSAGE_ID, "uuid:%s", msg_id.to_string().c_str());
+
+    // Send ACK
+    char* xml_buf = nullptr;
+    int xml_buf_size = 0;
+    ws_xml_dump_memory_enc(ack, &xml_buf, &xml_buf_size, "UTF-8");
+
+    sendResponse(conn, machine_id, remote_ip, xml_buf, xml_buf_size);
+
+    ws_xml_free_memory(xml_buf);
+    ws_xml_destroy_doc(ack);
+  }
+
+  return true;
+}
+
+void SourceInitiatedSubscriptionListener::onTrigger(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) {
+  logger_->log_trace("SourceInitiatedSubscriptionListener onTrigger called");
+}
+
+void SourceInitiatedSubscriptionListener::initialize() {
+  logger_->log_trace("Initializing SourceInitiatedSubscriptionListener");
+
+  // Set the supported properties
+  std::set<core::Property> properties;
+  properties.insert(ListenHostname);
+  properties.insert(ListenPort);
+  properties.insert(SubscriptionManagerPath);
+  properties.insert(SubscriptionsBasePath);
+  properties.insert(SSLCertificate);
+  properties.insert(SSLCertificateAuthority);
+  properties.insert(SSLVerifyPeer);
+  properties.insert(XPathXmlQuery);
+  properties.insert(InitialExistingEventsStrategy);
+  properties.insert(SubscriptionExpirationInterval);
+  properties.insert(HeartbeatInterval);
+  properties.insert(MaxElements);
+  properties.insert(MaxLatency);
+  properties.insert(ConnectionRetryInterval);
+  properties.insert(ConnectionRetryCount);
+  setSupportedProperties(properties);
+
+  // Set the supported relationships
+  std::set<core::Relationship> relationships;
+  relationships.insert(Success);
+  setSupportedRelationships(relationships);
+}
+
+void SourceInitiatedSubscriptionListener::onSchedule(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSessionFactory> &sessionFactory) {
+  std::string ssl_certificate_file;
+  std::string ssl_ca_file;
+  bool verify_peer = true;
+
+  state_manager_ = context->getStateManager();
+  if (state_manager_ == nullptr) {
+    throw Exception(PROCESSOR_EXCEPTION, "Failed to get StateManager");
+  }
+
+  std::string value;
+  context->getProperty(ListenHostname.getName(), listen_hostname_);
+  if (!context->getProperty(ListenPort.getName(), value)) {
+    throw Exception(PROCESSOR_EXCEPTION,"Listen Port attribute is missing or invalid");
+  } else {
+    core::Property::StringToInt(value, listen_port_);
+  }
+  context->getProperty(SubscriptionManagerPath.getName(), subscription_manager_path_);
+  context->getProperty(SubscriptionsBasePath.getName(), subscriptions_base_path_);
+  if (!context->getProperty(SSLCertificate.getName(), ssl_certificate_file)) {
+    throw Exception(PROCESSOR_EXCEPTION,"SSL Certificate attribute is missing");
+  }
+  if (!context->getProperty(SSLCertificateAuthority.getName(), ssl_ca_file)) {
+    throw Exception(PROCESSOR_EXCEPTION,"SSL Certificate Authority attribute is missing");
+  }
+  if (!context->getProperty(SSLVerifyPeer.getName(), value)) {
+    throw Exception(PROCESSOR_EXCEPTION,"SSL Verify Peer attribute is missing or invalid");
+  } else {
+    utils::StringUtils::StringToBool(value, verify_peer);
+  }
+  context->getProperty(XPathXmlQuery.getName(), xpath_xml_query_);
+  if (!context->getProperty(InitialExistingEventsStrategy.getName(), initial_existing_events_strategy_)) {
+    throw Exception(PROCESSOR_EXCEPTION,"Initial Existing Events Strategy attribute is missing or invalid");
+  }
+  if (!context->getProperty(SubscriptionExpirationInterval.getName(), value)) {
+    throw Exception(PROCESSOR_EXCEPTION,"Subscription Expiration Interval attribute is missing or invalid");
+  } else {
+    core::TimeUnit unit;
+    if (!core::Property::StringToTime(value, subscription_expiration_interval_, unit) ||
+        !core::Property::ConvertTimeUnitToMS(subscription_expiration_interval_, unit, subscription_expiration_interval_)) {
+      throw Exception(PROCESSOR_EXCEPTION,"Subscription Expiration Interval attribute is invalid");
+    }
+  }
+  if (!context->getProperty(HeartbeatInterval.getName(), value)) {
+    throw Exception(PROCESSOR_EXCEPTION,"Heartbeat Interval attribute is missing or invalid");
+  } else {
+    core::TimeUnit unit;
+    if (!core::Property::StringToTime(value, heartbeat_interval_, unit) || !core::Property::ConvertTimeUnitToMS(heartbeat_interval_, unit, heartbeat_interval_)) {
+      throw Exception(PROCESSOR_EXCEPTION,"Heartbeat Interval attribute is invalid");
+    }
+  }
+  if (!context->getProperty(MaxElements.getName(), value)) {
+    throw Exception(PROCESSOR_EXCEPTION,"Max Elements attribute is missing or invalid");
+  } else if (!core::Property::StringToInt(value, max_elements_)) {
+    throw Exception(PROCESSOR_EXCEPTION,"Max Elements attribute is invalid");
+  }
+  if (!context->getProperty(MaxLatency.getName(), value)) {
+    throw Exception(PROCESSOR_EXCEPTION,"Max Latency attribute is missing or invalid");
+  } else {
+    core::TimeUnit unit;
+    if (!core::Property::StringToTime(value, max_latency_, unit) || !core::Property::ConvertTimeUnitToMS(max_latency_, unit, max_latency_)) {
+      throw Exception(PROCESSOR_EXCEPTION,"Max Latency attribute is invalid");
+    }
+  }
+  if (!context->getProperty(ConnectionRetryInterval.getName(), value)) {
+    throw Exception(PROCESSOR_EXCEPTION,"Connection Retry Interval attribute is missing or invalid");
+  } else {
+    core::TimeUnit unit;
+    if (!core::Property::StringToTime(value, connection_retry_interval_, unit) || !core::Property::ConvertTimeUnitToMS(connection_retry_interval_, unit, connection_retry_interval_)) {
+      throw Exception(PROCESSOR_EXCEPTION,"Connection Retry Interval attribute is invalid");
+    }
+  }
+  if (!context->getProperty(ConnectionRetryCount.getName(), value)) {
+    throw Exception(PROCESSOR_EXCEPTION,"Connection Retry Count attribute is missing or invalid");
+  } else if (!core::Property::StringToInt(value, connection_retry_count_)) {
+    throw Exception(PROCESSOR_EXCEPTION,"Connection Retry Count attribute is invalid");
+  }
+
+  FILE* fp = fopen(ssl_ca_file.c_str(), "rb");
+  if (fp == nullptr) {
+    throw Exception(PROCESSOR_EXCEPTION,"Failed to open file specified by SSL Certificate Authority attribute");
+  }
+  X509* ca = nullptr;
+  PEM_read_X509(fp, &ca, nullptr, nullptr);
+  fclose(fp);
+  if (ca == nullptr) {
+    throw Exception(PROCESSOR_EXCEPTION,"Failed to parse file specified by SSL Certificate Authority attribute");
+  }
+  std::array<uint8_t, 20U> hash_buf;
+  int ret = X509_digest(ca, EVP_sha1(), hash_buf.data(), nullptr);
+  X509_free(ca);
+  if (ret != 1) {
+    throw Exception(PROCESSOR_EXCEPTION,"Failed to get fingerprint for CA specified by SSL Certificate Authority attribute");
+  }
+  ssl_ca_cert_thumbprint_ = utils::StringUtils::to_hex(hash_buf.data(), hash_buf.size(), true /*uppercase*/);
+  logger_->log_debug("%s SHA-1 thumbprint is %s", ssl_ca_file.c_str(), ssl_ca_cert_thumbprint_.c_str());
+
+  session_factory_ = sessionFactory;
+
+  // Load state
+  loadState();
+
+  // Start server
+  std::vector<std::string> options;
+  options.emplace_back("enable_keep_alive");
+  options.emplace_back("yes");
+  options.emplace_back("keep_alive_timeout_ms");
+  options.emplace_back("15000");
+  options.emplace_back("num_threads");
+  options.emplace_back("1");
+  options.emplace_back("listening_ports");
+  options.emplace_back(std::to_string(listen_port_) + "s");
+  options.emplace_back("ssl_certificate");
+  options.emplace_back(ssl_certificate_file);
+  options.emplace_back("ssl_ca_file");
+  options.emplace_back(ssl_ca_file);
+  options.emplace_back("ssl_verify_peer");
+  options.emplace_back(verify_peer ? "yes" : "no");
+
+  try {
+    server_ = std::unique_ptr<CivetServer>(new CivetServer(options));
+  } catch (const std::exception& e) {
+    throw Exception(PROCESSOR_EXCEPTION, std::string("Failed to initialize server, error: ") + e.what());
+  } catch (...) {
+    throw Exception(PROCESSOR_EXCEPTION,"Failed to initialize server");
+  }
+  handler_ = std::unique_ptr<Handler>(new Handler(*this));
+  server_->addHandler("**", *handler_);
+}
+
+void SourceInitiatedSubscriptionListener::notifyStop() {
+  server_.release();
+}
+
+} /* namespace processors */
+} /* namespace minifi */
+} /* namespace nifi */
+} /* namespace apache */
+} /* namespace org */
diff --git a/extensions/openwsman/processors/SourceInitiatedSubscriptionListener.h b/extensions/openwsman/processors/SourceInitiatedSubscriptionListener.h
new file mode 100644
index 0000000..70ef004
--- /dev/null
+++ b/extensions/openwsman/processors/SourceInitiatedSubscriptionListener.h
@@ -0,0 +1,173 @@
+/**
+ *
+ * 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 __SOURCE_INITIATED_SUBSCRIPTION_PROCESSOR_H__
+#define __SOURCE_INITIATED_SUBSCRIPTION_PROCESSOR_H__
+
+#include <memory>
+#include <string>
+#include <list>
+#include <map>
+#include <mutex>
+#include <thread>
+
+#include <CivetServer.h>
+extern "C" {
+#include "wsman-xml.h"
+}
+
+#include "utils/ByteArrayCallback.h"
+#include "FlowFileRecord.h"
+#include "core/Processor.h"
+#include "core/ProcessSession.h"
+#include "core/Core.h"
+#include "core/Property.h"
+#include "core/Resource.h"
+#include "controllers/SSLContextService.h"
+#include "core/logging/LoggerConfiguration.h"
+#include "utils/Id.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace processors {
+
+class SourceInitiatedSubscriptionListener : public core::Processor {
+ public:
+  static constexpr char const *INITIAL_EXISTING_EVENTS_STRATEGY_NONE = "None";
+  static constexpr char const *INITIAL_EXISTING_EVENTS_STRATEGY_ALL = "All";
+
+  static constexpr char const* ProcessorName = "SourceInitiatedSubscriptionListener";
+
+  explicit SourceInitiatedSubscriptionListener(std::string name, utils::Identifier uuid = utils::Identifier());
+
+  // Supported Properties
+  static core::Property ListenHostname;
+  static core::Property ListenPort;
+  static core::Property SubscriptionManagerPath;
+  static core::Property SubscriptionsBasePath;
+  static core::Property SSLCertificate;
+  static core::Property SSLCertificateAuthority;
+  static core::Property SSLVerifyPeer;
+  static core::Property XPathXmlQuery;
+  static core::Property InitialExistingEventsStrategy;
+  static core::Property SubscriptionExpirationInterval;
+  static core::Property HeartbeatInterval;
+  static core::Property MaxElements;
+  static core::Property MaxLatency;
+  static core::Property ConnectionRetryInterval;
+  static core::Property ConnectionRetryCount;
+
+  // Supported Relationships
+  static core::Relationship Success;
+
+  // Writes Attributes
+  static constexpr char const* ATTRIBUTE_WEF_REMOTE_MACHINEID = "wef.remote.machineid";
+  static constexpr char const* ATTRIBUTE_WEF_REMOTE_IP = "wef.remote.ip";
+
+  void onTrigger(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) override;
+  void initialize() override;
+  void onSchedule(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSessionFactory> &sessionFactory) override;
+  void notifyStop() override;
+  
+  class Handler: public CivetHandler {
+   public:
+    explicit Handler(SourceInitiatedSubscriptionListener& processor);
+    bool handlePost(CivetServer* server, struct mg_connection* conn);
+    
+    class WriteCallback : public OutputStreamCallback {
+     public:
+      explicit WriteCallback(char* text);
+      int64_t process(std::shared_ptr<io::BaseStream> stream);
+
+     private:
+      char* text_;
+    };
+
+   private:
+    SourceInitiatedSubscriptionListener& processor_;
+    
+    bool handleSubscriptionManager(struct mg_connection* conn, const std::string& endpoint, WsXmlDocH request);
+    bool handleSubscriptions(struct mg_connection* conn, const std::string& endpoint, WsXmlDocH request);
+    
+    static int enumerateEventCallback(WsXmlNodeH node, void* data);
+    std::string getSoapAction(WsXmlDocH doc);
+    std::string getMachineId(WsXmlDocH doc);
+    bool isAckRequested(WsXmlDocH doc);
+    void sendResponse(struct mg_connection* conn, const std::string& machineId, const std::string& remoteIp, char* xml_buf, size_t xml_buf_size);
+    
+    static std::string millisecondsToXsdDuration(int64_t milliseconds);
+  };
+
+ protected:
+  std::shared_ptr<logging::Logger> logger_;
+
+  std::shared_ptr<core::CoreComponentStateManager> state_manager_;
+
+  std::shared_ptr<core::ProcessSessionFactory> session_factory_;
+
+  std::string listen_hostname_;
+  uint16_t listen_port_;
+  std::string subscription_manager_path_;
+  std::string subscriptions_base_path_;
+  std::string ssl_ca_cert_thumbprint_;
+  std::string xpath_xml_query_;
+  std::string initial_existing_events_strategy_;
+  int64_t subscription_expiration_interval_;
+  int64_t heartbeat_interval_;
+  uint32_t max_elements_;
+  int64_t max_latency_;
+  int64_t connection_retry_interval_;
+  uint32_t connection_retry_count_;
+
+  std::unique_ptr<CivetServer> server_;
+  std::unique_ptr<Handler> handler_;
+  
+  struct SubscriberData {
+      WsXmlDocH bookmark_;
+      std::string subscription_version_;
+      WsXmlDocH subscription_;
+      std::string subscription_endpoint_;
+      std::string subscription_identifier_;
+
+      SubscriberData();
+      ~SubscriberData();
+
+      void setSubscription(const std::string& subscription_version, WsXmlDocH subscription, const std::string& subscription_endpoint, const std::string& subscription_identifier);
+      void clearSubscription();
+      void setBookmark(WsXmlDocH bookmark);
+      void clearBookmark();
+  };
+
+  std::mutex mutex_;
+  std::map<std::string /*machineId*/, SubscriberData> subscribers_;
+
+  bool persistState() const;
+  bool loadState();
+};
+
+REGISTER_RESOURCE(SourceInitiatedSubscriptionListener, "This processor implements a Windows Event Forwarding Source Initiated Subscription server with the help of OpenWSMAN. "
+                                                       "Windows hosts can be set up to connect and forward Event Logs to this processor.")
+
+} /* namespace processors */
+} /* namespace minifi */
+} /* namespace nifi */
+} /* namespace apache */
+} /* namespace org */
+
+#endif // __SOURCE_INITIATED_SUBSCRIPTION_PROCESSOR_H__
diff --git a/extensions/sftp/processors/ListSFTP.cpp b/extensions/sftp/processors/ListSFTP.cpp
index 6ac89f2..dd49beb 100644
--- a/extensions/sftp/processors/ListSFTP.cpp
+++ b/extensions/sftp/processors/ListSFTP.cpp
@@ -794,8 +794,7 @@
     return false;
   }
 
-  size_t i = 0;
-  while (true) {
+  for (size_t i = 0U;; i++) {
     std::string name;
     try {
       name = state_map.at("entity." + std::to_string(i) + ".name");
@@ -812,7 +811,6 @@
       logger_->log_error("State for entity \"%s\" is missing or invalid, skipping", name);
       continue;
     }
-    ++i;
   }
 
   if (state_listing_strategy != listing_strategy_ ||
diff --git a/extensions/standard-processors/processors/TailFile.cpp b/extensions/standard-processors/processors/TailFile.cpp
index 0a52a93..84fc92a 100644
--- a/extensions/standard-processors/processors/TailFile.cpp
+++ b/extensions/standard-processors/processors/TailFile.cpp
@@ -246,8 +246,7 @@
   std::unordered_map<std::string, std::string> state_map;
   if (state_manager_->get(state_map)) {
     std::map<std::string, TailState> new_tail_states;
-    size_t i = 0;
-    while (true) {
+    for (size_t i = 0U;; i++) {
       std::string name;
       try {
         name = state_map.at("file." + std::to_string(i) + ".name");
@@ -268,7 +267,6 @@
       } catch (...) {
         continue;
       }
-      ++i;
     }
     state_load_success = true;
     tail_states_ = std::move(new_tail_states);
diff --git a/thirdparty/libxml2/libxml2-win.patch b/thirdparty/libxml2/libxml2-win.patch
new file mode 100644
index 0000000..67162bc
--- /dev/null
+++ b/thirdparty/libxml2/libxml2-win.patch
@@ -0,0 +1,126 @@
+diff -rupN orig/CMakeLists.txt patched/CMakeLists.txt
+--- orig/CMakeLists.txt	1970-01-01 01:00:00.000000000 +0100
++++ patched/CMakeLists.txt	2020-04-17 14:16:13.000000000 +0200
+@@ -0,0 +1,65 @@
++cmake_minimum_required(VERSION 3.7)
++
++project(libxml2)
++
++set(SOURCES buf.c
++            c14n.c
++            catalog.c
++            chvalid.c
++            debugXML.c
++            dict.c
++            DOCBparser.c
++            encoding.c
++            entities.c
++            error.c
++            globals.c
++            hash.c
++            HTMLparser.c
++            HTMLtree.c
++            legacy.c
++            list.c
++            nanoftp.c
++            nanohttp.c
++            parser.c
++            parserInternals.c
++            pattern.c
++            relaxng.c
++            SAX.c
++            SAX2.c
++            schematron.c
++            threads.c
++            tree.c
++            uri.c
++            valid.c
++            xinclude.c
++            xlink.c
++            xmlcatalog.c
++            xmlIO.c
++            xmlmemory.c
++            xmlmodule.c
++            xmlreader.c
++            xmlregexp.c
++            xmlsave.c
++            xmlschemas.c
++            xmlschemastypes.c
++            xmlstring.c
++            xmlunicode.c
++            xmlwriter.c
++            xpath.c
++            xpointer.c)
++
++add_library(xml2 STATIC ${SOURCES})
++
++set_property(TARGET xml2 PROPERTY POSITION_INDEPENDENT_CODE ON)
++
++target_include_directories(xml2
++                            PRIVATE
++                              include
++                              win32/VC10)
++
++install(TARGETS xml2
++    ARCHIVE DESTINATION lib
++)
++
++install(DIRECTORY include/libxml DESTINATION include/libxml2
++          FILES_MATCHING PATTERN "*.h")
+diff -rupN orig/include/libxml/xmlversion.h patched/include/libxml/xmlversion.h
+--- orig/include/libxml/xmlversion.h	2019-10-30 20:14:29.000000000 +0100
++++ patched/include/libxml/xmlversion.h	2020-04-17 14:52:58.000000000 +0200
+@@ -50,7 +50,7 @@ XMLPUBFUN void XMLCALL xmlCheckVersion(i
+  *
+  * extra version information, used to show a CVS compilation
+  */
+-#define LIBXML_VERSION_EXTRA "-GITv2.9.10-rc1-2-ga5bb6aaa2"
++#define LIBXML_VERSION_EXTRA ""
+ 
+ /**
+  * LIBXML_TEST_VERSION:
+@@ -171,7 +171,7 @@ XMLPUBFUN void XMLCALL xmlCheckVersion(i
+  *
+  * Whether the FTP support is configured in
+  */
+-#if 1
++#if 0
+ #define LIBXML_FTP_ENABLED
+ #endif
+ 
+@@ -180,7 +180,7 @@ XMLPUBFUN void XMLCALL xmlCheckVersion(i
+  *
+  * Whether the HTTP support is configured in
+  */
+-#if 1
++#if 0
+ #define LIBXML_HTTP_ENABLED
+ #endif
+ 
+@@ -270,7 +270,7 @@ XMLPUBFUN void XMLCALL xmlCheckVersion(i
+  *
+  * Whether iconv support is available
+  */
+-#if 1
++#if 0
+ #define LIBXML_ICONV_ENABLED
+ #endif
+ 
+@@ -395,7 +395,7 @@ XMLPUBFUN void XMLCALL xmlCheckVersion(i
+  *
+  * Whether the Zlib support is compiled in
+  */
+-#if 1
++#if 0
+ #define LIBXML_ZLIB_ENABLED
+ #endif
+ 
+@@ -404,7 +404,7 @@ XMLPUBFUN void XMLCALL xmlCheckVersion(i
+  *
+  * Whether the Lzma support is compiled in
+  */
+-#if 1
++#if 0
+ #define LIBXML_LZMA_ENABLED
+ #endif
+ 
diff --git a/thirdparty/openwsman/openwsman.patch b/thirdparty/openwsman/openwsman.patch
new file mode 100644
index 0000000..5fff4b7
--- /dev/null
+++ b/thirdparty/openwsman/openwsman.patch
@@ -0,0 +1,81 @@
+diff -rupN orig/CMakeLists.txt patched/CMakeLists.txt
+--- orig/CMakeLists.txt	2019-09-17 11:38:38.000000000 +0200
++++ patched/CMakeLists.txt	2020-04-16 23:43:22.000000000 +0200
+@@ -24,7 +24,7 @@ if(COMMAND cmake_policy)
+ endif(COMMAND cmake_policy)
+ 	      
+ # where to look first for cmake modules, before ${CMAKE_ROOT}/Modules/ is checked
+-SET(CMAKE_MODULE_PATH ${CMAKE_SOURCE_DIR}/cmake/modules ${CMAKE_MODULE_PATH})
++LIST(APPEND CMAKE_MODULE_PATH ${CMAKE_SOURCE_DIR}/cmake/modules)
+ 
+ INCLUDE( ${CMAKE_SOURCE_DIR}/VERSION.cmake )
+ SET(VERSION "${OPENWSMAN_MAJOR}.${OPENWSMAN_MINOR}.${OPENWSMAN_PATCH}")
+@@ -168,8 +168,14 @@ ENDIF( USE_PAM )
+ 
+ INCLUDE(FindOpenSSL)
+ IF(OPENSSL_FOUND)
++  MESSAGE("OpenSSL found")
++  MESSAGE("OPENSSL_INCLUDE_DIR: ${OPENSSL_INCLUDE_DIR}")
++  MESSAGE("OPENSSL_LIBRARIES: ${OPENSSL_LIBRARIES}")
+   SET(HAVE_SSL 1)
+   SET(USE_OPENSSL 1)
++  INCLUDE_DIRECTORIES(${OPENSSL_INCLUDE_DIR})
++ELSE(OPENSSL_FOUND)
++  SET(HAVE_SSL 0)
+ ENDIF(OPENSSL_FOUND)
+ 
+ IF( BUILD_RUBY )
+@@ -256,6 +262,9 @@ IF(UNIX)
+   IF ( NOT CURL_FOUND)
+    MESSAGE( FATAL_ERROR " curl not found" )
+   ELSE ( NOT CURL_FOUND)
++    MESSAGE("cURL found")
++    MESSAGE("CURL_INCLUDE_DIR: ${CURL_INCLUDE_DIR}")
++    MESSAGE("CURL_LIBRARIES: ${CURL_LIBRARIES}")
+     INCLUDE_DIRECTORIES(${CURL_INCLUDE_DIR})
+     IF(CURL_VERSION_STRING)
+       STRING(COMPARE LESS ${CURL_VERSION_STRING} "7.12.0" result)
+@@ -272,6 +281,9 @@ INCLUDE(FindLibXml2)
+ IF ( NOT LIBXML2_FOUND)
+   MESSAGE( FATAL_ERROR " libxml2 not found" )
+ ELSE ( NOT LIBXML2_FOUND)
++  MESSAGE("Libxml2 found")
++  MESSAGE("LIBXML2_INCLUDE_DIR: ${LIBXML2_INCLUDE_DIR}")
++  MESSAGE("LIBXML2_LIBRARIES: ${LIBXML2_LIBRARIES}")
+   INCLUDE_DIRECTORIES(${LIBXML2_INCLUDE_DIR})
+ ENDIF( NOT LIBXML2_FOUND)
+ 
+diff -rupN orig/src/lib/wsman-soap.c patched/src/lib/wsman-soap.c
+--- orig/src/lib/wsman-soap.c	2019-09-17 11:38:38.000000000 +0200
++++ patched/src/lib/wsman-soap.c	2020-04-16 23:21:49.000000000 +0200
+@@ -991,10 +991,12 @@ unsigned long get_total_enum_context(WsC
+  * preset, hence marking them as weak symbols and testing to see
+  * if they are resolved before using them.
+  */
++#if 0
+ #pragma weak wsmand_options_get_max_threads
+ extern int wsmand_options_get_max_threads(void);
+ #pragma weak wsmand_options_get_max_connections_per_thread
+ extern int wsmand_options_get_max_connections_per_thread(void);
++#endif
+ 
+ /**
+  * Enumeration Stub for processing enumeration requests
+@@ -1030,9 +1032,17 @@ wsenum_enumerate_stub(SoapOpH op,
+         int max_threads = 0;
+         int max_connections_per_thread = 0;
+         int(* fptr)(void);
++#if 0
+         if((fptr = wsmand_options_get_max_threads) != 0){
++#else
++        if(0){
++#endif
+                 max_threads = (* fptr)();
++#if 0
+                 if((fptr = wsmand_options_get_max_connections_per_thread) != 0){
++#else
++                if(0){
++#endif
+                         max_connections_per_thread = (* fptr)();
+                 }
+                 else{