text: /**
   *
   * 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 <fcntl.h>
  #include <stdio.h>
  #include <semaphore.h>
  #include <signal.h>
  #include <vector>
  #include <queue>
  #include <map>
  #include <iostream>
  #include "io/BaseStream.h"
  
  #include "core/Core.h"
  
  #include "core/FlowConfiguration.h"
  #include "core/ConfigurationFactory.h"
  #include "core/RepositoryFactory.h"
  #include "FlowController.h"
  #include "MainHelper.h"
  #include "properties/Configure.h"
  #include "Controller.h"
  #include "c2/ControllerSocketProtocol.h"
  
  #include "cxxopts.hpp"
  
  namespace minifi = org::apache::nifi::minifi;
  
  int main(int argc, char **argv) {
    const auto logger = minifi::core::logging::LoggerConfiguration::getConfiguration().getLogger("controller");
  
    const std::string minifiHome = determineMinifiHome(logger);
    if (minifiHome.empty()) {
      // determineMinifiHome already logged everything we need
      return -1;
    }
  
    const auto configuration = std::make_shared<minifi::Configure>();
    configuration->setHome(minifiHome);
    configuration->loadConfigureFile(DEFAULT_NIFI_PROPERTIES_FILE);
  
    const auto log_properties = std::make_shared<minifi::core::logging::LoggerProperties>();
    log_properties->setHome(minifiHome);
    log_properties->loadConfigureFile(DEFAULT_LOG_PROPERTIES_FILE);
    minifi::core::logging::LoggerConfiguration::getConfiguration().initialize(log_properties);
  
    std::string context_name;
  
    std::shared_ptr<minifi::controllers::SSLContextService> secure_context = nullptr;
  
    // if the user wishes to use a controller service we need to instantiate the flow
    if (configuration->get("controller.ssl.context.service", context_name)) {
      const auto service = getControllerService(configuration, context_name);
      if (nullptr != service) {
        secure_context = std::static_pointer_cast<minifi::controllers::SSLContextService>(service);
      }
    }
  
    if (nullptr == secure_context) {
      std::string secureStr;
      if (configuration->get(minifi::Configure::nifi_remote_input_secure, secureStr) && minifi::utils::StringUtils::toBool(secureStr).value_or(false)) {
        secure_context = std::make_shared<minifi::controllers::SSLContextService>("ControllerSocketProtocolSSL", configuration);
        secure_context->onEnable();
      }
    }
  
    std::string value;
  
    auto stream_factory_ = minifi::io::StreamFactory::getInstance(configuration);
  
    std::string host = "localhost", portStr, caCert;
    int port = -1;
  
    cxxopts::Options options("MiNiFiController", "MiNiFi local agent controller");
    options.positional_help("[optional args]").show_positional_help();
  
    options.add_options()  //NOLINT
    ("h,help", "Shows Help")  //NOLINT
    ("host", "Specifies connecting host name", cxxopts::value<std::string>())  //NOLINT
    ("port", "Specifies connecting host port", cxxopts::value<int>())  //NOLINT
    ("stop", "Shuts down the provided component", cxxopts::value<std::vector<std::string>>())  //NOLINT
    ("start", "Starts provided component", cxxopts::value<std::vector<std::string>>())  //NOLINT
    ("l,list", "Provides a list of connections or processors", cxxopts::value<std::string>())  //NOLINT
    ("c,clear", "Clears the associated connection queue", cxxopts::value<std::vector<std::string>>())  //NOLINT
    ("getsize", "Reports the size of the associated connection queue", cxxopts::value<std::vector<std::string>>())  //NOLINT
    ("updateflow", "Updates the flow of the agent using the provided flow file", cxxopts::value<std::string>())  //NOLINT
    ("getfull", "Reports a list of full connections")  //NOLINT
    ("jstack", "Returns backtraces from the agent")  //NOLINT
    ("manifest", "Generates a manifest for the current binary")  //NOLINT
    ("noheaders", "Removes headers from output streams");
  
    bool show_headers = true;
  
    try {
      auto result = options.parse(argc, argv);
  
      if (result.count("help")) {
        std::cout << options.help({ "", "Group" }) << std::endl;
        exit(0);
      }
  
      if (result.count("host")) {
        host = result["host"].as<std::string>();
      } else {
        configuration->get("controller.socket.host", host);
      }
  
      if (result.count("port")) {
        port = result["port"].as<int>();
      } else {
        if (port == -1 && configuration->get("controller.socket.port", portStr)) {
          port = std::stoi(portStr);
        }
      }
  
      if ((IsNullOrEmpty(host) && port == -1)) {
        std::cout << "MiNiFi Controller is disabled" << std::endl;
        exit(0);
      }
      if (result.count("noheaders")) {
        show_headers = false;
      }
  
      if (result.count("stop") > 0) {
        auto& components = result["stop"].as<std::vector<std::string>>();
        for (const auto& component : components) {
          auto socket = secure_context != nullptr ? stream_factory_->createSecureSocket(host, port, secure_context) : stream_factory_->createSocket(host, port);
          if (stopComponent(std::move(socket), component))
            std::cout << component << " requested to stop" << std::endl;
          else
            std::cout << "Could not connect to remote host " << host << ":" << port << std::endl;
        }
      }
  
      if (result.count("start") > 0) {
        auto& components = result["start"].as<std::vector<std::string>>();
        for (const auto& component : components) {
          auto socket = secure_context != nullptr ? stream_factory_->createSecureSocket(host, port, secure_context) : stream_factory_->createSocket(host, port);
          if (startComponent(std::move(socket), component))
            std::cout << component << " requested to start" << std::endl;
          else
            std::cout << "Could not connect to remote host " << host << ":" << port << std::endl;
        }
      }
  
      if (result.count("c") > 0) {
        auto& components = result["c"].as<std::vector<std::string>>();
        for (const auto& connection : components) {
          auto socket = secure_context != nullptr ? stream_factory_->createSecureSocket(host, port, secure_context)
                                                  : stream_factory_->createSocket(host, port);
          if (clearConnection(std::move(socket), connection)) {
            std::cout << "Sent clear command to " << connection << ". Size before clear operation sent: " << std::endl;
            socket = secure_context != nullptr ? stream_factory_->createSecureSocket(host, port, secure_context)
                                               : stream_factory_->createSocket(host, port);
            if (getConnectionSize(std::move(socket), std::cout, connection) < 0)
              std::cout << "Could not connect to remote host " << host << ":" << port << std::endl;
          } else {
            std::cout << "Could not connect to remote host " << host << ":" << port << std::endl;
          }
        }
      }
  
      if (result.count("getsize") > 0) {
        auto& components = result["getsize"].as<std::vector<std::string>>();
        for (const auto& component : components) {
          auto socket = secure_context != nullptr ? stream_factory_->createSecureSocket(host, port, secure_context) : stream_factory_->createSocket(host, port);
          if (getConnectionSize(std::move(socket), std::cout, component) < 0)
            std::cout << "Could not connect to remote host " << host << ":" << port << std::endl;
        }
      }
      if (result.count("l") > 0) {
        auto& option = result["l"].as<std::string>();
        auto socket = secure_context != nullptr ? stream_factory_->createSecureSocket(host, port, secure_context) : stream_factory_->createSocket(host, port);
        if (option == "components") {
          if (listComponents(std::move(socket), std::cout, show_headers) < 0)
            std::cout << "Could not connect to remote host " << host << ":" << port << std::endl;
        } else if (option == "connections") {
          if (listConnections(std::move(socket), std::cout, show_headers) < 0)
            std::cout << "Could not connect to remote host " << host << ":" << port << std::endl;
        }
      }
      if (result.count("getfull") > 0) {
        auto socket = secure_context != nullptr ? stream_factory_->createSecureSocket(host, port, secure_context) : stream_factory_->createSocket(host, port);
        if (getFullConnections(std::move(socket), std::cout) < 0)
          std::cout << "Could not connect to remote host " << host << ":" << port << std::endl;
      }
  
      if (result.count("jstack") > 0) {
        auto socket = secure_context != nullptr ? stream_factory_->createSecureSocket(host, port, secure_context) : stream_factory_->createSocket(host, port);
        if (getJstacks(std::move(socket), std::cout) < 0)
          std::cout << "Could not connect to remote host " << host << ":" << port << std::endl;
      }
  
      if (result.count("updateflow") > 0) {
        auto& flow_file = result["updateflow"].as<std::string>();
        auto socket = secure_context != nullptr ? stream_factory_->createSecureSocket(host, port, secure_context) : stream_factory_->createSocket(host, port);
        if (updateFlow(std::move(socket), std::cout, flow_file) < 0)
          std::cout << "Could not connect to remote host " << host << ":" << port << std::endl;
      }
  
      if (result.count("manifest") > 0) {
        printManifest(configuration);
      }
    } catch (const std::exception &exc) {
      // catch anything thrown within try block that derives from std::exception
      std::cerr << exc.what() << std::endl;
    } catch (...) {
      std::cout << options.help({ "", "Group" }) << std::endl;
      exit(0);
    }
    return 0;
  }
  