You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ab...@apache.org on 2021/02/08 16:30:14 UTC
[nifi-minifi-cpp] branch main updated (b977773 -> 85f05b0)
This is an automated email from the ASF dual-hosted git repository.
aboda pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git.
from b977773 MINIFICPP-1434: Fix flaky CSite2SiteTests on macOS
new 1e4c820 MINIFICPP-1477 - fix StringUtils::trim
new 85f05b0 MINIFICPP-1455 - Fix FileStream error handling and reporting
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
extensions/expression-language/Expression.cpp | 5 +-
libminifi/include/io/FileStream.h | 3 +
libminifi/include/utils/StringUtils.h | 4 +-
libminifi/include/utils/ValueParser.h | 2 +-
libminifi/src/io/FileStream.cpp | 82 ++++++++++++++++++++-------
libminifi/test/TestBase.h | 4 ++
libminifi/test/unit/FileStreamTests.cpp | 81 ++++++++++++++++++++++++++
7 files changed, 154 insertions(+), 27 deletions(-)
[nifi-minifi-cpp] 01/02: MINIFICPP-1477 - fix StringUtils::trim
Posted by ab...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aboda pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 1e4c820481cec72d6543fe26f40a5085184d91de
Author: Marton Szasz <sz...@gmail.com>
AuthorDate: Thu Feb 4 17:04:20 2021 +0100
MINIFICPP-1477 - fix StringUtils::trim
Signed-off-by: Arpad Boda <ab...@apache.org>
This closes #990
---
extensions/expression-language/Expression.cpp | 5 +----
libminifi/include/utils/StringUtils.h | 4 ++--
libminifi/include/utils/ValueParser.h | 2 +-
3 files changed, 4 insertions(+), 7 deletions(-)
diff --git a/extensions/expression-language/Expression.cpp b/extensions/expression-language/Expression.cpp
index 545983f..5ce6f4d 100644
--- a/extensions/expression-language/Expression.cpp
+++ b/extensions/expression-language/Expression.cpp
@@ -840,10 +840,7 @@ Value expr_find(const std::vector<Value> &args) {
#endif // EXPRESSION_LANGUAGE_USE_REGEX
Value expr_trim(const std::vector<Value> &args) {
- std::string result = args[0].asString();
- auto ws_front = std::find_if_not(result.begin(), result.end(), [](int c) {return std::isspace(c);});
- auto ws_back = std::find_if_not(result.rbegin(), result.rend(), [](int c) {return std::isspace(c);}).base();
- return (ws_back <= ws_front ? Value(std::string()) : Value(std::string(ws_front, ws_back)));
+ return Value{utils::StringUtils::trim(args[0].asString())};
}
Value expr_append(const std::vector<Value> &args) {
diff --git a/libminifi/include/utils/StringUtils.h b/libminifi/include/utils/StringUtils.h
index 7dcfc49..54bab6c 100644
--- a/libminifi/include/utils/StringUtils.h
+++ b/libminifi/include/utils/StringUtils.h
@@ -99,7 +99,7 @@ class StringUtils {
* @returns modified string
*/
static inline std::string trimLeft(std::string s) {
- s.erase(s.begin(), std::find_if(s.begin(), s.end(), [](char c) -> bool { return !isspace(c); }));
+ s.erase(s.begin(), std::find_if(s.begin(), s.end(), [](unsigned char c) -> bool { return !isspace(c); }));
return s;
}
@@ -110,7 +110,7 @@ class StringUtils {
*/
static inline std::string trimRight(std::string s) {
- s.erase(std::find_if(s.rbegin(), s.rend(), [](char c) -> bool { return !isspace(c); }).base(), s.end());
+ s.erase(std::find_if(s.rbegin(), s.rend(), [](unsigned char c) -> bool { return !isspace(c); }).base(), s.end());
return s;
}
diff --git a/libminifi/include/utils/ValueParser.h b/libminifi/include/utils/ValueParser.h
index f7b11ba..ce4f0cc 100644
--- a/libminifi/include/utils/ValueParser.h
+++ b/libminifi/include/utils/ValueParser.h
@@ -148,7 +148,7 @@ class ValueParser {
}
void skipWhitespace() {
- while (offset < str.length() && std::isspace(str[offset])) {
+ while (offset < str.length() && std::isspace(static_cast<unsigned char>(str[offset]))) {
++offset;
}
}
[nifi-minifi-cpp] 02/02: MINIFICPP-1455 - Fix FileStream error
handling and reporting
Posted by ab...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aboda pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 85f05b08ed8f5d10b2745159d0f64475aa741230
Author: Martin Zink <ma...@protonmail.com>
AuthorDate: Tue Jan 26 19:27:06 2021 +0100
MINIFICPP-1455 - Fix FileStream error handling and reporting
Signed-off-by: Arpad Boda <ab...@apache.org>
This closes #987
---
libminifi/include/io/FileStream.h | 3 ++
libminifi/src/io/FileStream.cpp | 82 +++++++++++++++++++++++++--------
libminifi/test/TestBase.h | 4 ++
libminifi/test/unit/FileStreamTests.cpp | 81 ++++++++++++++++++++++++++++++++
4 files changed, 150 insertions(+), 20 deletions(-)
diff --git a/libminifi/include/io/FileStream.h b/libminifi/include/io/FileStream.h
index 54b8d59..3156a30 100644
--- a/libminifi/include/io/FileStream.h
+++ b/libminifi/include/io/FileStream.h
@@ -86,6 +86,9 @@ class FileStream : public io::BaseStream {
*/
int write(const uint8_t *value, int size) override;
+ private:
+ void seekToEndOfFile(const char* caller_error_msg);
+
std::mutex file_lock_;
std::unique_ptr<std::fstream> file_stream_;
size_t offset_;
diff --git a/libminifi/src/io/FileStream.cpp b/libminifi/src/io/FileStream.cpp
index 700ec9f..20d4e6d 100644
--- a/libminifi/src/io/FileStream.cpp
+++ b/libminifi/src/io/FileStream.cpp
@@ -31,6 +31,19 @@ namespace nifi {
namespace minifi {
namespace io {
+constexpr const char *FILE_OPENING_ERROR_MSG = "Error opening file: ";
+constexpr const char *READ_ERROR_MSG = "Error reading from file: ";
+constexpr const char *WRITE_ERROR_MSG = "Error writing to file: ";
+constexpr const char *SEEK_ERROR_MSG = "Error seeking in file: ";
+constexpr const char *INVALID_FILE_STREAM_ERROR_MSG = "invalid file stream";
+constexpr const char *TELLG_CALL_ERROR_MSG = "tellg call on file stream failed";
+constexpr const char *INVALID_BUFFER_ERROR_MSG = "invalid buffer";
+constexpr const char *FLUSH_CALL_ERROR_MSG = "flush call on file stream failed";
+constexpr const char *WRITE_CALL_ERROR_MSG = "write call on file stream failed";
+constexpr const char *EMPTY_MESSAGE_ERROR_MSG = "empty message";
+constexpr const char *SEEKG_CALL_ERROR_MSG = "seekg call on file stream failed";
+constexpr const char *SEEKP_CALL_ERROR_MSG = "seekp call on file stream failed";
+
FileStream::FileStream(const std::string &path, bool append)
: logger_(logging::LoggerFactory<FileStream>::getLogger()),
path_(path),
@@ -38,14 +51,22 @@ FileStream::FileStream(const std::string &path, bool append)
file_stream_ = std::unique_ptr<std::fstream>(new std::fstream());
if (append) {
file_stream_->open(path.c_str(), std::fstream::in | std::fstream::out | std::fstream::app | std::fstream::binary);
- file_stream_->seekg(0, file_stream_->end);
- file_stream_->seekp(0, file_stream_->end);
- std::streamoff len = file_stream_->tellg();
- length_ = len > 0 ? gsl::narrow<size_t>(len) : 0;
- seek(offset_);
+ if (file_stream_->is_open()) {
+ seekToEndOfFile(FILE_OPENING_ERROR_MSG);
+ auto len = file_stream_->tellg();
+ if (len == std::streampos(-1))
+ logging::LOG_ERROR(logger_) << FILE_OPENING_ERROR_MSG << TELLG_CALL_ERROR_MSG;
+ length_ = len > 0 ? gsl::narrow<size_t>(len) : 0;
+ seek(offset_);
+ } else {
+ logging::LOG_ERROR(logger_) << FILE_OPENING_ERROR_MSG << path << " " << strerror(errno);
+ }
} else {
file_stream_->open(path.c_str(), std::fstream::out | std::fstream::binary);
length_ = 0;
+ if (!file_stream_->is_open()) {
+ logging::LOG_ERROR(logger_) << FILE_OPENING_ERROR_MSG << path << " " << strerror(errno);
+ }
}
}
@@ -59,15 +80,16 @@ FileStream::FileStream(const std::string &path, uint32_t offset, bool write_enab
} else {
file_stream_->open(path.c_str(), std::fstream::in | std::fstream::binary);
}
- file_stream_->seekg(0, file_stream_->end);
- file_stream_->seekp(0, file_stream_->end);
- std::streamoff len = file_stream_->tellg();
- if (len > 0) {
- length_ = gsl::narrow<size_t>(len);
+ if (file_stream_->is_open()) {
+ seekToEndOfFile(FILE_OPENING_ERROR_MSG);
+ auto len = file_stream_->tellg();
+ if (len == std::streampos(-1))
+ logging::LOG_ERROR(logger_) << FILE_OPENING_ERROR_MSG << TELLG_CALL_ERROR_MSG;
+ length_ = len > 0 ? gsl::narrow<size_t>(len) : 0;
+ seek(offset_);
} else {
- length_ = 0;
+ logging::LOG_ERROR(logger_) << FILE_OPENING_ERROR_MSG << path << " " << strerror(errno);
}
- seek(offset);
}
void FileStream::close() {
@@ -77,10 +99,16 @@ void FileStream::close() {
void FileStream::seek(uint64_t offset) {
std::lock_guard<std::mutex> lock(file_lock_);
+ if (file_stream_ == nullptr || !file_stream_->is_open()) {
+ logging::LOG_ERROR(logger_) << SEEK_ERROR_MSG << INVALID_FILE_STREAM_ERROR_MSG;
+ return;
+ }
offset_ = gsl::narrow<size_t>(offset);
file_stream_->clear();
- file_stream_->seekg(offset_);
- file_stream_->seekp(offset_);
+ if (!file_stream_->seekg(offset_))
+ logging::LOG_ERROR(logger_) << SEEK_ERROR_MSG << SEEKG_CALL_ERROR_MSG;
+ if (!file_stream_->seekp(offset_))
+ logging::LOG_ERROR(logger_) << SEEK_ERROR_MSG << SEEKP_CALL_ERROR_MSG;
}
int FileStream::write(const uint8_t *value, int size) {
@@ -90,19 +118,26 @@ int FileStream::write(const uint8_t *value, int size) {
}
if (!IsNullOrEmpty(value)) {
std::lock_guard<std::mutex> lock(file_lock_);
+ if (file_stream_ == nullptr || !file_stream_->is_open()) {
+ logging::LOG_ERROR(logger_) << WRITE_ERROR_MSG << INVALID_FILE_STREAM_ERROR_MSG;
+ return -1;
+ }
if (file_stream_->write(reinterpret_cast<const char*>(value), size)) {
offset_ += size;
if (offset_ > length_) {
length_ = offset_;
}
if (!file_stream_->flush()) {
+ logging::LOG_ERROR(logger_) << WRITE_ERROR_MSG << FLUSH_CALL_ERROR_MSG;
return -1;
}
return size;
} else {
+ logging::LOG_ERROR(logger_) << WRITE_ERROR_MSG << WRITE_CALL_ERROR_MSG;
return -1;
}
} else {
+ logging::LOG_ERROR(logger_) << WRITE_ERROR_MSG << EMPTY_MESSAGE_ERROR_MSG;
return -1;
}
}
@@ -114,17 +149,17 @@ int FileStream::read(uint8_t *buf, int buflen) {
}
if (!IsNullOrEmpty(buf)) {
std::lock_guard<std::mutex> lock(file_lock_);
- if (!file_stream_) {
+ if (file_stream_ == nullptr || !file_stream_->is_open()) {
+ logging::LOG_ERROR(logger_) << READ_ERROR_MSG << INVALID_FILE_STREAM_ERROR_MSG;
return -1;
}
file_stream_->read(reinterpret_cast<char*>(buf), buflen);
- if ((file_stream_->rdstate() & (file_stream_->eofbit | file_stream_->failbit)) != 0) {
+ if (file_stream_->eof() || file_stream_->fail()) {
file_stream_->clear();
- file_stream_->seekg(0, file_stream_->end);
- file_stream_->seekp(0, file_stream_->end);
+ seekToEndOfFile(READ_ERROR_MSG);
auto tellg_result = file_stream_->tellg();
- if (tellg_result < 0) {
- logging::LOG_ERROR(logger_) << "Tellg call on file stream failed.";
+ if (tellg_result == std::streampos(-1)) {
+ logging::LOG_ERROR(logger_) << READ_ERROR_MSG << TELLG_CALL_ERROR_MSG;
return -1;
}
size_t len = gsl::narrow<size_t>(tellg_result);
@@ -140,10 +175,17 @@ int FileStream::read(uint8_t *buf, int buflen) {
}
} else {
+ logging::LOG_ERROR(logger_) << READ_ERROR_MSG << INVALID_BUFFER_ERROR_MSG;
return -1;
}
}
+void FileStream::seekToEndOfFile(const char *caller_error_msg) {
+ if (!file_stream_->seekg(0, file_stream_->end))
+ logging::LOG_ERROR(logger_) << caller_error_msg << SEEKG_CALL_ERROR_MSG;
+ if (!file_stream_->seekp(0, file_stream_->end))
+ logging::LOG_ERROR(logger_) << caller_error_msg << SEEKP_CALL_ERROR_MSG;
+}
} /* namespace io */
} /* namespace minifi */
} /* namespace nifi */
diff --git a/libminifi/test/TestBase.h b/libminifi/test/TestBase.h
index 9d0cdab..1a7d960 100644
--- a/libminifi/test/TestBase.h
+++ b/libminifi/test/TestBase.h
@@ -400,6 +400,10 @@ class TestController {
return log.logger_;
}
+ const LogTestController& getLog() const {
+ return log;
+ }
+
~TestController() {
for (const auto& dir : directories) {
utils::file::FileUtils::delete_dir(dir, true);
diff --git a/libminifi/test/unit/FileStreamTests.cpp b/libminifi/test/unit/FileStreamTests.cpp
index 5a74035..b13dcfa 100644
--- a/libminifi/test/unit/FileStreamTests.cpp
+++ b/libminifi/test/unit/FileStreamTests.cpp
@@ -24,6 +24,11 @@
#include "io/FileStream.h"
#include "../TestBase.h"
#include "utils/gsl.h"
+#include "utils/file/FileUtils.h"
+
+#ifdef USE_BOOST
+#include <boost/filesystem.hpp>
+#endif
TEST_CASE("TestFileOverWrite", "[TestFiles]") {
TestController testController;
@@ -263,3 +268,79 @@ TEST_CASE("Read zero bytes") {
minifi::io::FileStream stream(utils::file::concat_path(dir, "test.txt"), 0, true);
REQUIRE(stream.read(nullptr, 0) == 0);
}
+
+TEST_CASE("Non-existing file read/write test") {
+ TestController test_controller;
+ char format[] = "/tmp/gt.XXXXXX";
+ auto dir = test_controller.createTempDirectory(format);
+ minifi::io::FileStream stream(utils::file::concat_path(dir, "non_existing_file.txt"), 0, true);
+ REQUIRE(test_controller.getLog().getInstance().contains("Error opening file", std::chrono::seconds(0)));
+ REQUIRE(test_controller.getLog().getInstance().contains("No such file or directory", std::chrono::seconds(0)));
+ REQUIRE(stream.write("lorem ipsum", false) == -1);
+ REQUIRE(test_controller.getLog().getInstance().contains("Error writing to file: invalid file stream", std::chrono::seconds(0)));
+ std::vector<uint8_t> readBuffer;
+ stream.seek(0);
+ REQUIRE(stream.read(readBuffer, 1) == -1);
+ REQUIRE(test_controller.getLog().getInstance().contains("Error reading from file: invalid file stream", std::chrono::seconds(0)));
+}
+
+TEST_CASE("Existing file read/write test") {
+ TestController test_controller;
+ char format[] = "/tmp/gt.XXXXXX";
+ auto dir = test_controller.createTempDirectory(format);
+ std::string path_to_existing_file(utils::file::concat_path(dir, "existing_file.txt"));
+ {
+ std::ofstream outfile(path_to_existing_file);
+ outfile << "lorem ipsum" << std::endl;
+ outfile.close();
+ }
+ minifi::io::FileStream stream(path_to_existing_file, 0, true);
+ REQUIRE_FALSE(test_controller.getLog().getInstance().contains("Error opening file", std::chrono::seconds(0)));
+ REQUIRE_FALSE(stream.write("dolor sit amet", false) == -1);
+ REQUIRE_FALSE(test_controller.getLog().getInstance().contains("Error writing to file", std::chrono::seconds(0)));
+ std::vector<uint8_t> readBuffer;
+ stream.seek(0);
+ REQUIRE_FALSE(stream.read(readBuffer, 11) == -1);
+ REQUIRE_FALSE(test_controller.getLog().getInstance().contains("Error reading from file", std::chrono::seconds(0)));
+ stream.seek(0);
+ REQUIRE(stream.read(nullptr, 11) == -1);
+ REQUIRE(test_controller.getLog().getInstance().contains("Error reading from file: invalid buffer", std::chrono::seconds(0)));
+}
+
+#if !defined(WIN32) || defined(USE_BOOST)
+// This could be simplified with C++17 std::filesystem
+TEST_CASE("Opening file without permission creates error logs") {
+ TestController test_controller;
+ char format[] = "/tmp/gt.XXXXXX";
+ auto dir = test_controller.createTempDirectory(format);
+ std::string path_to_permissionless_file(utils::file::concat_path(dir, "permissionless_file.txt"));
+ {
+ std::ofstream outfile(path_to_permissionless_file);
+ outfile << "this file has been just created" << std::endl;
+ outfile.close();
+#ifndef WIN32
+ utils::file::FileUtils::set_permissions(path_to_permissionless_file, 0);
+#else
+ boost::filesystem::permissions(path_to_permissionless_file, boost::filesystem::no_perms);
+#endif
+ }
+ minifi::io::FileStream stream(path_to_permissionless_file, 0, false);
+ REQUIRE(test_controller.getLog().getInstance().contains("Error opening file", std::chrono::seconds(0)));
+ REQUIRE(test_controller.getLog().getInstance().contains("Permission denied", std::chrono::seconds(0)));
+}
+#endif
+
+TEST_CASE("Readonly filestream write test") {
+ TestController test_controller;
+ char format[] = "/tmp/gt.XXXXXX";
+ auto dir = test_controller.createTempDirectory(format);
+ std::string path_to_file(utils::file::concat_path(dir, "file_to_seek_in.txt"));
+ {
+ std::ofstream outfile(path_to_file);
+ outfile << "lorem ipsum" << std::endl;
+ outfile.close();
+ }
+ minifi::io::FileStream stream(path_to_file, 0, false);
+ REQUIRE(stream.write("dolor sit amet", false) == -1);
+ REQUIRE(test_controller.getLog().getInstance().contains("Error writing to file: write call on file stream failed", std::chrono::seconds(0)));
+}