You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by bq...@apache.org on 2017/11/14 16:02:28 UTC
nifi-minifi-cpp git commit: MINIFICPP-297: Remove Boost dependencies
since we have very little reliance on it
Repository: nifi-minifi-cpp
Updated Branches:
refs/heads/master c0a788b35 -> a698084cb
MINIFICPP-297: Remove Boost dependencies since we have very little reliance on it
MINIFICPP-297: Ensure posix compliance
This closes #181.
Signed-off-by: Bin Qiu <be...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/commit/a698084c
Tree: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/tree/a698084c
Diff: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/diff/a698084c
Branch: refs/heads/master
Commit: a698084cb5ad62e1deeb93b7904368652cac8741
Parents: c0a788b
Author: Marc Parisi <ph...@apache.org>
Authored: Wed Nov 8 14:03:22 2017 -0500
Committer: Bin Qiu <be...@gmail.com>
Committed: Tue Nov 14 08:01:28 2017 -0800
----------------------------------------------------------------------
CMakeLists.txt | 1 -
cmake/BuildTests.cmake | 10 +-
extensions/libarchive/CMakeLists.txt | 4 -
extensions/libarchive/FocusArchiveEntry.cpp | 83 +++----
extensions/libarchive/FocusArchiveEntry.h | 4 +-
extensions/libarchive/UnfocusArchiveEntry.cpp | 244 ++++++++++-----------
extensions/script/CMakeLists.txt | 5 -
libminifi/include/core/Core.h | 4 +
libminifi/include/utils/StringUtils.h | 10 +
libminifi/include/utils/file/FileManager.h | 113 ++++++++++
libminifi/include/utils/file/FileUtils.h | 69 ++++++
libminifi/test/unit/PutFileTests.cpp | 10 +-
12 files changed, 348 insertions(+), 209 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a698084c/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 5a9c97b..57636fc 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -110,7 +110,6 @@ add_subdirectory(libminifi)
if (EXCLUDE_BOOST)
message(" -- Excluding Boost dependent packages...")
- set(DISABLE_LIBARCHIVE "ON")
endif(EXCLUDE_BOOST)
#### EXTENSION
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a698084c/cmake/BuildTests.cmake
----------------------------------------------------------------------
diff --git a/cmake/BuildTests.cmake b/cmake/BuildTests.cmake
index 5b5fa0f..7045c3d 100644
--- a/cmake/BuildTests.cmake
+++ b/cmake/BuildTests.cmake
@@ -29,7 +29,9 @@ MACRO(GETSOURCEFILES result curdir)
SET(${result} ${dirlist})
ENDMACRO()
-find_package(Boost COMPONENTS system filesystem REQUIRED)
+if(NOT EXCLUDE_BOOST)
+ find_package(Boost COMPONENTS system filesystem)
+endif()
function(createTests testName)
message ("-- Adding test: ${testName}")
@@ -56,8 +58,10 @@ function(createTests testName)
target_include_directories(${testName} BEFORE PRIVATE "${Boost_INCLUDE_DIRS}")
target_link_libraries(${testName} ${SPD_LIB} ${TEST_BASE_LIB})
target_link_libraries(${testName} ${CMAKE_THREAD_LIBS_INIT} ${OPENSSL_LIBRARIES} core-minifi minifi yaml-cpp c-library civetweb-cpp ${JSON_CPP_LIB})
- target_link_libraries(${testName} ${Boost_SYSTEM_LIBRARY})
- target_link_libraries(${testName} ${Boost_FILESYSTEM_LIBRARY})
+ if (Boost_FOUND)
+ target_link_libraries(${testName} ${Boost_SYSTEM_LIBRARY})
+ target_link_libraries(${testName} ${Boost_FILESYSTEM_LIBRARY})
+ endif()
endfunction()
enable_testing(test)
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a698084c/extensions/libarchive/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/extensions/libarchive/CMakeLists.txt b/extensions/libarchive/CMakeLists.txt
index 0dd24de..9643809 100644
--- a/extensions/libarchive/CMakeLists.txt
+++ b/extensions/libarchive/CMakeLists.txt
@@ -53,10 +53,6 @@ endif()
find_package(ZLIB REQUIRED)
include_directories(${ZLIB_INCLUDE_DIRS})
target_link_libraries (minifi-archive-extensions ${ZLIB_LIBRARIES})
-find_package(Boost COMPONENTS system filesystem REQUIRED)
-include_directories(${Boost_INCLUDE_DIRS})
-target_link_libraries(minifi-archive-extensions ${Boost_SYSTEM_LIBRARY})
-target_link_libraries(minifi-archive-extensions ${Boost_FILESYSTEM_LIBRARY})
if (WIN32)
set_target_properties(minifi-archive-extensions PROPERTIES
LINK_FLAGS "/WHOLEARCHIVE"
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a698084c/extensions/libarchive/FocusArchiveEntry.cpp
----------------------------------------------------------------------
diff --git a/extensions/libarchive/FocusArchiveEntry.cpp b/extensions/libarchive/FocusArchiveEntry.cpp
index 2dee317..b3ed1fa 100644
--- a/extensions/libarchive/FocusArchiveEntry.cpp
+++ b/extensions/libarchive/FocusArchiveEntry.cpp
@@ -24,8 +24,6 @@
#include <string.h>
-#include <boost/filesystem.hpp>
-
#include <string>
#include <set>
@@ -47,13 +45,8 @@ namespace processors {
std::shared_ptr<utils::IdGenerator> FocusArchiveEntry::id_generator_ = utils::IdGenerator::getIdGenerator();
-core::Property FocusArchiveEntry::Path(
- "Path",
- "The path within the archive to focus (\"/\" to focus the total archive)",
- "");
-core::Relationship FocusArchiveEntry::Success(
- "success",
- "success operational on the flow record");
+core::Property FocusArchiveEntry::Path("Path", "The path within the archive to focus (\"/\" to focus the total archive)", "");
+core::Relationship FocusArchiveEntry::Success("success", "success operational on the flow record");
bool FocusArchiveEntry::set_or_update_attr(std::shared_ptr<core::FlowFile> flowFile, const std::string& key, const std::string& value) const {
if (flowFile->updateAttribute(key, value))
@@ -73,8 +66,7 @@ void FocusArchiveEntry::initialize() {
setSupportedRelationships(relationships);
}
-void FocusArchiveEntry::onTrigger(core::ProcessContext *context,
- core::ProcessSession *session) {
+void FocusArchiveEntry::onTrigger(core::ProcessContext *context, core::ProcessSession *session) {
auto flowFile = session->get();
std::shared_ptr<FlowFileRecord> flowFileRecord = std::static_pointer_cast<FlowFileRecord>(flowFile);
@@ -82,13 +74,15 @@ void FocusArchiveEntry::onTrigger(core::ProcessContext *context,
return;
}
+ fileutils::FileManager file_man;
+
std::string targetEntry;
context->getProperty(Path.getName(), targetEntry);
// Extract archive contents
ArchiveMetadata archiveMetadata;
archiveMetadata.focusedEntry = targetEntry;
- ReadCallback cb(this, &archiveMetadata);
+ ReadCallback cb(this, &file_man, &archiveMetadata);
session->read(flowFile, &cb);
// For each extracted entry, import & stash to key
@@ -96,18 +90,13 @@ void FocusArchiveEntry::onTrigger(core::ProcessContext *context,
for (auto &entryMetadata : archiveMetadata.entryMetadata) {
if (entryMetadata.entryType == AE_IFREG) {
- logger_->log_info("FocusArchiveEntry importing %s from %s",
- entryMetadata.entryName.c_str(),
- entryMetadata.tmpFileName.c_str());
+ logger_->log_info("FocusArchiveEntry importing %s from %s", entryMetadata.entryName.c_str(), entryMetadata.tmpFileName.c_str());
session->import(entryMetadata.tmpFileName, flowFile, false, 0);
char stashKey[37];
uuid_t stashKeyUuid;
id_generator_->generate(stashKeyUuid);
uuid_unparse_lower(stashKeyUuid, stashKey);
- logger_->log_debug(
- "FocusArchiveEntry generated stash key %s for entry %s",
- stashKey,
- entryMetadata.entryName.c_str());
+ logger_->log_debug("FocusArchiveEntry generated stash key %s for entry %s", stashKey, entryMetadata.entryName.c_str());
entryMetadata.stashKey.assign(stashKey);
if (entryMetadata.entryName == targetEntry) {
@@ -123,9 +112,7 @@ void FocusArchiveEntry::onTrigger(core::ProcessContext *context,
if (targetEntryStashKey != "") {
session->restore(targetEntryStashKey, flowFile);
} else {
- logger_->log_warn(
- "FocusArchiveEntry failed to locate target entry: %s",
- targetEntry.c_str());
+ logger_->log_warn("FocusArchiveEntry failed to locate target entry: %s", targetEntry.c_str());
}
// Set new/updated lens stack to attribute
@@ -138,8 +125,7 @@ void FocusArchiveEntry::onTrigger(core::ProcessContext *context,
if (flowFile->getAttribute("lens.archive.stack", existingLensStack)) {
logger_->log_info("FocusArchiveEntry loading existing lens context");
if (!reader.parse(existingLensStack, lensStack)) {
- logger_->log_error("FocusArchiveEntry JSON parse error: %s",
- reader.getFormattedErrorMessages());
+ logger_->log_error("FocusArchiveEntry JSON parse error: %s", reader.getFormattedErrorMessages());
context->yield();
return;
}
@@ -161,17 +147,17 @@ void FocusArchiveEntry::onTrigger(core::ProcessContext *context,
entryVal["entry_mtime_nsec"] = Json::Value(entryMetadata.entryMTimeNsec);
if (entryMetadata.entryType == AE_IFREG) {
- entryVal["stash_key"] = Json::Value(entryMetadata.stashKey);
+ entryVal["stash_key"] = Json::Value(entryMetadata.stashKey);
}
structVal.append(entryVal);
}
std::string archivenameStr;
- Json::Value archiveName {Json::nullValue};
+ Json::Value archiveName { Json::nullValue };
if (flowFile->getAttribute("filename", archivenameStr)) {
- archiveName = Json::Value(archivenameStr);
+ archiveName = Json::Value(archivenameStr);
}
Json::Value lensVal(Json::objectValue);
@@ -221,8 +207,8 @@ ssize_t FocusArchiveEntry::ReadCallback::read_cb(struct archive * a, void *d, co
} while (data->processor->isRunning() && last_read > 0 && read < 8196);
if (!data->processor->isRunning()) {
- archive_set_error(a, EINTR, "Processor shut down during read");
- return -1;
+ archive_set_error(a, EINTR, "Processor shut down during read");
+ return -1;
}
return read;
@@ -244,10 +230,8 @@ int64_t FocusArchiveEntry::ReadCallback::process(std::shared_ptr<io::BaseStream>
int res;
if ((res = archive_read_open(inputArchive, &data, ok_cb, read_cb, ok_cb))) {
- logger_->log_error(
- "FocusArchiveEntry can't open due to archive error: %s",
- archive_error_string(inputArchive));
- return nlen;
+ logger_->log_error("FocusArchiveEntry can't open due to archive error: %s", archive_error_string(inputArchive));
+ return nlen;
}
while (isRunning()) {
@@ -258,22 +242,17 @@ int64_t FocusArchiveEntry::ReadCallback::process(std::shared_ptr<io::BaseStream>
}
if (res < ARCHIVE_OK) {
- logger_->log_error(
- "FocusArchiveEntry can't read header due to archive error: %s",
- archive_error_string(inputArchive));
+ logger_->log_error("FocusArchiveEntry can't read header due to archive error: %s", archive_error_string(inputArchive));
return nlen;
}
if (res < ARCHIVE_WARN) {
- logger_->log_warn(
- "FocusArchiveEntry got archive warning while reading header: %s",
- archive_error_string(inputArchive));
+ logger_->log_warn("FocusArchiveEntry got archive warning while reading header: %s", archive_error_string(inputArchive));
return nlen;
}
auto entryName = archive_entry_pathname(entry);
- (*_archiveMetadata).archiveFormatName.assign(
- archive_format_name(inputArchive));
+ (*_archiveMetadata).archiveFormatName.assign(archive_format_name(inputArchive));
(*_archiveMetadata).archiveFormat = archive_format(inputArchive);
// Record entry metadata
@@ -289,24 +268,15 @@ int64_t FocusArchiveEntry::ReadCallback::process(std::shared_ptr<io::BaseStream>
metadata.entryMTime = archive_entry_mtime(entry);
metadata.entryMTimeNsec = archive_entry_mtime_nsec(entry);
- logger_->log_info(
- "FocusArchiveEntry entry type of %s is: %d",
- entryName,
- metadata.entryType);
- logger_->log_info(
- "FocusArchiveEntry entry perm of %s is: %d",
- entryName,
- metadata.entryPerm);
+ logger_->log_info("FocusArchiveEntry entry type of %s is: %d", entryName, metadata.entryType);
+ logger_->log_info("FocusArchiveEntry entry perm of %s is: %d", entryName, metadata.entryPerm);
// Write content to tmp file
if (entryType == AE_IFREG) {
- auto tmpFileName = boost::filesystem::unique_path().native();
+ auto tmpFileName = file_man_->unique_file(true);
metadata.tmpFileName = tmpFileName;
metadata.entryType = entryType;
- logger_->log_info(
- "FocusArchiveEntry extracting %s to: %s",
- entryName,
- tmpFileName.c_str());
+ logger_->log_info("FocusArchiveEntry extracting %s to: %s", entryName, tmpFileName.c_str());
auto fd = fopen(tmpFileName.c_str(), "w");
@@ -325,9 +295,8 @@ int64_t FocusArchiveEntry::ReadCallback::process(std::shared_ptr<io::BaseStream>
return nlen;
}
-FocusArchiveEntry::ReadCallback::ReadCallback(
- core::Processor *processor,
- ArchiveMetadata *archiveMetadata) : proc_(processor) {
+FocusArchiveEntry::ReadCallback::ReadCallback(core::Processor *processor,fileutils::FileManager *file_man, ArchiveMetadata *archiveMetadata)
+ : proc_(processor), file_man_(file_man) {
logger_ = logging::LoggerFactory<FocusArchiveEntry>::getLogger();
_archiveMetadata = archiveMetadata;
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a698084c/extensions/libarchive/FocusArchiveEntry.h
----------------------------------------------------------------------
diff --git a/extensions/libarchive/FocusArchiveEntry.h b/extensions/libarchive/FocusArchiveEntry.h
index aeef06a..55b44cc 100644
--- a/extensions/libarchive/FocusArchiveEntry.h
+++ b/extensions/libarchive/FocusArchiveEntry.h
@@ -32,6 +32,7 @@
#include "core/Core.h"
#include "core/logging/LoggerConfiguration.h"
#include "core/Resource.h"
+#include "utils/file/FileManager.h"
namespace org {
namespace apache {
@@ -90,12 +91,13 @@ class FocusArchiveEntry : public core::Processor {
class ReadCallback : public InputStreamCallback {
public:
- explicit ReadCallback(core::Processor*, ArchiveMetadata *archiveMetadata);
+ explicit ReadCallback(core::Processor*,fileutils::FileManager *file_man , ArchiveMetadata *archiveMetadata);
~ReadCallback();
virtual int64_t process(std::shared_ptr<io::BaseStream> stream);
bool isRunning() {return proc_->isRunning();}
private:
+ fileutils::FileManager *file_man_;
core::Processor * const proc_;
std::shared_ptr<logging::Logger> logger_;
ArchiveMetadata *_archiveMetadata;
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a698084c/extensions/libarchive/UnfocusArchiveEntry.cpp
----------------------------------------------------------------------
diff --git a/extensions/libarchive/UnfocusArchiveEntry.cpp b/extensions/libarchive/UnfocusArchiveEntry.cpp
index 1e433f0..840ad89 100644
--- a/extensions/libarchive/UnfocusArchiveEntry.cpp
+++ b/extensions/libarchive/UnfocusArchiveEntry.cpp
@@ -26,11 +26,9 @@
#include <string>
#include <set>
-#include <boost/filesystem.hpp>
-
#include <archive.h>
#include <archive_entry.h>
-
+#include "utils/file/FileManager.h"
#include "core/ProcessContext.h"
#include "core/ProcessSession.h"
#include "json/json.h"
@@ -60,134 +58,128 @@ void UnfocusArchiveEntry::initialize() {
setSupportedRelationships(relationships);
}
-void UnfocusArchiveEntry::onTrigger(
- core::ProcessContext *context,
- core::ProcessSession *session) {
+void UnfocusArchiveEntry::onTrigger(core::ProcessContext *context, core::ProcessSession *session) {
- auto flowFile = session->get();
+ auto flowFile = session->get();
- if (!flowFile) {
- return;
- }
+ if (!flowFile) {
+ return;
+ }
- // Get lens stack from attribute
- ArchiveMetadata lensArchiveMetadata;
+ fileutils::FileManager file_man;
- Json::Value lensStack;
- Json::Reader reader;
+ // Get lens stack from attribute
+ ArchiveMetadata lensArchiveMetadata;
- std::string existingLensStack;
+ Json::Value lensStack;
+ Json::Reader reader;
- if (flowFile->getAttribute("lens.archive.stack", existingLensStack)) {
- logger_->log_info("UnfocusArchiveEntry loading existing lens context");
+ std::string existingLensStack;
- // TODO(calebj) handle any exceptions that might arise from working with JSON data
- if (!reader.parse(existingLensStack, lensStack)) {
- logger_->log_error("UnfocusArchiveEntry JSON parse error: %s",
- reader.getFormattedErrorMessages());
- context->yield();
- return;
- }
+ if (flowFile->getAttribute("lens.archive.stack", existingLensStack)) {
+ logger_->log_info("UnfocusArchiveEntry loading existing lens context");
- } else {
- logger_->log_error("UnfocusArchiveEntry lens metadata not found");
+ // TODO(calebj) handle any exceptions that might arise from working with JSON data
+ if (!reader.parse(existingLensStack, lensStack)) {
+ logger_->log_error("UnfocusArchiveEntry JSON parse error: %s", reader.getFormattedErrorMessages());
context->yield();
return;
}
- Json::Value metadataDoc = lensStack[lensStack.size() - 1];
- lensStack.resize(lensStack.size() - 1);
-
- lensArchiveMetadata.archiveFormatName.assign(
- metadataDoc["archive_format_name"].asString());
- lensArchiveMetadata.archiveFormat =
- metadataDoc["archive_format"].asUInt64();
- lensArchiveMetadata.focusedEntry =
- metadataDoc["focused_entry"].asString();
-
- for (auto itr = metadataDoc["archive_structure"].begin();
- itr != metadataDoc["archive_structure"].end(); ++itr) {
- const auto &entryVal = *itr;
- ArchiveEntryMetadata metadata;
- metadata.tmpFileName = boost::filesystem::unique_path("/tmp/%%%%-%%%%-%%%%-%%%%").native();
- metadata.entryName.assign(entryVal["entry_name"].asString());
- metadata.entryType = entryVal["entry_type"].asUInt64();
- metadata.entryPerm = entryVal["entry_perm"].asUInt64();
- metadata.entrySize = entryVal["entry_size"].asUInt64();
- metadata.entryUID = entryVal["entry_uid"].asUInt64();
- metadata.entryGID = entryVal["entry_gid"].asUInt64();
- metadata.entryMTime = entryVal["entry_mtime"].asUInt64();
- metadata.entryMTimeNsec = entryVal["entry_mtime_nsec"].asInt64();
-
- if (metadata.entryType == AE_IFREG)
- metadata.stashKey.assign(entryVal["stash_key"].asString());
-
- lensArchiveMetadata.entryMetadata.push_back(metadata);
- }
+ } else {
+ logger_->log_error("UnfocusArchiveEntry lens metadata not found");
+ context->yield();
+ return;
+ }
- // Export focused entry to tmp file
- for (const auto &entry : lensArchiveMetadata.entryMetadata) {
- if (entry.entryType != AE_IFREG || entry.entrySize == 0) {
- continue;
- }
+ Json::Value metadataDoc = lensStack[lensStack.size() - 1];
+ lensStack.resize(lensStack.size() - 1);
+
+ lensArchiveMetadata.archiveFormatName.assign(metadataDoc["archive_format_name"].asString());
+ lensArchiveMetadata.archiveFormat = metadataDoc["archive_format"].asUInt64();
+ lensArchiveMetadata.focusedEntry = metadataDoc["focused_entry"].asString();
+
+ for (auto itr = metadataDoc["archive_structure"].begin(); itr != metadataDoc["archive_structure"].end(); ++itr) {
+ const auto &entryVal = *itr;
+ ArchiveEntryMetadata metadata;
+ metadata.tmpFileName = file_man.unique_file("/tmp/", true);
+ metadata.entryName.assign(entryVal["entry_name"].asString());
+ metadata.entryType = entryVal["entry_type"].asUInt64();
+ metadata.entryPerm = entryVal["entry_perm"].asUInt64();
+ metadata.entrySize = entryVal["entry_size"].asUInt64();
+ metadata.entryUID = entryVal["entry_uid"].asUInt64();
+ metadata.entryGID = entryVal["entry_gid"].asUInt64();
+ metadata.entryMTime = entryVal["entry_mtime"].asUInt64();
+ metadata.entryMTimeNsec = entryVal["entry_mtime_nsec"].asInt64();
+
+ if (metadata.entryType == AE_IFREG)
+ metadata.stashKey.assign(entryVal["stash_key"].asString());
+
+ lensArchiveMetadata.entryMetadata.push_back(metadata);
+ }
- if (entry.entryName == lensArchiveMetadata.focusedEntry) {
- logger_->log_debug("UnfocusArchiveEntry exporting focused entry to %s", entry.tmpFileName);
- session->exportContent(entry.tmpFileName, flowFile, false);
- }
+ // Export focused entry to tmp file
+ for (const auto &entry : lensArchiveMetadata.entryMetadata) {
+ if (entry.entryType != AE_IFREG || entry.entrySize == 0) {
+ continue;
}
- // Restore/export entries from stash, one-by-one, to tmp files
- for (const auto &entry : lensArchiveMetadata.entryMetadata) {
- if (entry.entryType != AE_IFREG || entry.entrySize == 0) {
- continue;
- }
+ if (entry.entryName == lensArchiveMetadata.focusedEntry) {
+ logger_->log_debug("UnfocusArchiveEntry exporting focused entry to %s", entry.tmpFileName);
+ session->exportContent(entry.tmpFileName, flowFile, false);
+ }
+ }
- if (entry.entryName == lensArchiveMetadata.focusedEntry) {
- continue;
- }
+ // Restore/export entries from stash, one-by-one, to tmp files
+ for (const auto &entry : lensArchiveMetadata.entryMetadata) {
+ if (entry.entryType != AE_IFREG || entry.entrySize == 0) {
+ continue;
+ }
- logger_->log_debug("UnfocusArchiveEntry exporting entry %s to %s", entry.stashKey, entry.tmpFileName);
- session->restore(entry.stashKey, flowFile);
- // TODO(calebj) implement copy export/don't worry about multiple claims/optimal efficiency for *now*
- session->exportContent(entry.tmpFileName, flowFile, false);
+ if (entry.entryName == lensArchiveMetadata.focusedEntry) {
+ continue;
}
- // Create archive by restoring each entry in the archive from tmp files
- WriteCallback cb(&lensArchiveMetadata);
- session->write(flowFile, &cb);
+ logger_->log_debug("UnfocusArchiveEntry exporting entry %s to %s", entry.stashKey, entry.tmpFileName);
+ session->restore(entry.stashKey, flowFile);
+ // TODO(calebj) implement copy export/don't worry about multiple claims/optimal efficiency for *now*
+ session->exportContent(entry.tmpFileName, flowFile, false);
+ }
- // Set new/updated lens stack to attribute (already 'popped' with resize())
- Json::FastWriter writer;
- std::string stackStr = writer.write(lensStack);
+ // Create archive by restoring each entry in the archive from tmp files
+ WriteCallback cb(&lensArchiveMetadata);
+ session->write(flowFile, &cb);
- if (!flowFile->updateAttribute("lens.archive.stack", stackStr)) {
- flowFile->addAttribute("lens.archive.stack", stackStr);
- }
+ // Set new/updated lens stack to attribute (already 'popped' with resize())
+ Json::FastWriter writer;
+ std::string stackStr = writer.write(lensStack);
- // Update filename to that of original archive
- Json::Value archiveName = metadataDoc.get("archive_name", Json::Value(Json::nullValue));
-
- if (archiveName.isNull()) {
- flowFile->removeAttribute("filename");
- flowFile->removeAttribute("path");
- flowFile->removeAttribute("absolute.path");
- } else {
- std::string abs_path = archiveName.asString();
- std::size_t found = abs_path.find_last_of("/\\");
- std::string path = abs_path.substr(0, found);
- std::string name = abs_path.substr(found + 1);
- set_or_update_attr(flowFile, "filename", name);
- set_or_update_attr(flowFile, "path", path);
- set_or_update_attr(flowFile, "absolute.path", abs_path);
- }
+ if (!flowFile->updateAttribute("lens.archive.stack", stackStr)) {
+ flowFile->addAttribute("lens.archive.stack", stackStr);
+ }
+
+ // Update filename to that of original archive
+ Json::Value archiveName = metadataDoc.get("archive_name", Json::Value(Json::nullValue));
+
+ if (archiveName.isNull()) {
+ flowFile->removeAttribute("filename");
+ flowFile->removeAttribute("path");
+ flowFile->removeAttribute("absolute.path");
+ } else {
+ std::string abs_path = archiveName.asString();
+ std::size_t found = abs_path.find_last_of("/\\");
+ std::string path = abs_path.substr(0, found);
+ std::string name = abs_path.substr(found + 1);
+ set_or_update_attr(flowFile, "filename", name);
+ set_or_update_attr(flowFile, "path", path);
+ set_or_update_attr(flowFile, "absolute.path", abs_path);
+ }
- // Transfer to the relationship
- session->transfer(flowFile, Success);
+ // Transfer to the relationship
+ session->transfer(flowFile, Success);
}
-UnfocusArchiveEntry::WriteCallback::WriteCallback(
- ArchiveMetadata *archiveMetadata) {
+UnfocusArchiveEntry::WriteCallback::WriteCallback(ArchiveMetadata *archiveMetadata) {
logger_ = logging::LoggerFactory<UnfocusArchiveEntry>::getLogger();
_archiveMetadata = archiveMetadata;
}
@@ -219,17 +211,15 @@ int64_t UnfocusArchiveEntry::WriteCallback::process(std::shared_ptr<io::BaseStre
struct archive_entry* entry;
for (const auto &entryMetadata : _archiveMetadata->entryMetadata) {
- entry = archive_entry_new();
- logger_->log_info("UnfocusArchiveEntry writing entry %s",
- entryMetadata.entryName.c_str());
-
+ entry = archive_entry_new();
+ logger_->log_info("UnfocusArchiveEntry writing entry %s", entryMetadata.entryName.c_str());
if (entryMetadata.entryType == AE_IFREG && entryMetadata.entrySize > 0) {
- size_t stat_ok = stat(entryMetadata.tmpFileName.c_str(), &st);
- if (stat_ok != 0) {
- logger_->log_error("Error statting %s: %d", entryMetadata.tmpFileName.c_str(), stat_ok);
- }
- archive_entry_copy_stat(entry, &st);
+ size_t stat_ok = stat(entryMetadata.tmpFileName.c_str(), &st);
+ if (stat_ok != 0) {
+ logger_->log_error("Error statting %s: %d", entryMetadata.tmpFileName, stat_ok);
+ }
+ archive_entry_copy_stat(entry, &st);
}
archive_entry_set_filetype(entry, entryMetadata.entryType);
@@ -240,28 +230,17 @@ int64_t UnfocusArchiveEntry::WriteCallback::process(std::shared_ptr<io::BaseStre
archive_entry_set_gid(entry, entryMetadata.entryGID);
archive_entry_set_mtime(entry, entryMetadata.entryMTime, entryMetadata.entryMTimeNsec);
- logger_->log_info("Writing %s with type %d, perms %d, size %d, uid %d, gid %d, mtime %d,%d",
- entryMetadata.entryName.c_str(),
- entryMetadata.entryType,
- entryMetadata.entryPerm,
- entryMetadata.entrySize,
- entryMetadata.entryUID,
- entryMetadata.entryGID,
- entryMetadata.entryMTime,
- entryMetadata.entryMTimeNsec);
+ logger_->log_info("Writing %s with type %d, perms %d, size %d, uid %d, gid %d, mtime %d,%d", entryMetadata.entryName.c_str(), entryMetadata.entryType, entryMetadata.entryPerm,
+ entryMetadata.entrySize, entryMetadata.entryUID, entryMetadata.entryGID, entryMetadata.entryMTime, entryMetadata.entryMTimeNsec);
archive_write_header(outputArchive, entry);
// If entry is regular file, copy entry contents
if (entryMetadata.entryType == AE_IFREG && entryMetadata.entrySize > 0) {
logger_->log_info("UnfocusArchiveEntry writing %d bytes of "
- "data from tmp file %s to archive entry %s",
- st.st_size,
- entryMetadata.tmpFileName.c_str(),
- entryMetadata.entryName.c_str());
- std::ifstream ifs(
- entryMetadata.tmpFileName,
- std::ifstream::in | std::ios::binary);
+ "data from tmp file %s to archive entry %s",
+ st.st_size, entryMetadata.tmpFileName.c_str(), entryMetadata.entryName.c_str());
+ std::ifstream ifs(entryMetadata.tmpFileName, std::ifstream::in | std::ios::binary);
while (ifs.good()) {
ifs.read(buf, sizeof(buf));
@@ -269,9 +248,8 @@ int64_t UnfocusArchiveEntry::WriteCallback::process(std::shared_ptr<io::BaseStre
int64_t written = archive_write_data(outputArchive, buf, len);
if (written < 0) {
logger_->log_error("UnfocusArchiveEntry failed to write data to "
- "archive entry %s due to error: %s",
- entryMetadata.entryName.c_str(),
- archive_error_string(outputArchive));
+ "archive entry %s due to error: %s",
+ entryMetadata.entryName.c_str(), archive_error_string(outputArchive));
} else {
nlen += written;
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a698084c/extensions/script/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/extensions/script/CMakeLists.txt b/extensions/script/CMakeLists.txt
index 11f38df..7475871 100644
--- a/extensions/script/CMakeLists.txt
+++ b/extensions/script/CMakeLists.txt
@@ -92,11 +92,6 @@ endif()
find_package(ZLIB REQUIRED)
include_directories(${ZLIB_INCLUDE_DIRS})
target_link_libraries (minifi-script-extensions ${ZLIB_LIBRARIES})
-find_package(Boost COMPONENTS system filesystem REQUIRED)
-include_directories(${Boost_INCLUDE_DIRS})
-target_link_libraries(minifi-script-extensions ${Boost_SYSTEM_LIBRARY})
-target_link_libraries(minifi-script-extensions ${Boost_FILESYSTEM_LIBRARY})
-
if (WIN32)
set_target_properties(minifi-script-extensions PROPERTIES
LINK_FLAGS "/WHOLEARCHIVE"
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a698084c/libminifi/include/core/Core.h
----------------------------------------------------------------------
diff --git a/libminifi/include/core/Core.h b/libminifi/include/core/Core.h
index 80b1ca4..ef31ecc 100644
--- a/libminifi/include/core/Core.h
+++ b/libminifi/include/core/Core.h
@@ -35,6 +35,8 @@ namespace apache {
namespace nifi {
namespace minifi {
namespace utils {
+namespace file{
+}
}
namespace processors {
}
@@ -179,6 +181,8 @@ namespace logging {
}
}
+namespace fileutils = org::apache::nifi::minifi::utils::file;
+
namespace minifi = org::apache::nifi::minifi;
namespace core = org::apache::nifi::minifi::core;
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a698084c/libminifi/include/utils/StringUtils.h
----------------------------------------------------------------------
diff --git a/libminifi/include/utils/StringUtils.h b/libminifi/include/utils/StringUtils.h
index b404928..df48086 100644
--- a/libminifi/include/utils/StringUtils.h
+++ b/libminifi/include/utils/StringUtils.h
@@ -141,6 +141,16 @@ class StringUtils {
}
+ static std::string& replaceAll(std::string& source_string, const std::string &from_string, const std::string &to_string) {
+ std::size_t loc = 0;
+ std::size_t lastFound;
+ while ((lastFound = source_string.find(from_string, loc)) != std::string::npos) {
+ source_string.replace(lastFound, from_string.size(), to_string);
+ loc = lastFound + to_string.size();
+ }
+ return source_string;
+ }
+
};
} /* namespace utils */
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a698084c/libminifi/include/utils/file/FileManager.h
----------------------------------------------------------------------
diff --git a/libminifi/include/utils/file/FileManager.h b/libminifi/include/utils/file/FileManager.h
new file mode 100644
index 0000000..f6a4312
--- /dev/null
+++ b/libminifi/include/utils/file/FileManager.h
@@ -0,0 +1,113 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#ifndef LIBMINIFI_INCLUDE_UTILS_FILEMANAGER_H_
+#define LIBMINIFI_INCLUDE_UTILS_FILEMANAGER_H_
+
+#ifdef BOOST_VERSION
+#include <boost/filesystem.hpp>
+#else
+#include <cstdlib>
+#endif
+#include <cstdio>
+#include <unistd.h>
+#include <fcntl.h>
+#include "io/validation.h"
+#include "utils/Id.h"
+#include "utils/StringUtils.h"
+#ifdef WIN32
+#define stat _stat
+#endif
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace utils {
+namespace file {
+
+/**
+ * Simple implementation of simple file manager utilities.
+ *
+ * unique_file is not a static implementation so that we can support scope driven temporary files.
+ */
+class FileManager {
+ public:
+
+ FileManager() {
+ }
+
+ ~FileManager() {
+ for (auto file : unique_files_) {
+ unlink(file.c_str());
+ }
+ }
+ std::string unique_file(const std::string &location, bool keep = false) {
+
+ if (!IsNullOrEmpty(location)) {
+ std::string file_name = location + "/" + non_repeating_string_generator_.generate();
+ while (!verify_not_exist(file_name)) {
+ file_name = location + "/" + non_repeating_string_generator_.generate();
+ }
+ if (!keep)
+ unique_files_.push_back(file_name);
+ return file_name;
+ } else {
+ std::string file_name = "/tmp/" + non_repeating_string_generator_.generate();
+ while (!verify_not_exist(file_name)) {
+ file_name = "/tmp/" + non_repeating_string_generator_.generate();
+ }
+ if (!keep)
+ unique_files_.push_back(file_name);
+ return file_name;
+ }
+ }
+
+ std::string unique_file(bool keep = false) {
+#ifdef BOOST_VERSION
+ return boost::filesystem::unique_path().native();
+#else
+ std::string file_name = "/tmp/" + non_repeating_string_generator_.generate();
+ while (!verify_not_exist(file_name)) {
+ file_name = "/tmp/" + non_repeating_string_generator_.generate();
+ }
+ if (!keep)
+ unique_files_.push_back(file_name);
+ return file_name;
+#endif
+ }
+
+
+ protected:
+
+ inline bool verify_not_exist(const std::string& name) {
+ struct stat buffer;
+ return (stat(name.c_str(), &buffer) != 0);
+ }
+
+ utils::NonRepeatingStringGenerator non_repeating_string_generator_;
+
+ std::vector<std::string> unique_files_;
+};
+
+} /* namespace file */
+} /* namespace utils */
+} /* namespace minifi */
+} /* namespace nifi */
+} /* namespace apache */
+} /* namespace org */
+
+#endif /* LIBMINIFI_INCLUDE_UTILS_FILEMANAGER_H_ */
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a698084c/libminifi/include/utils/file/FileUtils.h
----------------------------------------------------------------------
diff --git a/libminifi/include/utils/file/FileUtils.h b/libminifi/include/utils/file/FileUtils.h
new file mode 100644
index 0000000..2011d7e
--- /dev/null
+++ b/libminifi/include/utils/file/FileUtils.h
@@ -0,0 +1,69 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#ifndef LIBMINIFI_INCLUDE_UTILS_FILEUTILS_H_
+#define LIBMINIFI_INCLUDE_UTILS_FILEUTILS_H_
+
+#ifdef BOOST_VERSION
+#include <boost/filesystem.hpp>
+#else
+#include <cstdlib>
+#endif
+#include <cstdio>
+#include <unistd.h>
+#include <fcntl.h>
+#ifdef WIN32
+#define stat _stat
+#endif
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace utils {
+namespace file {
+
+/**
+ * Simple implementation of some file system utilities.
+ *
+ */
+class FileUtils {
+ public:
+
+ FileUtils() = delete;
+
+ static uint64_t last_write_time(const std::string &path) {
+#ifdef BOOST_VERSION
+ return boost::filesystem::last_write_time(movedFile.str());
+#else
+ struct stat result;
+ if (stat(path.c_str(), &result) == 0) {
+ return result.st_mtime;
+ }
+#endif
+ return 0;
+ }
+
+};
+
+} /* namespace file */
+} /* namespace utils */
+} /* namespace minifi */
+} /* namespace nifi */
+} /* namespace apache */
+} /* namespace org */
+
+#endif /* LIBMINIFI_INCLUDE_UTILS_FILEUTILS_H_ */
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a698084c/libminifi/test/unit/PutFileTests.cpp
----------------------------------------------------------------------
diff --git a/libminifi/test/unit/PutFileTests.cpp b/libminifi/test/unit/PutFileTests.cpp
index 37f451a..fa3d185 100644
--- a/libminifi/test/unit/PutFileTests.cpp
+++ b/libminifi/test/unit/PutFileTests.cpp
@@ -24,8 +24,8 @@
#include <set>
#include <fstream>
-#include <boost/filesystem.hpp>
+#include "utils/file/FileUtils.h"
#include "../TestBase.h"
#include "processors/ListenHTTP.h"
#include "processors/LogAttribute.h"
@@ -220,7 +220,7 @@ TEST_CASE("PutFileTestFileExistsIgnore", "[getfileputpfile]") {
file.open(movedFile.str(), std::ios::out);
file << "tempFile";
file.close();
- auto filemodtime = boost::filesystem::last_write_time(movedFile.str());
+ auto filemodtime = fileutils::FileUtils::last_write_time(movedFile.str());
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
plan->reset();
@@ -241,7 +241,7 @@ TEST_CASE("PutFileTestFileExistsIgnore", "[getfileputpfile]") {
// verify that the fle was moved
REQUIRE(false == std::ifstream(ss.str()).good());
REQUIRE(true == std::ifstream(movedFile.str()).good());
- REQUIRE(filemodtime == boost::filesystem::last_write_time(movedFile.str()));
+ REQUIRE(filemodtime == fileutils::FileUtils::last_write_time(movedFile.str()));
LogTestController::getInstance().reset();
}
@@ -288,7 +288,7 @@ TEST_CASE("PutFileTestFileExistsReplace", "[getfileputpfile]") {
file.open(movedFile.str(), std::ios::out);
file << "tempFile";
file.close();
- auto filemodtime = boost::filesystem::last_write_time(movedFile.str());
+ auto filemodtime = fileutils::FileUtils::last_write_time(movedFile.str());
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
plan->reset();
@@ -309,7 +309,7 @@ TEST_CASE("PutFileTestFileExistsReplace", "[getfileputpfile]") {
// verify that the fle was moved
REQUIRE(false == std::ifstream(ss.str()).good());
REQUIRE(true == std::ifstream(movedFile.str()).good());
- REQUIRE(filemodtime != boost::filesystem::last_write_time(movedFile.str()));
+ REQUIRE(filemodtime != fileutils::FileUtils::last_write_time(movedFile.str()));
LogTestController::getInstance().reset();
}