| /** |
| * |
| * 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 <fstream> |
| #include <map> |
| #include <memory> |
| #include <utility> |
| #include <string> |
| #include <set> |
| #include <random> |
| #include <sstream> |
| #include <iostream> |
| #include "FlowController.h" |
| #include "../TestBase.h" |
| #include "core/Core.h" |
| #include "../../include/core/FlowFile.h" |
| #include "../unit/ProvenanceTestHelper.h" |
| #include "core/Processor.h" |
| #include "core/ProcessContext.h" |
| #include "core/ProcessSession.h" |
| #include "core/ProcessorNode.h" |
| #include "CompressContent.h" |
| #include "io/FileStream.h" |
| #include "FlowFileRecord.h" |
| #include "processors/LogAttribute.h" |
| #include "processors/PutFile.h" |
| #include "utils/file/FileUtils.h" |
| |
| class ReadCallback: public org::apache::nifi::minifi::InputStreamCallback { |
| public: |
| explicit ReadCallback(size_t size) : |
| read_size_(0) { |
| buffer_size_ = size; |
| buffer_ = new uint8_t[buffer_size_]; |
| archive_buffer_ = nullptr; |
| archive_buffer_size_ = 0; |
| } |
| ~ReadCallback() { |
| if (buffer_) |
| delete[] buffer_; |
| if (archive_buffer_) |
| delete[] archive_buffer_; |
| } |
| int64_t process(std::shared_ptr<org::apache::nifi::minifi::io::BaseStream> stream) { |
| int64_t total_read = 0; |
| int64_t ret = 0; |
| do { |
| ret = stream->read(buffer_ + read_size_, gsl::narrow<int>(buffer_size_ - read_size_)); |
| if (ret == 0) break; |
| if (ret < 0) return ret; |
| read_size_ += gsl::narrow<size_t>(ret); |
| total_read += ret; |
| } while (buffer_size_ != read_size_); |
| return total_read; |
| } |
| void archive_read() { |
| struct archive *a; |
| a = archive_read_new(); |
| archive_read_support_format_all(a); |
| archive_read_support_filter_all(a); |
| archive_read_open_memory(a, buffer_, read_size_); |
| struct archive_entry *ae; |
| |
| REQUIRE(archive_read_next_header(a, &ae) == ARCHIVE_OK); |
| int size = gsl::narrow<int>(archive_entry_size(ae)); |
| archive_buffer_ = new char[size]; |
| archive_buffer_size_ = size; |
| archive_read_data(a, archive_buffer_, size); |
| archive_read_free(a); |
| } |
| |
| uint8_t *buffer_; |
| size_t buffer_size_; |
| size_t read_size_; |
| char *archive_buffer_; |
| int archive_buffer_size_; |
| }; |
| |
| /** |
| * There is strong coupling between these compression and decompression |
| * tests. Some compression tests also set up the stage for the subsequent |
| * decompression test. Each such test controller should either be |
| * CompressTestController or a DecompressTestController. |
| */ |
| class CompressDecompressionTestController : public TestController{ |
| protected: |
| static std::string tempDir_; |
| static std::string raw_content_path_; |
| static std::string compressed_content_path_; |
| static TestController& get_global_controller() { |
| static TestController controller; |
| return controller; |
| } |
| |
| void setupFlow() { |
| LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::CompressContent>(); |
| LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::LogAttribute>(); |
| LogTestController::getInstance().setTrace<core::ProcessSession>(); |
| LogTestController::getInstance().setTrace<core::ProcessContext>(); |
| LogTestController::getInstance().setTrace<core::repository::VolatileContentRepository>(); |
| LogTestController::getInstance().setTrace<org::apache::nifi::minifi::Connection>(); |
| LogTestController::getInstance().setTrace<org::apache::nifi::minifi::core::Connectable>(); |
| LogTestController::getInstance().setTrace<org::apache::nifi::minifi::io::FileStream>(); |
| |
| std::shared_ptr<TestRepository> repo = std::make_shared<TestRepository>(); |
| |
| processor = std::make_shared<org::apache::nifi::minifi::processors::CompressContent>("compresscontent"); |
| processor->initialize(); |
| utils::Identifier processoruuid; |
| REQUIRE(true == processor->getUUID(processoruuid)); |
| |
| std::shared_ptr<core::ContentRepository> content_repo = std::make_shared<core::repository::VolatileContentRepository>(); |
| content_repo->initialize(std::make_shared<org::apache::nifi::minifi::Configure>()); |
| // connection from compress processor to log attribute |
| output = std::make_shared<minifi::Connection>(repo, content_repo, "Output"); |
| output->addRelationship(core::Relationship("success", "compress successful output")); |
| output->setSource(processor); |
| output->setSourceUUID(processoruuid); |
| processor->addConnection(output); |
| // connection to compress processor |
| input = std::make_shared<minifi::Connection>(repo, content_repo, "Input"); |
| input->setDestination(processor); |
| input->setDestinationUUID(processoruuid); |
| processor->addConnection(input); |
| |
| processor->setAutoTerminatedRelationships({{"failure", ""}}); |
| |
| processor->incrementActiveTasks(); |
| processor->setScheduledState(core::ScheduledState::RUNNING); |
| |
| context = std::make_shared<core::ProcessContext>(std::make_shared<core::ProcessorNode>(processor), nullptr, repo, repo, content_repo); |
| } |
| |
| public: |
| class RawContent{ |
| std::string content_; |
| explicit RawContent(std::string&& content_): content_(std::move(content_)) {} |
| friend class CompressDecompressionTestController; |
| public: |
| bool operator==(const std::string& actual) const noexcept { |
| return content_ == actual; |
| } |
| bool operator!=(const std::string& actual) const noexcept { |
| return content_ != actual; |
| } |
| }; |
| |
| std::string rawContentPath() const { |
| return raw_content_path_; |
| } |
| |
| std::string compressedPath() const { |
| return compressed_content_path_; |
| } |
| |
| RawContent getRawContent() const { |
| std::ifstream file; |
| file.open(raw_content_path_, std::ios::binary); |
| std::string contents{std::istreambuf_iterator<char>(file), std::istreambuf_iterator<char>()}; |
| return RawContent{std::move(contents)}; |
| } |
| |
| virtual ~CompressDecompressionTestController() = 0; |
| |
| std::shared_ptr<core::Processor> processor; |
| std::shared_ptr<core::ProcessContext> context; |
| std::shared_ptr<minifi::Connection> output; |
| std::shared_ptr<minifi::Connection> input; |
| }; |
| |
| CompressDecompressionTestController::~CompressDecompressionTestController() { |
| LogTestController::getInstance().reset(); |
| } |
| |
| std::string CompressDecompressionTestController::tempDir_; |
| std::string CompressDecompressionTestController::raw_content_path_; |
| std::string CompressDecompressionTestController::compressed_content_path_; |
| |
| class CompressTestController : public CompressDecompressionTestController { |
| static void initContentWithRandomData() { |
| int random_seed = 0x454; |
| std::ofstream file; |
| file.open(raw_content_path_, std::ios::binary); |
| |
| std::mt19937 gen(random_seed); |
| std::uniform_int_distribution<> dis(0, 99); |
| for (int i = 0; i < 100000; i++) { |
| file << std::to_string(dis(gen)); |
| } |
| } |
| |
| public: |
| CompressTestController() { |
| char format[] = "/tmp/test.XXXXXX"; |
| tempDir_ = get_global_controller().createTempDirectory(format); |
| REQUIRE(!tempDir_.empty()); |
| raw_content_path_ = utils::file::FileUtils::concat_path(tempDir_, "minifi-expect-compresscontent.txt"); |
| compressed_content_path_ = utils::file::FileUtils::concat_path(tempDir_, "minifi-compresscontent"); |
| initContentWithRandomData(); |
| setupFlow(); |
| } |
| |
| template<class ...Args> |
| void writeCompressed(Args&& ...args) { |
| std::ofstream file(compressed_content_path_, std::ios::binary); |
| file.write(std::forward<Args>(args)...); |
| } |
| }; |
| |
| class DecompressTestController : public CompressDecompressionTestController{ |
| public: |
| DecompressTestController() { |
| setupFlow(); |
| } |
| ~DecompressTestController() { |
| tempDir_ = ""; |
| raw_content_path_ = ""; |
| compressed_content_path_ = ""; |
| } |
| }; |
| |
| TEST_CASE_METHOD(CompressTestController, "CompressFileGZip", "[compressfiletest1]") { |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressMode, MODE_COMPRESS); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressFormat, COMPRESSION_FORMAT_GZIP); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressLevel, "9"); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::UpdateFileName, "true"); |
| |
| core::ProcessSession sessionGenFlowFile(context); |
| std::shared_ptr<core::FlowFile> flow = std::static_pointer_cast < core::FlowFile > (sessionGenFlowFile.create()); |
| sessionGenFlowFile.import(rawContentPath(), flow, true, 0); |
| sessionGenFlowFile.flushContent(); |
| input->put(flow); |
| |
| REQUIRE(processor->getName() == "compresscontent"); |
| auto factory = std::make_shared<core::ProcessSessionFactory>(context); |
| processor->onSchedule(context, factory); |
| auto session = std::make_shared<core::ProcessSession>(context); |
| processor->onTrigger(context, session); |
| session->commit(); |
| |
| // validate the compress content |
| std::set<std::shared_ptr<core::FlowFile>> expiredFlowRecords; |
| std::shared_ptr<core::FlowFile> flow1 = output->poll(expiredFlowRecords); |
| REQUIRE(flow1->getSize() > 0); |
| { |
| REQUIRE(flow1->getSize() != flow->getSize()); |
| std::string mime; |
| flow1->getAttribute(org::apache::nifi::minifi::core::SpecialFlowAttribute::MIME_TYPE, mime); |
| REQUIRE(mime == "application/gzip"); |
| ReadCallback callback(gsl::narrow<size_t>(flow1->getSize())); |
| sessionGenFlowFile.read(flow1, &callback); |
| callback.archive_read(); |
| std::string content(reinterpret_cast<char *> (callback.archive_buffer_), callback.archive_buffer_size_); |
| REQUIRE(getRawContent() == content); |
| // write the compress content for next test |
| writeCompressed(reinterpret_cast<char *> (callback.buffer_), callback.read_size_); |
| } |
| } |
| |
| TEST_CASE_METHOD(DecompressTestController, "DecompressFileGZip", "[compressfiletest2]") { |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressMode, MODE_DECOMPRESS); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressFormat, COMPRESSION_FORMAT_GZIP); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressLevel, "9"); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::UpdateFileName, "true"); |
| |
| core::ProcessSession sessionGenFlowFile(context); |
| std::shared_ptr<core::FlowFile> flow = std::static_pointer_cast < core::FlowFile > (sessionGenFlowFile.create()); |
| sessionGenFlowFile.import(compressedPath(), flow, true, 0); |
| sessionGenFlowFile.flushContent(); |
| input->put(flow); |
| |
| REQUIRE(processor->getName() == "compresscontent"); |
| auto factory = std::make_shared<core::ProcessSessionFactory>(context); |
| processor->onSchedule(context, factory); |
| auto session = std::make_shared<core::ProcessSession>(context); |
| processor->onTrigger(context, session); |
| session->commit(); |
| |
| // validate the compress content |
| std::set<std::shared_ptr<core::FlowFile>> expiredFlowRecords; |
| std::shared_ptr<core::FlowFile> flow1 = output->poll(expiredFlowRecords); |
| REQUIRE(flow1->getSize() > 0); |
| { |
| REQUIRE(flow1->getSize() != flow->getSize()); |
| std::string mime; |
| REQUIRE(flow1->getAttribute(org::apache::nifi::minifi::core::SpecialFlowAttribute::MIME_TYPE, mime) == false); |
| ReadCallback callback(gsl::narrow<size_t>(flow1->getSize())); |
| sessionGenFlowFile.read(flow1, &callback); |
| std::string content(reinterpret_cast<char *> (callback.buffer_), callback.read_size_); |
| REQUIRE(getRawContent() == content); |
| } |
| } |
| |
| TEST_CASE_METHOD(CompressTestController, "CompressFileBZip", "[compressfiletest3]") { |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressMode, MODE_COMPRESS); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressFormat, COMPRESSION_FORMAT_BZIP2); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressLevel, "9"); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::UpdateFileName, "true"); |
| |
| core::ProcessSession sessionGenFlowFile(context); |
| std::shared_ptr<core::FlowFile> flow = std::static_pointer_cast < core::FlowFile > (sessionGenFlowFile.create()); |
| sessionGenFlowFile.import(rawContentPath(), flow, true, 0); |
| sessionGenFlowFile.flushContent(); |
| input->put(flow); |
| |
| REQUIRE(processor->getName() == "compresscontent"); |
| auto factory = std::make_shared<core::ProcessSessionFactory>(context); |
| processor->onSchedule(context, factory); |
| auto session = std::make_shared<core::ProcessSession>(context); |
| processor->onTrigger(context, session); |
| session->commit(); |
| |
| // validate the compress content |
| std::set<std::shared_ptr<core::FlowFile>> expiredFlowRecords; |
| std::shared_ptr<core::FlowFile> flow1 = output->poll(expiredFlowRecords); |
| REQUIRE(flow1->getSize() > 0); |
| { |
| REQUIRE(flow1->getSize() != flow->getSize()); |
| std::string mime; |
| flow1->getAttribute(org::apache::nifi::minifi::core::SpecialFlowAttribute::MIME_TYPE, mime); |
| REQUIRE(mime == "application/bzip2"); |
| ReadCallback callback(gsl::narrow<size_t>(flow1->getSize())); |
| sessionGenFlowFile.read(flow1, &callback); |
| callback.archive_read(); |
| std::string contents(reinterpret_cast<char *> (callback.archive_buffer_), callback.archive_buffer_size_); |
| REQUIRE(getRawContent() == contents); |
| // write the compress content for next test |
| writeCompressed(reinterpret_cast<char *> (callback.buffer_), callback.read_size_); |
| } |
| } |
| |
| |
| TEST_CASE_METHOD(DecompressTestController, "DecompressFileBZip", "[compressfiletest4]") { |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressMode, MODE_DECOMPRESS); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressFormat, COMPRESSION_FORMAT_BZIP2); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressLevel, "9"); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::UpdateFileName, "true"); |
| |
| core::ProcessSession sessionGenFlowFile(context); |
| std::shared_ptr<core::FlowFile> flow = std::static_pointer_cast < core::FlowFile > (sessionGenFlowFile.create()); |
| sessionGenFlowFile.import(compressedPath(), flow, true, 0); |
| sessionGenFlowFile.flushContent(); |
| input->put(flow); |
| |
| REQUIRE(processor->getName() == "compresscontent"); |
| auto factory = std::make_shared<core::ProcessSessionFactory>(context); |
| processor->onSchedule(context, factory); |
| auto session = std::make_shared<core::ProcessSession>(context); |
| processor->onTrigger(context, session); |
| session->commit(); |
| |
| // validate the compress content |
| std::set<std::shared_ptr<core::FlowFile>> expiredFlowRecords; |
| std::shared_ptr<core::FlowFile> flow1 = output->poll(expiredFlowRecords); |
| REQUIRE(flow1->getSize() > 0); |
| { |
| REQUIRE(flow1->getSize() != flow->getSize()); |
| std::string mime; |
| REQUIRE(flow1->getAttribute(org::apache::nifi::minifi::core::SpecialFlowAttribute::MIME_TYPE, mime) == false); |
| ReadCallback callback(gsl::narrow<size_t>(flow1->getSize())); |
| sessionGenFlowFile.read(flow1, &callback); |
| std::string contents(reinterpret_cast<char *> (callback.buffer_), callback.read_size_); |
| REQUIRE(getRawContent() == contents); |
| } |
| } |
| |
| TEST_CASE_METHOD(CompressTestController, "CompressFileLZMA", "[compressfiletest5]") { |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressMode, MODE_COMPRESS); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressFormat, COMPRESSION_FORMAT_LZMA); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressLevel, "9"); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::UpdateFileName, "true"); |
| |
| core::ProcessSession sessionGenFlowFile(context); |
| std::shared_ptr<core::FlowFile> flow = std::static_pointer_cast < core::FlowFile > (sessionGenFlowFile.create()); |
| sessionGenFlowFile.import(rawContentPath(), flow, true, 0); |
| sessionGenFlowFile.flushContent(); |
| input->put(flow); |
| |
| REQUIRE(processor->getName() == "compresscontent"); |
| auto factory = std::make_shared<core::ProcessSessionFactory>(context); |
| processor->onSchedule(context, factory); |
| auto session = std::make_shared<core::ProcessSession>(context); |
| processor->onTrigger(context, session); |
| session->commit(); |
| |
| if (LogTestController::getInstance().contains("compression not supported on this platform")) { |
| // platform not support LZMA |
| LogTestController::getInstance().reset(); |
| return; |
| } |
| |
| // validate the compress content |
| std::set<std::shared_ptr<core::FlowFile>> expiredFlowRecords; |
| std::shared_ptr<core::FlowFile> flow1 = output->poll(expiredFlowRecords); |
| REQUIRE(flow1->getSize() > 0); |
| { |
| REQUIRE(flow1->getSize() != flow->getSize()); |
| std::string mime; |
| flow1->getAttribute(org::apache::nifi::minifi::core::SpecialFlowAttribute::MIME_TYPE, mime); |
| REQUIRE(mime == "application/x-lzma"); |
| ReadCallback callback(gsl::narrow<size_t>(flow1->getSize())); |
| sessionGenFlowFile.read(flow1, &callback); |
| callback.archive_read(); |
| std::string contents(reinterpret_cast<char *> (callback.archive_buffer_), callback.archive_buffer_size_); |
| REQUIRE(getRawContent() == contents); |
| // write the compress content for next test |
| writeCompressed(reinterpret_cast<char *> (callback.buffer_), callback.read_size_); |
| } |
| } |
| |
| |
| TEST_CASE_METHOD(DecompressTestController, "DecompressFileLZMA", "[compressfiletest6]") { |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressMode, MODE_DECOMPRESS); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressFormat, COMPRESSION_FORMAT_ATTRIBUTE); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressLevel, "9"); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::UpdateFileName, "true"); |
| |
| core::ProcessSession sessionGenFlowFile(context); |
| std::shared_ptr<core::FlowFile> flow = std::static_pointer_cast < core::FlowFile > (sessionGenFlowFile.create()); |
| sessionGenFlowFile.import(compressedPath(), flow, true, 0); |
| flow->setAttribute(org::apache::nifi::minifi::core::SpecialFlowAttribute::MIME_TYPE, "application/x-lzma"); |
| sessionGenFlowFile.flushContent(); |
| input->put(flow); |
| |
| REQUIRE(processor->getName() == "compresscontent"); |
| auto factory = std::make_shared<core::ProcessSessionFactory>(context); |
| processor->onSchedule(context, factory); |
| auto session = std::make_shared<core::ProcessSession>(context); |
| processor->onTrigger(context, session); |
| session->commit(); |
| |
| if (LogTestController::getInstance().contains("compression not supported on this platform")) { |
| // platform not support LZMA |
| LogTestController::getInstance().reset(); |
| return; |
| } |
| |
| // validate the compress content |
| std::set<std::shared_ptr<core::FlowFile>> expiredFlowRecords; |
| std::shared_ptr<core::FlowFile> flow1 = output->poll(expiredFlowRecords); |
| REQUIRE(flow1->getSize() > 0); |
| { |
| REQUIRE(flow1->getSize() != flow->getSize()); |
| std::string mime; |
| REQUIRE(flow1->getAttribute(org::apache::nifi::minifi::core::SpecialFlowAttribute::MIME_TYPE, mime) == false); |
| ReadCallback callback(gsl::narrow<size_t>(flow1->getSize())); |
| sessionGenFlowFile.read(flow1, &callback); |
| std::string contents(reinterpret_cast<char *> (callback.buffer_), callback.read_size_); |
| REQUIRE(getRawContent() == contents); |
| } |
| } |
| |
| TEST_CASE_METHOD(CompressTestController, "CompressFileXYLZMA", "[compressfiletest7]") { |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressMode, MODE_COMPRESS); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressFormat, COMPRESSION_FORMAT_XZ_LZMA2); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressLevel, "9"); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::UpdateFileName, "true"); |
| |
| core::ProcessSession sessionGenFlowFile(context); |
| std::shared_ptr<core::FlowFile> flow = std::static_pointer_cast < core::FlowFile > (sessionGenFlowFile.create()); |
| sessionGenFlowFile.import(rawContentPath(), flow, true, 0); |
| sessionGenFlowFile.flushContent(); |
| input->put(flow); |
| |
| REQUIRE(processor->getName() == "compresscontent"); |
| auto factory = std::make_shared<core::ProcessSessionFactory>(context); |
| processor->onSchedule(context, factory); |
| auto session = std::make_shared<core::ProcessSession>(context); |
| processor->onTrigger(context, session); |
| session->commit(); |
| |
| if (LogTestController::getInstance().contains("compression not supported on this platform")) { |
| // platform not support LZMA |
| LogTestController::getInstance().reset(); |
| return; |
| } |
| |
| // validate the compress content |
| std::set<std::shared_ptr<core::FlowFile>> expiredFlowRecords; |
| std::shared_ptr<core::FlowFile> flow1 = output->poll(expiredFlowRecords); |
| REQUIRE(flow1->getSize() > 0); |
| { |
| REQUIRE(flow1->getSize() != flow->getSize()); |
| std::string mime; |
| flow1->getAttribute(org::apache::nifi::minifi::core::SpecialFlowAttribute::MIME_TYPE, mime); |
| REQUIRE(mime == "application/x-xz"); |
| ReadCallback callback(gsl::narrow<size_t>(flow1->getSize())); |
| sessionGenFlowFile.read(flow1, &callback); |
| callback.archive_read(); |
| std::string contents(reinterpret_cast<char *> (callback.archive_buffer_), callback.archive_buffer_size_); |
| REQUIRE(getRawContent() == contents); |
| // write the compress content for next test |
| writeCompressed(reinterpret_cast<char *> (callback.buffer_), callback.read_size_); |
| } |
| } |
| |
| |
| TEST_CASE_METHOD(DecompressTestController, "DecompressFileXYLZMA", "[compressfiletest8]") { |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressMode, MODE_DECOMPRESS); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressFormat, COMPRESSION_FORMAT_ATTRIBUTE); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::CompressLevel, "9"); |
| context->setProperty(org::apache::nifi::minifi::processors::CompressContent::UpdateFileName, "true"); |
| |
| core::ProcessSession sessionGenFlowFile(context); |
| std::shared_ptr<core::FlowFile> flow = std::static_pointer_cast < core::FlowFile > (sessionGenFlowFile.create()); |
| sessionGenFlowFile.import(compressedPath(), flow, true, 0); |
| flow->setAttribute(org::apache::nifi::minifi::core::SpecialFlowAttribute::MIME_TYPE, "application/x-xz"); |
| sessionGenFlowFile.flushContent(); |
| input->put(flow); |
| |
| REQUIRE(processor->getName() == "compresscontent"); |
| auto factory = std::make_shared<core::ProcessSessionFactory>(context); |
| processor->onSchedule(context, factory); |
| auto session = std::make_shared<core::ProcessSession>(context); |
| processor->onTrigger(context, session); |
| session->commit(); |
| |
| if (LogTestController::getInstance().contains("compression not supported on this platform")) { |
| // platform not support LZMA |
| LogTestController::getInstance().reset(); |
| return; |
| } |
| |
| // validate the compress content |
| std::set<std::shared_ptr<core::FlowFile>> expiredFlowRecords; |
| std::shared_ptr<core::FlowFile> flow1 = output->poll(expiredFlowRecords); |
| REQUIRE(flow1->getSize() > 0); |
| { |
| REQUIRE(flow1->getSize() != flow->getSize()); |
| std::string mime; |
| REQUIRE(flow1->getAttribute(org::apache::nifi::minifi::core::SpecialFlowAttribute::MIME_TYPE, mime) == false); |
| ReadCallback callback(gsl::narrow<size_t>(flow1->getSize())); |
| sessionGenFlowFile.read(flow1, &callback); |
| std::string contents(reinterpret_cast<char *> (callback.buffer_), callback.read_size_); |
| REQUIRE(getRawContent() == contents); |
| } |
| } |
| |
| TEST_CASE_METHOD(TestController, "RawGzipCompressionDecompression", "[compressfiletest8]") { |
| LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::CompressContent>(); |
| LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::PutFile>(); |
| |
| // Create temporary directories |
| char format_src[] = "/tmp/archives.XXXXXX"; |
| std::string src_dir = createTempDirectory(format_src); |
| REQUIRE(!src_dir.empty()); |
| |
| char format_dst[] = "/tmp/archived.XXXXXX"; |
| std::string dst_dir = createTempDirectory(format_dst); |
| REQUIRE(!dst_dir.empty()); |
| |
| // Define files |
| std::string src_file = utils::file::FileUtils::concat_path(src_dir, "src.txt"); |
| std::string compressed_file = utils::file::FileUtils::concat_path(dst_dir, "src.txt.gz"); |
| std::string decompressed_file = utils::file::FileUtils::concat_path(dst_dir, "src.txt"); |
| |
| // Build MiNiFi processing graph |
| auto plan = createPlan(); |
| auto get_file = plan->addProcessor( |
| "GetFile", |
| "GetFile"); |
| auto compress_content = plan->addProcessor( |
| "CompressContent", |
| "CompressContent", |
| core::Relationship("success", "d"), |
| true); |
| auto put_compressed = plan->addProcessor( |
| "PutFile", |
| "PutFile", |
| core::Relationship("success", "d"), |
| true); |
| auto decompress_content = plan->addProcessor( |
| "CompressContent", |
| "CompressContent", |
| core::Relationship("success", "d"), |
| true); |
| auto put_decompressed = plan->addProcessor( |
| "PutFile", |
| "PutFile", |
| core::Relationship("success", "d"), |
| true); |
| |
| // Configure GetFile processor |
| plan->setProperty(get_file, "Input Directory", src_dir); |
| |
| // Configure CompressContent processor for compression |
| plan->setProperty(compress_content, "Mode", MODE_COMPRESS); |
| plan->setProperty(compress_content, "Compression Format", COMPRESSION_FORMAT_GZIP); |
| plan->setProperty(compress_content, "Update Filename", "true"); |
| plan->setProperty(compress_content, "Encapsulate in TAR", "false"); |
| |
| // Configure first PutFile processor |
| plan->setProperty(put_compressed, "Directory", dst_dir); |
| |
| // Configure CompressContent processor for decompression |
| plan->setProperty(decompress_content, "Mode", MODE_DECOMPRESS); |
| plan->setProperty(decompress_content, "Compression Format", COMPRESSION_FORMAT_GZIP); |
| plan->setProperty(decompress_content, "Update Filename", "true"); |
| plan->setProperty(decompress_content, "Encapsulate in TAR", "false"); |
| |
| // Configure second PutFile processor |
| plan->setProperty(put_decompressed, "Directory", dst_dir); |
| |
| // Create source file |
| std::string content; |
| SECTION("Empty content") { |
| } |
| SECTION("Short content") { |
| content = "Repeated repeated repeated repeated repeated stuff."; |
| } |
| SECTION("Long content") { |
| std::stringstream content_ss; |
| // if only this part fails VolatileContentRepository's fixed max size |
| // and some change in the cleanup logic might interfere |
| for (size_t i = 0U; i < 512 * 1024U; i++) { |
| content_ss << "foobar"; |
| } |
| content = content_ss.str(); |
| } |
| |
| std::ofstream{ src_file } << content; |
| |
| // Run flow |
| runSession(plan, true); |
| |
| // Check compressed file |
| std::ifstream compressed(compressed_file, std::ios::in | std::ios::binary); |
| std::vector<uint8_t> compressed_content((std::istreambuf_iterator<char>(compressed)), std::istreambuf_iterator<char>()); |
| REQUIRE(2 < compressed_content.size()); |
| // gzip magic number |
| REQUIRE(0x1f == compressed_content[0]); |
| REQUIRE(0x8b == compressed_content[1]); |
| |
| // Check decompressed file |
| std::ifstream decompressed(decompressed_file, std::ios::in | std::ios::binary); |
| std::string decompressed_content((std::istreambuf_iterator<char>(decompressed)), std::istreambuf_iterator<char>()); |
| REQUIRE(content == decompressed_content); |
| |
| LogTestController::getInstance().reset(); |
| } |