You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by sz...@apache.org on 2021/06/23 16:55:54 UTC
[nifi-minifi-cpp] 06/06: MINIFICPP-1494 Allow InvokeHTTP GET
requests without incoming flowfile
This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 31fd0999ac656f28600a952d0b5aad66365d816b
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Wed Jun 23 18:45:06 2021 +0200
MINIFICPP-1494 Allow InvokeHTTP GET requests without incoming flowfile
- Refactor HTTPGetIntegrationTest
- Fix flaky HTTP test
- Fix InvokeHTTP relationship documentation
- Add documentation of relationships of InvokeHTTP
Closes #1082
Signed-off-by: Marton Szasz <sz...@apache.org>
---
PROCESSORS.md | 6 +-
extensions/http-curl/processors/InvokeHTTP.cpp | 7 +-
extensions/http-curl/processors/InvokeHTTP.h | 1 +
extensions/http-curl/tests/CMakeLists.txt | 8 +-
extensions/http-curl/tests/HTTPHandlers.h | 22 ++++
.../http-curl/tests/HttpGetIntegrationTest.cpp | 139 ---------------------
extensions/http-curl/tests/VerifyInvokeHTTP.h | 131 +++++++++++++++++++
.../http-curl/tests/VerifyInvokeHTTPGetTest.cpp | 63 ++++++++++
...keHTTPTest.cpp => VerifyInvokeHTTPPostTest.cpp} | 121 ++----------------
libminifi/test/resources/TestHTTPGet.yml | 20 ++-
libminifi/test/resources/TestHTTPGetSecure.yml | 19 ++-
.../test/resources/TestHTTPPostChunkedEncoding.yml | 14 +--
12 files changed, 273 insertions(+), 278 deletions(-)
diff --git a/PROCESSORS.md b/PROCESSORS.md
index 3cbf541..4fe1f06 100644
--- a/PROCESSORS.md
+++ b/PROCESSORS.md
@@ -728,7 +728,11 @@ In the list below, the names of required properties appear in bold. Any other pr
| Name | Description |
| - | - |
-|success|All files are routed to success|
+|success|The original FlowFile will be routed upon success (2xx status codes). It will have new attributes detailing the success of the request.|
+|response|A Response FlowFile will be routed upon success (2xx status codes). If the 'Always Output Response' property is true then the response will be sent to this relationship regardless of the status code received.|
+|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.|
+|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.|
+|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.|
## ListSFTP
diff --git a/extensions/http-curl/processors/InvokeHTTP.cpp b/extensions/http-curl/processors/InvokeHTTP.cpp
index 9ac3c07..34b4db1 100644
--- a/extensions/http-curl/processors/InvokeHTTP.cpp
+++ b/extensions/http-curl/processors/InvokeHTTP.cpp
@@ -123,9 +123,12 @@ 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::Success("success", "The original FlowFile will be routed upon success (2xx status codes). "
+ "It will have new attributes detailing the success of the request.");
-core::Relationship InvokeHTTP::RelResponse("response", "Represents a response flowfile");
+core::Relationship InvokeHTTP::RelResponse("response", "A Response FlowFile will be routed upon success (2xx status codes). "
+ "If the 'Always Output Response' property is true then the response will be sent "
+ "to this relationship regardless of the status code received.");
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.");
diff --git a/extensions/http-curl/processors/InvokeHTTP.h b/extensions/http-curl/processors/InvokeHTTP.h
index 99f54a0..766729f 100644
--- a/extensions/http-curl/processors/InvokeHTTP.h
+++ b/extensions/http-curl/processors/InvokeHTTP.h
@@ -47,6 +47,7 @@ class InvokeHTTP : public core::Processor {
*/
explicit InvokeHTTP(const std::string& name, const utils::Identifier& uuid = {})
: Processor(name, uuid) {
+ setTriggerWhenEmpty(true);
}
// Destructor
virtual ~InvokeHTTP();
diff --git a/extensions/http-curl/tests/CMakeLists.txt b/extensions/http-curl/tests/CMakeLists.txt
index 025392d..d31e400 100644
--- a/extensions/http-curl/tests/CMakeLists.txt
+++ b/extensions/http-curl/tests/CMakeLists.txt
@@ -66,7 +66,7 @@ ENDFOREACH()
message("-- Finished building ${CURL_INT_TEST_COUNT} libcURL integration test file(s)...")
-add_test(NAME HttpGetIntegrationTest COMMAND HttpGetIntegrationTest "${TEST_RESOURCES}/TestHTTPGet.yml" "${TEST_RESOURCES}/")
+add_test(NAME VerifyInvokeHTTPGetTest COMMAND VerifyInvokeHTTPGetTest "${TEST_RESOURCES}/TestHTTPGet.yml")
add_test(NAME C2UpdateTest COMMAND C2UpdateTest "${TEST_RESOURCES}/TestHTTPGet.yml" "${TEST_RESOURCES}/")
add_test(NAME C2FetchFlowIfMissingTest COMMAND C2FetchFlowIfMissingTest "${TEST_RESOURCES}/TestEmpty.yml" "${TEST_RESOURCES}/")
add_test(NAME C2ConfigEncryption COMMAND C2ConfigEncryption "${TEST_RESOURCES}/decrypted.config.yml" "${TEST_RESOURCES}/")
@@ -77,9 +77,9 @@ add_test(NAME C2FailedUpdateTest COMMAND C2FailedUpdateTest "${TEST_RESOURCES}/T
add_test(NAME C2NullConfiguration COMMAND C2NullConfiguration "${TEST_RESOURCES}/TestNull.yml" "${TEST_RESOURCES}/")
add_test(NAME C2RequestClassTest COMMAND C2RequestClassTest)
if (NOT OPENSSL_OFF)
- add_test(NAME HttpGetIntegrationTestSecure COMMAND HttpGetIntegrationTest "${TEST_RESOURCES}/TestHTTPGetSecure.yml" "${TEST_RESOURCES}/")
+ add_test(NAME VerifyInvokeHTTPGetTestSecure COMMAND VerifyInvokeHTTPGetTest "${TEST_RESOURCES}/TestHTTPGetSecure.yml" "${TEST_RESOURCES}/")
add_test(NAME C2VerifyHeartbeatAndStopSecure COMMAND C2VerifyHeartbeatAndStop "${TEST_RESOURCES}/C2VerifyHeartbeatAndStopSecure.yml" "${TEST_RESOURCES}/")
- add_test(NAME VerifyInvokeHTTPTestSecure COMMAND VerifyInvokeHTTPTest "${TEST_RESOURCES}/TestInvokeHTTPPostSecure.yml" "${TEST_RESOURCES}/")
+ add_test(NAME VerifyInvokeHTTPPostTestSecure COMMAND VerifyInvokeHTTPPostTest "${TEST_RESOURCES}/TestInvokeHTTPPostSecure.yml" "${TEST_RESOURCES}/")
endif()
add_test(NAME HttpPostIntegrationTest COMMAND HttpPostIntegrationTest "${TEST_RESOURCES}/TestHTTPPost.yml" "${TEST_RESOURCES}/")
if (NOT APPLE)
@@ -92,7 +92,7 @@ add_test(NAME HTTPSiteToSiteTests COMMAND HTTPSiteToSiteTests "${TEST_RESOURCES}
add_test(NAME TimeoutHTTPSiteToSiteTests COMMAND TimeoutHTTPSiteToSiteTests "${TEST_RESOURCES}/TestTimeoutHTTPSiteToSite.yml" "${TEST_RESOURCES}/" "http://localhost:8098/nifi-api")
add_test(NAME SiteToSiteRestTest COMMAND SiteToSiteRestTest "${TEST_RESOURCES}/TestSite2SiteRest.yml" "${TEST_RESOURCES}/" "http://localhost:8077/nifi-api/site-to-site")
add_test(NAME ControllerServiceIntegrationTests COMMAND ControllerServiceIntegrationTests "${TEST_RESOURCES}/TestControllerServices.yml" "${TEST_RESOURCES}/")
-add_test(NAME VerifyInvokeHTTPTest COMMAND VerifyInvokeHTTPTest "${TEST_RESOURCES}/TestInvokeHTTPPost.yml")
+add_test(NAME VerifyInvokeHTTPPostTest COMMAND VerifyInvokeHTTPPostTest "${TEST_RESOURCES}/TestInvokeHTTPPost.yml")
add_test(NAME AbsoluteTimeoutTest COMMAND AbsoluteTimeoutTest)
add_test(NAME C2PauseResumeTest COMMAND C2PauseResumeTest "${TEST_RESOURCES}/C2PauseResumeTest.yml" "${TEST_RESOURCES}/")
add_test(NAME C2LogHeartbeatTest COMMAND C2LogHeartbeatTest)
diff --git a/extensions/http-curl/tests/HTTPHandlers.h b/extensions/http-curl/tests/HTTPHandlers.h
index e797b37..8cdc2d3 100644
--- a/extensions/http-curl/tests/HTTPHandlers.h
+++ b/extensions/http-curl/tests/HTTPHandlers.h
@@ -624,3 +624,25 @@ class TimeoutingHTTPHandler : public ServerAwareHandler {
}
std::vector<std::chrono::milliseconds> wait_times_;
};
+
+class HttpGetResponder : public ServerAwareHandler {
+ public:
+ bool handleGet(CivetServer* /*server*/, struct mg_connection *conn) override {
+ puts("handle get");
+ static const std::string site2site_rest_resp = "hi this is a get test";
+ mg_printf(conn, "HTTP/1.1 200 OK\r\nContent-Type: "
+ "text/plain\r\nContent-Length: %lu\r\nConnection: close\r\n\r\n",
+ site2site_rest_resp.length());
+ mg_printf(conn, "%s", site2site_rest_resp.c_str());
+ return true;
+ }
+};
+
+class RetryHttpGetResponder : public ServerAwareHandler {
+ public:
+ bool handleGet(CivetServer* /*server*/, struct mg_connection *conn) override {
+ puts("handle get with retry");
+ mg_printf(conn, "HTTP/1.1 501 Not Implemented\r\nContent-Type: text/plain\r\nContent-Length: 0\r\nConnection: close\r\n\r\n");
+ return true;
+ }
+};
diff --git a/extensions/http-curl/tests/HttpGetIntegrationTest.cpp b/extensions/http-curl/tests/HttpGetIntegrationTest.cpp
deleted file mode 100644
index 1cd4b5b..0000000
--- a/extensions/http-curl/tests/HttpGetIntegrationTest.cpp
+++ /dev/null
@@ -1,139 +0,0 @@
-/**
- *
- * 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.
- */
-
-#define CURLOPT_SSL_VERIFYPEER_DISABLE 1
-#undef NDEBUG
-#include <cassert>
-#include <utility>
-#include <chrono>
-#include <memory>
-#include <string>
-#include <vector>
-#include "HTTPClient.h"
-#include "InvokeHTTP.h"
-#include "TestServer.h"
-#include "TestBase.h"
-#include "utils/StringUtils.h"
-#include "core/yaml/YamlConfiguration.h"
-#include "FlowController.h"
-#include "properties/Configure.h"
-#include "unit/ProvenanceTestHelper.h"
-#include "io/StreamFactory.h"
-#include "processors/LogAttribute.h"
-#include "integration/IntegrationBase.h"
-#include "utils/IntegrationTestUtils.h"
-
-int log_message(const struct mg_connection* /*conn*/, const char *message) {
- puts(message);
- return 1;
-}
-
-int ssl_enable(void* /*ssl_context*/, void* /*user_data*/) {
- puts("Enable ssl");
- return 0;
-}
-
-class HttpResponder : public CivetHandler {
- private:
- public:
- bool handleGet(CivetServer* /*server*/, struct mg_connection *conn) override {
- puts("handle get");
- static const std::string site2site_rest_resp = "hi this is a get test";
- mg_printf(conn, "HTTP/1.1 200 OK\r\nContent-Type: "
- "text/plain\r\nContent-Length: %lu\r\nConnection: close\r\n\r\n",
- site2site_rest_resp.length());
- mg_printf(conn, "%s", site2site_rest_resp.c_str());
- return true;
- }
-};
-
-int main(int argc, char **argv) {
- using org::apache::nifi::minifi::utils::verifyLogLinePresenceInPollTime;
- const cmd_args args = parse_cmdline_args(argc, argv);
-
- LogTestController::getInstance().setDebug<core::Processor>();
- LogTestController::getInstance().setDebug<core::ProcessSession>();
- LogTestController::getInstance().setDebug<utils::HTTPClient>();
- LogTestController::getInstance().setDebug<minifi::controllers::SSLContextService>();
- LogTestController::getInstance().setDebug<minifi::processors::InvokeHTTP>();
- LogTestController::getInstance().setDebug<minifi::processors::LogAttribute>();
-
- std::shared_ptr<minifi::Configure> configuration = std::make_shared<minifi::Configure>();
- configuration->set(minifi::Configure::nifi_default_directory, args.key_dir);
-
- 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, args.test_file);
-
- std::shared_ptr<minifi::io::StreamFactory> stream_factory = minifi::io::StreamFactory::getInstance(configuration);
-
- std::shared_ptr<core::ContentRepository> content_repo = std::make_shared<core::repository::VolatileContentRepository>();
-
- content_repo->initialize(configuration);
-
- std::unique_ptr<core::FlowConfiguration> yaml_ptr = std::unique_ptr<core::YamlConfiguration>(
- new core::YamlConfiguration(test_repo, test_repo, content_repo, stream_factory, configuration, args.test_file));
- 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, configuration, std::move(yaml_ptr), content_repo, DEFAULT_ROOT_GROUP_NAME, true);
-
- core::YamlConfiguration yaml_config(test_repo, test_repo, content_repo, stream_factory, configuration, args.test_file);
-
- std::shared_ptr<core::Processor> proc = yaml_config.getRoot()->findProcessorByName("invoke");
- assert(proc != nullptr);
-
- const auto inv = std::dynamic_pointer_cast<minifi::processors::InvokeHTTP>(proc);
- assert(inv != nullptr);
-
- std::string url;
- inv->getProperty(minifi::processors::InvokeHTTP::URL.getName(), url);
- HttpResponder h_ex;
- std::string port, scheme, path;
- std::unique_ptr<TestServer> server;
- parse_http_components(url, port, scheme, path);
- CivetCallbacks callback{};
- if (scheme == "https") {
- std::string cert;
- cert = args.key_dir + "nifi-cert.pem";
- callback.init_ssl = ssl_enable;
- std::string https_port = port + "s";
- callback.log_message = log_message;
- server = utils::make_unique<TestServer>(https_port, path, &h_ex, &callback, cert, cert);
- } else {
- server = utils::make_unique<TestServer>(port, path, &h_ex);
- }
- controller->load();
- controller->start();
-
- assert(verifyLogLinePresenceInPollTime(
- std::chrono::seconds(10),
- "key:invokehttp.request.url value:" + url,
- "key:invokehttp.status.code value:200",
- "key:flow.id"));
-
- controller->waitUnload(60000);
- if (url.find("localhost") == std::string::npos) {
- server.reset();
- exit(1);
- }
-
- LogTestController::getInstance().reset();
- return 0;
-}
diff --git a/extensions/http-curl/tests/VerifyInvokeHTTP.h b/extensions/http-curl/tests/VerifyInvokeHTTP.h
new file mode 100644
index 0000000..6546173
--- /dev/null
+++ b/extensions/http-curl/tests/VerifyInvokeHTTP.h
@@ -0,0 +1,131 @@
+/**
+ *
+ * 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.
+ */
+#pragma once
+
+#undef NDEBUG
+
+#include <memory>
+#include <utility>
+#include <string>
+
+#include "TestBase.h"
+#include "HTTPClient.h"
+#include "InvokeHTTP.h"
+#include "processors/LogAttribute.h"
+#include "core/state/ProcessorController.h"
+#include "HTTPIntegrationBase.h"
+#include "utils/GeneralUtils.h"
+
+class VerifyInvokeHTTP : public HTTPIntegrationBase {
+ public:
+ VerifyInvokeHTTP()
+ : HTTPIntegrationBase(6000) {
+ }
+
+ void testSetup() override {
+ LogTestController::getInstance().setDebug<utils::HTTPClient>();
+ LogTestController::getInstance().setDebug<LogTestController>();
+ LogTestController::getInstance().setTrace<minifi::processors::InvokeHTTP>();
+ LogTestController::getInstance().setTrace<minifi::processors::LogAttribute>();
+ LogTestController::getInstance().setDebug<core::Processor>();
+ LogTestController::getInstance().setDebug<core::ProcessSession>();
+ }
+
+ void setUrl(const std::string &url, ServerAwareHandler *handler) override {
+ if (path_) {
+ throw std::logic_error("Url is already set");
+ }
+ std::string port, scheme, path;
+ parse_http_components(url, port, scheme, path);
+ path_ = path;
+ HTTPIntegrationBase::setUrl(url, handler);
+ }
+
+ void setProperties(const std::shared_ptr<core::Processor>& proc) {
+ std::string url = scheme + "://localhost:" + getWebPort() + *path_;
+ proc->setProperty(minifi::processors::InvokeHTTP::URL.getName(), url);
+ }
+
+ void setProperty(const std::string& property, const std::string& value) {
+ const auto components = flowController_->getComponents("InvokeHTTP");
+ assert(!components.empty());
+
+ const auto stateController = components[0];
+ assert(stateController);
+ const auto processorController = std::dynamic_pointer_cast<minifi::state::ProcessorController>(stateController);
+ assert(processorController);
+ auto proc = processorController->getProcessor();
+ proc->setProperty(property, value);
+ }
+
+ virtual void setupFlow(const utils::optional<std::string>& flow_yml_path) {
+ testSetup();
+
+ std::shared_ptr<core::Repository> test_repo = std::make_shared<TestRepository>();
+ std::shared_ptr<core::Repository> test_flow_repo = std::make_shared<TestFlowRepository>();
+
+ if (flow_yml_path) {
+ configuration->set(minifi::Configure::nifi_flow_configuration_file, *flow_yml_path);
+ }
+ configuration->set("c2.agent.heartbeat.period", "200");
+ std::shared_ptr<core::ContentRepository> content_repo = std::make_shared<core::repository::VolatileContentRepository>();
+ content_repo->initialize(configuration);
+ std::shared_ptr<minifi::io::StreamFactory> stream_factory = minifi::io::StreamFactory::getInstance(configuration);
+ std::unique_ptr<core::FlowConfiguration> yaml_ptr =
+ minifi::utils::make_unique<core::YamlConfiguration>(test_repo, test_repo, content_repo, stream_factory, configuration, flow_yml_path);
+
+ flowController_ = std::make_shared<minifi::FlowController>(test_repo, test_flow_repo, configuration, std::move(yaml_ptr), content_repo, DEFAULT_ROOT_GROUP_NAME, true);
+ flowController_->load();
+
+ std::string url = scheme + "://localhost:" + getWebPort() + *path_;
+ setProperty(minifi::processors::InvokeHTTP::URL.getName(), url);
+ }
+
+ void run(const utils::optional<std::string>& flow_yml_path = {}, const utils::optional<std::string>& = {}) override {
+ setupFlow(flow_yml_path);
+ startFlowController();
+
+ runAssertions();
+
+ shutdownBeforeFlowController();
+ stopFlowController();
+ }
+
+ void run(const std::string& url,
+ const std::string& test_file_location,
+ const std::string& key_dir,
+ ServerAwareHandler* handler) {
+ setKeyDir(key_dir);
+ setUrl(url, handler);
+ run(test_file_location);
+ }
+
+ void startFlowController() {
+ flowController_->start();
+ }
+
+ void stopFlowController() {
+ flowController_->unload();
+ flowController_->stopC2();
+
+ cleanup();
+ }
+
+ private:
+ utils::optional<std::string> path_;
+};
diff --git a/extensions/http-curl/tests/VerifyInvokeHTTPGetTest.cpp b/extensions/http-curl/tests/VerifyInvokeHTTPGetTest.cpp
new file mode 100644
index 0000000..3af0c06
--- /dev/null
+++ b/extensions/http-curl/tests/VerifyInvokeHTTPGetTest.cpp
@@ -0,0 +1,63 @@
+/**
+ *
+ * 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 "VerifyInvokeHTTP.h"
+
+#include "HTTPHandlers.h"
+#include "utils/IntegrationTestUtils.h"
+
+#define CURLOPT_SSL_VERIFYPEER_DISABLE 1
+
+class VerifyHTTPGet : public VerifyInvokeHTTP {
+ public:
+ virtual void runAssertions() {
+ assert(org::apache::nifi::minifi::utils::verifyLogLinePresenceInPollTime(
+ std::chrono::seconds(10),
+ "key:invokehttp.status.code value:200",
+ "key:flow.id"));
+ }
+};
+
+class VerifyRetryHTTPGet : public VerifyInvokeHTTP {
+ public:
+ void runAssertions() override {
+ assert(org::apache::nifi::minifi::utils::verifyLogLinePresenceInPollTime(
+ std::chrono::seconds(10),
+ "isSuccess: 0, response code 501"));
+ assert(org::apache::nifi::minifi::utils::verifyLogLinePresenceInPollTime(
+ std::chrono::seconds(10),
+ "from InvokeHTTP to relationship retry"));
+ }
+};
+
+int main(int argc, char **argv) {
+ const cmd_args args = parse_cmdline_args(argc, argv);
+
+ {
+ HttpGetResponder http_handler;
+ VerifyHTTPGet harness;
+ harness.run(args.url, args.test_file, args.key_dir, &http_handler);
+ }
+
+ {
+ RetryHttpGetResponder http_handler;
+ VerifyRetryHTTPGet harness;
+ harness.run(args.url, args.test_file, args.key_dir, &http_handler);
+ }
+
+ return 0;
+}
diff --git a/extensions/http-curl/tests/VerifyInvokeHTTPTest.cpp b/extensions/http-curl/tests/VerifyInvokeHTTPPostTest.cpp
similarity index 52%
rename from extensions/http-curl/tests/VerifyInvokeHTTPTest.cpp
rename to extensions/http-curl/tests/VerifyInvokeHTTPPostTest.cpp
index 6fa4713..7090346 100644
--- a/extensions/http-curl/tests/VerifyInvokeHTTPTest.cpp
+++ b/extensions/http-curl/tests/VerifyInvokeHTTPPostTest.cpp
@@ -15,107 +15,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+#include "VerifyInvokeHTTP.h"
-#undef NDEBUG
-#include "TestBase.h"
#include "HTTPHandlers.h"
-#include "HTTPClient.h"
-#include "InvokeHTTP.h"
-#include "processors/LogAttribute.h"
-#include "core/state/ProcessorController.h"
-
-#include "CivetServer.h"
-#include "HTTPIntegrationBase.h"
#include "utils/IntegrationTestUtils.h"
-class VerifyInvokeHTTP : public HTTPIntegrationBase {
- public:
- VerifyInvokeHTTP()
- : HTTPIntegrationBase(6000) {
- }
-
- void testSetup() override {
- LogTestController::getInstance().setDebug<utils::HTTPClient>();
- LogTestController::getInstance().setDebug<LogTestController>();
- LogTestController::getInstance().setTrace<minifi::processors::InvokeHTTP>();
- LogTestController::getInstance().setTrace<minifi::processors::LogAttribute>();
- }
-
- void setUrl(const std::string &url, ServerAwareHandler *handler) override {
- if (path_) {
- throw std::logic_error("Url is already set");
- }
- std::string port, scheme, path;
- parse_http_components(url, port, scheme, path);
- path_ = path;
- HTTPIntegrationBase::setUrl(url, handler);
- }
-
- void setProperties(std::shared_ptr<core::Processor> proc) {
- std::string url = scheme + "://localhost:" + getWebPort() + *path_;
- proc->setProperty(minifi::processors::InvokeHTTP::URL.getName(), url);
- }
-
- void setProperty(const std::string& property, const std::string& value) {
- const auto components = flowController_->getComponents("InvokeHTTP");
- assert(!components.empty());
-
- const auto stateController = components.at(0);
- assert(stateController);
- const auto processorController = std::dynamic_pointer_cast<minifi::state::ProcessorController>(stateController);
- assert(processorController);
- auto proc = processorController->getProcessor();
- proc->setProperty(property, value);
- }
-
- virtual void setupFlow(const utils::optional<std::string>& flow_yml_path) {
- testSetup();
-
- std::shared_ptr<core::Repository> test_repo = std::make_shared<TestRepository>();
- std::shared_ptr<core::Repository> test_flow_repo = std::make_shared<TestFlowRepository>();
-
- if (flow_yml_path) {
- configuration->set(minifi::Configure::nifi_flow_configuration_file, *flow_yml_path);
- }
- configuration->set("c2.agent.heartbeat.period", "200");
- std::shared_ptr<core::ContentRepository> content_repo = std::make_shared<core::repository::VolatileContentRepository>();
- content_repo->initialize(configuration);
- std::shared_ptr<minifi::io::StreamFactory> stream_factory = minifi::io::StreamFactory::getInstance(configuration);
- std::unique_ptr<core::FlowConfiguration> yaml_ptr = std::unique_ptr<core::YamlConfiguration>(
- new core::YamlConfiguration(test_repo, test_repo, content_repo, stream_factory, configuration, flow_yml_path));
-
- flowController_ = std::make_shared<minifi::FlowController>(test_repo, test_flow_repo, configuration, std::move(yaml_ptr), content_repo, DEFAULT_ROOT_GROUP_NAME, true);
- flowController_->load();
-
- std::string url = scheme + "://localhost:" + getWebPort() + *path_;
- setProperty(minifi::processors::InvokeHTTP::URL.getName(), url);
- }
-
- void run(const utils::optional<std::string>& flow_yml_path = {}, const utils::optional<std::string>& = {}) override {
- setupFlow(flow_yml_path);
- startFlowController();
-
- runAssertions();
-
- shutdownBeforeFlowController();
- stopFlowController();
- }
-
- void startFlowController() {
- flowController_->start();
- }
-
- void stopFlowController() {
- flowController_->unload();
- flowController_->stopC2();
-
- cleanup();
- }
-
- private:
- utils::optional<std::string> path_;
-};
-
class VerifyInvokeHTTPOKResponse : public VerifyInvokeHTTP {
public:
void runAssertions() override {
@@ -189,17 +93,6 @@ class VerifyRWTimeoutInvokeHTTP : public VerifyInvokeHTTP {
}
};
-void run(VerifyInvokeHTTP& harness,
- const std::string& url,
- const std::string& test_file_location,
- const std::string& key_dir,
- ServerAwareHandler * handler) {
-
- harness.setKeyDir(key_dir);
- harness.setUrl(url, handler);
- harness.run(test_file_location);
-}
-
int main(int argc, char ** argv) {
const cmd_args args = parse_cmdline_args(argc, argv);
@@ -220,37 +113,37 @@ int main(int argc, char ** argv) {
{
InvokeHTTPResponseOKHandler handler;
VerifyInvokeHTTPOKResponse harness;
- run(harness, args.url, args.test_file, args.key_dir, &handler);
+ harness.run(args.url, args.test_file, args.key_dir, &handler);
}
{
InvokeHTTPRedirectHandler handler;
VerifyInvokeHTTPOK200Response harness;
- run(harness, args.url, args.test_file, args.key_dir, &handler);
+ harness.run(args.url, args.test_file, args.key_dir, &handler);
}
{
InvokeHTTPRedirectHandler handler;
VerifyInvokeHTTPRedirectResponse harness;
- run(harness, args.url, args.test_file, args.key_dir, &handler);
+ harness.run(args.url, args.test_file, args.key_dir, &handler);
}
{
InvokeHTTPResponse404Handler handler;
VerifyNoRetryInvokeHTTP harness;
- run(harness, args.url, args.test_file, args.key_dir, &handler);
+ harness.run(args.url, args.test_file, args.key_dir, &handler);
}
{
InvokeHTTPResponse501Handler handler;
VerifyRetryInvokeHTTP harness;
- run(harness, args.url, args.test_file, args.key_dir, &handler);
+ harness.run(args.url, args.test_file, args.key_dir, &handler);
}
{
TimeoutingHTTPHandler handler({std::chrono::seconds(2)});
VerifyRWTimeoutInvokeHTTP harness;
- run(harness, args.url, args.test_file, args.key_dir, &handler);
+ harness.run(args.url, args.test_file, args.key_dir, &handler);
}
return 0;
diff --git a/libminifi/test/resources/TestHTTPGet.yml b/libminifi/test/resources/TestHTTPGet.yml
index fc4a07f..bfddd3c 100644
--- a/libminifi/test/resources/TestHTTPGet.yml
+++ b/libminifi/test/resources/TestHTTPGet.yml
@@ -20,7 +20,7 @@ Flow Controller:
name: MiNiFi Flow
id: 2438e3c8-015a-1000-79ca-83af40ec1990
Processors:
- - name: invoke
+ - name: InvokeHTTP
id: 2438e3c8-015a-1000-79ca-83af40ec1991
class: org.apache.nifi.processors.standard.InvokeHTTP
max concurrent tasks: 1
@@ -30,7 +30,6 @@ Processors:
yield period: 1 sec
run duration nanos: 0
auto-terminated relationships list:
- - retry
- no retry
- response
- failure
@@ -54,7 +53,7 @@ Processors:
Connections:
- name: TransferFilesToRPG
id: 2438e3c8-015a-1000-79ca-83af40ec1997
- source name: invoke
+ source name: InvokeHTTP
source id: 2438e3c8-015a-1000-79ca-83af40ec1991
source relationship name: success
destination name: LogAttribute
@@ -67,11 +66,22 @@ Connections:
source name: LogAttribute
source id: 2438e3c8-015a-1000-79ca-83af40ec1992
destination name: LogAttribute
- destination id: 2438e3c8-015a-1000-79ca-83af40ec1992
+ 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
+ - name: RetryInvokeHTTP
+ id: c8a96f1d-48cd-4ee3-8a2f-081c3e3c7bcd
+ source name: InvokeHTTP
+ source id: 2438e3c8-015a-1000-79ca-83af40ec1991
+ source relationship names: retry
+ destination name: InvokeHTTP
+ destination id: 2438e3c8-015a-1000-79ca-83af40ec1991
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 seconds
+
Remote Processing Groups:
-
+
diff --git a/libminifi/test/resources/TestHTTPGetSecure.yml b/libminifi/test/resources/TestHTTPGetSecure.yml
index d90770d..194c4ff 100644
--- a/libminifi/test/resources/TestHTTPGetSecure.yml
+++ b/libminifi/test/resources/TestHTTPGetSecure.yml
@@ -20,7 +20,7 @@ Flow Controller:
name: MiNiFi Flow
id: 2438e3c8-015a-1000-79ca-83af40ec1990
Processors:
- - name: invoke
+ - name: InvokeHTTP
id: 2438e3c8-015a-1000-79ca-83af40ec1991
class: org.apache.nifi.processors.standard.InvokeHTTP
max concurrent tasks: 1
@@ -31,7 +31,6 @@ Processors:
run duration nanos: 0
auto-terminated relationships list:
- failure
- - retry
- no retry
- response
Properties:
@@ -56,7 +55,7 @@ Processors:
Connections:
- name: TransferFilesToRPG
id: 2438e3c8-015a-1000-79ca-83af40ec1997
- source name: invoke
+ source name: InvokeHTTP
source id: 2438e3c8-015a-1000-79ca-83af40ec1991
source relationship name: success
destination name: LogAttribute
@@ -69,11 +68,21 @@ Connections:
source name: LogAttribute
source id: 2438e3c8-015a-1000-79ca-83af40ec1992
destination name: LogAttribute
- destination id: 2438e3c8-015a-1000-79ca-83af40ec1992
+ 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
+ - name: RetryInvokeHTTP
+ id: c8a96f1d-48cd-4ee3-8a2f-081c3e3c7bcd
+ source name: InvokeHTTP
+ source id: 2438e3c8-015a-1000-79ca-83af40ec1991
+ source relationship names: retry
+ destination name: InvokeHTTP
+ destination id: 2438e3c8-015a-1000-79ca-83af40ec1991
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 seconds
Controller Services:
- name: SSLContextService
@@ -90,4 +99,4 @@ Controller Services:
- value: nifi-cert.pem
Remote Processing Groups:
-
+
diff --git a/libminifi/test/resources/TestHTTPPostChunkedEncoding.yml b/libminifi/test/resources/TestHTTPPostChunkedEncoding.yml
index 993a289..532771e 100644
--- a/libminifi/test/resources/TestHTTPPostChunkedEncoding.yml
+++ b/libminifi/test/resources/TestHTTPPostChunkedEncoding.yml
@@ -48,12 +48,11 @@ Processors:
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
+ scheduling strategy: EVENT_DRIVEN
penalization period: 30 sec
yield period: 1 sec
run duration nanos: 0
- auto-terminated relationships list:
+ auto-terminated relationships list:
- success
- retry
- failure
@@ -68,17 +67,16 @@ Processors:
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
+ scheduling strategy: EVENT_DRIVEN
penalization period: 30 sec
yield period: 1 sec
run duration nanos: 0
- auto-terminated relationships list:
+ auto-terminated relationships list:
- success
Properties:
LogLevel: debug
-Connections:
+Connections:
- name: GenerateFlowFile/Invoke
id: 2438e3c8-015a-1000-79ca-83af40ec1997
source name: invoke
@@ -98,4 +96,4 @@ Connections:
max work queue data size: 1 MB
flowfile expiration: 60 sec
Remote Processing Groups:
-
+