You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by GitBox <gi...@apache.org> on 2021/04/27 15:26:21 UTC
[GitHub] [nifi-minifi-cpp] fgerlits opened a new pull request #1064: MINIFICPP-987 Improve error handling in GetFile
fgerlits opened a new pull request #1064:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1064
https://issues.apache.org/jira/browse/MINIFICPP-987
* Get rid of the access() calls and catch file read exceptions inside the loop in onTrigger().
* Get rid of the ProcessSession::import() call and replace it with write().
* Remove the unused FileOutputCallback class.
* Log file removal failures. In this case, we will keep reading and sending the file, but that is better than not sending some files.
* Add a few more unit tests.
---
Thank you for submitting a contribution to Apache NiFi - MiNiFi C++.
In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:
### For all changes:
- [x] Is there a JIRA ticket associated with this PR? Is it referenced
in the commit message?
- [x] Does your PR title start with MINIFICPP-XXXX where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.
- [x] Has your PR been rebased against the latest commit within the target branch (typically main)?
- [x] Is your initial contribution a single, squashed commit?
### For code changes:
- [ ] If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under [ASF 2.0](http://www.apache.org/legal/resolved.html#category-a)?
- [ ] If applicable, have you updated the LICENSE file?
- [ ] If applicable, have you updated the NOTICE file?
### For documentation related changes:
- [ ] Have you ensured that format looks appropriate for the output in which it is rendered?
### Note:
Please ensure that once the PR is submitted, you check GitHub Actions CI results for build issues and submit an update to your PR as soon as possible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #1064: MINIFICPP-987 Improve error handling in GetFile
Posted by GitBox <gi...@apache.org>.
adamdebreceni commented on a change in pull request #1064:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1064#discussion_r631033715
##########
File path: extensions/standard-processors/processors/GetFile.cpp
##########
@@ -158,76 +147,89 @@ void GetFile::onSchedule(core::ProcessContext *context, core::ProcessSessionFact
request_.inputDirectory = value;
}
-void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession *session) {
- // Perform directory list
-
+void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession* session) {
metrics_->iterations_++;
- const bool isDirEmptyBeforePoll = isListingEmpty();
- logger_->log_debug("Is listing empty before polling directory %i", isDirEmptyBeforePoll);
- if (isDirEmptyBeforePoll) {
+ const bool is_dir_empty_before_poll = isListingEmpty();
+ logger_->log_debug("Listing is %s before polling directory", is_dir_empty_before_poll ? "empty" : "not empty");
+ if (is_dir_empty_before_poll) {
if (request_.pollInterval == 0 || (utils::timeutils::getTimeMillis() - last_listing_time_) > request_.pollInterval) {
performListing(request_);
last_listing_time_.store(utils::timeutils::getTimeMillis());
}
}
- const bool isDirEmptyAfterPoll = isListingEmpty();
- logger_->log_debug("Is listing empty after polling directory %i", isDirEmptyAfterPoll);
-
- if (!isDirEmptyAfterPoll) {
- try {
- std::queue<std::string> list;
- pollListing(list, request_);
- while (!list.empty()) {
- std::string fileName = list.front();
- list.pop();
- logger_->log_info("GetFile process %s", fileName);
- auto flowFile = session->create();
- if (flowFile == nullptr)
- return;
- std::size_t found = fileName.find_last_of("/\\");
- std::string path = fileName.substr(0, found);
- std::string name = fileName.substr(found + 1);
- flowFile->setAttribute(core::SpecialFlowAttribute::FILENAME, name);
- flowFile->setAttribute(core::SpecialFlowAttribute::PATH, path);
- flowFile->addAttribute(core::SpecialFlowAttribute::ABSOLUTE_PATH, fileName);
- session->import(fileName, flowFile, request_.keepSourceFile);
- session->transfer(flowFile, Success);
+ const bool is_dir_empty_after_poll = isListingEmpty();
+ logger_->log_debug("Listing is %s after polling directory", is_dir_empty_after_poll ? "empty" : "not empty");
+ if (is_dir_empty_after_poll) {
+ yield();
+ return;
+ }
+
+ std::queue<std::string> list_of_file_names = pollListing(request_.batchSize);
+ while (!list_of_file_names.empty()) {
+ std::string file_name = list_of_file_names.front();
+ list_of_file_names.pop();
+ getSingleFile(*session, file_name);
+ }
+}
+
+void GetFile::getSingleFile(core::ProcessSession& session, const std::string& file_name) const {
+ logger_->log_info("GetFile process %s", file_name);
+ auto flow_file = session.create();
+ gsl_Expects(flow_file);
+ std::string path;
+ std::string name;
+ std::tie(path, name) = utils::file::split_path(file_name);
+ flow_file->setAttribute(core::SpecialFlowAttribute::FILENAME, name);
+ flow_file->setAttribute(core::SpecialFlowAttribute::PATH, path);
+ flow_file->addAttribute(core::SpecialFlowAttribute::ABSOLUTE_PATH, file_name);
+
+ try {
+ utils::FileReaderCallback file_reader_callback{file_name};
+ session.write(flow_file, &file_reader_callback);
+ session.transfer(flow_file, Success);
+ if (!request_.keepSourceFile) {
+ auto remove_status = remove(file_name.c_str());
+ if (remove_status != 0) {
+ logger_->log_error("GetFile could not delete file '%s', error %d: %s", file_name, errno, strerror(errno));
}
- } catch (std::exception &exception) {
- logger_->log_debug("GetFile Caught Exception %s", exception.what());
- throw;
- } catch (...) {
- throw;
}
+ } catch (const std::exception& exception) {
Review comment:
we could limit the kinds of exceptions we consider "recoverable" (i.e. discard that specific flowfile, but continue processor execution) we could go for `std::ios_base::failure`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] szaszm commented on pull request #1064: MINIFICPP-987 Improve error handling in GetFile
Posted by GitBox <gi...@apache.org>.
szaszm commented on pull request #1064:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1064#issuecomment-852872338
Should've been closed by 34603d0, but didn't close due to a github glitch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] szaszm closed pull request #1064: MINIFICPP-987 Improve error handling in GetFile
Posted by GitBox <gi...@apache.org>.
szaszm closed pull request #1064:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1064
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] fgerlits commented on a change in pull request #1064: MINIFICPP-987 Improve error handling in GetFile
Posted by GitBox <gi...@apache.org>.
fgerlits commented on a change in pull request #1064:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1064#discussion_r634441394
##########
File path: libminifi/src/utils/FileReaderCallback.cpp
##########
@@ -0,0 +1,72 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "utils/FileReaderCallback.h"
+
+#include <cinttypes>
+
+#include "Exception.h"
+#include "core/logging/LoggerConfiguration.h"
+
+namespace {
+
+constexpr std::size_t BUFFER_SIZE = 4096;
+
+} // namespace
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace utils {
+
+FileReaderCallback::FileReaderCallback(const std::string& file_name)
+ : logger_(core::logging::LoggerFactory<FileReaderCallback>::getLogger()) {
+ logger_->log_debug("Opening %s", file_name);
+ input_stream_.open(file_name.c_str(), std::fstream::in | std::fstream::binary);
+ if (!input_stream_.is_open()) {
+ throw Exception(FILE_OPERATION_EXCEPTION, "Could not open file: " + file_name);
+ }
+}
+
+int64_t FileReaderCallback::process(const std::shared_ptr<io::BaseStream>& output_stream) {
+ std::array<char, BUFFER_SIZE> buffer;
+
+ uint64_t num_bytes_written = 0;
+
+ while (input_stream_.good()) {
+ input_stream_.read(buffer.data(), buffer.size());
+
+ const auto num_bytes_read = input_stream_.gcount();
+ logger_->log_trace("Read %jd bytes of input", std::intmax_t{num_bytes_read});
+
+ const int len = gsl::narrow<int>(num_bytes_read);
+
+ output_stream->write(reinterpret_cast<uint8_t*>(buffer.data()), len);
+ num_bytes_written += len;
+ }
+ input_stream_.close();
+
+ logger_->log_debug("Finished reading %" PRIu64 " bytes from the file", num_bytes_written);
+ return num_bytes_written;
Review comment:
There are problems with `exceptions()`, mainly that `read()` will set the failbit when it reads the last chunk of the file. I ended up following the advice of https://gehrcke.de/2011/06/reading-files-in-c-using-ifstream-dealing-correctly-with-badbit-failbit-eofbit-and-perror/ in https://github.com/apache/nifi-minifi-cpp/pull/1064/commits/0ea95c9fd071ab701a730c29abed60806b171b34 (with some modifications because `read()` and `getline()` work slightly differently).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] fgerlits commented on a change in pull request #1064: MINIFICPP-987 Improve error handling in GetFile
Posted by GitBox <gi...@apache.org>.
fgerlits commented on a change in pull request #1064:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1064#discussion_r629371618
##########
File path: extensions/standard-processors/tests/unit/GetFileTests.cpp
##########
@@ -91,12 +113,30 @@ TEST_CASE("GetFile: MaxSize", "[getFileFifo]") { // NOLINT
#endif
}
-
-TEST_CASE("GetFile: Directory", "[getFileDir]") {
- TestController testController;
+TEST_CASE("GetFile onSchedule() throws if the required Directory property is not set", "[GetFile]") {
+ TestController test_controller;
LogTestController::getInstance().setTrace<TestPlan>();
LogTestController::getInstance().setTrace<processors::GetFile>();
- auto plan = testController.createPlan();
+ auto plan = test_controller.createPlan();
auto get_file = plan->addProcessor("GetFile", "Get");
REQUIRE_THROWS_AS(plan->runNextProcessor(), minifi::Exception&);
}
+
+TEST_CASE("GetFile removes the source file if KeepSourceFile is false") {
+ GetFileTestController test_controller;
+ SECTION("KeepSourceFile is not set, so defaults to false") {}
+ SECTION("KeepSourceFile is set to false explicitly") { test_controller.setProperty(processors::GetFile::KeepSourceFile, "false"); }
+
+ test_controller.runSession();
+
+ REQUIRE_FALSE(utils::file::FileUtils::exists(test_controller.input_file_name_));
+}
+
+TEST_CASE("GetFile keeps the source file if KeepSourceFile is true") {
+ GetFileTestController test_controller;
+ test_controller.setProperty(processors::GetFile::KeepSourceFile, "true");
+
+ test_controller.runSession();
+
+ REQUIRE(utils::file::FileUtils::exists(test_controller.input_file_name_));
+}
Review comment:
I have created a new Jira for this: https://issues.apache.org/jira/browse/MINIFICPP-1555
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] fgerlits commented on a change in pull request #1064: MINIFICPP-987 Improve error handling in GetFile
Posted by GitBox <gi...@apache.org>.
fgerlits commented on a change in pull request #1064:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1064#discussion_r630911063
##########
File path: extensions/standard-processors/processors/GetFile.cpp
##########
@@ -158,76 +147,89 @@ void GetFile::onSchedule(core::ProcessContext *context, core::ProcessSessionFact
request_.inputDirectory = value;
}
-void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession *session) {
- // Perform directory list
-
+void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession* session) {
metrics_->iterations_++;
- const bool isDirEmptyBeforePoll = isListingEmpty();
- logger_->log_debug("Is listing empty before polling directory %i", isDirEmptyBeforePoll);
- if (isDirEmptyBeforePoll) {
+ const bool is_dir_empty_before_poll = isListingEmpty();
+ logger_->log_debug("Listing is %s before polling directory", is_dir_empty_before_poll ? "empty" : "not empty");
+ if (is_dir_empty_before_poll) {
if (request_.pollInterval == 0 || (utils::timeutils::getTimeMillis() - last_listing_time_) > request_.pollInterval) {
performListing(request_);
last_listing_time_.store(utils::timeutils::getTimeMillis());
}
}
- const bool isDirEmptyAfterPoll = isListingEmpty();
- logger_->log_debug("Is listing empty after polling directory %i", isDirEmptyAfterPoll);
-
- if (!isDirEmptyAfterPoll) {
- try {
- std::queue<std::string> list;
- pollListing(list, request_);
- while (!list.empty()) {
- std::string fileName = list.front();
- list.pop();
- logger_->log_info("GetFile process %s", fileName);
- auto flowFile = session->create();
- if (flowFile == nullptr)
- return;
- std::size_t found = fileName.find_last_of("/\\");
- std::string path = fileName.substr(0, found);
- std::string name = fileName.substr(found + 1);
- flowFile->setAttribute(core::SpecialFlowAttribute::FILENAME, name);
- flowFile->setAttribute(core::SpecialFlowAttribute::PATH, path);
- flowFile->addAttribute(core::SpecialFlowAttribute::ABSOLUTE_PATH, fileName);
- session->import(fileName, flowFile, request_.keepSourceFile);
- session->transfer(flowFile, Success);
+ const bool is_dir_empty_after_poll = isListingEmpty();
+ logger_->log_debug("Listing is %s after polling directory", is_dir_empty_after_poll ? "empty" : "not empty");
+ if (is_dir_empty_after_poll) {
+ yield();
+ return;
+ }
+
+ std::queue<std::string> list_of_file_names = pollListing(request_.batchSize);
+ while (!list_of_file_names.empty()) {
+ std::string file_name = list_of_file_names.front();
+ list_of_file_names.pop();
+ getSingleFile(*session, file_name);
+ }
+}
+
+void GetFile::getSingleFile(core::ProcessSession& session, const std::string& file_name) const {
+ logger_->log_info("GetFile process %s", file_name);
+ auto flow_file = session.create();
+ gsl_Expects(flow_file);
+ std::string path;
+ std::string name;
+ std::tie(path, name) = utils::file::split_path(file_name);
+ flow_file->setAttribute(core::SpecialFlowAttribute::FILENAME, name);
+ flow_file->setAttribute(core::SpecialFlowAttribute::PATH, path);
+ flow_file->addAttribute(core::SpecialFlowAttribute::ABSOLUTE_PATH, file_name);
+
+ try {
+ utils::FileReaderCallback file_reader_callback{file_name};
+ session.write(flow_file, &file_reader_callback);
+ session.transfer(flow_file, Success);
+ if (!request_.keepSourceFile) {
+ auto remove_status = remove(file_name.c_str());
+ if (remove_status != 0) {
+ logger_->log_error("GetFile could not delete file '%s', error %d: %s", file_name, errno, strerror(errno));
}
- } catch (std::exception &exception) {
- logger_->log_debug("GetFile Caught Exception %s", exception.what());
- throw;
- } catch (...) {
- throw;
}
+ } catch (const std::exception& exception) {
+ logger_->log_error("GetFile caught exception while processing file '%s': %s", file_name, exception.what());
+ flow_file->setDeleted(true);
Review comment:
We don't want to roll back if reading one file failed but another file succeeded; in this case, we want to ignore the failed file but process the successful one.
My understanding is that every flow file we create with `session->create()` has to either be transferred to an output relationship or marked as deleted, otherwise `routeFlowFile(flow_file)`, and therefore `session->commit()`, will throw.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] arpadboda commented on a change in pull request #1064: MINIFICPP-987 Improve error handling in GetFile
Posted by GitBox <gi...@apache.org>.
arpadboda commented on a change in pull request #1064:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1064#discussion_r630885656
##########
File path: extensions/standard-processors/processors/GetFile.cpp
##########
@@ -158,76 +147,89 @@ void GetFile::onSchedule(core::ProcessContext *context, core::ProcessSessionFact
request_.inputDirectory = value;
}
-void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession *session) {
- // Perform directory list
-
+void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession* session) {
metrics_->iterations_++;
- const bool isDirEmptyBeforePoll = isListingEmpty();
- logger_->log_debug("Is listing empty before polling directory %i", isDirEmptyBeforePoll);
- if (isDirEmptyBeforePoll) {
+ const bool is_dir_empty_before_poll = isListingEmpty();
+ logger_->log_debug("Listing is %s before polling directory", is_dir_empty_before_poll ? "empty" : "not empty");
+ if (is_dir_empty_before_poll) {
if (request_.pollInterval == 0 || (utils::timeutils::getTimeMillis() - last_listing_time_) > request_.pollInterval) {
performListing(request_);
last_listing_time_.store(utils::timeutils::getTimeMillis());
}
}
- const bool isDirEmptyAfterPoll = isListingEmpty();
- logger_->log_debug("Is listing empty after polling directory %i", isDirEmptyAfterPoll);
-
- if (!isDirEmptyAfterPoll) {
- try {
- std::queue<std::string> list;
- pollListing(list, request_);
- while (!list.empty()) {
- std::string fileName = list.front();
- list.pop();
- logger_->log_info("GetFile process %s", fileName);
- auto flowFile = session->create();
- if (flowFile == nullptr)
- return;
- std::size_t found = fileName.find_last_of("/\\");
- std::string path = fileName.substr(0, found);
- std::string name = fileName.substr(found + 1);
- flowFile->setAttribute(core::SpecialFlowAttribute::FILENAME, name);
- flowFile->setAttribute(core::SpecialFlowAttribute::PATH, path);
- flowFile->addAttribute(core::SpecialFlowAttribute::ABSOLUTE_PATH, fileName);
- session->import(fileName, flowFile, request_.keepSourceFile);
- session->transfer(flowFile, Success);
+ const bool is_dir_empty_after_poll = isListingEmpty();
+ logger_->log_debug("Listing is %s after polling directory", is_dir_empty_after_poll ? "empty" : "not empty");
+ if (is_dir_empty_after_poll) {
+ yield();
+ return;
+ }
+
+ std::queue<std::string> list_of_file_names = pollListing(request_.batchSize);
+ while (!list_of_file_names.empty()) {
+ std::string file_name = list_of_file_names.front();
+ list_of_file_names.pop();
+ getSingleFile(*session, file_name);
+ }
+}
+
+void GetFile::getSingleFile(core::ProcessSession& session, const std::string& file_name) const {
+ logger_->log_info("GetFile process %s", file_name);
+ auto flow_file = session.create();
+ gsl_Expects(flow_file);
+ std::string path;
+ std::string name;
+ std::tie(path, name) = utils::file::split_path(file_name);
+ flow_file->setAttribute(core::SpecialFlowAttribute::FILENAME, name);
+ flow_file->setAttribute(core::SpecialFlowAttribute::PATH, path);
+ flow_file->addAttribute(core::SpecialFlowAttribute::ABSOLUTE_PATH, file_name);
+
+ try {
+ utils::FileReaderCallback file_reader_callback{file_name};
+ session.write(flow_file, &file_reader_callback);
+ session.transfer(flow_file, Success);
+ if (!request_.keepSourceFile) {
+ auto remove_status = remove(file_name.c_str());
+ if (remove_status != 0) {
+ logger_->log_error("GetFile could not delete file '%s', error %d: %s", file_name, errno, strerror(errno));
}
- } catch (std::exception &exception) {
- logger_->log_debug("GetFile Caught Exception %s", exception.what());
- throw;
- } catch (...) {
- throw;
}
+ } catch (const std::exception& exception) {
+ logger_->log_error("GetFile caught exception while processing file '%s': %s", file_name, exception.what());
+ flow_file->setDeleted(true);
Review comment:
Do we need this?
I would expect session->rollback() later to clean up all the "mess" we made in an interrupted onTrigger
##########
File path: extensions/standard-processors/processors/GetFile.cpp
##########
@@ -158,76 +147,89 @@ void GetFile::onSchedule(core::ProcessContext *context, core::ProcessSessionFact
request_.inputDirectory = value;
}
-void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession *session) {
- // Perform directory list
-
+void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession* session) {
metrics_->iterations_++;
- const bool isDirEmptyBeforePoll = isListingEmpty();
- logger_->log_debug("Is listing empty before polling directory %i", isDirEmptyBeforePoll);
- if (isDirEmptyBeforePoll) {
+ const bool is_dir_empty_before_poll = isListingEmpty();
+ logger_->log_debug("Listing is %s before polling directory", is_dir_empty_before_poll ? "empty" : "not empty");
Review comment:
Nitpicking: "empty" can be part of the format string, so this line can be 5 chars shorter :))
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a change in pull request #1064: MINIFICPP-987 Improve error handling in GetFile
Posted by GitBox <gi...@apache.org>.
lordgamez commented on a change in pull request #1064:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1064#discussion_r622308397
##########
File path: extensions/standard-processors/tests/unit/GetFileTests.cpp
##########
@@ -91,12 +113,30 @@ TEST_CASE("GetFile: MaxSize", "[getFileFifo]") { // NOLINT
#endif
}
-
-TEST_CASE("GetFile: Directory", "[getFileDir]") {
- TestController testController;
+TEST_CASE("GetFile onSchedule() throws if the required Directory property is not set", "[GetFile]") {
+ TestController test_controller;
LogTestController::getInstance().setTrace<TestPlan>();
LogTestController::getInstance().setTrace<processors::GetFile>();
- auto plan = testController.createPlan();
+ auto plan = test_controller.createPlan();
auto get_file = plan->addProcessor("GetFile", "Get");
REQUIRE_THROWS_AS(plan->runNextProcessor(), minifi::Exception&);
}
+
+TEST_CASE("GetFile removes the source file if KeepSourceFile is false") {
+ GetFileTestController test_controller;
+ SECTION("KeepSourceFile is not set, so defaults to false") {}
+ SECTION("KeepSourceFile is set to false explicitly") { test_controller.setProperty(processors::GetFile::KeepSourceFile, "false"); }
+
+ test_controller.runSession();
+
+ REQUIRE_FALSE(utils::file::FileUtils::exists(test_controller.input_file_name_));
+}
+
+TEST_CASE("GetFile keeps the source file if KeepSourceFile is true") {
+ GetFileTestController test_controller;
+ test_controller.setProperty(processors::GetFile::KeepSourceFile, "true");
+
+ test_controller.runSession();
+
+ REQUIRE(utils::file::FileUtils::exists(test_controller.input_file_name_));
+}
Review comment:
There are some uncovered properties in the test suite like Recurse, *Age, BatchSize, FileFilter etc. We could add some tests for those, or as it is not really part of this issue, have it in separate PR with a newly opened issue.
##########
File path: extensions/standard-processors/processors/GetFile.cpp
##########
@@ -159,47 +148,54 @@ void GetFile::onSchedule(core::ProcessContext *context, core::ProcessSessionFact
}
void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession *session) {
- // Perform directory list
-
metrics_->iterations_++;
- const bool isDirEmptyBeforePoll = isListingEmpty();
- logger_->log_debug("Is listing empty before polling directory %i", isDirEmptyBeforePoll);
- if (isDirEmptyBeforePoll) {
+ const bool is_dir_empty_before_poll = isListingEmpty();
+ logger_->log_debug("Listing is %s before polling directory", is_dir_empty_before_poll ? "empty" : "not empty");
+ if (is_dir_empty_before_poll) {
if (request_.pollInterval == 0 || (utils::timeutils::getTimeMillis() - last_listing_time_) > request_.pollInterval) {
performListing(request_);
last_listing_time_.store(utils::timeutils::getTimeMillis());
}
}
- const bool isDirEmptyAfterPoll = isListingEmpty();
- logger_->log_debug("Is listing empty after polling directory %i", isDirEmptyAfterPoll);
+ const bool is_dir_empty_after_poll = isListingEmpty();
+ logger_->log_debug("Listing is %s after polling directory", is_dir_empty_after_poll ? "empty" : "not empty");
+ if (is_dir_empty_after_poll) {
+ yield();
+ return;
+ }
+
+ std::queue<std::string> list_of_file_names = pollListing(request_.batchSize);
+ while (!list_of_file_names.empty()) {
Review comment:
Minor: I may extract the core of the loop as a separate function to keep onTrigger shorter and more clear.
##########
File path: libminifi/src/utils/FileReaderCallback.cpp
##########
@@ -0,0 +1,72 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "utils/FileReaderCallback.h"
+
+#include <cinttypes>
+
+#include "Exception.h"
+#include "core/logging/LoggerConfiguration.h"
+
+namespace {
+
+constexpr std::size_t BUFFER_SIZE = 4096;
+
+} // namespace
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace utils {
+
+FileReaderCallback::FileReaderCallback(const std::string& file_name)
+ : logger_(core::logging::LoggerFactory<FileReaderCallback>::getLogger()) {
+ logger_->log_debug("Opening %s", file_name);
+ input_stream_.open(file_name.c_str(), std::fstream::in | std::fstream::binary);
+ if (!input_stream_.is_open() || !input_stream_.good()) {
Review comment:
Nit: Could the `input_stream_.good()` be enough in this condition? Or is it possible that we get a good input stream, but the `is_open` returns false?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] fgerlits commented on a change in pull request #1064: MINIFICPP-987 Improve error handling in GetFile
Posted by GitBox <gi...@apache.org>.
fgerlits commented on a change in pull request #1064:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1064#discussion_r629374185
##########
File path: libminifi/src/utils/FileReaderCallback.cpp
##########
@@ -0,0 +1,72 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "utils/FileReaderCallback.h"
+
+#include <cinttypes>
+
+#include "Exception.h"
+#include "core/logging/LoggerConfiguration.h"
+
+namespace {
+
+constexpr std::size_t BUFFER_SIZE = 4096;
+
+} // namespace
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace utils {
+
+FileReaderCallback::FileReaderCallback(const std::string& file_name)
+ : logger_(core::logging::LoggerFactory<FileReaderCallback>::getLogger()) {
+ logger_->log_debug("Opening %s", file_name);
+ input_stream_.open(file_name.c_str(), std::fstream::in | std::fstream::binary);
+ if (!input_stream_.is_open() || !input_stream_.good()) {
Review comment:
I copied this mindlessly from `ProcessSession`, but I think you're right, `is_open()` and `good()` are the same immediately after `open()`, so we were checking it twice. I have removed the `good()` check, because I think `is_open()` reads better in combination with `open()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #1064: MINIFICPP-987 Improve error handling in GetFile
Posted by GitBox <gi...@apache.org>.
adamdebreceni commented on a change in pull request #1064:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1064#discussion_r631034571
##########
File path: libminifi/src/utils/FileReaderCallback.cpp
##########
@@ -0,0 +1,72 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "utils/FileReaderCallback.h"
+
+#include <cinttypes>
+
+#include "Exception.h"
+#include "core/logging/LoggerConfiguration.h"
+
+namespace {
+
+constexpr std::size_t BUFFER_SIZE = 4096;
+
+} // namespace
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace utils {
+
+FileReaderCallback::FileReaderCallback(const std::string& file_name)
+ : logger_(core::logging::LoggerFactory<FileReaderCallback>::getLogger()) {
+ logger_->log_debug("Opening %s", file_name);
+ input_stream_.open(file_name.c_str(), std::fstream::in | std::fstream::binary);
+ if (!input_stream_.is_open()) {
+ throw Exception(FILE_OPERATION_EXCEPTION, "Could not open file: " + file_name);
+ }
+}
+
+int64_t FileReaderCallback::process(const std::shared_ptr<io::BaseStream>& output_stream) {
+ std::array<char, BUFFER_SIZE> buffer;
+
+ uint64_t num_bytes_written = 0;
+
+ while (input_stream_.good()) {
+ input_stream_.read(buffer.data(), buffer.size());
+
+ const auto num_bytes_read = input_stream_.gcount();
+ logger_->log_trace("Read %jd bytes of input", std::intmax_t{num_bytes_read});
+
+ const int len = gsl::narrow<int>(num_bytes_read);
+
+ output_stream->write(reinterpret_cast<uint8_t*>(buffer.data()), len);
+ num_bytes_written += len;
+ }
+ input_stream_.close();
+
+ logger_->log_debug("Finished reading %" PRIu64 " bytes from the file", num_bytes_written);
+ return num_bytes_written;
Review comment:
do we set the stream somewhere to throw on `failure`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] fgerlits commented on a change in pull request #1064: MINIFICPP-987 Improve error handling in GetFile
Posted by GitBox <gi...@apache.org>.
fgerlits commented on a change in pull request #1064:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1064#discussion_r634437724
##########
File path: extensions/standard-processors/processors/GetFile.cpp
##########
@@ -158,76 +147,89 @@ void GetFile::onSchedule(core::ProcessContext *context, core::ProcessSessionFact
request_.inputDirectory = value;
}
-void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession *session) {
- // Perform directory list
-
+void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession* session) {
metrics_->iterations_++;
- const bool isDirEmptyBeforePoll = isListingEmpty();
- logger_->log_debug("Is listing empty before polling directory %i", isDirEmptyBeforePoll);
- if (isDirEmptyBeforePoll) {
+ const bool is_dir_empty_before_poll = isListingEmpty();
+ logger_->log_debug("Listing is %s before polling directory", is_dir_empty_before_poll ? "empty" : "not empty");
+ if (is_dir_empty_before_poll) {
if (request_.pollInterval == 0 || (utils::timeutils::getTimeMillis() - last_listing_time_) > request_.pollInterval) {
performListing(request_);
last_listing_time_.store(utils::timeutils::getTimeMillis());
}
}
- const bool isDirEmptyAfterPoll = isListingEmpty();
- logger_->log_debug("Is listing empty after polling directory %i", isDirEmptyAfterPoll);
-
- if (!isDirEmptyAfterPoll) {
- try {
- std::queue<std::string> list;
- pollListing(list, request_);
- while (!list.empty()) {
- std::string fileName = list.front();
- list.pop();
- logger_->log_info("GetFile process %s", fileName);
- auto flowFile = session->create();
- if (flowFile == nullptr)
- return;
- std::size_t found = fileName.find_last_of("/\\");
- std::string path = fileName.substr(0, found);
- std::string name = fileName.substr(found + 1);
- flowFile->setAttribute(core::SpecialFlowAttribute::FILENAME, name);
- flowFile->setAttribute(core::SpecialFlowAttribute::PATH, path);
- flowFile->addAttribute(core::SpecialFlowAttribute::ABSOLUTE_PATH, fileName);
- session->import(fileName, flowFile, request_.keepSourceFile);
- session->transfer(flowFile, Success);
+ const bool is_dir_empty_after_poll = isListingEmpty();
+ logger_->log_debug("Listing is %s after polling directory", is_dir_empty_after_poll ? "empty" : "not empty");
+ if (is_dir_empty_after_poll) {
+ yield();
+ return;
+ }
+
+ std::queue<std::string> list_of_file_names = pollListing(request_.batchSize);
+ while (!list_of_file_names.empty()) {
+ std::string file_name = list_of_file_names.front();
+ list_of_file_names.pop();
+ getSingleFile(*session, file_name);
+ }
+}
+
+void GetFile::getSingleFile(core::ProcessSession& session, const std::string& file_name) const {
+ logger_->log_info("GetFile process %s", file_name);
+ auto flow_file = session.create();
+ gsl_Expects(flow_file);
+ std::string path;
+ std::string name;
+ std::tie(path, name) = utils::file::split_path(file_name);
+ flow_file->setAttribute(core::SpecialFlowAttribute::FILENAME, name);
+ flow_file->setAttribute(core::SpecialFlowAttribute::PATH, path);
+ flow_file->addAttribute(core::SpecialFlowAttribute::ABSOLUTE_PATH, file_name);
+
+ try {
+ utils::FileReaderCallback file_reader_callback{file_name};
+ session.write(flow_file, &file_reader_callback);
+ session.transfer(flow_file, Success);
+ if (!request_.keepSourceFile) {
+ auto remove_status = remove(file_name.c_str());
+ if (remove_status != 0) {
+ logger_->log_error("GetFile could not delete file '%s', error %d: %s", file_name, errno, strerror(errno));
}
- } catch (std::exception &exception) {
- logger_->log_debug("GetFile Caught Exception %s", exception.what());
- throw;
- } catch (...) {
- throw;
}
+ } catch (const std::exception& exception) {
Review comment:
Good point. Fixed in https://github.com/apache/nifi-minifi-cpp/pull/1064/commits/0ea95c9fd071ab701a730c29abed60806b171b34; now non-IO exceptions are no longer caught, they will cause a rollback.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] fgerlits commented on a change in pull request #1064: MINIFICPP-987 Improve error handling in GetFile
Posted by GitBox <gi...@apache.org>.
fgerlits commented on a change in pull request #1064:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1064#discussion_r630912957
##########
File path: extensions/standard-processors/processors/GetFile.cpp
##########
@@ -158,76 +147,89 @@ void GetFile::onSchedule(core::ProcessContext *context, core::ProcessSessionFact
request_.inputDirectory = value;
}
-void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession *session) {
- // Perform directory list
-
+void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession* session) {
metrics_->iterations_++;
- const bool isDirEmptyBeforePoll = isListingEmpty();
- logger_->log_debug("Is listing empty before polling directory %i", isDirEmptyBeforePoll);
- if (isDirEmptyBeforePoll) {
+ const bool is_dir_empty_before_poll = isListingEmpty();
+ logger_->log_debug("Listing is %s before polling directory", is_dir_empty_before_poll ? "empty" : "not empty");
Review comment:
I'll change it if you think saving 5 chars in the line (and 5 bytes in the code) is worth it, but I think it is more readable this way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] arpadboda commented on a change in pull request #1064: MINIFICPP-987 Improve error handling in GetFile
Posted by GitBox <gi...@apache.org>.
arpadboda commented on a change in pull request #1064:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1064#discussion_r630913410
##########
File path: extensions/standard-processors/processors/GetFile.cpp
##########
@@ -158,76 +147,89 @@ void GetFile::onSchedule(core::ProcessContext *context, core::ProcessSessionFact
request_.inputDirectory = value;
}
-void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession *session) {
- // Perform directory list
-
+void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession* session) {
metrics_->iterations_++;
- const bool isDirEmptyBeforePoll = isListingEmpty();
- logger_->log_debug("Is listing empty before polling directory %i", isDirEmptyBeforePoll);
- if (isDirEmptyBeforePoll) {
+ const bool is_dir_empty_before_poll = isListingEmpty();
+ logger_->log_debug("Listing is %s before polling directory", is_dir_empty_before_poll ? "empty" : "not empty");
+ if (is_dir_empty_before_poll) {
if (request_.pollInterval == 0 || (utils::timeutils::getTimeMillis() - last_listing_time_) > request_.pollInterval) {
performListing(request_);
last_listing_time_.store(utils::timeutils::getTimeMillis());
}
}
- const bool isDirEmptyAfterPoll = isListingEmpty();
- logger_->log_debug("Is listing empty after polling directory %i", isDirEmptyAfterPoll);
-
- if (!isDirEmptyAfterPoll) {
- try {
- std::queue<std::string> list;
- pollListing(list, request_);
- while (!list.empty()) {
- std::string fileName = list.front();
- list.pop();
- logger_->log_info("GetFile process %s", fileName);
- auto flowFile = session->create();
- if (flowFile == nullptr)
- return;
- std::size_t found = fileName.find_last_of("/\\");
- std::string path = fileName.substr(0, found);
- std::string name = fileName.substr(found + 1);
- flowFile->setAttribute(core::SpecialFlowAttribute::FILENAME, name);
- flowFile->setAttribute(core::SpecialFlowAttribute::PATH, path);
- flowFile->addAttribute(core::SpecialFlowAttribute::ABSOLUTE_PATH, fileName);
- session->import(fileName, flowFile, request_.keepSourceFile);
- session->transfer(flowFile, Success);
+ const bool is_dir_empty_after_poll = isListingEmpty();
+ logger_->log_debug("Listing is %s after polling directory", is_dir_empty_after_poll ? "empty" : "not empty");
+ if (is_dir_empty_after_poll) {
+ yield();
+ return;
+ }
+
+ std::queue<std::string> list_of_file_names = pollListing(request_.batchSize);
+ while (!list_of_file_names.empty()) {
+ std::string file_name = list_of_file_names.front();
+ list_of_file_names.pop();
+ getSingleFile(*session, file_name);
+ }
+}
+
+void GetFile::getSingleFile(core::ProcessSession& session, const std::string& file_name) const {
+ logger_->log_info("GetFile process %s", file_name);
+ auto flow_file = session.create();
+ gsl_Expects(flow_file);
+ std::string path;
+ std::string name;
+ std::tie(path, name) = utils::file::split_path(file_name);
+ flow_file->setAttribute(core::SpecialFlowAttribute::FILENAME, name);
+ flow_file->setAttribute(core::SpecialFlowAttribute::PATH, path);
+ flow_file->addAttribute(core::SpecialFlowAttribute::ABSOLUTE_PATH, file_name);
+
+ try {
+ utils::FileReaderCallback file_reader_callback{file_name};
+ session.write(flow_file, &file_reader_callback);
+ session.transfer(flow_file, Success);
+ if (!request_.keepSourceFile) {
+ auto remove_status = remove(file_name.c_str());
+ if (remove_status != 0) {
+ logger_->log_error("GetFile could not delete file '%s', error %d: %s", file_name, errno, strerror(errno));
}
- } catch (std::exception &exception) {
- logger_->log_debug("GetFile Caught Exception %s", exception.what());
- throw;
- } catch (...) {
- throw;
}
+ } catch (const std::exception& exception) {
+ logger_->log_error("GetFile caught exception while processing file '%s': %s", file_name, exception.what());
+ flow_file->setDeleted(true);
Review comment:
Okay, I need more coffee, didn't realise that this is a different scenario. :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] arpadboda commented on a change in pull request #1064: MINIFICPP-987 Improve error handling in GetFile
Posted by GitBox <gi...@apache.org>.
arpadboda commented on a change in pull request #1064:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1064#discussion_r630913672
##########
File path: extensions/standard-processors/processors/GetFile.cpp
##########
@@ -158,76 +147,89 @@ void GetFile::onSchedule(core::ProcessContext *context, core::ProcessSessionFact
request_.inputDirectory = value;
}
-void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession *session) {
- // Perform directory list
-
+void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession* session) {
metrics_->iterations_++;
- const bool isDirEmptyBeforePoll = isListingEmpty();
- logger_->log_debug("Is listing empty before polling directory %i", isDirEmptyBeforePoll);
- if (isDirEmptyBeforePoll) {
+ const bool is_dir_empty_before_poll = isListingEmpty();
+ logger_->log_debug("Listing is %s before polling directory", is_dir_empty_before_poll ? "empty" : "not empty");
Review comment:
No, deffinitely not, leave it as it is. :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] arpadboda commented on a change in pull request #1064: MINIFICPP-987 Improve error handling in GetFile
Posted by GitBox <gi...@apache.org>.
arpadboda commented on a change in pull request #1064:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1064#discussion_r630913672
##########
File path: extensions/standard-processors/processors/GetFile.cpp
##########
@@ -158,76 +147,89 @@ void GetFile::onSchedule(core::ProcessContext *context, core::ProcessSessionFact
request_.inputDirectory = value;
}
-void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession *session) {
- // Perform directory list
-
+void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession* session) {
metrics_->iterations_++;
- const bool isDirEmptyBeforePoll = isListingEmpty();
- logger_->log_debug("Is listing empty before polling directory %i", isDirEmptyBeforePoll);
- if (isDirEmptyBeforePoll) {
+ const bool is_dir_empty_before_poll = isListingEmpty();
+ logger_->log_debug("Listing is %s before polling directory", is_dir_empty_before_poll ? "empty" : "not empty");
Review comment:
No, definitely not, leave it as it is. :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #1064: MINIFICPP-987 Improve error handling in GetFile
Posted by GitBox <gi...@apache.org>.
adamdebreceni commented on a change in pull request #1064:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1064#discussion_r635948542
##########
File path: libminifi/src/utils/FileReaderCallback.cpp
##########
@@ -0,0 +1,72 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "utils/FileReaderCallback.h"
+
+#include <cinttypes>
+
+#include "Exception.h"
+#include "core/logging/LoggerConfiguration.h"
+
+namespace {
+
+constexpr std::size_t BUFFER_SIZE = 4096;
+
+} // namespace
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace utils {
+
+FileReaderCallback::FileReaderCallback(const std::string& file_name)
+ : logger_(core::logging::LoggerFactory<FileReaderCallback>::getLogger()) {
+ logger_->log_debug("Opening %s", file_name);
+ input_stream_.open(file_name.c_str(), std::fstream::in | std::fstream::binary);
+ if (!input_stream_.is_open()) {
+ throw Exception(FILE_OPERATION_EXCEPTION, "Could not open file: " + file_name);
+ }
+}
+
+int64_t FileReaderCallback::process(const std::shared_ptr<io::BaseStream>& output_stream) {
+ std::array<char, BUFFER_SIZE> buffer;
+
+ uint64_t num_bytes_written = 0;
+
+ while (input_stream_.good()) {
+ input_stream_.read(buffer.data(), buffer.size());
+
+ const auto num_bytes_read = input_stream_.gcount();
+ logger_->log_trace("Read %jd bytes of input", std::intmax_t{num_bytes_read});
+
+ const int len = gsl::narrow<int>(num_bytes_read);
+
+ output_stream->write(reinterpret_cast<uint8_t*>(buffer.data()), len);
+ num_bytes_written += len;
+ }
+ input_stream_.close();
+
+ logger_->log_debug("Finished reading %" PRIu64 " bytes from the file", num_bytes_written);
+ return num_bytes_written;
Review comment:
okay, now I see that an error during open sets the `failbit` (so we would ask for an exception on that bit) but `failbit` is also set during `eof`, but then we wouldn't want an exception
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] fgerlits commented on a change in pull request #1064: MINIFICPP-987 Improve error handling in GetFile
Posted by GitBox <gi...@apache.org>.
fgerlits commented on a change in pull request #1064:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1064#discussion_r629371159
##########
File path: extensions/standard-processors/processors/GetFile.cpp
##########
@@ -159,47 +148,54 @@ void GetFile::onSchedule(core::ProcessContext *context, core::ProcessSessionFact
}
void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession *session) {
- // Perform directory list
-
metrics_->iterations_++;
- const bool isDirEmptyBeforePoll = isListingEmpty();
- logger_->log_debug("Is listing empty before polling directory %i", isDirEmptyBeforePoll);
- if (isDirEmptyBeforePoll) {
+ const bool is_dir_empty_before_poll = isListingEmpty();
+ logger_->log_debug("Listing is %s before polling directory", is_dir_empty_before_poll ? "empty" : "not empty");
+ if (is_dir_empty_before_poll) {
if (request_.pollInterval == 0 || (utils::timeutils::getTimeMillis() - last_listing_time_) > request_.pollInterval) {
performListing(request_);
last_listing_time_.store(utils::timeutils::getTimeMillis());
}
}
- const bool isDirEmptyAfterPoll = isListingEmpty();
- logger_->log_debug("Is listing empty after polling directory %i", isDirEmptyAfterPoll);
+ const bool is_dir_empty_after_poll = isListingEmpty();
+ logger_->log_debug("Listing is %s after polling directory", is_dir_empty_after_poll ? "empty" : "not empty");
+ if (is_dir_empty_after_poll) {
+ yield();
+ return;
+ }
+
+ std::queue<std::string> list_of_file_names = pollListing(request_.batchSize);
+ while (!list_of_file_names.empty()) {
Review comment:
done in https://github.com/apache/nifi-minifi-cpp/pull/1064/commits/3a154cc5756502d6b20e953b6eb31edd157d7e78
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org