/**
 *
 * 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 "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 "MergeContent.h"
#include <sstream>
#include <iostream>
#include "processors/LogAttribute.h"

static const char* FLOW_FILE = "/tmp/minifi-mergecontent";
static const char* EXPECT_MERGE_CONTENT_FIRST = "/tmp/minifi-expect-mergecontent1.txt";
static const char* EXPECT_MERGE_CONTENT_SECOND = "/tmp/minifi-expect-mergecontent2.txt";
static const char* HEADER_FILE = "/tmp/minifi-mergecontent.header";
static const char* FOOTER_FILE = "/tmp/minifi-mergecontent.footer";
static const char* DEMARCATOR_FILE = "/tmp/minifi-mergecontent.demarcator";

class ReadCallback: public org::apache::nifi::minifi::InputStreamCallback {
 public:
  explicit ReadCallback(uint64_t size) :
      read_size_(0) {
    buffer_size_ = size;
    buffer_ = new uint8_t[buffer_size_];
    archive_buffer_num_ = 0;
  }
  ~ReadCallback() {
    if (buffer_)
      delete[] buffer_;
    for (int i = 0; i < archive_buffer_num_; i++) {
      delete[] archive_buffer_[i];
    }
  }
  int64_t process(std::shared_ptr<org::apache::nifi::minifi::io::BaseStream> stream) {
    int64_t ret = 0;
    ret = stream->read(buffer_, buffer_size_);
    if (stream)
      read_size_ = stream->getSize();
    else
      read_size_ = buffer_size_;
    return ret;
  }
  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;

    while (archive_read_next_header(a, &ae) == ARCHIVE_OK && archive_buffer_num_ < 10) {
      int size = archive_entry_size(ae);
      archive_buffer_[archive_buffer_num_] = new char[size];
      archive_buffer_size_[archive_buffer_num_] = size;
      archive_read_data(a, archive_buffer_[archive_buffer_num_], size);
      archive_buffer_num_++;
    }
  }

  uint8_t *buffer_;
  uint64_t buffer_size_;
  uint64_t read_size_;
  char *archive_buffer_[10];
  int archive_buffer_size_[10];
  int archive_buffer_num_;
};

TEST_CASE("MergeFileDefragment", "[mergefiletest1]") {
  try {
    std::ofstream expectfileFirst;
    std::ofstream expectfileSecond;

    expectfileFirst.open(EXPECT_MERGE_CONTENT_FIRST);
    expectfileSecond.open(EXPECT_MERGE_CONTENT_SECOND);

    // Create and write to the test file
    for (int i = 0; i < 6; i++) {
      std::ofstream tmpfile;
      std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
      tmpfile.open(flowFileName.c_str());
      for (int j = 0; j < 32; j++) {
        tmpfile << std::to_string(i);
        if (i < 3)
          expectfileFirst << std::to_string(i);
        else
          expectfileSecond << std::to_string(i);
      }
      tmpfile.close();
    }
    expectfileFirst.close();
    expectfileSecond.close();

    TestController testController;
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::MergeContent>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::LogAttribute>();
    LogTestController::getInstance().setTrace<core::ProcessSession>();
    LogTestController::getInstance().setTrace<core::repository::VolatileContentRepository>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::BinFiles>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::Bin>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::BinManager>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::Connection>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::core::Connectable>();

    std::shared_ptr<TestRepository> repo = std::make_shared<TestRepository>();

    std::shared_ptr<core::Processor> processor = std::make_shared<org::apache::nifi::minifi::processors::MergeContent>("mergecontent");
    std::shared_ptr<core::Processor> logAttributeProcessor = std::make_shared<org::apache::nifi::minifi::processors::LogAttribute>("logattribute");
    processor->initialize();
    utils::Identifier processoruuid;
    REQUIRE(true == processor->getUUID(processoruuid));
    utils::Identifier logAttributeuuid;
    REQUIRE(true == logAttributeProcessor->getUUID(logAttributeuuid));

    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 merge processor to log attribute
    std::shared_ptr<minifi::Connection> connection = std::make_shared<minifi::Connection>(repo, content_repo, "logattributeconnection");
    connection->addRelationship(core::Relationship("merged", "Merge successful output"));
    connection->setSource(processor);
    connection->setDestination(logAttributeProcessor);
    connection->setSourceUUID(processoruuid);
    connection->setDestinationUUID(logAttributeuuid);
    processor->addConnection(connection);
    // connection to merge processor
    std::shared_ptr<minifi::Connection> mergeconnection = std::make_shared<minifi::Connection>(repo, content_repo, "mergeconnection");
    mergeconnection->setDestination(processor);
    mergeconnection->setDestinationUUID(processoruuid);
    processor->addConnection(mergeconnection);

    std::set<core::Relationship> autoTerminatedRelationships;
    core::Relationship original("original", "");
    core::Relationship failure("failure", "");
    autoTerminatedRelationships.insert(original);
    autoTerminatedRelationships.insert(failure);
    processor->setAutoTerminatedRelationships(autoTerminatedRelationships);

    processor->incrementActiveTasks();
    processor->setScheduledState(core::ScheduledState::RUNNING);
    logAttributeProcessor->incrementActiveTasks();
    logAttributeProcessor->setScheduledState(core::ScheduledState::RUNNING);

    std::shared_ptr<core::ProcessorNode> node = std::make_shared<core::ProcessorNode>(processor);
    std::shared_ptr<core::controller::ControllerServiceProvider> controller_services_provider = nullptr;
    auto context = std::make_shared<core::ProcessContext>(node, controller_services_provider, repo, repo, content_repo);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::MergeFormat, MERGE_FORMAT_CONCAT_VALUE);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::MergeStrategy, MERGE_STRATEGY_DEFRAGMENT);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::DelimiterStratgey, DELIMITER_STRATEGY_TEXT);

    core::ProcessSession sessionGenFlowFile(context);
    std::shared_ptr<core::FlowFile> record[6];

    // Generate 6 flowfiles, first threes merged to one, second thress merged to one
    std::shared_ptr<core::Connectable> income = node->getNextIncomingConnection();
    std::shared_ptr<minifi::Connection> income_connection = std::static_pointer_cast<minifi::Connection>(income);
    for (int i = 0; i < 6; i++) {
      std::shared_ptr<core::FlowFile> flow = std::static_pointer_cast < core::FlowFile > (sessionGenFlowFile.create());
      std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
      sessionGenFlowFile.import(flowFileName, flow, true, 0);
      // three bundle
      if (i < 3)
        flow->setAttribute(processors::BinFiles::FRAGMENT_ID_ATTRIBUTE, std::to_string(0));
      else
        flow->setAttribute(processors::BinFiles::FRAGMENT_ID_ATTRIBUTE, std::to_string(1));
      if (i < 3)
        flow->setAttribute(processors::BinFiles::FRAGMENT_INDEX_ATTRIBUTE, std::to_string(i));
      else
        flow->setAttribute(processors::BinFiles::FRAGMENT_INDEX_ATTRIBUTE, std::to_string(i-3));
      flow->setAttribute(processors::BinFiles::FRAGMENT_COUNT_ATTRIBUTE, std::to_string(3));
      record[i] = flow;
    }
    income_connection->put(record[0]);
    income_connection->put(record[2]);
    income_connection->put(record[5]);
    income_connection->put(record[4]);
    income_connection->put(record[1]);
    income_connection->put(record[3]);

    REQUIRE(processor->getName() == "mergecontent");
    auto factory = std::make_shared<core::ProcessSessionFactory>(context);
    processor->onSchedule(context, factory);
    for (int i = 0; i < 6; i++) {
      auto session = std::make_shared<core::ProcessSession>(context);
      processor->onTrigger(context, session);
      session->commit();
    }
    // validate the merge content
    std::set<std::shared_ptr<core::FlowFile>> expiredFlowRecords;
    std::shared_ptr<core::FlowFile> flow1 = connection->poll(expiredFlowRecords);
    std::shared_ptr<core::FlowFile> flow2 = connection->poll(expiredFlowRecords);
    REQUIRE(flow1->getSize() == 96);
    {
      ReadCallback callback(flow1->getSize());
      sessionGenFlowFile.read(flow1, &callback);
      std::ifstream file1;
      file1.open(EXPECT_MERGE_CONTENT_FIRST, std::ios::in);
      std::string contents((std::istreambuf_iterator<char>(file1)), std::istreambuf_iterator<char>());
      std::string expectContents(reinterpret_cast<char *> (callback.buffer_), callback.read_size_);
      REQUIRE(expectContents == contents);
      file1.close();
    }
    REQUIRE(flow2->getSize() == 96);
    {
      ReadCallback callback(flow2->getSize());
      sessionGenFlowFile.read(flow2, &callback);
      std::ifstream file2;
      file2.open(EXPECT_MERGE_CONTENT_SECOND, std::ios::in);
      std::string contents((std::istreambuf_iterator<char>(file2)), std::istreambuf_iterator<char>());
      std::string expectContents(reinterpret_cast<char *> (callback.buffer_), callback.read_size_);
      REQUIRE(expectContents == contents);
      file2.close();
    }
    LogTestController::getInstance().reset();
    for (int i = 0; i < 6; i++) {
      std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
      unlink(flowFileName.c_str());
    }
    unlink(EXPECT_MERGE_CONTENT_FIRST);
    unlink(EXPECT_MERGE_CONTENT_SECOND);
  } catch (...) {
  }
}

TEST_CASE("MergeFileDefragmentDelimiter", "[mergefiletest2]") {
  try {
    std::ofstream expectfileFirst;
    std::ofstream expectfileSecond;
    std::ofstream headerfile, footerfile, demarcatorfile;
    expectfileFirst.open(EXPECT_MERGE_CONTENT_FIRST);
    expectfileSecond.open(EXPECT_MERGE_CONTENT_SECOND);
    headerfile.open(HEADER_FILE);
    headerfile << "header";
    expectfileFirst << "header";
    expectfileSecond << "header";
    headerfile.close();
    footerfile.open(FOOTER_FILE);
    footerfile << "footer";
    footerfile.close();
    demarcatorfile.open(DEMARCATOR_FILE);
    demarcatorfile << "demarcator";
    demarcatorfile.close();

    // Create and write to the test file
    for (int i = 0; i < 6; i++) {
      if (i != 0 && i <= 2)
        expectfileFirst << "demarcator";
      if (i != 3 && i >= 4)
        expectfileSecond << "demarcator";
      std::ofstream tmpfile;
      std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
      tmpfile.open(flowFileName.c_str());
      for (int j = 0; j < 32; j++) {
        tmpfile << std::to_string(i);
        if (i < 3)
          expectfileFirst << std::to_string(i);
        else
          expectfileSecond << std::to_string(i);
      }
      tmpfile.close();
    }
    expectfileFirst << "footer";
    expectfileSecond << "footer";
    expectfileFirst.close();
    expectfileSecond.close();

    TestController testController;
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::MergeContent>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::LogAttribute>();
    LogTestController::getInstance().setTrace<core::ProcessSession>();
    LogTestController::getInstance().setTrace<core::repository::VolatileContentRepository>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::BinFiles>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::Bin>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::BinManager>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::Connection>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::core::Connectable>();

    std::shared_ptr<TestRepository> repo = std::make_shared<TestRepository>();

    std::shared_ptr<core::Processor> processor = std::make_shared<org::apache::nifi::minifi::processors::MergeContent>("mergecontent");
    std::shared_ptr<core::Processor> logAttributeProcessor = std::make_shared<org::apache::nifi::minifi::processors::LogAttribute>("logattribute");
    processor->initialize();
    utils::Identifier processoruuid;
    REQUIRE(true == processor->getUUID(processoruuid));
    utils::Identifier logAttributeuuid;
    REQUIRE(true == logAttributeProcessor->getUUID(logAttributeuuid));

    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 merge processor to log attribute
    std::shared_ptr<minifi::Connection> connection = std::make_shared<minifi::Connection>(repo, content_repo, "logattributeconnection");
    connection->addRelationship(core::Relationship("merged", "Merge successful output"));
    connection->setSource(processor);
    connection->setDestination(logAttributeProcessor);
    connection->setSourceUUID(processoruuid);
    connection->setDestinationUUID(logAttributeuuid);
    processor->addConnection(connection);
    // connection to merge processor
    std::shared_ptr<minifi::Connection> mergeconnection = std::make_shared<minifi::Connection>(repo, content_repo, "mergeconnection");
    mergeconnection->setDestination(processor);
    mergeconnection->setDestinationUUID(processoruuid);
    processor->addConnection(mergeconnection);

    std::set<core::Relationship> autoTerminatedRelationships;
    core::Relationship original("original", "");
    core::Relationship failure("failure", "");
    autoTerminatedRelationships.insert(original);
    autoTerminatedRelationships.insert(failure);
    processor->setAutoTerminatedRelationships(autoTerminatedRelationships);

    processor->incrementActiveTasks();
    processor->setScheduledState(core::ScheduledState::RUNNING);
    logAttributeProcessor->incrementActiveTasks();
    logAttributeProcessor->setScheduledState(core::ScheduledState::RUNNING);

    std::shared_ptr<core::ProcessorNode> node = std::make_shared<core::ProcessorNode>(processor);
    std::shared_ptr<core::controller::ControllerServiceProvider> controller_services_provider = nullptr;
    auto context = std::make_shared<core::ProcessContext>(node, controller_services_provider, repo, repo, content_repo);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::MergeFormat, MERGE_FORMAT_CONCAT_VALUE);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::MergeStrategy, MERGE_STRATEGY_DEFRAGMENT);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::DelimiterStratgey, DELIMITER_STRATEGY_FILENAME);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::Header, "/tmp/minifi-mergecontent.header");
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::Footer, "/tmp/minifi-mergecontent.footer");
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::Demarcator, "/tmp/minifi-mergecontent.demarcator");

    core::ProcessSession sessionGenFlowFile(context);
    std::shared_ptr<core::FlowFile> record[6];

    // Generate 6 flowfiles, first threes merged to one, second thress merged to one
    std::shared_ptr<core::Connectable> income = node->getNextIncomingConnection();
    std::shared_ptr<minifi::Connection> income_connection = std::static_pointer_cast<minifi::Connection>(income);
    for (int i = 0; i < 6; i++) {
      std::shared_ptr<core::FlowFile> flow = std::static_pointer_cast < core::FlowFile > (sessionGenFlowFile.create());
      std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
      sessionGenFlowFile.import(flowFileName, flow, true, 0);
      // three bundle
      if (i < 3)
        flow->setAttribute(processors::BinFiles::FRAGMENT_ID_ATTRIBUTE, std::to_string(0));
      else
        flow->setAttribute(processors::BinFiles::FRAGMENT_ID_ATTRIBUTE, std::to_string(1));
      if (i < 3)
        flow->setAttribute(processors::BinFiles::FRAGMENT_INDEX_ATTRIBUTE, std::to_string(i));
      else
        flow->setAttribute(processors::BinFiles::FRAGMENT_INDEX_ATTRIBUTE, std::to_string(i-3));
      flow->setAttribute(processors::BinFiles::FRAGMENT_COUNT_ATTRIBUTE, std::to_string(3));
      record[i] = flow;
    }
    income_connection->put(record[0]);
    income_connection->put(record[2]);
    income_connection->put(record[5]);
    income_connection->put(record[4]);
    income_connection->put(record[1]);
    income_connection->put(record[3]);

    REQUIRE(processor->getName() == "mergecontent");
    auto factory = std::make_shared<core::ProcessSessionFactory>(context);
    processor->onSchedule(context, factory);
    for (int i = 0; i < 6; i++) {
      auto session = std::make_shared<core::ProcessSession>(context);
      processor->onTrigger(context, session);
      session->commit();
    }
    // validate the merge content
    std::set<std::shared_ptr<core::FlowFile>> expiredFlowRecords;
    std::shared_ptr<core::FlowFile> flow1 = connection->poll(expiredFlowRecords);
    std::shared_ptr<core::FlowFile> flow2 = connection->poll(expiredFlowRecords);
    REQUIRE(flow1->getSize() == 128);
    {
      ReadCallback callback(flow1->getSize());
      sessionGenFlowFile.read(flow1, &callback);
      std::ifstream file1;
      file1.open(EXPECT_MERGE_CONTENT_FIRST, std::ios::in);
      std::string contents((std::istreambuf_iterator<char>(file1)), std::istreambuf_iterator<char>());
      std::string expectContents(reinterpret_cast<char *> (callback.buffer_), callback.read_size_);
      REQUIRE(expectContents == contents);
      file1.close();
    }
    REQUIRE(flow2->getSize() == 128);
    {
      ReadCallback callback(flow2->getSize());
      sessionGenFlowFile.read(flow2, &callback);
      std::ifstream file2;
      file2.open(EXPECT_MERGE_CONTENT_SECOND, std::ios::in);
      std::string contents((std::istreambuf_iterator<char>(file2)), std::istreambuf_iterator<char>());
      std::string expectContents(reinterpret_cast<char *> (callback.buffer_), callback.read_size_);
      REQUIRE(expectContents == contents);
      file2.close();
    }
    LogTestController::getInstance().reset();
    for (int i = 0; i < 6; i++) {
      std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
      unlink(flowFileName.c_str());
    }
    unlink(EXPECT_MERGE_CONTENT_FIRST);
    unlink(EXPECT_MERGE_CONTENT_SECOND);
    unlink(FOOTER_FILE);
    unlink(HEADER_FILE);
    unlink(DEMARCATOR_FILE);
  } catch (...) {
  }
}

TEST_CASE("MergeFileDefragmentDropFlow", "[mergefiletest3]") {
  try {
    std::ofstream expectfileFirst;
    std::ofstream expectfileSecond;

    expectfileFirst.open(EXPECT_MERGE_CONTENT_FIRST);
    expectfileSecond.open(EXPECT_MERGE_CONTENT_SECOND);

    // Create and write to the test file, drop record 4
    for (int i = 0; i < 6; i++) {
      if (i == 4)
        continue;
      std::ofstream tmpfile;
      std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
      tmpfile.open(flowFileName.c_str());
      for (int j = 0; j < 32; j++) {
        tmpfile << std::to_string(i);
        if (i < 3)
          expectfileFirst << std::to_string(i);
        else
          expectfileSecond << std::to_string(i);
      }
      tmpfile.close();
    }
    expectfileFirst.close();
    expectfileSecond.close();

    TestController testController;
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::MergeContent>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::LogAttribute>();
    LogTestController::getInstance().setTrace<core::ProcessSession>();
    LogTestController::getInstance().setTrace<core::repository::VolatileContentRepository>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::BinFiles>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::Bin>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::BinManager>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::Connection>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::core::Connectable>();

    std::shared_ptr<TestRepository> repo = std::make_shared<TestRepository>();

    std::shared_ptr<core::Processor> processor = std::make_shared<org::apache::nifi::minifi::processors::MergeContent>("mergecontent");
    std::shared_ptr<core::Processor> logAttributeProcessor = std::make_shared<org::apache::nifi::minifi::processors::LogAttribute>("logattribute");
    processor->initialize();
    utils::Identifier processoruuid;
    REQUIRE(true == processor->getUUID(processoruuid));
    utils::Identifier logAttributeuuid;
    REQUIRE(true == logAttributeProcessor->getUUID(logAttributeuuid));

    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 merge processor to log attribute
    std::shared_ptr<minifi::Connection> connection = std::make_shared<minifi::Connection>(repo, content_repo, "logattributeconnection");
    connection->addRelationship(core::Relationship("merged", "Merge successful output"));
    connection->setSource(processor);
    connection->setDestination(logAttributeProcessor);
    connection->setSourceUUID(processoruuid);
    connection->setDestinationUUID(logAttributeuuid);
    processor->addConnection(connection);
    // connection to merge processor
    std::shared_ptr<minifi::Connection> mergeconnection = std::make_shared<minifi::Connection>(repo, content_repo, "mergeconnection");
    mergeconnection->setDestination(processor);
    mergeconnection->setDestinationUUID(processoruuid);
    processor->addConnection(mergeconnection);

    std::set<core::Relationship> autoTerminatedRelationships;
    core::Relationship original("original", "");
    core::Relationship failure("failure", "");
    autoTerminatedRelationships.insert(original);
    autoTerminatedRelationships.insert(failure);
    processor->setAutoTerminatedRelationships(autoTerminatedRelationships);

    processor->incrementActiveTasks();
    processor->setScheduledState(core::ScheduledState::RUNNING);
    logAttributeProcessor->incrementActiveTasks();
    logAttributeProcessor->setScheduledState(core::ScheduledState::RUNNING);

    std::shared_ptr<core::ProcessorNode> node = std::make_shared<core::ProcessorNode>(processor);
    std::shared_ptr<core::controller::ControllerServiceProvider> controller_services_provider = nullptr;
    auto context = std::make_shared<core::ProcessContext>(node, controller_services_provider, repo, repo, content_repo);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::MergeFormat, MERGE_FORMAT_CONCAT_VALUE);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::MergeStrategy, MERGE_STRATEGY_DEFRAGMENT);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::DelimiterStratgey, DELIMITER_STRATEGY_TEXT);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::MaxBinAge, "1 sec");

    core::ProcessSession sessionGenFlowFile(context);
    std::shared_ptr<core::FlowFile> record[6];

    // Generate 6 flowfiles, first threes merged to one, second thress merged to one
    std::shared_ptr<core::Connectable> income = node->getNextIncomingConnection();
    std::shared_ptr<minifi::Connection> income_connection = std::static_pointer_cast<minifi::Connection>(income);
    for (int i = 0; i < 6; i++) {
      if (i == 4)
        continue;
      std::shared_ptr<core::FlowFile> flow = std::static_pointer_cast < core::FlowFile > (sessionGenFlowFile.create());
      std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
      sessionGenFlowFile.import(flowFileName, flow, true, 0);
      // three bundle
      if (i < 3)
        flow->setAttribute(processors::BinFiles::FRAGMENT_ID_ATTRIBUTE, std::to_string(0));
      else
        flow->setAttribute(processors::BinFiles::FRAGMENT_ID_ATTRIBUTE, std::to_string(1));
      if (i < 3)
        flow->setAttribute(processors::BinFiles::FRAGMENT_INDEX_ATTRIBUTE, std::to_string(i));
      else
        flow->setAttribute(processors::BinFiles::FRAGMENT_INDEX_ATTRIBUTE, std::to_string(i-3));
      flow->setAttribute(processors::BinFiles::FRAGMENT_COUNT_ATTRIBUTE, std::to_string(3));
      record[i] = flow;
    }
    income_connection->put(record[0]);
    income_connection->put(record[2]);
    income_connection->put(record[5]);
    income_connection->put(record[1]);
    income_connection->put(record[3]);

    REQUIRE(processor->getName() == "mergecontent");
    auto factory = std::make_shared<core::ProcessSessionFactory>(context);
    processor->onSchedule(context, factory);
    for (int i = 0; i < 6; i++) {
      if (i == 4)
        continue;
      auto session = std::make_shared<core::ProcessSession>(context);
      processor->onTrigger(context, session);
      session->commit();
    }
    std::this_thread::sleep_for(std::chrono::milliseconds(2000));
    {
      auto session = std::make_shared<core::ProcessSession>(context);
      processor->onTrigger(context, session);
    }
    // validate the merge content
    std::set<std::shared_ptr<core::FlowFile>> expiredFlowRecords;
    std::shared_ptr<core::FlowFile> flow1 = connection->poll(expiredFlowRecords);
    std::shared_ptr<core::FlowFile> flow2 = connection->poll(expiredFlowRecords);
    REQUIRE(flow1->getSize() == 96);
    {
      ReadCallback callback(flow1->getSize());
      sessionGenFlowFile.read(flow1, &callback);
      std::ifstream file1;
      file1.open(EXPECT_MERGE_CONTENT_FIRST, std::ios::in);
      std::string contents((std::istreambuf_iterator<char>(file1)), std::istreambuf_iterator<char>());
      std::string expectContents(reinterpret_cast<char *> (callback.buffer_), callback.read_size_);
      REQUIRE(expectContents == contents);
      file1.close();
    }
    REQUIRE(flow2->getSize() == 64);
    {
      ReadCallback callback(flow2->getSize());
      sessionGenFlowFile.read(flow2, &callback);
      std::ifstream file2;
      file2.open(EXPECT_MERGE_CONTENT_SECOND, std::ios::in);
      std::string contents((std::istreambuf_iterator<char>(file2)), std::istreambuf_iterator<char>());
      std::string expectContents(reinterpret_cast<char *> (callback.buffer_), callback.read_size_);
      REQUIRE(expectContents == contents);
      file2.close();
    }
    LogTestController::getInstance().reset();
    for (int i = 0; i < 6; i++) {
      if (i == 4)
        continue;
      std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
      unlink(flowFileName.c_str());
    }
    unlink(EXPECT_MERGE_CONTENT_FIRST);
    unlink(EXPECT_MERGE_CONTENT_SECOND);
  } catch (...) {
  }
}

TEST_CASE("MergeFileBinPack", "[mergefiletest4]") {
  try {
    std::ofstream expectfileFirst;
    std::ofstream expectfileSecond;
    expectfileFirst.open(EXPECT_MERGE_CONTENT_FIRST);
    expectfileSecond.open(EXPECT_MERGE_CONTENT_SECOND);

    // Create and write to the test file
    for (int i = 0; i < 6; i++) {
      std::ofstream tmpfile;
      std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
      tmpfile.open(flowFileName.c_str());
      for (int j = 0; j < 32; j++) {
        tmpfile << std::to_string(i);
        if (i < 3)
          expectfileFirst << std::to_string(i);
        else
          expectfileSecond << std::to_string(i);
      }
      tmpfile.close();
    }
    expectfileFirst.close();
    expectfileSecond.close();

    TestController testController;
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::MergeContent>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::LogAttribute>();
    LogTestController::getInstance().setTrace<core::ProcessSession>();
    LogTestController::getInstance().setTrace<core::repository::VolatileContentRepository>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::BinFiles>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::Bin>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::BinManager>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::Connection>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::core::Connectable>();

    std::shared_ptr<TestRepository> repo = std::make_shared<TestRepository>();

    std::shared_ptr<core::Processor> processor = std::make_shared<org::apache::nifi::minifi::processors::MergeContent>("mergecontent");
    std::shared_ptr<core::Processor> logAttributeProcessor = std::make_shared<org::apache::nifi::minifi::processors::LogAttribute>("logattribute");
    processor->initialize();
    utils::Identifier processoruuid;
    REQUIRE(true == processor->getUUID(processoruuid));
    utils::Identifier logAttributeuuid;
    REQUIRE(true == logAttributeProcessor->getUUID(logAttributeuuid));

    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 merge processor to log attribute
    std::shared_ptr<minifi::Connection> connection = std::make_shared<minifi::Connection>(repo, content_repo, "logattributeconnection");
    connection->addRelationship(core::Relationship("merged", "Merge successful output"));
    connection->setSource(processor);
    connection->setDestination(logAttributeProcessor);
    connection->setSourceUUID(processoruuid);
    connection->setDestinationUUID(logAttributeuuid);
    processor->addConnection(connection);
    // connection to merge processor
    std::shared_ptr<minifi::Connection> mergeconnection = std::make_shared<minifi::Connection>(repo, content_repo, "mergeconnection");
    mergeconnection->setDestination(processor);
    mergeconnection->setDestinationUUID(processoruuid);
    processor->addConnection(mergeconnection);

    std::set<core::Relationship> autoTerminatedRelationships;
    core::Relationship original("original", "");
    core::Relationship failure("failure", "");
    autoTerminatedRelationships.insert(original);
    autoTerminatedRelationships.insert(failure);
    processor->setAutoTerminatedRelationships(autoTerminatedRelationships);

    processor->incrementActiveTasks();
    processor->setScheduledState(core::ScheduledState::RUNNING);
    logAttributeProcessor->incrementActiveTasks();
    logAttributeProcessor->setScheduledState(core::ScheduledState::RUNNING);

    std::shared_ptr<core::ProcessorNode> node = std::make_shared<core::ProcessorNode>(processor);
    std::shared_ptr<core::controller::ControllerServiceProvider> controller_services_provider = nullptr;
    auto context = std::make_shared<core::ProcessContext>(node, controller_services_provider, repo, repo, content_repo);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::MergeFormat, MERGE_FORMAT_CONCAT_VALUE);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::MergeStrategy, MERGE_STRATEGY_BIN_PACK);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::DelimiterStratgey, DELIMITER_STRATEGY_TEXT);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::MinSize, "96");
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::CorrelationAttributeName, "tag");

    core::ProcessSession sessionGenFlowFile(context);
    std::shared_ptr<core::FlowFile> record[6];

    // Generate 6 flowfiles, first threes merged to one, second thress merged to one
    std::shared_ptr<core::Connectable> income = node->getNextIncomingConnection();
    std::shared_ptr<minifi::Connection> income_connection = std::static_pointer_cast<minifi::Connection>(income);
    for (int i = 0; i < 6; i++) {
      std::shared_ptr<core::FlowFile> flow = std::static_pointer_cast < core::FlowFile > (sessionGenFlowFile.create());
      std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
      sessionGenFlowFile.import(flowFileName, flow, true, 0);
      flow->setAttribute("tag", "tag");
      record[i] = flow;
    }
    income_connection->put(record[0]);
    income_connection->put(record[1]);
    income_connection->put(record[2]);
    income_connection->put(record[3]);
    income_connection->put(record[4]);
    income_connection->put(record[5]);

    REQUIRE(processor->getName() == "mergecontent");
    auto factory = std::make_shared<core::ProcessSessionFactory>(context);
    processor->onSchedule(context, factory);
    for (int i = 0; i < 6; i++) {
      auto session = std::make_shared<core::ProcessSession>(context);
      processor->onTrigger(context, session);
      session->commit();
    }
    // validate the merge content
    std::set<std::shared_ptr<core::FlowFile>> expiredFlowRecords;
    std::shared_ptr<core::FlowFile> flow1 = connection->poll(expiredFlowRecords);
    std::shared_ptr<core::FlowFile> flow2 = connection->poll(expiredFlowRecords);
    REQUIRE(flow1->getSize() == 96);
    {
      ReadCallback callback(flow1->getSize());
      sessionGenFlowFile.read(flow1, &callback);
      std::ifstream file1;
      file1.open(EXPECT_MERGE_CONTENT_FIRST, std::ios::in);
      std::string contents((std::istreambuf_iterator<char>(file1)), std::istreambuf_iterator<char>());
      std::string expectContents(reinterpret_cast<char *> (callback.buffer_), callback.read_size_);
      REQUIRE(expectContents == contents);
      file1.close();
    }
    REQUIRE(flow2->getSize() == 96);
    {
      ReadCallback callback(flow2->getSize());
      sessionGenFlowFile.read(flow2, &callback);
      std::ifstream file2;
      file2.open(EXPECT_MERGE_CONTENT_SECOND, std::ios::in);
      std::string contents((std::istreambuf_iterator<char>(file2)), std::istreambuf_iterator<char>());
      std::string expectContents(reinterpret_cast<char *> (callback.buffer_), callback.read_size_);
      REQUIRE(expectContents == contents);
      file2.close();
    }
    LogTestController::getInstance().reset();
    for (int i = 0; i < 6; i++) {
      std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
      unlink(flowFileName.c_str());
    }
    unlink(EXPECT_MERGE_CONTENT_FIRST);
    unlink(EXPECT_MERGE_CONTENT_SECOND);
  } catch (...) {
  }
}


TEST_CASE("MergeFileTar", "[mergefiletest4]") {
  try {
    std::ofstream expectfileFirst;
    std::ofstream expectfileSecond;
    expectfileFirst.open(EXPECT_MERGE_CONTENT_FIRST);
    expectfileSecond.open(EXPECT_MERGE_CONTENT_SECOND);

    // Create and write to the test file
    for (int i = 0; i < 6; i++) {
      std::ofstream tmpfile;
      std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
      tmpfile.open(flowFileName.c_str());
      for (int j = 0; j < 32; j++) {
        tmpfile << std::to_string(i);
        if (i < 3)
          expectfileFirst << std::to_string(i);
        else
          expectfileSecond << std::to_string(i);
      }
      tmpfile.close();
    }
    expectfileFirst.close();
    expectfileSecond.close();

    TestController testController;
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::MergeContent>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::LogAttribute>();
    LogTestController::getInstance().setTrace<core::ProcessSession>();
    LogTestController::getInstance().setTrace<core::repository::VolatileContentRepository>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::BinFiles>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::Bin>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::BinManager>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::Connection>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::core::Connectable>();

    std::shared_ptr<TestRepository> repo = std::make_shared<TestRepository>();

    std::shared_ptr<core::Processor> processor = std::make_shared<org::apache::nifi::minifi::processors::MergeContent>("mergecontent");
    std::shared_ptr<core::Processor> logAttributeProcessor = std::make_shared<org::apache::nifi::minifi::processors::LogAttribute>("logattribute");
    processor->initialize();
    utils::Identifier processoruuid;
    REQUIRE(true == processor->getUUID(processoruuid));
    utils::Identifier logAttributeuuid;
    REQUIRE(true == logAttributeProcessor->getUUID(logAttributeuuid));

    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 merge processor to log attribute
    std::shared_ptr<minifi::Connection> connection = std::make_shared<minifi::Connection>(repo, content_repo, "logattributeconnection");
    connection->addRelationship(core::Relationship("merged", "Merge successful output"));
    connection->setSource(processor);
    connection->setDestination(logAttributeProcessor);
    connection->setSourceUUID(processoruuid);
    connection->setDestinationUUID(logAttributeuuid);
    processor->addConnection(connection);
    // connection to merge processor
    std::shared_ptr<minifi::Connection> mergeconnection = std::make_shared<minifi::Connection>(repo, content_repo, "mergeconnection");
    mergeconnection->setDestination(processor);
    mergeconnection->setDestinationUUID(processoruuid);
    processor->addConnection(mergeconnection);

    std::set<core::Relationship> autoTerminatedRelationships;
    core::Relationship original("original", "");
    core::Relationship failure("failure", "");
    autoTerminatedRelationships.insert(original);
    autoTerminatedRelationships.insert(failure);
    processor->setAutoTerminatedRelationships(autoTerminatedRelationships);

    processor->incrementActiveTasks();
    processor->setScheduledState(core::ScheduledState::RUNNING);
    logAttributeProcessor->incrementActiveTasks();
    logAttributeProcessor->setScheduledState(core::ScheduledState::RUNNING);

    std::shared_ptr<core::ProcessorNode> node = std::make_shared<core::ProcessorNode>(processor);
    std::shared_ptr<core::controller::ControllerServiceProvider> controller_services_provider = nullptr;
    auto context = std::make_shared<core::ProcessContext>(node, controller_services_provider, repo, repo, content_repo);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::MergeFormat, MERGE_FORMAT_TAR_VALUE);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::MergeStrategy, MERGE_STRATEGY_BIN_PACK);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::DelimiterStratgey, DELIMITER_STRATEGY_TEXT);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::MinSize, "96");
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::CorrelationAttributeName, "tag");

    core::ProcessSession sessionGenFlowFile(context);
    std::shared_ptr<core::FlowFile> record[6];

    // Generate 6 flowfiles, first threes merged to one, second thress merged to one
    std::shared_ptr<core::Connectable> income = node->getNextIncomingConnection();
    std::shared_ptr<minifi::Connection> income_connection = std::static_pointer_cast<minifi::Connection>(income);
    for (int i = 0; i < 6; i++) {
      std::shared_ptr<core::FlowFile> flow = std::static_pointer_cast < core::FlowFile > (sessionGenFlowFile.create());
      std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
      sessionGenFlowFile.import(flowFileName, flow, true, 0);
      flow->setAttribute("tag", "tag");
      record[i] = flow;
    }
    income_connection->put(record[0]);
    income_connection->put(record[1]);
    income_connection->put(record[2]);
    income_connection->put(record[3]);
    income_connection->put(record[4]);
    income_connection->put(record[5]);

    REQUIRE(processor->getName() == "mergecontent");
    auto factory = std::make_shared<core::ProcessSessionFactory>(context);
    processor->onSchedule(context, factory);
    for (int i = 0; i < 6; i++) {
      auto session = std::make_shared<core::ProcessSession>(context);
      processor->onTrigger(context, session);
      session->commit();
    }
    // validate the merge content
    std::set<std::shared_ptr<core::FlowFile>> expiredFlowRecords;
    std::shared_ptr<core::FlowFile> flow1 = connection->poll(expiredFlowRecords);
    std::shared_ptr<core::FlowFile> flow2 = connection->poll(expiredFlowRecords);
    REQUIRE(flow1->getSize() > 0);
    {
      ReadCallback callback(flow1->getSize());
      sessionGenFlowFile.read(flow1, &callback);
      callback.archive_read();
      REQUIRE(callback.archive_buffer_num_ == 3);
      for (int i = 0; i < 3; i++) {
        std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
        std::ifstream file1;
        file1.open(flowFileName, std::ios::in);
        std::string contents((std::istreambuf_iterator<char>(file1)), std::istreambuf_iterator<char>());
        std::string expectContents(reinterpret_cast<char *> (callback.archive_buffer_[i]), callback.archive_buffer_size_[i]);
        REQUIRE(expectContents == contents);
        file1.close();
      }
    }
    REQUIRE(flow2->getSize() > 0);
    {
      ReadCallback callback(flow2->getSize());
      sessionGenFlowFile.read(flow2, &callback);
      callback.archive_read();
      REQUIRE(callback.archive_buffer_num_ == 3);
      for (int i = 3; i < 6; i++) {
        std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
        std::ifstream file1;
        file1.open(flowFileName, std::ios::in);
        std::string contents((std::istreambuf_iterator<char>(file1)), std::istreambuf_iterator<char>());
        std::string expectContents(reinterpret_cast<char *> (callback.archive_buffer_[i-3]), callback.archive_buffer_size_[i-3]);
        REQUIRE(expectContents == contents);
        file1.close();
      }
    }
    LogTestController::getInstance().reset();
    for (int i = 0; i < 6; i++) {
      std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
      unlink(flowFileName.c_str());
    }
    unlink(EXPECT_MERGE_CONTENT_FIRST);
    unlink(EXPECT_MERGE_CONTENT_SECOND);
  } catch (...) {
  }
}

TEST_CASE("MergeFileZip", "[mergefiletest5]") {
  try {
    std::ofstream expectfileFirst;
    std::ofstream expectfileSecond;
    expectfileFirst.open(EXPECT_MERGE_CONTENT_FIRST);
    expectfileSecond.open(EXPECT_MERGE_CONTENT_SECOND);

    // Create and write to the test file
    for (int i = 0; i < 6; i++) {
      std::ofstream tmpfile;
      std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
      tmpfile.open(flowFileName.c_str());
      for (int j = 0; j < 32; j++) {
        tmpfile << std::to_string(i);
        if (i < 3)
          expectfileFirst << std::to_string(i);
        else
          expectfileSecond << std::to_string(i);
      }
      tmpfile.close();
    }
    expectfileFirst.close();
    expectfileSecond.close();

    TestController testController;
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::MergeContent>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::LogAttribute>();
    LogTestController::getInstance().setTrace<core::ProcessSession>();
    LogTestController::getInstance().setTrace<core::repository::VolatileContentRepository>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::BinFiles>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::Bin>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::processors::BinManager>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::Connection>();
    LogTestController::getInstance().setTrace<org::apache::nifi::minifi::core::Connectable>();

    std::shared_ptr<TestRepository> repo = std::make_shared<TestRepository>();

    std::shared_ptr<core::Processor> processor = std::make_shared<org::apache::nifi::minifi::processors::MergeContent>("mergecontent");
    std::shared_ptr<core::Processor> logAttributeProcessor = std::make_shared<org::apache::nifi::minifi::processors::LogAttribute>("logattribute");
    processor->initialize();
    utils::Identifier processoruuid;
    REQUIRE(true == processor->getUUID(processoruuid));
    utils::Identifier logAttributeuuid;
    REQUIRE(true == logAttributeProcessor->getUUID(logAttributeuuid));

    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 merge processor to log attribute
    std::shared_ptr<minifi::Connection> connection = std::make_shared<minifi::Connection>(repo, content_repo, "logattributeconnection");
    connection->addRelationship(core::Relationship("merged", "Merge successful output"));
    connection->setSource(processor);
    connection->setDestination(logAttributeProcessor);
    connection->setSourceUUID(processoruuid);
    connection->setDestinationUUID(logAttributeuuid);
    processor->addConnection(connection);
    // connection to merge processor
    std::shared_ptr<minifi::Connection> mergeconnection = std::make_shared<minifi::Connection>(repo, content_repo, "mergeconnection");
    mergeconnection->setDestination(processor);
    mergeconnection->setDestinationUUID(processoruuid);
    processor->addConnection(mergeconnection);

    std::set<core::Relationship> autoTerminatedRelationships;
    core::Relationship original("original", "");
    core::Relationship failure("failure", "");
    autoTerminatedRelationships.insert(original);
    autoTerminatedRelationships.insert(failure);
    processor->setAutoTerminatedRelationships(autoTerminatedRelationships);

    processor->incrementActiveTasks();
    processor->setScheduledState(core::ScheduledState::RUNNING);
    logAttributeProcessor->incrementActiveTasks();
    logAttributeProcessor->setScheduledState(core::ScheduledState::RUNNING);

    std::shared_ptr<core::ProcessorNode> node = std::make_shared<core::ProcessorNode>(processor);
    std::shared_ptr<core::controller::ControllerServiceProvider> controller_services_provider = nullptr;
    auto context = std::make_shared<core::ProcessContext>(node, controller_services_provider, repo, repo, content_repo);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::MergeFormat, MERGE_FORMAT_ZIP_VALUE);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::MergeStrategy, MERGE_STRATEGY_BIN_PACK);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::DelimiterStratgey, DELIMITER_STRATEGY_TEXT);
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::MinSize, "96");
    context->setProperty(org::apache::nifi::minifi::processors::MergeContent::CorrelationAttributeName, "tag");

    core::ProcessSession sessionGenFlowFile(context);
    std::shared_ptr<core::FlowFile> record[6];

    // Generate 6 flowfiles, first threes merged to one, second thress merged to one
    std::shared_ptr<core::Connectable> income = node->getNextIncomingConnection();
    std::shared_ptr<minifi::Connection> income_connection = std::static_pointer_cast<minifi::Connection>(income);
    for (int i = 0; i < 6; i++) {
      std::shared_ptr<core::FlowFile> flow = std::static_pointer_cast < core::FlowFile > (sessionGenFlowFile.create());
      std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
      sessionGenFlowFile.import(flowFileName, flow, true, 0);
      flow->setAttribute("tag", "tag");
      record[i] = flow;
    }
    income_connection->put(record[0]);
    income_connection->put(record[1]);
    income_connection->put(record[2]);
    income_connection->put(record[3]);
    income_connection->put(record[4]);
    income_connection->put(record[5]);

    REQUIRE(processor->getName() == "mergecontent");
    auto factory = std::make_shared<core::ProcessSessionFactory>(context);
    processor->onSchedule(context, factory);
    for (int i = 0; i < 6; i++) {
      auto session = std::make_shared<core::ProcessSession>(context);
      processor->onTrigger(context, session);
      session->commit();
    }
    // validate the merge content
    std::set<std::shared_ptr<core::FlowFile>> expiredFlowRecords;
    std::shared_ptr<core::FlowFile> flow1 = connection->poll(expiredFlowRecords);
    std::shared_ptr<core::FlowFile> flow2 = connection->poll(expiredFlowRecords);
    REQUIRE(flow1->getSize() > 0);
    {
      ReadCallback callback(flow1->getSize());
      sessionGenFlowFile.read(flow1, &callback);
      callback.archive_read();
      REQUIRE(callback.archive_buffer_num_ == 3);
      for (int i = 0; i < 3; i++) {
        std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
        std::ifstream file1;
        file1.open(flowFileName, std::ios::in);
        std::string contents((std::istreambuf_iterator<char>(file1)), std::istreambuf_iterator<char>());
        std::string expectContents(reinterpret_cast<char *> (callback.archive_buffer_[i]), callback.archive_buffer_size_[i]);
        REQUIRE(expectContents == contents);
        file1.close();
      }
    }
    REQUIRE(flow2->getSize() > 0);
    {
      ReadCallback callback(flow2->getSize());
      sessionGenFlowFile.read(flow2, &callback);
      callback.archive_read();
      REQUIRE(callback.archive_buffer_num_ == 3);
      for (int i = 3; i < 6; i++) {
        std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
        std::ifstream file1;
        file1.open(flowFileName, std::ios::in);
        std::string contents((std::istreambuf_iterator<char>(file1)), std::istreambuf_iterator<char>());
        std::string expectContents(reinterpret_cast<char *> (callback.archive_buffer_[i-3]), callback.archive_buffer_size_[i-3]);
        REQUIRE(expectContents == contents);
        file1.close();
      }
    }
    LogTestController::getInstance().reset();
    for (int i = 0; i < 6; i++) {
      std::string flowFileName = std::string(FLOW_FILE) + "." + std::to_string(i) + ".txt";
      unlink(flowFileName.c_str());
    }
    unlink(EXPECT_MERGE_CONTENT_FIRST);
    unlink(EXPECT_MERGE_CONTENT_SECOND);
  } catch (...) {
  }
}