This is an automated email from the ASF dual-hosted git repository. szaszm pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 24d9f3a53e66c9ea2f3c24f5445c164cce01a727 Author: Adam Debreceni <[email protected]> AuthorDate: Tue Jan 4 15:48:37 2022 +0100 MINIFICPP-1698 Make archive read/write agent-wide available Closes #1224 Signed-off-by: Marton Szasz <[email protected]> --- extensions/libarchive/ArchiveStreamProvider.cpp | 49 ++++ extensions/libarchive/CompressContent.cpp | 77 ++++-- extensions/libarchive/CompressContent.h | 276 +-------------------- extensions/libarchive/ReadArchiveStream.cpp | 88 +++++++ extensions/libarchive/ReadArchiveStream.h | 89 +++++++ extensions/libarchive/WriteArchiveStream.cpp | 131 ++++++++++ extensions/libarchive/WriteArchiveStream.h | 86 +++++++ libminifi/include/core/ProcessSession.h | 8 + libminifi/include/io/ArchiveStream.h | 54 ++++ libminifi/include/io/StreamPipe.h | 28 ++- .../test/archive-tests/ArchiveStreamTests.cpp | 55 ++++ 11 files changed, 646 insertions(+), 295 deletions(-) diff --git a/extensions/libarchive/ArchiveStreamProvider.cpp b/extensions/libarchive/ArchiveStreamProvider.cpp new file mode 100644 index 0000000..776a104 --- /dev/null +++ b/extensions/libarchive/ArchiveStreamProvider.cpp @@ -0,0 +1,49 @@ +/** + * + * 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 <memory> + +#include "core/Resource.h" +#include "WriteArchiveStream.h" +#include "ReadArchiveStream.h" + +namespace org::apache::nifi::minifi::io { + +class ArchiveStreamProviderImpl : public ArchiveStreamProvider { + public: + using ArchiveStreamProvider::ArchiveStreamProvider; + std::unique_ptr<WriteArchiveStream> createWriteStream(int compress_level, const std::string& compress_format, + std::shared_ptr<OutputStream> sink, std::shared_ptr<core::logging::Logger> logger) override { + CompressionFormat format = CompressionFormat::parse(compress_format.c_str(), CompressionFormat{}); + if (!format) { + if (logger) { + logger->log_error("Unrecognized compression format '%s'", compress_format); + } + return nullptr; + } + return std::make_unique<WriteArchiveStreamImpl>(compress_level, format, std::move(sink)); + } + + std::unique_ptr<ReadArchiveStream> createReadStream(std::shared_ptr<InputStream> archive_stream) override { + return std::make_unique<ReadArchiveStreamImpl>(std::move(archive_stream)); + } +}; + +REGISTER_INTERNAL_RESOURCE_AS(ArchiveStreamProviderImpl, ("ArchiveStreamProvider")); + +} // namespace org::apache::nifi::minifi::io diff --git a/extensions/libarchive/CompressContent.cpp b/extensions/libarchive/CompressContent.cpp index 12e2016..421bc8d 100644 --- a/extensions/libarchive/CompressContent.cpp +++ b/extensions/libarchive/CompressContent.cpp @@ -27,6 +27,7 @@ #include "core/ProcessSession.h" #include "utils/StringUtils.h" #include "core/Resource.h" +#include "io/StreamPipe.h" namespace org { namespace apache { @@ -66,19 +67,19 @@ core::Relationship CompressContent::Failure("failure", "FlowFiles will be transf const std::string CompressContent::TAR_EXT = ".tar"; -const std::map<std::string, CompressContent::CompressionFormat> CompressContent::compressionFormatMimeTypeMap_{ - {"application/gzip", CompressionFormat::GZIP}, - {"application/bzip2", CompressionFormat::BZIP2}, - {"application/x-bzip2", CompressionFormat::BZIP2}, - {"application/x-lzma", CompressionFormat::LZMA}, - {"application/x-xz", CompressionFormat::XZ_LZMA2} +const std::map<std::string, io::CompressionFormat> CompressContent::compressionFormatMimeTypeMap_{ + {"application/gzip", io::CompressionFormat::GZIP}, + {"application/bzip2", io::CompressionFormat::BZIP2}, + {"application/x-bzip2", io::CompressionFormat::BZIP2}, + {"application/x-lzma", io::CompressionFormat::LZMA}, + {"application/x-xz", io::CompressionFormat::XZ_LZMA2} }; -const std::map<CompressContent::CompressionFormat, std::string> CompressContent::fileExtension_{ - {CompressionFormat::GZIP, ".gz"}, - {CompressionFormat::LZMA, ".lzma"}, - {CompressionFormat::BZIP2, ".bz2"}, - {CompressionFormat::XZ_LZMA2, ".xz"} +const std::map<io::CompressionFormat, std::string> CompressContent::fileExtension_{ + {io::CompressionFormat::GZIP, ".gz"}, + {io::CompressionFormat::LZMA, ".lzma"}, + {io::CompressionFormat::BZIP2, ".bz2"}, + {io::CompressionFormat::XZ_LZMA2, ".xz"} }; void CompressContent::initialize() { @@ -129,7 +130,7 @@ void CompressContent::onTrigger(const std::shared_ptr<core::ProcessContext> &con void CompressContent::processFlowFile(const std::shared_ptr<core::FlowFile>& flowFile, const std::shared_ptr<core::ProcessSession>& session) { session->remove(flowFile); - CompressionFormat compressFormat; + io::CompressionFormat compressFormat; if (compressFormat_ == ExtendedCompressionFormat::USE_MIME_TYPE) { std::string attr; flowFile->getAttribute(core::SpecialFlowAttribute::MIME_TYPE, attr); @@ -147,22 +148,22 @@ void CompressContent::processFlowFile(const std::shared_ptr<core::FlowFile>& flo return; } } else { - compressFormat = compressFormat_.cast<CompressionFormat>(); + compressFormat = compressFormat_.cast<io::CompressionFormat>(); } std::string mimeType = toMimeType(compressFormat); // Validate - if (!encapsulateInTar_ && compressFormat != CompressionFormat::GZIP) { + if (!encapsulateInTar_ && compressFormat != io::CompressionFormat::GZIP) { logger_->log_error("non-TAR encapsulated format only supports GZIP compression"); session->transfer(flowFile, Failure); return; } - if (compressFormat == CompressionFormat::BZIP2 && archive_bzlib_version() == nullptr) { + if (compressFormat == io::CompressionFormat::BZIP2 && archive_bzlib_version() == nullptr) { logger_->log_error("%s compression format is requested, but the agent was compiled without BZip2 support", compressFormat.toString()); session->transfer(flowFile, Failure); return; } - if ((compressFormat == CompressionFormat::LZMA || compressFormat == CompressionFormat::XZ_LZMA2) && archive_liblzma_version() == nullptr) { + if ((compressFormat == io::CompressionFormat::LZMA || compressFormat == io::CompressionFormat::XZ_LZMA2) && archive_liblzma_version() == nullptr) { logger_->log_error("%s compression format is requested, but the agent was compiled without LZMA support ", compressFormat.toString()); session->transfer(flowFile, Failure); return; @@ -176,9 +177,37 @@ void CompressContent::processFlowFile(const std::shared_ptr<core::FlowFile>& flo std::shared_ptr<core::FlowFile> result = session->create(flowFile); bool success = false; if (encapsulateInTar_) { - CompressContent::WriteCallback callback(compressMode_, compressLevel_, compressFormat, flowFile, session); - session->write(result, &callback); - success = callback.status_ >= 0; + std::function<int64_t(const std::shared_ptr<io::InputStream>&, const std::shared_ptr<io::OutputStream>&)> transformer; + + if (compressMode_ == CompressionMode::Compress) { + std::string filename; + flowFile->getAttribute(core::SpecialFlowAttribute::FILENAME, filename); + transformer = [&, filename] (const std::shared_ptr<io::InputStream>& in, const std::shared_ptr<io::OutputStream>& out) -> int64_t { + io::WriteArchiveStreamImpl compressor(compressLevel_, compressFormat, out); + if (!compressor.newEntry({filename, in->size()})) { + return -1; + } + return internal::pipe(in.get(), &compressor); + }; + } else { + transformer = [&] (const std::shared_ptr<io::InputStream>& in, const std::shared_ptr<io::OutputStream>& out) -> int64_t { + io::ReadArchiveStreamImpl decompressor(in); + if (!decompressor.nextEntry()) { + return -1; + } + return internal::pipe(&decompressor, out.get()); + }; + } + session->write(result, FunctionOutputStreamCallback([&] (const auto& out) { + return session->read(flowFile, FunctionInputStreamCallback([&] (const auto& in) { + return transformer(in, out); + })); + })); + // TODO(adebreceni): previous attempt to handle a malformed archive were in vain + // as the session->read threw anyway rolling back the flowfile, we should correctly + // forward a malformed archive to failure + // https://issues.apache.org/jira/browse/MINIFICPP-1708 + success = true; } else { CompressContent::GzipWriteCallback callback(compressMode_, compressLevel_, flowFile, session); session->write(result, &callback); @@ -218,12 +247,12 @@ void CompressContent::processFlowFile(const std::shared_ptr<core::FlowFile>& flo } } -std::string CompressContent::toMimeType(CompressionFormat format) { +std::string CompressContent::toMimeType(io::CompressionFormat format) { switch (format.value()) { - case CompressionFormat::GZIP: return "application/gzip"; - case CompressionFormat::BZIP2: return "application/bzip2"; - case CompressionFormat::LZMA: return "application/x-lzma"; - case CompressionFormat::XZ_LZMA2: return "application/x-xz"; + case io::CompressionFormat::GZIP: return "application/gzip"; + case io::CompressionFormat::BZIP2: return "application/bzip2"; + case io::CompressionFormat::LZMA: return "application/x-lzma"; + case io::CompressionFormat::XZ_LZMA2: return "application/x-xz"; } throw Exception(GENERAL_EXCEPTION, "Invalid compression format"); } diff --git a/extensions/libarchive/CompressContent.h b/extensions/libarchive/CompressContent.h index bd0c3ba..7e9358b 100644 --- a/extensions/libarchive/CompressContent.h +++ b/extensions/libarchive/CompressContent.h @@ -39,6 +39,8 @@ #include "utils/Enum.h" #include "utils/gsl.h" #include "utils/Export.h" +#include "WriteArchiveStream.h" +#include "ReadArchiveStream.h" namespace org { namespace apache { @@ -81,277 +83,11 @@ class CompressContent : public core::Processor { (Decompress, "decompress") ) - SMART_ENUM(CompressionFormat, - (GZIP, "gzip"), - (LZMA, "lzma"), - (XZ_LZMA2, "xz-lzma2"), - (BZIP2, "bzip2") - ) - - SMART_ENUM_EXTEND(ExtendedCompressionFormat, CompressionFormat, (GZIP, LZMA, XZ_LZMA2, BZIP2), + SMART_ENUM_EXTEND(ExtendedCompressionFormat, io::CompressionFormat, (GZIP, LZMA, XZ_LZMA2, BZIP2), (USE_MIME_TYPE, "use mime.type attribute") ) public: - // Nest Callback Class for read stream from flow for compress - class ReadCallbackCompress: public InputStreamCallback { - public: - ReadCallbackCompress(std::shared_ptr<core::FlowFile> &flow, struct archive *arch, struct archive_entry *entry) : - flow_(flow), arch_(arch), entry_(entry), status_(0) { - } - ~ReadCallbackCompress() override = default; - int64_t process(const std::shared_ptr<io::BaseStream>& stream) override { - uint8_t buffer[4096U]; - int64_t ret = 0; - uint64_t read_size = 0; - - ret = archive_write_header(arch_, entry_); - if (ret != ARCHIVE_OK) { - logger_->log_error("Compress Content archive error %s", archive_error_string(arch_)); - status_ = -1; - return -1; - } - while (read_size < flow_->getSize()) { - const auto readret = stream->read(buffer, sizeof(buffer)); - if (io::isError(readret)) { - status_ = -1; - return -1; - } - if (readret > 0) { - ret = archive_write_data(arch_, buffer, readret); - if (ret < 0) { - logger_->log_error("Compress Content archive error %s", archive_error_string(arch_)); - status_ = -1; - return -1; - } - read_size += gsl::narrow<uint64_t>(ret); - } else { - break; - } - } - return gsl::narrow<int64_t>(read_size); - } - std::shared_ptr<core::FlowFile> flow_; - struct archive *arch_; - struct archive_entry *entry_; - int status_; - std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<CompressContent>::getLogger(); - }; - // Nest Callback Class for read stream from flow for decompress - struct ReadCallbackDecompress : InputStreamCallback { - explicit ReadCallbackDecompress(std::shared_ptr<core::FlowFile> flow) : - flow_file(std::move(flow)) { - } - ~ReadCallbackDecompress() override = default; - int64_t process(const std::shared_ptr<io::BaseStream>& stream) override { - stream->seek(offset); - const auto readRet = stream->read(buffer, sizeof(buffer)); - stream_read_result = readRet; - if (!io::isError(readRet)) { - offset += readRet; - } - return gsl::narrow<int64_t>(readRet); - } - size_t stream_read_result = 0; // read size or error code, to be checked with io::isError - uint8_t buffer[8192] = {0}; - size_t offset = 0; - std::shared_ptr<core::FlowFile> flow_file; - }; - // Nest Callback Class for write stream - class WriteCallback: public OutputStreamCallback { - public: - WriteCallback(CompressionMode compress_mode, int compress_level, CompressionFormat compress_format, - const std::shared_ptr<core::FlowFile> &flow, const std::shared_ptr<core::ProcessSession> &session) : - compress_mode_(compress_mode), compress_level_(compress_level), compress_format_(compress_format), - flow_(flow), session_(session), - readDecompressCb_(flow) { - size_ = 0; - stream_ = nullptr; - status_ = 0; - } - ~WriteCallback() = default; - - CompressionMode compress_mode_; - int compress_level_; - CompressionFormat compress_format_; - std::shared_ptr<core::FlowFile> flow_; - std::shared_ptr<core::ProcessSession> session_; - std::shared_ptr<io::BaseStream> stream_; - int64_t size_; - std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<CompressContent>::getLogger(); - CompressContent::ReadCallbackDecompress readDecompressCb_; - int status_; - - static la_ssize_t archive_write(struct archive* /*arch*/, void *context, const void *buff, size_t size) { - auto* const callback = static_cast<WriteCallback*>(context); - const auto ret = callback->stream_->write(reinterpret_cast<const uint8_t*>(buff), size); - if (!io::isError(ret)) callback->size_ += gsl::narrow<int64_t>(ret); - return io::isError(ret) ? -1 : gsl::narrow<la_ssize_t>(ret); - } - - static la_ssize_t archive_read(struct archive* archive, void *context, const void **buff) { - auto *callback = reinterpret_cast<WriteCallback *>(context); - callback->session_->read(callback->flow_, &callback->readDecompressCb_); - *buff = callback->readDecompressCb_.buffer; - if (io::isError(callback->readDecompressCb_.stream_read_result)) { - archive_set_error(archive, EIO, "Error reading flowfile"); - return -1; - } - return gsl::narrow<la_ssize_t>(callback->readDecompressCb_.stream_read_result); - } - - static la_int64_t archive_skip(struct archive* /*a*/, void* /*client_data*/, la_int64_t /*request*/) { - return 0; - } - - void archive_write_log_error_cleanup(struct archive *arch) { - logger_->log_error("Compress Content archive write error %s", archive_error_string(arch)); - status_ = -1; - archive_write_free(arch); - } - - void archive_read_log_error_cleanup(struct archive *arch) { - logger_->log_error("Compress Content archive read error %s", archive_error_string(arch)); - status_ = -1; - archive_read_free(arch); - } - - int64_t process(const std::shared_ptr<io::BaseStream>& stream) { - struct archive *arch; - int r; - - if (compress_mode_ == CompressionMode::Compress) { - arch = archive_write_new(); - if (!arch) { - status_ = -1; - return -1; - } - r = archive_write_set_format_ustar(arch); - if (r != ARCHIVE_OK) { - archive_write_log_error_cleanup(arch); - return -1; - } - if (compress_format_ == CompressionFormat::GZIP) { - r = archive_write_add_filter_gzip(arch); - if (r != ARCHIVE_OK) { - archive_write_log_error_cleanup(arch); - return -1; - } - std::string option; - option = "gzip:compression-level=" + std::to_string(compress_level_); - r = archive_write_set_options(arch, option.c_str()); - if (r != ARCHIVE_OK) { - archive_write_log_error_cleanup(arch); - return -1; - } - } else if (compress_format_ == CompressionFormat::BZIP2) { - r = archive_write_add_filter_bzip2(arch); - if (r != ARCHIVE_OK) { - archive_write_log_error_cleanup(arch); - return -1; - } - } else if (compress_format_ == CompressionFormat::LZMA) { - r = archive_write_add_filter_lzma(arch); - if (r != ARCHIVE_OK) { - archive_write_log_error_cleanup(arch); - return -1; - } - } else if (compress_format_ == CompressionFormat::XZ_LZMA2) { - r = archive_write_add_filter_xz(arch); - if (r != ARCHIVE_OK) { - archive_write_log_error_cleanup(arch); - return -1; - } - } else { - archive_write_log_error_cleanup(arch); - return -1; - } - r = archive_write_set_bytes_per_block(arch, 0); - if (r != ARCHIVE_OK) { - archive_write_log_error_cleanup(arch); - return -1; - } - this->stream_ = stream; - r = archive_write_open(arch, this, NULL, archive_write, NULL); - if (r != ARCHIVE_OK) { - archive_write_log_error_cleanup(arch); - return -1; - } - struct archive_entry *entry = archive_entry_new(); - if (!entry) { - archive_write_log_error_cleanup(arch); - return -1; - } - std::string fileName; - flow_->getAttribute(core::SpecialFlowAttribute::FILENAME, fileName); - archive_entry_set_pathname(entry, fileName.c_str()); - archive_entry_set_size(entry, flow_->getSize()); - archive_entry_set_mode(entry, S_IFREG | 0755); - ReadCallbackCompress readCb(flow_, arch, entry); - session_->read(flow_, &readCb); - if (readCb.status_ < 0) { - archive_entry_free(entry); - archive_write_log_error_cleanup(arch); - status_ = -1; - return -1; - } - archive_entry_free(entry); - archive_write_close(arch); - archive_write_free(arch); - return size_; - } else { - arch = archive_read_new(); - if (!arch) { - status_ = -1; - return -1; - } - r = archive_read_support_format_all(arch); - if (r != ARCHIVE_OK) { - archive_read_log_error_cleanup(arch); - return -1; - } - r = archive_read_support_filter_all(arch); - if (r != ARCHIVE_OK) { - archive_read_log_error_cleanup(arch); - return -1; - } - this->stream_ = stream; - r = archive_read_open2(arch, this, NULL, archive_read, archive_skip, NULL); - if (r != ARCHIVE_OK) { - archive_read_log_error_cleanup(arch); - return -1; - } - struct archive_entry *entry; - if (archive_read_next_header(arch, &entry) != ARCHIVE_OK) { - archive_read_log_error_cleanup(arch); - return -1; - } - int64_t entry_size = archive_entry_size(entry); - logger_->log_debug("Decompress Content archive entry size %" PRId64, entry_size); - size_ = 0; - while (size_ < entry_size) { - char buffer[8192]; - const auto read_result = archive_read_data(arch, buffer, sizeof(buffer)); - if (read_result < 0) { - archive_read_log_error_cleanup(arch); - return -1; - } - if (read_result == 0) - break; - size_ += read_result; - const auto write_result = stream_->write(reinterpret_cast<uint8_t*>(buffer), gsl::narrow<size_t>(read_result)); - if (io::isError(write_result)) { - archive_read_log_error_cleanup(arch); - return -1; - } - } - archive_read_close(arch); - archive_read_free(arch); - return size_; - } - } - }; - class GzipWriteCallback : public OutputStreamCallback { public: GzipWriteCallback(CompressionMode compress_mode, int compress_level, std::shared_ptr<core::FlowFile> flow, std::shared_ptr<core::ProcessSession> session) @@ -433,7 +169,7 @@ class CompressContent : public core::Processor { void initialize() override; private: - static std::string toMimeType(CompressionFormat format); + static std::string toMimeType(io::CompressionFormat format); void processFlowFile(const std::shared_ptr<core::FlowFile>& flowFile, const std::shared_ptr<core::ProcessSession>& session); @@ -448,8 +184,8 @@ class CompressContent : public core::Processor { bool updateFileName_; bool encapsulateInTar_; uint32_t batchSize_{1}; - static const std::map<std::string, CompressionFormat> compressionFormatMimeTypeMap_; - static const std::map<CompressionFormat, std::string> fileExtension_; + static const std::map<std::string, io::CompressionFormat> compressionFormatMimeTypeMap_; + static const std::map<io::CompressionFormat, std::string> fileExtension_; }; } /* namespace processors */ diff --git a/extensions/libarchive/ReadArchiveStream.cpp b/extensions/libarchive/ReadArchiveStream.cpp new file mode 100644 index 0000000..109e0e7 --- /dev/null +++ b/extensions/libarchive/ReadArchiveStream.cpp @@ -0,0 +1,88 @@ +/** + * + * 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 "ReadArchiveStream.h" + +namespace org::apache::nifi::minifi::io { + +ReadArchiveStreamImpl::archive_ptr ReadArchiveStreamImpl::createReadArchive() { + archive_ptr arch{archive_read_new()}; + if (!arch) { + logger_->log_error("Failed to create read archive"); + return nullptr; + } + + int result; + + result = archive_read_support_format_all(arch.get()); + if (result != ARCHIVE_OK) { + logger_->log_error("Archive read support format all error %s", archive_error_string(arch.get())); + return nullptr; + } + result = archive_read_support_filter_all(arch.get()); + if (result != ARCHIVE_OK) { + logger_->log_error("Archive read support filter all error %s", archive_error_string(arch.get())); + return nullptr; + } + result = archive_read_open2(arch.get(), &reader_, nullptr, archive_read, nullptr, nullptr); + if (result != ARCHIVE_OK) { + logger_->log_error("Archive read open error %s", archive_error_string(arch.get())); + return nullptr; + } + return arch; +} + +std::optional<EntryInfo> ReadArchiveStreamImpl::nextEntry() { + if (!arch_) { + return std::nullopt; + } + entry_size_.reset(); + struct archive_entry *entry; + int result = archive_read_next_header(arch_.get(), &entry); + if (result != ARCHIVE_OK) { + if (result != ARCHIVE_EOF) { + logger_->log_error("Archive read next header error %s", archive_error_string(arch_.get())); + } + return std::nullopt; + } + entry_size_ = gsl::narrow<size_t>(archive_entry_size(entry)); + logger_->log_debug("Archive entry size %zu", entry_size_.value()); + return EntryInfo{archive_entry_pathname(entry), entry_size_.value()}; +} + +size_t ReadArchiveStreamImpl::read(uint8_t* buf, size_t len) { + if (!arch_ || !entry_size_) { + return STREAM_ERROR; + } + + if (len == 0) { + return 0; + } + gsl_Expects(buf); + + const la_ssize_t result = archive_read_data(arch_.get(), buf, len); + if (result < 0) { + logger_->log_error("Archive read data error %s", archive_error_string(arch_.get())); + entry_size_.reset(); + arch_.reset(); + return STREAM_ERROR; + } + return result; +} + +} // namespace org::apache::nifi::minifi::io diff --git a/extensions/libarchive/ReadArchiveStream.h b/extensions/libarchive/ReadArchiveStream.h new file mode 100644 index 0000000..7ffaa0c --- /dev/null +++ b/extensions/libarchive/ReadArchiveStream.h @@ -0,0 +1,89 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include <memory> +#include <utility> + +#include "io/OutputStream.h" +#include "io/ArchiveStream.h" +#include "core/logging/LoggerConfiguration.h" + +#include "archive_entry.h" +#include "archive.h" + +namespace org::apache::nifi::minifi::io { + +class ReadArchiveStreamImpl : public ReadArchiveStream { + struct archive_read_deleter { + int operator()(struct archive* ptr) const { + return archive_read_free(ptr); + } + }; + using archive_ptr = std::unique_ptr<struct archive, archive_read_deleter>; + + class BufferedReader { + public: + explicit BufferedReader(std::shared_ptr<InputStream> input) : input_(std::move(input)) {} + + std::optional<gsl::span<const uint8_t>> readChunk() { + size_t result = input_->read(buffer_.data(), buffer_.size()); + if (io::isError(result)) { + return std::nullopt; + } + return gsl::span<const uint8_t>(buffer_.data(), result); + } + + private: + std::shared_ptr<InputStream> input_; + std::array<uint8_t, 4096> buffer_; + }; + + archive_ptr createReadArchive(); + + public: + explicit ReadArchiveStreamImpl(std::shared_ptr<InputStream> input) : reader_(std::move(input)) { + arch_ = createReadArchive(); + } + + std::optional<EntryInfo> nextEntry() override; + + using InputStream::read; + + size_t read(uint8_t* buf, size_t len) override; + + private: + static la_ssize_t archive_read(struct archive* archive, void *context, const void **buff) { + auto* const input = reinterpret_cast<BufferedReader*>(context); + auto opt_buffer = input->readChunk(); + if (!opt_buffer) { + archive_set_error(archive, EIO, "Error reading archive"); + return -1; + } + *buff = opt_buffer->data(); + return gsl::narrow<la_ssize_t>(opt_buffer->size()); + } + + std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<ReadArchiveStream>::getLogger(); + BufferedReader reader_; + archive_ptr arch_; + std::optional<size_t> entry_size_; +}; + +} // namespace org::apache::nifi::minifi::io diff --git a/extensions/libarchive/WriteArchiveStream.cpp b/extensions/libarchive/WriteArchiveStream.cpp new file mode 100644 index 0000000..af375bf --- /dev/null +++ b/extensions/libarchive/WriteArchiveStream.cpp @@ -0,0 +1,131 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "WriteArchiveStream.h" + +#include <utility> +#include <string> + +namespace org::apache::nifi::minifi::io { + +WriteArchiveStreamImpl::archive_ptr WriteArchiveStreamImpl::createWriteArchive() { + archive_ptr arch{archive_write_new()}; + if (!arch) { + logger_->log_error("Failed to create write archive"); + return nullptr; + } + + int result; + + result = archive_write_set_format_ustar(arch.get()); + if (result != ARCHIVE_OK) { + logger_->log_error("Archive write set format ustar error %s", archive_error_string(arch.get())); + return nullptr; + } + if (compress_format_ == CompressionFormat::GZIP) { + result = archive_write_add_filter_gzip(arch.get()); + if (result != ARCHIVE_OK) { + logger_->log_error("Archive write add filter gzip error %s", archive_error_string(arch.get())); + return nullptr; + } + std::string option = "gzip:compression-level=" + std::to_string(compress_level_); + result = archive_write_set_options(arch.get(), option.c_str()); + if (result != ARCHIVE_OK) { + logger_->log_error("Archive write set options error %s", archive_error_string(arch.get())); + return nullptr; + } + } else if (compress_format_ == CompressionFormat::BZIP2) { + result = archive_write_add_filter_bzip2(arch.get()); + if (result != ARCHIVE_OK) { + logger_->log_error("Archive write add filter bzip2 error %s", archive_error_string(arch.get())); + return nullptr; + } + } else if (compress_format_ == CompressionFormat::LZMA) { + result = archive_write_add_filter_lzma(arch.get()); + if (result != ARCHIVE_OK) { + logger_->log_error("Archive write add filter lzma error %s", archive_error_string(arch.get())); + return nullptr; + } + } else if (compress_format_ == CompressionFormat::XZ_LZMA2) { + result = archive_write_add_filter_xz(arch.get()); + if (result != ARCHIVE_OK) { + logger_->log_error("Archive write add filter xz error %s", archive_error_string(arch.get())); + return nullptr; + } + } else { + logger_->log_error("Archive write unsupported compression format"); + return nullptr; + } + result = archive_write_set_bytes_per_block(arch.get(), 0); + if (result != ARCHIVE_OK) { + logger_->log_error("Archive write set bytes per block error %s", archive_error_string(arch.get())); + return nullptr; + } + result = archive_write_open(arch.get(), sink_.get(), nullptr, archive_write, nullptr); + if (result != ARCHIVE_OK) { + logger_->log_error("Archive write open error %s", archive_error_string(arch.get())); + return nullptr; + } + return arch; +} + +bool WriteArchiveStreamImpl::newEntry(const EntryInfo& info) { + if (!arch_) { + return false; + } + arch_entry_.reset(archive_entry_new()); + if (!arch_entry_) { + logger_->log_error("Failed to create archive entry"); + return false; + } + archive_entry_set_pathname(arch_entry_.get(), info.filename.c_str()); + archive_entry_set_size(arch_entry_.get(), info.size); + archive_entry_set_mode(arch_entry_.get(), S_IFREG | 0755); + + int result = archive_write_header(arch_.get(), arch_entry_.get()); + if (result != ARCHIVE_OK) { + logger_->log_error("Archive write header error %s", archive_error_string(arch_.get())); + return false; + } + return true; +} + +size_t WriteArchiveStreamImpl::write(const uint8_t* data, size_t len) { + if (!arch_ || !arch_entry_) { + return STREAM_ERROR; + } + + if (len == 0) { + return 0; + } + gsl_Expects(data); + + int result = archive_write_data(arch_.get(), data, len); + if (result < 0) { + logger_->log_error("Archive write data error %s", archive_error_string(arch_.get())); + arch_entry_.reset(); + arch_.reset(); + return STREAM_ERROR; + } + + return result; +} + +} // namespace org::apache::nifi::minifi::io + + diff --git a/extensions/libarchive/WriteArchiveStream.h b/extensions/libarchive/WriteArchiveStream.h new file mode 100644 index 0000000..12e175a --- /dev/null +++ b/extensions/libarchive/WriteArchiveStream.h @@ -0,0 +1,86 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include <memory> +#include <utility> +#include <string> + +#include "io/ArchiveStream.h" +#include "archive_entry.h" +#include "archive.h" +#include "utils/Enum.h" +#include "core/Core.h" +#include "logging/LoggerConfiguration.h" + +namespace org::apache::nifi::minifi::io { + +SMART_ENUM(CompressionFormat, + (GZIP, "gzip"), + (LZMA, "lzma"), + (XZ_LZMA2, "xz-lzma2"), + (BZIP2, "bzip2") +) + +class WriteArchiveStreamImpl: public WriteArchiveStream { + struct archive_write_deleter { + int operator()(struct archive* ptr) const { + return archive_write_free(ptr); + } + }; + using archive_ptr = std::unique_ptr<struct archive, archive_write_deleter>; + struct archive_entry_deleter { + void operator()(struct archive_entry* ptr) const { + archive_entry_free(ptr); + } + }; + using archive_entry_ptr = std::unique_ptr<struct archive_entry, archive_entry_deleter>; + + archive_ptr createWriteArchive(); + + public: + WriteArchiveStreamImpl(int compress_level, CompressionFormat compress_format, std::shared_ptr<OutputStream> sink) + : compress_level_(compress_level), + compress_format_(compress_format), + sink_(std::move(sink)) { + arch_ = createWriteArchive(); + } + + using OutputStream::write; + + bool newEntry(const EntryInfo& info) override; + + size_t write(const uint8_t* data, size_t len) override; + + private: + static la_ssize_t archive_write(struct archive* /*arch*/, void *context, const void *buff, size_t size) { + auto* const output = static_cast<OutputStream*>(context); + const auto ret = output->write(reinterpret_cast<const uint8_t*>(buff), size); + return io::isError(ret) ? -1 : gsl::narrow<la_ssize_t>(ret); + } + + int compress_level_; + CompressionFormat compress_format_; + std::shared_ptr<io::OutputStream> sink_; + archive_ptr arch_; + archive_entry_ptr arch_entry_; + std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<WriteArchiveStreamImpl>::getLogger(); +}; + +} // namespace org::apache::nifi::minifi::io diff --git a/libminifi/include/core/ProcessSession.h b/libminifi/include/core/ProcessSession.h index c80351c..af54e95 100644 --- a/libminifi/include/core/ProcessSession.h +++ b/libminifi/include/core/ProcessSession.h @@ -88,8 +88,16 @@ class ProcessSession : public ReferenceContainer { void remove(const std::shared_ptr<core::FlowFile> &flow); // Execute the given read callback against the content int64_t read(const std::shared_ptr<core::FlowFile> &flow, InputStreamCallback *callback); + + int64_t read(const std::shared_ptr<core::FlowFile> &flow, InputStreamCallback&& callback) { + return read(flow, &callback); + } // Execute the given write callback against the content void write(const std::shared_ptr<core::FlowFile> &flow, OutputStreamCallback *callback); + + void write(const std::shared_ptr<core::FlowFile> &flow, OutputStreamCallback&& callback) { + return write(flow, &callback); + } // Read and write the flow file at the same time (eg. for processing it line by line) int64_t readWrite(const std::shared_ptr<core::FlowFile> &flow, InputOutputStreamCallback *callback); // Replace content with buffer diff --git a/libminifi/include/io/ArchiveStream.h b/libminifi/include/io/ArchiveStream.h new file mode 100644 index 0000000..def8838 --- /dev/null +++ b/libminifi/include/io/ArchiveStream.h @@ -0,0 +1,54 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include <memory> +#include <string> + +#include "OutputStream.h" +#include "InputStream.h" +#include "core/Core.h" +#include "core/logging/Logger.h" + +namespace org::apache::nifi::minifi::io { + +struct EntryInfo { + std::string filename; + size_t size; +}; + +class WriteArchiveStream : public OutputStream { + public: + virtual bool newEntry(const EntryInfo& info) = 0; +}; + +class ReadArchiveStream : public InputStream { + public: + virtual std::optional<EntryInfo> nextEntry() = 0; +}; + +class ArchiveStreamProvider : public core::CoreComponent { + public: + using CoreComponent::CoreComponent; + virtual std::unique_ptr<WriteArchiveStream> createWriteStream(int compress_level, const std::string& compress_format, + std::shared_ptr<OutputStream> sink, std::shared_ptr<core::logging::Logger> logger) = 0; + virtual std::unique_ptr<ReadArchiveStream> createReadStream(std::shared_ptr<InputStream> archive_stream) = 0; +}; + +} // namespace org::apache::nifi::minifi::io diff --git a/libminifi/include/io/StreamPipe.h b/libminifi/include/io/StreamPipe.h index 0956f8d..bc30aca 100644 --- a/libminifi/include/io/StreamPipe.h +++ b/libminifi/include/io/StreamPipe.h @@ -48,9 +48,31 @@ class InputOutputStreamCallback { virtual int64_t process(const std::shared_ptr<io::BaseStream>& input, const std::shared_ptr<io::BaseStream>& output) = 0; }; +class FunctionOutputStreamCallback : public OutputStreamCallback { + public: + explicit FunctionOutputStreamCallback(std::function<int64_t(const std::shared_ptr<io::OutputStream>&)> fn) : fn_(std::move(fn)) {} + + int64_t process(const std::shared_ptr<io::BaseStream>& stream) override { + return fn_(stream); + } + private: + std::function<int64_t(const std::shared_ptr<io::OutputStream>&)> fn_; +}; + +class FunctionInputStreamCallback : public InputStreamCallback { + public: + explicit FunctionInputStreamCallback(std::function<int64_t(const std::shared_ptr<io::InputStream>&)> fn) : fn_(std::move(fn)) {} + + int64_t process(const std::shared_ptr<io::BaseStream>& stream) override { + return fn_(stream); + } + private: + std::function<int64_t(const std::shared_ptr<io::InputStream>&)> fn_; +}; + namespace internal { -inline int64_t pipe(const std::shared_ptr<io::InputStream>& src, const std::shared_ptr<io::OutputStream>& dst) { +inline int64_t pipe(io::InputStream* src, io::OutputStream* dst) { uint8_t buffer[4096U]; int64_t totalTransferred = 0; while (true) { @@ -77,6 +99,10 @@ inline int64_t pipe(const std::shared_ptr<io::InputStream>& src, const std::shar return totalTransferred; } +inline int64_t pipe(const std::shared_ptr<io::InputStream>& src, const std::shared_ptr<io::OutputStream>& dst) { + return pipe(src.get(), dst.get()); +} + } // namespace internal class InputStreamPipe : public InputStreamCallback { diff --git a/libminifi/test/archive-tests/ArchiveStreamTests.cpp b/libminifi/test/archive-tests/ArchiveStreamTests.cpp new file mode 100644 index 0000000..3c8c715 --- /dev/null +++ b/libminifi/test/archive-tests/ArchiveStreamTests.cpp @@ -0,0 +1,55 @@ +/** + * + * 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 "../TestBase.h" +#include "WriteArchiveStream.h" +#include "ReadArchiveStream.h" + +using namespace minifi; + +TEST_CASE("Create and read archive") { + std::map<std::string, std::string> files{ + {"a.txt", "hello, I'm file A"}, + {"b.txt", "hello, I'm file B"} + }; + + auto archive = std::make_shared<io::BufferStream>(); + + { + io::WriteArchiveStreamImpl compressor(9, io::CompressionFormat::GZIP, archive); + + for (const auto& [filename, content] : files) { + REQUIRE(compressor.newEntry({filename, content.length()})); + REQUIRE(compressor.write(reinterpret_cast<const uint8_t*>(content.data()), content.length()) == content.length()); + } + } + + { + io::ReadArchiveStreamImpl decompressor(archive); + + size_t extracted_entries = 0; + while (auto info = decompressor.nextEntry()) { + ++extracted_entries; + std::string file_content; + file_content.resize(info->size); + REQUIRE(decompressor.read(reinterpret_cast<uint8_t*>(file_content.data()), file_content.length()) == file_content.length()); + REQUIRE(files[info->filename] == file_content); + } + REQUIRE(extracted_entries == files.size()); + } +}
