Repository: nifi-minifi-cpp Updated Branches: refs/heads/master 573c511f7 -> 071ac1e98
MINIFI-239: Add InvokeHTTP Processor and corresponding tests This closes #80. Signed-off-by: Aldrin Piri <[email protected]> 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/071ac1e9 Tree: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/tree/071ac1e9 Diff: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/diff/071ac1e9 Branch: refs/heads/master Commit: 071ac1e9884d0335f33098cfcf313f396cdd6276 Parents: 573c511 Author: Marc Parisi <[email protected]> Authored: Fri Apr 21 19:55:56 2017 -0400 Committer: Aldrin Piri <[email protected]> Committed: Fri Apr 28 11:06:23 2017 -0400 ---------------------------------------------------------------------- CMakeLists.txt | 55 ++- README.md | 5 + libminifi/CMakeLists.txt | 6 + libminifi/include/core/FlowConfiguration.h | 2 +- libminifi/include/processors/InvokeHTTP.h | 228 +++++++++ libminifi/include/utils/ByteInputCallBack.h | 74 +++ libminifi/src/core/FlowConfiguration.cpp | 4 + libminifi/src/processors/InvokeHTTP.cpp | 537 +++++++++++++++++++++ libminifi/test/HttpGetIntegrationTest.cpp | 118 +++++ libminifi/test/HttpPostIntegrationTest.cpp | 117 +++++ libminifi/test/unit/InvokeHTTPTests.cpp | 471 ++++++++++++++++++ libminifi/test/unit/ProvenanceTestHelper.h | 67 +++ libminifi/test/unit/resource/TestHTTPGet.yml | 73 +++ libminifi/test/unit/resource/TestHTTPPost.yml | 87 ++++ main/MiNiFiMain.cpp | 6 + 15 files changed, 1848 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/071ac1e9/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/CMakeLists.txt b/CMakeLists.txt index 519dae9..42c86d7 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -72,6 +72,7 @@ list(APPEND CMAKE_MODULE_PATH "${CMAKE_CURRENT_SOURCE_DIR}/cmake") # Include LevelDB find_package (Leveldb REQUIRED) +find_package(CURL REQUIRED) if (LEVELDB_FOUND) include_directories(${LEVELDB_INCLUDE_DIRS}) else () @@ -137,8 +138,56 @@ enable_testing(test) target_include_directories(tests PRIVATE BEFORE "libminifi/include/processors") target_include_directories(tests PRIVATE BEFORE "libminifi/include/provenance") target_link_libraries(tests ${CMAKE_THREAD_LIBS_INIT} ${UUID_LIBRARIES} ${LEVELDB_LIBRARIES} ${OPENSSL_LIBRARIES} minifi yaml-cpp c-library civetweb-cpp jsoncpp_lib_static) + if (CURL_FOUND) + target_include_directories(tests PRIVATE BEFORE ${CURL_INCLUDE_DIRS}) + target_link_libraries(tests ${CURL_LIBRARIES}) + endif(CURL_FOUND) add_test(NAME LibMinifiTests COMMAND tests) + + file(GLOB LIBMINIFI_TEST_CS "libminifi/test/HttpGetIntegrationTest.cpp") + add_executable(testHttpGet ${LIBMINIFI_TEST_CS} ${SPD_SOURCES}) + target_include_directories(testHttpGet PRIVATE BEFORE "thirdparty/catch") + target_include_directories(testHttpGet PRIVATE BEFORE "thirdparty/yaml-cpp-yaml-cpp-0.5.3/include") + target_include_directories(testHttpGet PRIVATE BEFORE ${LEVELDB_INCLUDE_DIRS}) + target_include_directories(testHttpGet PRIVATE BEFORE "include") + target_include_directories(testHttpGet PRIVATE BEFORE "libminifi/include/") + target_include_directories(testHttpGet PRIVATE BEFORE "libminifi/include/core") + target_include_directories(testHttpGet PRIVATE BEFORE "libminifi/include/core/controller") + target_include_directories(testHttpGet PRIVATE BEFORE "libminifi/include/core/repository") + target_include_directories(testHttpGet PRIVATE BEFORE "libminifi/include/io") + target_include_directories(testHttpGet PRIVATE BEFORE "libminifi/include/utils") + target_include_directories(testHttpGet PRIVATE BEFORE "libminifi/include/processors") + target_include_directories(testHttpGet PRIVATE BEFORE "libminifi/include/provenance") + target_link_libraries(testHttpGet ${CMAKE_THREAD_LIBS_INIT} ${UUID_LIBRARIES} ${LEVELDB_LIBRARIES} ${OPENSSL_LIBRARIES} minifi yaml-cpp c-library civetweb-cpp jsoncpp_lib_static) + if (CURL_FOUND) + target_include_directories(testHttpGet PRIVATE BEFORE ${CURL_INCLUDE_DIRS}) + target_link_libraries(testHttpGet ${CURL_LIBRARIES}) + endif(CURL_FOUND) + add_test(NAME testHttpGet COMMAND testHttpGet "${CMAKE_SOURCE_DIR}/libminifi/test/unit/resource/TestHTTPGet.yml") + + file(GLOB LIBMINIFI_TEST_CS "libminifi/test/HttpPostIntegrationTest.cpp") + add_executable(testHttpPost ${LIBMINIFI_TEST_CS} ${SPD_SOURCES}) + target_include_directories(testHttpPost PRIVATE BEFORE "thirdparty/catch") + target_include_directories(testHttpPost PRIVATE BEFORE "thirdparty/yaml-cpp-yaml-cpp-0.5.3/include") + target_include_directories(testHttpPost PRIVATE BEFORE ${LEVELDB_INCLUDE_DIRS}) + target_include_directories(testHttpPost PRIVATE BEFORE "include") + target_include_directories(testHttpPost PRIVATE BEFORE "libminifi/include/") + target_include_directories(testHttpPost PRIVATE BEFORE "libminifi/include/core") + target_include_directories(testHttpPost PRIVATE BEFORE "libminifi/include/core/controller") + target_include_directories(testHttpPost PRIVATE BEFORE "libminifi/include/core/repository") + target_include_directories(testHttpPost PRIVATE BEFORE "libminifi/include/io") + target_include_directories(testHttpPost PRIVATE BEFORE "libminifi/include/utils") + target_include_directories(testHttpPost PRIVATE BEFORE "libminifi/include/processors") + target_include_directories(testHttpPost PRIVATE BEFORE "libminifi/include/provenance") + target_link_libraries(testHttpPost ${CMAKE_THREAD_LIBS_INIT} ${UUID_LIBRARIES} ${LEVELDB_LIBRARIES} ${OPENSSL_LIBRARIES} minifi yaml-cpp c-library civetweb-cpp jsoncpp_lib_static) + if (CURL_FOUND) + target_include_directories(testHttpPost PRIVATE BEFORE ${CURL_INCLUDE_DIRS}) + target_link_libraries(testHttpPost ${CURL_LIBRARIES}) + endif(CURL_FOUND) + add_test(NAME testHttpPost COMMAND testHttpPost "${CMAKE_SOURCE_DIR}/libminifi/test/unit/resource/TestHTTPPost.yml") + + file(GLOB LIBMINIFI_TEST_EXECUTE_PROCESS "libminifi/test/TestExecuteProcess.cpp") add_executable(testExecuteProcess ${LIBMINIFI_TEST_EXECUTE_PROCESS} ${SPD_SOURCES}) target_include_directories(testExecuteProcess PRIVATE BEFORE "thirdparty/yaml-cpp-yaml-cpp-0.5.3/include") @@ -151,7 +200,11 @@ enable_testing(test) target_include_directories(testExecuteProcess PRIVATE BEFORE "libminifi/include/utils") target_include_directories(testExecuteProcess PRIVATE BEFORE "libminifi/include/processors") target_include_directories(testExecuteProcess PRIVATE BEFORE "libminifi/include/provenance") - target_link_libraries(testExecuteProcess ${CMAKE_THREAD_LIBS_INIT} ${UUID_LIBRARIES} ${LEVELDB_LIBRARIES} ${OPENSSL_LIBRARIES} minifi yaml-cpp c-library civetweb-cpp) + if (CURL_FOUND) + target_include_directories(testExecuteProcess PRIVATE BEFORE ${CURL_INCLUDE_DIRS}) + target_link_libraries(testExecuteProcess ${CURL_LIBRARIES}) + endif(CURL_FOUND) + target_link_libraries(testExecuteProcess ${CMAKE_THREAD_LIBS_INIT} ${UUID_LIBRARIES} ${LEVELDB_LIBRARIES} ${OPENSSL_LIBRARIES} minifi yaml-cpp c-library civetweb-cpp jsoncpp_lib_static) add_test(NAME ExecuteProcess COMMAND testExecuteProcess) # Create a custom build target called "docker" that will invoke DockerBuild.sh and create the NiFi-MiNiFi-CPP Docker image http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/071ac1e9/README.md ---------------------------------------------------------------------- diff --git a/README.md b/README.md index be2a400..2a83a9a 100644 --- a/README.md +++ b/README.md @@ -70,6 +70,7 @@ Perspectives of the role of MiNiFi should be from the perspective of the agent a #### Libraries / Development Headers * libboost and boost-devel * 1.48.0 or greater +* libCurl * libleveldb and libleveldb-devel * libuuid and uuid-dev * openssl @@ -79,6 +80,7 @@ Perspectives of the role of MiNiFi should be from the perspective of the agent a #### Libraries * libuuid * libleveldb +* libcurl * libssl and libcrypto from openssl The needed dependencies can be installed with the following commands for: @@ -88,6 +90,7 @@ Yum based Linux Distributions # ~/Development/code/apache/nifi-minifi-cpp on git:master $ yum install cmake \ gcc gcc-c++ \ + libcurl-devel \ leveldb-devel leveldb \ libuuid libuuid-devel \ boost-devel \ libssl-dev @@ -98,6 +101,7 @@ Aptitude based Linux Distributions # ~/Development/code/apache/nifi-minifi-cpp on git:master $ apt-get install cmake \ gcc g++ \ + libcurl-dev \ libleveldb-dev libleveldb1v5 \ uuid-dev uuid \ libboost-all-dev libssl-dev @@ -107,6 +111,7 @@ OS X Using Homebrew (with XCode Command Line Tools installed) ``` # ~/Development/code/apache/nifi-minifi-cpp on git:master $ brew install cmake \ + curl \ leveldb \ ossp-uuid \ boost \ openssl http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/071ac1e9/libminifi/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/libminifi/CMakeLists.txt b/libminifi/CMakeLists.txt index 5de0a87..320fc45 100644 --- a/libminifi/CMakeLists.txt +++ b/libminifi/CMakeLists.txt @@ -76,8 +76,14 @@ target_link_libraries (minifi ${ZLIB_LIBRARIES}) if (NOT IOS) # Include Boost System find_package(Boost COMPONENTS system REQUIRED) +find_package(CURL) target_link_libraries(minifi ${Boost_SYSTEM_LIBRARY}) +if (CURL_FOUND) + include_directories(${CURL_INCLUDE_DIRS}) + target_link_libraries (minifi ${CURL_LIBRARIES}) +endif(CURL_FOUND) + # Include LevelDB find_package (Leveldb REQUIRED) if (LEVELDB_FOUND) http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/071ac1e9/libminifi/include/core/FlowConfiguration.h ---------------------------------------------------------------------- diff --git a/libminifi/include/core/FlowConfiguration.h b/libminifi/include/core/FlowConfiguration.h index c8cb7eb..79e400d 100644 --- a/libminifi/include/core/FlowConfiguration.h +++ b/libminifi/include/core/FlowConfiguration.h @@ -26,9 +26,9 @@ #include "processors/GetFile.h" #include "processors/PutFile.h" #include "processors/TailFile.h" -#include "processors/ListenHTTP.h" #include "processors/ListenSyslog.h" #include "processors/GenerateFlowFile.h" +#include "processors/InvokeHTTP.h" #include "processors/ListenHTTP.h" #include "processors/LogAttribute.h" #include "processors/ExecuteProcess.h" http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/071ac1e9/libminifi/include/processors/InvokeHTTP.h ---------------------------------------------------------------------- diff --git a/libminifi/include/processors/InvokeHTTP.h b/libminifi/include/processors/InvokeHTTP.h new file mode 100644 index 0000000..789b3b5 --- /dev/null +++ b/libminifi/include/processors/InvokeHTTP.h @@ -0,0 +1,228 @@ +/** + * InvokeHTTP 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 __INVOKE_HTTP_H__ +#define __INVOKE_HTTP_H__ + +#include <memory> +#include <regex> + +#include <curl/curl.h> +#include "FlowFileRecord.h" +#include "core/Processor.h" +#include "core/ProcessSession.h" +#include "core/Core.h" +#include "core/Property.h" +#include "utils/ByteInputCallBack.h" + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace processors { + +struct CallBackPosition { + utils::ByteInputCallBack *ptr; + size_t pos; +}; + +/** + * HTTP Response object + */ +struct HTTPRequestResponse { + std::vector<char> data; + + /** + * Receive HTTP Response. + */ + static size_t recieve_write(char * data, size_t size, size_t nmemb, + void * p) { + return static_cast<HTTPRequestResponse*>(p)->write_content(data, size, + nmemb); + } + + /** + * Callback for post, put, and patch operations + * @param buffer + * @param size size of buffer + * @param nitems items to add + * @param insteam input stream object. + */ + + static size_t send_write(char * data, size_t size, size_t nmemb, void * p) { + if (p != 0) { + CallBackPosition *callback = (CallBackPosition*) p; + if (callback->pos <= callback->ptr->getBufferSize()) { + char *ptr = callback->ptr->getBuffer(); + int len = callback->ptr->getBufferSize() - callback->pos; + if (len <= 0) { + delete callback->ptr; + delete callback; + return 0; + } + if (len > size * nmemb) + len = size * nmemb; + memcpy(data, callback->ptr->getBuffer() + callback->pos, len); + callback->pos += len; + return len; + } + } else { + return CURL_READFUNC_ABORT; + } + + return 0; + } + + size_t write_content(char* ptr, size_t size, size_t nmemb) { + data.insert(data.end(), ptr, ptr + size * nmemb); + return size * nmemb; + } + +}; + +// InvokeHTTP Class +class InvokeHTTP : public core::Processor { + public: + + // Constructor + /*! + * Create a new processor + */ + InvokeHTTP(std::string name, uuid_t uuid = NULL) + : Processor(name, uuid), + date_header_include_(true), + connect_timeout_(20000), + penalize_no_retry_(false), + read_timeout_(20000), + always_output_response_(false) { + curl_global_init(CURL_GLOBAL_DEFAULT); + } + // Destructor + virtual ~InvokeHTTP(); + // Processor Name + static const char *ProcessorName; + // Supported Properties + static core::Property Method; + static core::Property URL; + static core::Property ConnectTimeout; + static core::Property ReadTimeout; + static core::Property DateHeader; + static core::Property FollowRedirects; + static core::Property AttributesToSend; + static core::Property SSLContext; + static core::Property ProxyHost; + static core::Property ProxyPort; + static core::Property ProxyUser; + static core::Property ProxyPassword; + static core::Property ContentType; + static core::Property SendBody; + + static core::Property PropPutOutputAttributes; + + static core::Property AlwaysOutputResponse; + + static core::Property PenalizeOnNoRetry; + + static const char* STATUS_CODE; + static const char* STATUS_MESSAGE; + static const char* RESPONSE_BODY; + static const char* REQUEST_URL; + static const char* TRANSACTION_ID; + static const char* REMOTE_DN; + static const char* EXCEPTION_CLASS; + static const char* EXCEPTION_MESSAGE; + // Supported Relationships + static core::Relationship Success; + static core::Relationship RelResponse; + static core::Relationship RelRetry; + static core::Relationship RelNoRetry; + static core::Relationship RelFailure; + + void onTrigger(core::ProcessContext *context, core::ProcessSession *session); + void initialize(); + void onSchedule(core::ProcessContext *context, + core::ProcessSessionFactory *sessionFactory); + + protected: + + /** + * Generate a transaction ID + * @return transaction ID string. + */ + std::string generateId(); + /** + * Set the request method on the curl struct. + * @param curl pointer to this instance. + * @param string request method + */ + void set_request_method(CURL *curl, const std::string &); + + struct curl_slist *build_header_list( + CURL *curl, std::string regex, + const std::map<std::string, std::string> &); + + bool matches(const std::string &value, const std::string &sregex); + + /** + * Routes the flowfile to the proper destination + * @param request request flow file record + * @param response response flow file record + * @param session process session + * @param context process context + * @param isSuccess success code or not + * @param statuscode http response code. + */ + void route(std::shared_ptr<FlowFileRecord> &request, + std::shared_ptr<FlowFileRecord> &response, + core::ProcessSession *session, core::ProcessContext *context, + bool isSuccess, int statusCode); + /** + * Determine if we should emit a new flowfile based on our activity + * @param method method type + * @return result of the evaluation. + */ + bool emitFlowFile(const std::string &method); + CURLcode res; + + // http method + std::string method_; + // url + std::string url_; + // include date in the header + bool date_header_include_; + // attribute to send regex + std::string attribute_to_send_regex_; + // connection timeout + int64_t connect_timeout_; + // read timeout. + int64_t read_timeout_; + // attribute in which response body will be added + std::string put_attribute_name_; + // determine if we always output a response. + bool always_output_response_; + // penalize on no retry + bool penalize_no_retry_; +}; + +} /* namespace processors */ +} /* namespace minifi */ +} /* namespace nifi */ +} /* namespace apache */ +} /* namespace org */ + +#endif http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/071ac1e9/libminifi/include/utils/ByteInputCallBack.h ---------------------------------------------------------------------- diff --git a/libminifi/include/utils/ByteInputCallBack.h b/libminifi/include/utils/ByteInputCallBack.h new file mode 100644 index 0000000..72303ff --- /dev/null +++ b/libminifi/include/utils/ByteInputCallBack.h @@ -0,0 +1,74 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#ifndef LIBMINIFI_INCLUDE_UTILS_BYTEINPUTCALLBACK_H_ +#define LIBMINIFI_INCLUDE_UTILS_BYTEINPUTCALLBACK_H_ + +#include <fstream> +#include <iterator> +#include "FlowFileRecord.h" + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace utils { + +/** + * General vector based uint8_t callback. + */ +class ByteInputCallBack : public InputStreamCallback { + public: + ByteInputCallBack() { + } + + virtual ~ByteInputCallBack() { + + } + + virtual void process(std::ifstream *stream) { + + std::vector<char> nv = std::vector<char>(std::istreambuf_iterator<char>(*stream), + std::istreambuf_iterator<char>()); + vec = std::move(nv); + + ptr = &vec[0]; + + } + + char *getBuffer() { + return ptr; + } + + + + + const size_t getBufferSize() { + return vec.size(); + } + + private: + char *ptr; + std::vector<char> vec; +}; + +} /* namespace utils */ +} /* namespace minifi */ +} /* namespace nifi */ +} /* namespace apache */ +} /* namespace org */ + +#endif /* LIBMINIFI_INCLUDE_UTILS_BYTEINPUTCALLBACK_H_ */ http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/071ac1e9/libminifi/src/core/FlowConfiguration.cpp ---------------------------------------------------------------------- diff --git a/libminifi/src/core/FlowConfiguration.cpp b/libminifi/src/core/FlowConfiguration.cpp index d2df002..f2b6f8b 100644 --- a/libminifi/src/core/FlowConfiguration.cpp +++ b/libminifi/src/core/FlowConfiguration.cpp @@ -64,6 +64,10 @@ std::shared_ptr<core::Processor> FlowConfiguration::createProcessor( processor = std::make_shared< org::apache::nifi::minifi::processors::ListenHTTP>(name, uuid); } else if (name + == org::apache::nifi::minifi::processors::InvokeHTTP::ProcessorName) { + processor = std::make_shared< + org::apache::nifi::minifi::processors::InvokeHTTP>(name, uuid); + } else if (name == org::apache::nifi::minifi::processors::ExecuteProcess::ProcessorName) { processor = std::make_shared< org::apache::nifi::minifi::processors::ExecuteProcess>(name, uuid); http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/071ac1e9/libminifi/src/processors/InvokeHTTP.cpp ---------------------------------------------------------------------- diff --git a/libminifi/src/processors/InvokeHTTP.cpp b/libminifi/src/processors/InvokeHTTP.cpp new file mode 100644 index 0000000..5a76751 --- /dev/null +++ b/libminifi/src/processors/InvokeHTTP.cpp @@ -0,0 +1,537 @@ +/** + * + * 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 "processors/InvokeHTTP.h" + +#include <curl/curlbuild.h> +#include <curl/easy.h> +#include <sys/_types/_size_t.h> +#include <sys/_types/_uuid_t.h> +#include <uuid/uuid.h> +#include <memory> +#include <algorithm> +#include <cctype> +#include <cstdint> +#include <cstring> +#include <iostream> +#include <iterator> +#include <map> +#include <set> +#include <string> +#include <utility> +#include <vector> + +#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" + +#if (__GNUC__ >= 4) +#if (__GNUC_MINOR__ < 9) +#include <regex.h> +#endif +#endif + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace processors { + +const char *InvokeHTTP::ProcessorName = "InvokeHTTP"; + +core::Property InvokeHTTP::Method( + "HTTP Method", + "HTTP request method (GET, POST, PUT, PATCH, DELETE, HEAD, OPTIONS). " + "Arbitrary methods are also supported. Methods other than POST, PUT and PATCH will be sent without a message body.", + "GET"); +core::Property InvokeHTTP::URL( + "Remote URL", + "Remote URL which will be connected to, including scheme, host, port, path.", + ""); +core::Property InvokeHTTP::ConnectTimeout( + "Connection Timeout", "Max wait time for connection to remote service.", + "5 secs"); +core::Property InvokeHTTP::ReadTimeout( + "Read Timeout", "Max wait time for response from remote service.", + "15 secs"); +core::Property InvokeHTTP::DateHeader( + "Include Date Header", "Include an RFC-2616 Date header in the request.", + "True"); +core::Property InvokeHTTP::FollowRedirects( + "Follow Redirects", "Follow HTTP redirects issued by remote server.", + "True"); +core::Property InvokeHTTP::AttributesToSend( + "Attributes to Send", + "Regular expression that defines which attributes to send as HTTP" + " headers in the request. If not defined, no attributes are sent as headers.", + ""); +core::Property InvokeHTTP::SSLContext( + "SSL Context Service", + "The SSL Context Service used to provide client certificate information for TLS/SSL (https) connections.", + ""); +core::Property InvokeHTTP::ProxyHost( + "Proxy Host", + "The fully qualified hostname or IP address of the proxy server", ""); +core::Property InvokeHTTP::ProxyPort("Proxy Port", + "The port of the proxy server", ""); +core::Property InvokeHTTP::ProxyUser( + "invokehttp-proxy-user", + "Username to set when authenticating against proxy", ""); +core::Property InvokeHTTP::ProxyPassword( + "invokehttp-proxy-password", + "Password to set when authenticating against proxy", ""); +core::Property InvokeHTTP::ContentType( + "Content-type", + "The Content-Type to specify for when content is being transmitted through a PUT, " + "POST or PATCH. In the case of an empty value after evaluating an expression language expression, " + "Content-Type defaults to", + "application/octet-stream"); +core::Property InvokeHTTP::SendBody( + "send-message-body", + "If true, sends the HTTP message body on POST/PUT/PATCH requests (default). " + "If false, suppresses the message body and content-type header for these requests.", + "true"); + +core::Property InvokeHTTP::PropPutOutputAttributes( + "Put Response Body in Attribute", + "If set, the response body received back will be put into an attribute of the original " + "FlowFile instead of a separate FlowFile. The attribute key to put to is determined by evaluating value of this property. ", + ""); +core::Property InvokeHTTP::AlwaysOutputResponse( + "Always Output Response", + "Will force a response FlowFile to be generated and routed to the 'Response' relationship " + "regardless of what the server status code received is ", + "false"); +core::Property InvokeHTTP::PenalizeOnNoRetry( + "Penalize on \"No Retry\"", + "Enabling this property will penalize FlowFiles that are routed to the \"No Retry\" relationship.", + "false"); + +const char* InvokeHTTP::STATUS_CODE = "invokehttp.status.code"; +const char* InvokeHTTP::STATUS_MESSAGE = "invokehttp.status.message"; +const char* InvokeHTTP::RESPONSE_BODY = "invokehttp.response.body"; +const char* InvokeHTTP::REQUEST_URL = "invokehttp.request.url"; +const char* InvokeHTTP::TRANSACTION_ID = "invokehttp.tx.id"; +const char* InvokeHTTP::REMOTE_DN = "invokehttp.remote.dn"; +const char* InvokeHTTP::EXCEPTION_CLASS = "invokehttp.java.exception.class"; +const char* InvokeHTTP::EXCEPTION_MESSAGE = "invokehttp.java.exception.message"; + +core::Relationship InvokeHTTP::Success("success", + "All files are routed to success"); + +core::Relationship InvokeHTTP::RelResponse("response", + "Represents a response flowfile"); + +core::Relationship InvokeHTTP::RelRetry( + "retry", + "The original FlowFile will be routed on any status code that can be retried " + "(5xx status codes). It will have new attributes detailing the request."); + +core::Relationship InvokeHTTP::RelNoRetry( + "no retry", + "The original FlowFile will be routed on any status code that should NOT " + "be retried (1xx, 3xx, 4xx status codes). It will have new attributes detailing the request."); + +core::Relationship InvokeHTTP::RelFailure( + "failure", + "The original FlowFile will be routed on any type of connection failure, " + "timeout or general exception. It will have new attributes detailing the request."); + +void InvokeHTTP::set_request_method(CURL *curl, const std::string &method) { + std::string my_method; + std::transform(method.begin(), method.end(), my_method.begin(), ::toupper); + if (my_method == "POST") { + curl_easy_setopt(curl, CURLOPT_POST, 1); + } else if (my_method == "PUT") { + curl_easy_setopt(curl, CURLOPT_UPLOAD, 1); + } else if (my_method == "GET") { + } else { + curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, my_method.c_str()); + } +} + +void InvokeHTTP::initialize() { + logger_->log_info("Initializing InvokeHTTP"); + + // Set the supported properties + std::set<core::Property> properties; + properties.insert(Method); + properties.insert(URL); + properties.insert(ConnectTimeout); + properties.insert(ReadTimeout); + properties.insert(DateHeader); + properties.insert(AttributesToSend); + properties.insert(SSLContext); + properties.insert(ProxyHost); + properties.insert(ProxyPort); + properties.insert(ProxyUser); + properties.insert(ProxyPassword); + properties.insert(ContentType); + properties.insert(SendBody); + properties.insert(AlwaysOutputResponse); + + setSupportedProperties(properties); + // Set the supported relationships + std::set<core::Relationship> relationships; + relationships.insert(Success); + setSupportedRelationships(relationships); +} + +void InvokeHTTP::onSchedule(core::ProcessContext *context, + core::ProcessSessionFactory *sessionFactory) { + if (!context->getProperty(Method.getName(), method_)) { + logger_->log_info( + "%s attribute is missing, so default value of %s will be used", + Method.getName().c_str(), Method.getValue().c_str()); + return; + } + + if (!context->getProperty(URL.getName(), url_)) { + logger_->log_info( + "%s attribute is missing, so default value of %s will be used", + URL.getName().c_str(), URL.getValue().c_str()); + return; + } + + std::string timeoutStr; + + if (context->getProperty(ConnectTimeout.getName(), timeoutStr)) { + core::Property::StringToInt(timeoutStr, connect_timeout_); + // set the timeout in curl options. + + } else { + logger_->log_info( + "%s attribute is missing, so default value of %s will be used", + ConnectTimeout.getName().c_str(), ConnectTimeout.getValue().c_str()); + + return; + } + + if (context->getProperty(ReadTimeout.getName(), timeoutStr)) { + core::Property::StringToInt(timeoutStr, read_timeout_); + + } else { + logger_->log_info( + "%s attribute is missing, so default value of %s will be used", + ReadTimeout.getName().c_str(), ReadTimeout.getValue().c_str()); + } + + std::string dateHeaderStr; + if (!context->getProperty(DateHeader.getName(), dateHeaderStr)) { + logger_->log_info( + "%s attribute is missing, so default value of %s will be used", + DateHeader.getName().c_str(), DateHeader.getValue().c_str()); + } + + date_header_include_ = utils::StringUtils::StringToBool(dateHeaderStr, + date_header_include_); + + if (!context->getProperty(PropPutOutputAttributes.getName(), + put_attribute_name_)) { + logger_->log_info( + "%s attribute is missing, so default value of %s will be used", + PropPutOutputAttributes.getName().c_str(), + PropPutOutputAttributes.getValue().c_str()); + } + + if (!context->getProperty(AttributesToSend.getName(), + attribute_to_send_regex_)) { + logger_->log_info( + "%s attribute is missing, so default value of %s will be used", + AttributesToSend.getName().c_str(), + AttributesToSend.getValue().c_str()); + } + + std::string always_output_response = "false"; + if (!context->getProperty(AlwaysOutputResponse.getName(), + always_output_response)) { + logger_->log_info( + "%s attribute is missing, so default value of %s will be used", + AttributesToSend.getName().c_str(), + AttributesToSend.getValue().c_str()); + } + + utils::StringUtils::StringToBool(always_output_response, + always_output_response_); + + std::string penalize_no_retry = "false"; + if (!context->getProperty(PenalizeOnNoRetry.getName(), penalize_no_retry)) { + logger_->log_info( + "%s attribute is missing, so default value of %s will be used", + AttributesToSend.getName().c_str(), + AttributesToSend.getValue().c_str()); + } + + utils::StringUtils::StringToBool(penalize_no_retry, penalize_no_retry_); +} + +InvokeHTTP::~InvokeHTTP() { + curl_global_cleanup(); +} + +inline bool InvokeHTTP::matches(const std::string &value, + const std::string &sregex) { +#ifdef __GNUC__ +#if (__GNUC__ >= 4) +#if (__GNUC_MINOR__ < 9) + regex_t regex; + int ret = regcomp(®ex, sregex.c_str(), 0); + if (ret) + return false; + ret = regexec(®ex, value.c_str(), (size_t) 0, NULL, 0); + regfree(®ex); + if (ret) + return false; +#else + try { + std::regex re(sregex); + + if (!std::regex_match(value, re)) { + return false; + } + } catch (std::regex_error e) { + logger_->log_error("Invalid File Filter regex: %s.", e.what()); + return false; + } +#endif +#endif +#else + logger_->log_info("Cannot support regex filtering"); + if (regex == ".*") + return true; +#endif + return true; +} + +std::string InvokeHTTP::generateId() { + uuid_t txId; + uuid_generate(txId); + char uuidStr[37]; + uuid_unparse_lower(txId, uuidStr); + return uuidStr; +} + +bool InvokeHTTP::emitFlowFile(const std::string &method) { + return ("POST" == method || "PUT" == method || "PATCH" == method); +} + +struct curl_slist *InvokeHTTP::build_header_list( + CURL *curl, std::string regex, + const std::map<std::string, std::string> &attributes) { + struct curl_slist *list = NULL; + if (curl) { + for (auto attribute : attributes) { + if (matches(attribute.first, regex)) { + std::string attr = attribute.first + ":" + attribute.second; + list = curl_slist_append(list, attr.c_str()); + } + } + } + return list; +} +void InvokeHTTP::onTrigger(core::ProcessContext *context, + core::ProcessSession *session) { + std::shared_ptr<FlowFileRecord> flowFile = std::static_pointer_cast< + FlowFileRecord>(session->get()); + + logger_->log_info("onTrigger InvokeHTTP with %s", method_.c_str()); + + if (flowFile == nullptr) { + if (!emitFlowFile(method_)) { + logger_->log_info("InvokeHTTP -- create flow file with %s", + method_.c_str()); + flowFile = std::static_pointer_cast<FlowFileRecord>(session->create()); + } else { + logger_->log_info("exiting because method is %s", method_.c_str()); + return; + } + } else { + logger_->log_info("InvokeHTTP -- Received flowfile "); + } + // create a transaction id + std::string tx_id = generateId(); + + CURL *http_session = curl_easy_init(); + // set the HTTP request method from libCURL + set_request_method(http_session, method_); + curl_easy_setopt(http_session, CURLOPT_URL, url_.c_str()); + + if (connect_timeout_ > 0) { + curl_easy_setopt(http_session, CURLOPT_TIMEOUT, connect_timeout_); + } + + if (read_timeout_ > 0) { + curl_easy_setopt(http_session, CURLOPT_TIMEOUT, read_timeout_); + } + HTTPRequestResponse content; + curl_easy_setopt(http_session, CURLOPT_WRITEFUNCTION, + &HTTPRequestResponse::recieve_write); + + curl_easy_setopt(http_session, CURLOPT_WRITEDATA, + static_cast<void*>(&content)); + + if (emitFlowFile(method_)) { + logger_->log_info("InvokeHTTP -- reading flowfile"); + std::shared_ptr<ResourceClaim> claim = flowFile->getResourceClaim(); + if (claim) { + utils::ByteInputCallBack *callback = new utils::ByteInputCallBack(); + session->read(flowFile, callback); + CallBackPosition *callbackObj = new CallBackPosition; + callbackObj->ptr = callback; + callbackObj->pos = 0; + logger_->log_info("InvokeHTTP -- Setting callback"); + curl_easy_setopt(http_session, CURLOPT_UPLOAD, 1L); + curl_easy_setopt(http_session, CURLOPT_INFILESIZE_LARGE, + (curl_off_t)callback->getBufferSize()); + curl_easy_setopt(http_session, CURLOPT_READFUNCTION, + &HTTPRequestResponse::send_write); + curl_easy_setopt(http_session, CURLOPT_READDATA, + static_cast<void*>(callbackObj)); + } else { + logger_->log_error("InvokeHTTP -- no resource claim"); + } + + } else { + logger_->log_info("InvokeHTTP -- Not emitting flowfile to HTTP Server"); + } + + // append all headers + struct curl_slist *headers = build_header_list(http_session, + attribute_to_send_regex_, + flowFile->getAttributes()); + curl_easy_setopt(http_session, CURLOPT_HTTPHEADER, headers); + + logger_->log_info("InvokeHTTP -- curl performed"); + res = curl_easy_perform(http_session); + + if (res == CURLE_OK) { + logger_->log_info("InvokeHTTP -- curl successful"); + + bool putToAttribute = !IsNullOrEmpty(put_attribute_name_); + + std::string response_body(content.data.begin(), content.data.end()); + int64_t http_code = 0; + curl_easy_getinfo(http_session, CURLINFO_RESPONSE_CODE, &http_code); + char *content_type; + /* ask for the content-type */ + curl_easy_getinfo(http_session, CURLINFO_CONTENT_TYPE, &content_type); + + flowFile->addAttribute(STATUS_CODE, std::to_string(http_code)); + flowFile->addAttribute(STATUS_MESSAGE, response_body); + flowFile->addAttribute(REQUEST_URL, url_); + flowFile->addAttribute(TRANSACTION_ID, tx_id); + + bool isSuccess = ((int32_t) (http_code / 100)) == 2 + && res != CURLE_ABORTED_BY_CALLBACK; + bool output_body_to_requestAttr = (!isSuccess || putToAttribute) + && flowFile != nullptr; + bool output_body_to_content = isSuccess && !putToAttribute; + bool body_empty = IsNullOrEmpty(content.data); + + logger_->log_info("isSuccess: %d", isSuccess); + std::shared_ptr<FlowFileRecord> response_flow = nullptr; + + if (output_body_to_content) { + if (flowFile != nullptr) { + response_flow = std::static_pointer_cast<FlowFileRecord>( + session->create(flowFile)); + } else { + response_flow = std::static_pointer_cast<FlowFileRecord>( + session->create()); + } + + std::string ct = content_type; + response_flow->addKeyedAttribute(MIME_TYPE, ct); + response_flow->addAttribute(STATUS_CODE, std::to_string(http_code)); + response_flow->addAttribute(STATUS_MESSAGE, response_body); + response_flow->addAttribute(REQUEST_URL, url_); + response_flow->addAttribute(TRANSACTION_ID, tx_id); + io::DataStream stream((const uint8_t*) content.data.data(), + content.data.size()); + // need an import from the data stream. + session->importFrom(stream, response_flow); + } else { + logger_->log_info("Cannot output body to content"); + response_flow = std::static_pointer_cast<FlowFileRecord>( + session->create()); + } + route(flowFile, response_flow, session, context, isSuccess, http_code); + } else { + logger_->log_error("InvokeHTTP -- curl_easy_perform() failed %s\n", + curl_easy_strerror(res)); + } + curl_slist_free_all(headers); + curl_easy_cleanup(http_session); +} + +void InvokeHTTP::route(std::shared_ptr<FlowFileRecord> &request, + std::shared_ptr<FlowFileRecord> &response, + core::ProcessSession *session, + core::ProcessContext *context, bool isSuccess, + int statusCode) { + // check if we should yield the processor + if (!isSuccess && request == nullptr) { + context->yield(); + } + + // If the property to output the response flowfile regardless of status code is set then transfer it + bool responseSent = false; + if (always_output_response_ && response != nullptr) { + session->transfer(response, Success); + responseSent = true; + } + + // transfer to the correct relationship + // 2xx -> SUCCESS + if (isSuccess) { + // we have two flowfiles to transfer + if (request != nullptr) { + session->transfer(request, Success); + } + if (response != nullptr && !responseSent) { + session->transfer(response, Success); + } + + // 5xx -> RETRY + } else if (statusCode / 100 == 5) { + if (request != nullptr) { + session->penalize(request); + session->transfer(request, RelRetry); + } + + // 1xx, 3xx, 4xx -> NO RETRY + } else { + if (request != nullptr) { + if (penalize_no_retry_) { + session->penalize(request); + } + session->transfer(request, RelNoRetry); + } + } +} + +} /* namespace processors */ +} /* namespace minifi */ +} /* namespace nifi */ +} /* namespace apache */ +} /* namespace org */ http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/071ac1e9/libminifi/test/HttpGetIntegrationTest.cpp ---------------------------------------------------------------------- diff --git a/libminifi/test/HttpGetIntegrationTest.cpp b/libminifi/test/HttpGetIntegrationTest.cpp new file mode 100644 index 0000000..04e6268 --- /dev/null +++ b/libminifi/test/HttpGetIntegrationTest.cpp @@ -0,0 +1,118 @@ +/** + * + * 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 <cassert> +#include <chrono> +#include <fstream> +#include <memory> +#include <string> +#include <thread> +#include <type_traits> +#include <vector> +#include <sys/stat.h> +#include "utils/StringUtils.h" +#include "../include/core/Core.h" +#include "../include/core/logging/LogAppenders.h" +#include "../include/core/logging/BaseLogger.h" +#include "../include/core/logging/Logger.h" +#include "../include/core/ProcessGroup.h" +#include "../include/core/yaml/YamlConfiguration.h" +#include "../include/FlowController.h" +#include "../include/properties/Configure.h" +#include "unit/ProvenanceTestHelper.h" + +std::string test_file_location; + +void waitToVerifyProcessor() { + std::this_thread::sleep_for(std::chrono::seconds(2)); +} + +int main(int argc, char **argv) { + + if (argc > 1) { + test_file_location = argv[1]; + } + mkdir("content_repository", S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH); + std::ostringstream oss; + std::unique_ptr<logging::BaseLogger> outputLogger = std::unique_ptr< + logging::BaseLogger>( + new org::apache::nifi::minifi::core::logging::OutputStreamAppender(oss, + 0)); + std::shared_ptr<logging::Logger> logger = logging::Logger::getLogger(); + logger->updateLogger(std::move(outputLogger)); + logger->setLogLevel("debug"); + + minifi::Configure *configuration = minifi::Configure::getConfigure(); + + std::shared_ptr<core::Repository> test_repo = + std::make_shared<TestRepository>(); + std::shared_ptr<core::Repository> test_flow_repo = std::make_shared< + TestFlowRepository>(); + + configuration->set(minifi::Configure::nifi_flow_configuration_file, + test_file_location); + + std::unique_ptr<core::FlowConfiguration> yaml_ptr = std::unique_ptr< + core::YamlConfiguration>( + new core::YamlConfiguration(test_repo, test_repo, test_file_location)); + std::shared_ptr<TestRepository> repo = + std::static_pointer_cast<TestRepository>(test_repo); + + std::shared_ptr<minifi::FlowController> controller = std::make_shared< + minifi::FlowController>(test_repo, test_flow_repo, std::move(yaml_ptr), + DEFAULT_ROOT_GROUP_NAME, + true); + + core::YamlConfiguration yaml_config(test_repo, test_repo, test_file_location); + + std::unique_ptr<core::ProcessGroup> ptr = yaml_config.getRoot( + test_file_location); + std::shared_ptr<core::ProcessGroup> pg = std::shared_ptr<core::ProcessGroup>( + ptr.get()); + ptr.release(); + + controller->load(); + controller->start(); + waitToVerifyProcessor(); + + controller->waitUnload(60000); + std::string logs = oss.str(); + assert(logs.find("key:filename value:") != std::string::npos); + assert( + logs.find( + "key:invokehttp.request.url value:https://curl.haxx.se/libcurl/c/httpput.html";) + != std::string::npos); + assert(logs.find("Size:8970 Offset:0") != std::string::npos); + assert( + logs.find("key:invokehttp.status.code value:200") != std::string::npos); + std::string stringtofind = "Resource Claim created ./content_repository/"; + + size_t loc = logs.find(stringtofind); + while (loc > 0) { + std::string id = logs.substr(loc + stringtofind.size(), 36); + + loc = logs.find(stringtofind, loc+1); + std::string path = "content_repository/" + id; + unlink(path.c_str()); + + if ( loc == std::string::npos) + break; + } + rmdir("./content_repository"); + return 0; +} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/071ac1e9/libminifi/test/HttpPostIntegrationTest.cpp ---------------------------------------------------------------------- diff --git a/libminifi/test/HttpPostIntegrationTest.cpp b/libminifi/test/HttpPostIntegrationTest.cpp new file mode 100644 index 0000000..2898611 --- /dev/null +++ b/libminifi/test/HttpPostIntegrationTest.cpp @@ -0,0 +1,117 @@ +/** + * + * 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 <cassert> +#include <chrono> +#include <fstream> +#include <memory> +#include <string> +#include <thread> +#include <type_traits> +#include <vector> +#include <sys/stat.h> +#include "utils/StringUtils.h" +#include "../include/core/Core.h" +#include "../include/core/logging/LogAppenders.h" +#include "../include/core/logging/BaseLogger.h" +#include "../include/core/logging/Logger.h" +#include "../include/core/ProcessGroup.h" +#include "../include/core/yaml/YamlConfiguration.h" +#include "../include/FlowController.h" +#include "../include/properties/Configure.h" +#include "unit/ProvenanceTestHelper.h" + +std::string test_file_location; + +void waitToVerifyProcessor() { + std::this_thread::sleep_for(std::chrono::seconds(2)); +} + +int main(int argc, char **argv) { + + if (argc > 1) { + test_file_location = argv[1]; + } + mkdir("/tmp/aljr39/",S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH); + std::ofstream myfile; + myfile.open ("/tmp/aljr39/example.txt"); + myfile << "Hello world" << std::endl; + myfile.close(); + mkdir("content_repository", S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH); + std::ostringstream oss; + std::unique_ptr<logging::BaseLogger> outputLogger = std::unique_ptr< + logging::BaseLogger>( + new org::apache::nifi::minifi::core::logging::OutputStreamAppender(oss, + 0)); + std::shared_ptr<logging::Logger> logger = logging::Logger::getLogger(); + logger->updateLogger(std::move(outputLogger)); + logger->setLogLevel("debug"); + + minifi::Configure *configuration = minifi::Configure::getConfigure(); + + std::shared_ptr<core::Repository> test_repo = + std::make_shared<TestRepository>(); + std::shared_ptr<core::Repository> test_flow_repo = std::make_shared< + TestFlowRepository>(); + + configuration->set(minifi::Configure::nifi_flow_configuration_file, + test_file_location); + + std::unique_ptr<core::FlowConfiguration> yaml_ptr = std::unique_ptr< + core::YamlConfiguration>( + new core::YamlConfiguration(test_repo, test_repo, test_file_location)); + std::shared_ptr<TestRepository> repo = + std::static_pointer_cast<TestRepository>(test_repo); + + std::shared_ptr<minifi::FlowController> controller = std::make_shared< + minifi::FlowController>(test_repo, test_flow_repo, std::move(yaml_ptr), + DEFAULT_ROOT_GROUP_NAME, + true); + + core::YamlConfiguration yaml_config(test_repo, test_repo, test_file_location); + + std::unique_ptr<core::ProcessGroup> ptr = yaml_config.getRoot( + test_file_location); + std::shared_ptr<core::ProcessGroup> pg = std::shared_ptr<core::ProcessGroup>( + ptr.get()); + ptr.release(); + + controller->load(); + controller->start(); + waitToVerifyProcessor(); + + controller->waitUnload(60000); + std::string logs = oss.str(); + assert(logs.find("curl performed") != std::string::npos); + assert(logs.find("Import offset 0 length 12") != std::string::npos); + + std::string stringtofind = "Resource Claim created ./content_repository/"; + + size_t loc = logs.find(stringtofind); + while (loc > 0 && loc != std::string::npos) { + std::string id = logs.substr(loc + stringtofind.size(), 36); + loc = logs.find(stringtofind, loc+1); + std::string path = "content_repository/" + id; + unlink(path.c_str()); + if ( loc == std::string::npos) + break; + } + + rmdir("./content_repository"); + return 0; +} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/071ac1e9/libminifi/test/unit/InvokeHTTPTests.cpp ---------------------------------------------------------------------- diff --git a/libminifi/test/unit/InvokeHTTPTests.cpp b/libminifi/test/unit/InvokeHTTPTests.cpp new file mode 100644 index 0000000..6b96549 --- /dev/null +++ b/libminifi/test/unit/InvokeHTTPTests.cpp @@ -0,0 +1,471 @@ +/** + * + * 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 <uuid/uuid.h> +#include <fstream> +#include "FlowController.h" +#include "ProvenanceTestHelper.h" +#include "../TestBase.h" +#include "core/logging/LogAppenders.h" +#include "core/logging/BaseLogger.h" +#include "processors/GetFile.h" +#include "core/Core.h" +#include "../../include/core/FlowFile.h" +#include "core/Processor.h" +#include "core/ProcessContext.h" +#include "core/ProcessSession.h" +#include "core/ProcessorNode.h" + +TEST_CASE("HTTPTestsPostNoResourceClaim", "[httptest1]") { + + std::stringstream oss; + std::unique_ptr<logging::BaseLogger> outputLogger = std::unique_ptr< + logging::BaseLogger>( + new org::apache::nifi::minifi::core::logging::OutputStreamAppender(oss,0)); + std::shared_ptr<logging::Logger> logger = logging::Logger::getLogger(); + logger->updateLogger(std::move(outputLogger)); + + TestController testController; + + testController.enableDebug(); + + + std::shared_ptr<TestRepository> repo = std::make_shared< + TestRepository>(); + + std::shared_ptr<core::Processor> processor = std::make_shared< + org::apache::nifi::minifi::processors::ListenHTTP>("listenhttp"); + + std::shared_ptr<core::Processor> invokehttp = std::make_shared< + org::apache::nifi::minifi::processors::InvokeHTTP>("invokehttp"); + uuid_t processoruuid; + REQUIRE(true == processor->getUUID(processoruuid)); + + uuid_t invokehttp_uuid; + REQUIRE(true == invokehttp->getUUID(invokehttp_uuid)); + + + std::shared_ptr<minifi::Connection> connection = std::make_shared< + minifi::Connection>(repo,"getfileCreate2Connection"); + connection->setRelationship(core::Relationship("success", "description")); + + std::shared_ptr<minifi::Connection> connection2 = std::make_shared< + minifi::Connection>(repo,"listenhttp"); + + connection2->setRelationship(core::Relationship("No Retry", "description")); + + // link the connections so that we can test results at the end for this + connection->setSource(processor); + + + // link the connections so that we can test results at the end for this + connection->setDestination(invokehttp); + + connection2->setSource(invokehttp); + + + connection2->setSourceUUID(invokehttp_uuid); + connection->setSourceUUID(processoruuid); + connection->setDestinationUUID(invokehttp_uuid); + + processor->addConnection(connection); + invokehttp->addConnection(connection); + invokehttp->addConnection(connection2); + + + core::ProcessorNode node(processor); + core::ProcessorNode node2(invokehttp); + + core::ProcessContext context(node, repo); + core::ProcessContext context2(node2, repo); + context.setProperty(org::apache::nifi::minifi::processors::ListenHTTP::Port, + "8685"); + context.setProperty(org::apache::nifi::minifi::processors::ListenHTTP::BasePath, + "/testytesttest"); + + context2.setProperty(org::apache::nifi::minifi::processors::InvokeHTTP::Method, + "POST"); + context2.setProperty(org::apache::nifi::minifi::processors::InvokeHTTP::URL, + "http://localhost:8685/testytesttest";); + core::ProcessSession session(&context); + core::ProcessSession session2(&context2); + + REQUIRE(processor->getName() == "listenhttp"); + + core::ProcessSessionFactory factory(&context); + + std::shared_ptr<core::FlowFile> record; + processor->setScheduledState(core::ScheduledState::RUNNING); + processor->onSchedule(&context, &factory); + processor->onTrigger(&context, &session); + + invokehttp->incrementActiveTasks(); + invokehttp->setScheduledState(core::ScheduledState::RUNNING); + core::ProcessSessionFactory factory2(&context2); + invokehttp->onSchedule(&context2, &factory2); + invokehttp->onTrigger(&context2, &session2); + + provenance::ProvenanceReporter *reporter = session.getProvenanceReporter(); + std::set<provenance::ProvenanceEventRecord*> records = reporter->getEvents(); + record = session.get(); + REQUIRE(record == nullptr); + REQUIRE(records.size() == 0); + + + processor->incrementActiveTasks(); + processor->setScheduledState(core::ScheduledState::RUNNING); + processor->onTrigger(&context, &session); + + reporter = session.getProvenanceReporter(); + + records = reporter->getEvents(); + session.commit(); + + invokehttp->incrementActiveTasks(); + invokehttp->setScheduledState(core::ScheduledState::RUNNING); + invokehttp->onTrigger(&context2, &session2); + + session2.commit(); + records = reporter->getEvents(); + + + + for (provenance::ProvenanceEventRecord *provEventRecord : records) { + REQUIRE(provEventRecord->getComponentType() == processor->getName()); + } + std::shared_ptr<core::FlowFile> ffr = session2.get(); + std::string log_attribute_output = oss.str(); +std::cout << log_attribute_output << std::endl; + REQUIRE( log_attribute_output.find("exiting because method is POST") != std::string::npos ); + +} + + +TEST_CASE("HTTPTestsWithNoResourceClaimPOST", "[httptest1]") { + + std::stringstream oss; + std::unique_ptr<logging::BaseLogger> outputLogger = std::unique_ptr< + logging::BaseLogger>( + new org::apache::nifi::minifi::core::logging::OutputStreamAppender(oss,0)); + std::shared_ptr<logging::Logger> logger = logging::Logger::getLogger(); + logger->updateLogger(std::move(outputLogger)); + + TestController testController; + + testController.enableDebug(); + + + + std::shared_ptr<TestRepository> repo = std::make_shared< + TestRepository>(); + + std::shared_ptr<core::Processor> getfileprocessor = std::make_shared< + org::apache::nifi::minifi::processors::GetFile>("getfileCreate2"); + + std::shared_ptr<core::Processor> logAttribute = std::make_shared< + org::apache::nifi::minifi::processors::LogAttribute>("logattribute"); + + char format[] = "/tmp/gt.XXXXXX"; + char *dir = testController.createTempDirectory(format); + + std::shared_ptr<core::Processor> listenhttp = std::make_shared< + org::apache::nifi::minifi::processors::ListenHTTP>("listenhttp"); + + std::shared_ptr<core::Processor> invokehttp = std::make_shared< + org::apache::nifi::minifi::processors::InvokeHTTP>("invokehttp"); + uuid_t processoruuid; + REQUIRE(true == listenhttp->getUUID(processoruuid)); + + uuid_t invokehttp_uuid; + REQUIRE(true == invokehttp->getUUID(invokehttp_uuid)); + + + std::shared_ptr<minifi::Connection> gcConnection = std::make_shared< + minifi::Connection>(repo, "getfileCreate2Connection"); + gcConnection->setRelationship(core::Relationship("success", "description")); + +std::shared_ptr<minifi::Connection> laConnection = std::make_shared< + minifi::Connection>(repo, "logattribute"); +laConnection->setRelationship(core::Relationship("success", "description")); + + + + std::shared_ptr<minifi::Connection> connection = std::make_shared< + minifi::Connection>(repo,"getfileCreate2Connection"); + connection->setRelationship(core::Relationship("success", "description")); + + std::shared_ptr<minifi::Connection> connection2 = std::make_shared< + minifi::Connection>(repo,"listenhttp"); + + connection2->setRelationship(core::Relationship("No Retry", "description")); + + // link the connections so that we can test results at the end for this + connection->setSource(listenhttp); + + + connection2->setSourceUUID(invokehttp_uuid); + connection->setSourceUUID(processoruuid); + connection->setDestinationUUID(invokehttp_uuid); + + listenhttp->addConnection(connection); + invokehttp->addConnection(connection); + invokehttp->addConnection(connection2); + + + core::ProcessorNode node(listenhttp); + core::ProcessorNode node2(invokehttp); + + core::ProcessContext context(node, repo); + core::ProcessContext context2(node2, repo); + context.setProperty(org::apache::nifi::minifi::processors::ListenHTTP::Port, + "8686"); + context.setProperty(org::apache::nifi::minifi::processors::ListenHTTP::BasePath, + "/testytesttest"); + + context2.setProperty(org::apache::nifi::minifi::processors::InvokeHTTP::Method, + "POST"); + context2.setProperty(org::apache::nifi::minifi::processors::InvokeHTTP::URL, + "http://localhost:8686/testytesttest";); + core::ProcessSession session(&context); + core::ProcessSession session2(&context2); + + REQUIRE(listenhttp->getName() == "listenhttp"); + + core::ProcessSessionFactory factory(&context); + + std::shared_ptr<core::FlowFile> record; + listenhttp->setScheduledState(core::ScheduledState::RUNNING); + listenhttp->onSchedule(&context, &factory); + listenhttp->onTrigger(&context, &session); + + invokehttp->incrementActiveTasks(); + invokehttp->setScheduledState(core::ScheduledState::RUNNING); + core::ProcessSessionFactory factory2(&context2); + invokehttp->onSchedule(&context2, &factory2); + invokehttp->onTrigger(&context2, &session2); + + provenance::ProvenanceReporter *reporter = session.getProvenanceReporter(); + std::set<provenance::ProvenanceEventRecord*> records = reporter->getEvents(); + record = session.get(); + REQUIRE(record == nullptr); + REQUIRE(records.size() == 0); + + + listenhttp->incrementActiveTasks(); + listenhttp->setScheduledState(core::ScheduledState::RUNNING); + listenhttp->onTrigger(&context, &session); + + reporter = session.getProvenanceReporter(); + + records = reporter->getEvents(); + session.commit(); + + invokehttp->incrementActiveTasks(); + invokehttp->setScheduledState(core::ScheduledState::RUNNING); + invokehttp->onTrigger(&context2, &session2); + + session2.commit(); + records = reporter->getEvents(); + + + + for (provenance::ProvenanceEventRecord *provEventRecord : records) { + REQUIRE(provEventRecord->getComponentType() == listenhttp->getName()); + } + std::shared_ptr<core::FlowFile> ffr = session2.get(); + std::string log_attribute_output = oss.str(); +std::cout << log_attribute_output << std::endl; + REQUIRE( log_attribute_output.find("exiting because method is POST") != std::string::npos ); + +} + + +class CallBack : public minifi::OutputStreamCallback +{ + public: + CallBack() + { + + } + virtual ~CallBack(){ + + } + virtual void process(std::ofstream *stream){ + std::string st = "we're gnna write some test stuff"; + stream->write(st.c_str(),st.length()); + } +}; + +TEST_CASE("HTTPTestsWithResourceClaimPOST", "[httptest1]") { + + std::stringstream oss; + std::unique_ptr<logging::BaseLogger> outputLogger = std::unique_ptr< + logging::BaseLogger>( + new org::apache::nifi::minifi::core::logging::OutputStreamAppender(oss,0)); + std::shared_ptr<logging::Logger> logger = logging::Logger::getLogger(); + logger->updateLogger(std::move(outputLogger)); + + TestController testController; + + testController.enableDebug(); + + + + std::shared_ptr<TestRepository> repo = std::make_shared< + TestRepository>(); + + std::shared_ptr<core::Processor> getfileprocessor = std::make_shared< + org::apache::nifi::minifi::processors::GetFile>("getfileCreate2"); + + std::shared_ptr<core::Processor> logAttribute = std::make_shared< + org::apache::nifi::minifi::processors::LogAttribute>("logattribute"); + + char format[] = "/tmp/gt.XXXXXX"; + char *dir = testController.createTempDirectory(format); + + std::shared_ptr<core::Processor> listenhttp = std::make_shared< + org::apache::nifi::minifi::processors::ListenHTTP>("listenhttp"); + + std::shared_ptr<core::Processor> invokehttp = std::make_shared< + org::apache::nifi::minifi::processors::InvokeHTTP>("invokehttp"); + uuid_t processoruuid; + REQUIRE(true == listenhttp->getUUID(processoruuid)); + + uuid_t invokehttp_uuid; + REQUIRE(true == invokehttp->getUUID(invokehttp_uuid)); + + + std::shared_ptr<minifi::Connection> gcConnection = std::make_shared< + minifi::Connection>(repo, "getfileCreate2Connection"); + gcConnection->setRelationship(core::Relationship("success", "description")); + +std::shared_ptr<minifi::Connection> laConnection = std::make_shared< + minifi::Connection>(repo, "logattribute"); +laConnection->setRelationship(core::Relationship("success", "description")); + + + + std::shared_ptr<minifi::Connection> connection = std::make_shared< + minifi::Connection>(repo,"getfileCreate2Connection"); + connection->setRelationship(core::Relationship("success", "description")); + + std::shared_ptr<minifi::Connection> connection2 = std::make_shared< + minifi::Connection>(repo,"listenhttp"); + + connection2->setRelationship(core::Relationship("No Retry", "description")); + + // link the connections so that we can test results at the end for this + connection->setSource(listenhttp); + + connection->setSourceUUID(invokehttp_uuid); + connection->setDestinationUUID(processoruuid); + + connection2->setSourceUUID(processoruuid); + connection2->setSourceUUID(processoruuid); + + + listenhttp->addConnection(connection); + invokehttp->addConnection(connection); + invokehttp->addConnection(connection2); + + + core::ProcessorNode node(invokehttp); + core::ProcessorNode node2(listenhttp); + + core::ProcessContext context(node, repo); + core::ProcessContext context2(node2, repo); + context.setProperty(org::apache::nifi::minifi::processors::ListenHTTP::Port, + "8680"); + context.setProperty(org::apache::nifi::minifi::processors::ListenHTTP::BasePath, + "/testytesttest"); + + context2.setProperty(org::apache::nifi::minifi::processors::InvokeHTTP::Method, + "POST"); + context2.setProperty(org::apache::nifi::minifi::processors::InvokeHTTP::URL, + "http://localhost:8680/testytesttest";); + core::ProcessSession session(&context); + core::ProcessSession session2(&context2); + + REQUIRE(listenhttp->getName() == "listenhttp"); + + core::ProcessSessionFactory factory(&context); + + std::shared_ptr<core::FlowFile> record; + + CallBack callback; + + /* + explicit FlowFileRecord(std::shared_ptr<core::Repository> flow_repository, + std::map<std::string, std::string> attributes, + std::shared_ptr<ResourceClaim> claim = nullptr); + */ + std::map<std::string,std::string> attributes; + attributes["testy"] = "test"; + std::shared_ptr<minifi::FlowFileRecord> flow = std::make_shared<minifi::FlowFileRecord>(repo,attributes); + session2.write(flow,&callback); + + invokehttp->incrementActiveTasks(); + invokehttp->setScheduledState(core::ScheduledState::RUNNING); + core::ProcessSessionFactory factory2(&context2); + invokehttp->onSchedule(&context2, &factory2); + invokehttp->onTrigger(&context2, &session2); + + listenhttp->incrementActiveTasks(); + listenhttp->setScheduledState(core::ScheduledState::RUNNING); + listenhttp->onSchedule(&context, &factory); + listenhttp->onTrigger(&context, &session); + + + + provenance::ProvenanceReporter *reporter = session.getProvenanceReporter(); + std::set<provenance::ProvenanceEventRecord*> records = reporter->getEvents(); + record = session.get(); + REQUIRE(record == nullptr); + REQUIRE(records.size() == 0); + + + listenhttp->incrementActiveTasks(); + listenhttp->setScheduledState(core::ScheduledState::RUNNING); + listenhttp->onTrigger(&context, &session); + + reporter = session.getProvenanceReporter(); + + records = reporter->getEvents(); + session.commit(); + + invokehttp->incrementActiveTasks(); + invokehttp->setScheduledState(core::ScheduledState::RUNNING); + invokehttp->onTrigger(&context2, &session2); + + session2.commit(); + records = reporter->getEvents(); + + + + for (provenance::ProvenanceEventRecord *provEventRecord : records) { + REQUIRE(provEventRecord->getComponentType() == listenhttp->getName()); + } + std::shared_ptr<core::FlowFile> ffr = session2.get(); + std::string log_attribute_output = oss.str(); +std::cout << log_attribute_output << std::endl; + REQUIRE( log_attribute_output.find("exiting because method is POST") != std::string::npos ); + +} + + + + http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/071ac1e9/libminifi/test/unit/ProvenanceTestHelper.h ---------------------------------------------------------------------- diff --git a/libminifi/test/unit/ProvenanceTestHelper.h b/libminifi/test/unit/ProvenanceTestHelper.h index 01df637..1a76be7 100644 --- a/libminifi/test/unit/ProvenanceTestHelper.h +++ b/libminifi/test/unit/ProvenanceTestHelper.h @@ -21,6 +21,7 @@ #include "provenance/Provenance.h" #include "FlowController.h" #include "core/Repository.h" +#include "core/repository/FlowFileRepository.h" #include "core/Core.h" /** * Test repository @@ -89,6 +90,72 @@ class TestRepository : public core::Repository { std::map<std::string, std::string> repositoryResults; }; +class TestFlowRepository : public core::repository::FlowFileRepository { + public: + TestFlowRepository() + : core::repository::FlowFileRepository("./", 1000, 100, 0) { + } + // initialize + bool initialize() { + return true; + } + + // Destructor + virtual ~TestFlowRepository() { + + } + + bool Put(std::string key, uint8_t *buf, int bufLen) { + repositoryResults.insert( + std::pair<std::string, std::string>( + key, std::string((const char*) buf, bufLen))); + return true; + } + // Delete + bool Delete(std::string key) { + repositoryResults.erase(key); + return true; + } + // Get + bool Get(std::string key, std::string &value) { + auto result = repositoryResults.find(key); + if (result != repositoryResults.end()) { + value = result->second; + return true; + } else { + return false; + } + } + + const std::map<std::string, std::string> &getRepoMap() const { + return repositoryResults; + } + + void getProvenanceRecord( + std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>> &records, + int maxSize) { + for (auto entry : repositoryResults) { + if (records.size() >= maxSize) + break; + std::shared_ptr<provenance::ProvenanceEventRecord> eventRead = + std::make_shared<provenance::ProvenanceEventRecord>(); + + if (eventRead->DeSerialize((uint8_t*) entry.second.data(), + entry.second.length())) { + records.push_back(eventRead); + } + } + } + + void run() { + // do nothing + } + protected: + std::map<std::string, std::string> repositoryResults; +}; + + + class TestFlowController : public minifi::FlowController { public: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/071ac1e9/libminifi/test/unit/resource/TestHTTPGet.yml ---------------------------------------------------------------------- diff --git a/libminifi/test/unit/resource/TestHTTPGet.yml b/libminifi/test/unit/resource/TestHTTPGet.yml new file mode 100644 index 0000000..0783b8e --- /dev/null +++ b/libminifi/test/unit/resource/TestHTTPGet.yml @@ -0,0 +1,73 @@ +# +# 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. +# +Flow Controller: + name: MiNiFi Flow + id: 2438e3c8-015a-1000-79ca-83af40ec1990 +Processors: + - name: invoke + id: 2438e3c8-015a-1000-79ca-83af40ec1991 + class: org.apache.nifi.processors.standard.InvokeHTTP + max concurrent tasks: 1 + scheduling strategy: TIMER_DRIVEN + scheduling period: 1 sec + penalization period: 30 sec + yield period: 1 sec + run duration nanos: 0 + auto-terminated relationships list: + Properties: + HTTP Method: GET + Remote URL: https://curl.haxx.se/libcurl/c/httpput.html + - name: OhJeez + id: 2438e3c8-015a-1000-79ca-83af40ec1992 + class: org.apache.nifi.processors.standard.LogAttribute + max concurrent tasks: 1 + scheduling strategy: TIMER_DRIVEN + scheduling period: 1 sec + penalization period: 30 sec + yield period: 1 sec + run duration nanos: 0 + auto-terminated relationships list: response + Properties: + Log Level: info + Log Payload: true + +Connections: + - name: TransferFilesToRPG + id: 2438e3c8-015a-1000-79ca-83af40ec1997 + source name: invoke + source id: 2438e3c8-015a-1000-79ca-83af40ec1991 + source relationship name: success + destination name: OhJeez + destination id: 2438e3c8-015a-1000-79ca-83af40ec1992 + max work queue size: 0 + max work queue data size: 1 MB + flowfile expiration: 60 sec + - name: TransferFilesToRPG2 + id: 2438e3c8-015a-1000-79ca-83af40ec1917 + source name: OhJeez + source id: 2438e3c8-015a-1000-79ca-83af40ec1992 + destination name: OhJeez + destination id: 2438e3c8-015a-1000-79ca-83af40ec1992 + source relationship name: success + max work queue size: 0 + max work queue data size: 1 MB + flowfile expiration: 60 sec + +Remote Processing Groups: + \ No newline at end of file http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/071ac1e9/libminifi/test/unit/resource/TestHTTPPost.yml ---------------------------------------------------------------------- diff --git a/libminifi/test/unit/resource/TestHTTPPost.yml b/libminifi/test/unit/resource/TestHTTPPost.yml new file mode 100644 index 0000000..837194d --- /dev/null +++ b/libminifi/test/unit/resource/TestHTTPPost.yml @@ -0,0 +1,87 @@ +# +# 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. +# +Flow Controller: + name: MiNiFi Flow + id: 2438e3c8-015a-1000-79ca-83af40ec1990 +Processors: + - name: invoke + id: 2438e3c8-015a-1000-79ca-83af40ec1991 + class: org.apache.nifi.processors.standard.GetFile + max concurrent tasks: 1 + scheduling strategy: TIMER_DRIVEN + scheduling period: 1 sec + penalization period: 30 sec + yield period: 1 sec + run duration nanos: 0 + auto-terminated relationships list: + Properties: + Input Directory: /tmp/aljr39 + Keep Source File: false + + - name: OhJeez + id: 2438e3c8-015a-1000-79ca-83af40ec1992 + class: org.apache.nifi.processors.standard.InvokeHTTP + max concurrent tasks: 1 + scheduling strategy: TIMER_DRIVEN + scheduling period: 1 sec + penalization period: 30 sec + yield period: 1 sec + run duration nanos: 0 + auto-terminated relationships list: response + Properties: + HTTP Method: POST + Remote URL: http://requestb.in/u8ax9uu8 + + - name: Loggit + id: 2438e3c8-015a-1000-79ca-83af40ec1993 + class: org.apache.nifi.processors.standard.LogAttribute + max concurrent tasks: 1 + scheduling strategy: TIMER_DRIVEN + scheduling period: 1 sec + penalization period: 30 sec + yield period: 1 sec + run duration nanos: 0 + auto-terminated relationships list: response + Properties: + LogLevel: info + +Connections: + - name: TransferFilesToRPG + id: 2438e3c8-015a-1000-79ca-83af40ec1997 + source name: invoke + source id: 2438e3c8-015a-1000-79ca-83af40ec1991 + source relationship name: success + destination name: OhJeez + destination id: 2438e3c8-015a-1000-79ca-83af40ec1992 + max work queue size: 0 + max work queue data size: 1 MB + flowfile expiration: 60 sec + - name: TransferFilesToRPG2 + id: 2438e3c8-015a-1000-79ca-83af40ec1917 + source name: OhJeez + source id: 2438e3c8-015a-1000-79ca-83af40ec1992 + destination name: OhJeez + destination id: 2438e3c8-015a-1000-79ca-83af40ec1993 + source relationship name: success + max work queue size: 0 + max work queue data size: 1 MB + flowfile expiration: 60 sec + +Remote Processing Groups: + \ No newline at end of file http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/071ac1e9/main/MiNiFiMain.cpp ---------------------------------------------------------------------- diff --git a/main/MiNiFiMain.cpp b/main/MiNiFiMain.cpp index 9e6a37f..f4ea89e 100644 --- a/main/MiNiFiMain.cpp +++ b/main/MiNiFiMain.cpp @@ -143,6 +143,12 @@ int main(int argc, char **argv) { STOP_WAIT_TIME_MS); } + std::string log_level; + if (configure->get(minifi::Configure::nifi_log_level, + log_level)) { + logger->setLogLevel(log_level); + } + // set the log configuration. std::unique_ptr<logging::BaseLogger> configured_logger = logging::LogInstance::getConfiguredLogger(configure);
