/**
 *
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#ifndef LIBMINIFI_INCLUDE_CAPI_INSTANCE_H_
#define LIBMINIFI_INCLUDE_CAPI_INSTANCE_H_

#include <memory>
#include <type_traits>
#include <string>
#include "core/Property.h"
#include "properties/Configure.h"
#include "io/StreamFactory.h"
#include "RemoteProcessorGroupPort.h"
#include "core/ContentRepository.h"
#include "core/repository/VolatileContentRepository.h"
#include "core/Repository.h"

#include "C2CallbackAgent.h"
#include "core/Connectable.h"
#include "core/ProcessorNode.h"
#include "core/ProcessContext.h"
#include "core/ProcessSession.h"
#include "core/ProcessSessionFactory.h"
#include "core/controller/ControllerServiceProvider.h"
#include "core/FlowConfiguration.h"
#include "ReflexiveSession.h"
#include "utils/ThreadPool.h"
#include "core/state/UpdateController.h"
namespace org {
namespace apache {
namespace nifi {
namespace minifi {

class ProcessorLink {
 public:
  explicit ProcessorLink(const std::shared_ptr<core::Processor> &processor)
      : processor_(processor) {

  }

  const std::shared_ptr<core::Processor> &getProcessor() {
    return processor_;
  }

 protected:
  std::shared_ptr<core::Processor> processor_;
};

class Instance {
 public:

  explicit Instance(const std::string &url, const std::string &port)
      : configure_(std::make_shared<Configure>()),
        url_(url),
        agent_(nullptr),
        rpgInitialized_(false),
        listener_thread_pool_(1),
        content_repo_(std::make_shared<minifi::core::repository::VolatileContentRepository>()),
        no_op_repo_(std::make_shared<minifi::core::Repository>()) {
    running_ = false;
    stream_factory_ = minifi::io::StreamFactory::getInstance(configure_);
    uuid_t uuid;
    uuid_parse(port.c_str(), uuid);
    rpg_ = std::make_shared<minifi::RemoteProcessorGroupPort>(stream_factory_, url, url, configure_, uuid);
    proc_node_ = std::make_shared<core::ProcessorNode>(rpg_);
    core::FlowConfiguration::initialize_static_functions();
    content_repo_->initialize(configure_);
  }

  ~Instance() {
    running_ = false;
    listener_thread_pool_.shutdown();
  }

  bool isRPGConfigured() {
    return rpgInitialized_;
  }

  void enableAsyncC2(C2_Server *server,c2_stop_callback *c1, c2_start_callback *c2, c2_update_callback *c3) {
    std::shared_ptr<core::controller::ControllerServiceProvider> controller_service_provider = nullptr;
    running_ = true;
    if (server->type != C2_Server_Type::MQTT){
      configure_->set("c2.rest.url",server->url);
      configure_->set("c2.rest.url.ack",server->ack_url);
    }
    agent_ = std::make_shared<c2::C2CallbackAgent>(controller_service_provider, nullptr, configure_);
    listener_thread_pool_.start();
    registerUpdateListener(agent_, 1000);
    agent_->setStopCallback(c1);
  }

  void setRemotePort(std::string remote_port) {
    rpg_->setProperty(minifi::RemoteProcessorGroupPort::portUUID, remote_port);
    rpg_->initialize();
    rpg_->setTransmitting(true);
    rpgInitialized_ = true;
  }

  std::shared_ptr<Configure> getConfiguration() {
    return configure_;
  }

  std::shared_ptr<minifi::core::Repository> getNoOpRepository() {
    return no_op_repo_;
  }

  std::shared_ptr<minifi::core::ContentRepository> getContentRepository() {
    return content_repo_;
  }

  void transfer(const std::shared_ptr<FlowFileRecord> &ff) {
    std::shared_ptr<core::controller::ControllerServiceProvider> controller_service_provider = nullptr;
    auto processContext = std::make_shared<core::ProcessContext>(proc_node_, controller_service_provider, no_op_repo_, no_op_repo_, content_repo_);
    auto sessionFactory = std::make_shared<core::ProcessSessionFactory>(processContext);

    rpg_->onSchedule(processContext, sessionFactory);

    auto session = std::make_shared<core::ReflexiveSession>(processContext);

    session->add(ff);

    rpg_->onTrigger(processContext, session);
  }

 protected:

  bool registerUpdateListener(const std::shared_ptr<state::UpdateController> &updateController, const int64_t &delay) {
    auto functions = updateController->getFunctions();
    // run all functions independently

    for (auto function : functions) {
      std::unique_ptr<utils::AfterExecute<state::Update>> after_execute = std::unique_ptr<utils::AfterExecute<state::Update>>(new state::UpdateRunner(running_, delay));
      utils::Worker<state::Update> functor(function, "listeners", std::move(after_execute));
      std::future<state::Update> future;
      if (!listener_thread_pool_.execute(std::move(functor), future)) {
        // denote failure
        return false;
      }
    }
    return true;
  }

  std::shared_ptr<c2::C2CallbackAgent> agent_;

  std::atomic<bool> running_;

  bool rpgInitialized_;

  std::shared_ptr<minifi::core::Repository> no_op_repo_;

  std::shared_ptr<minifi::core::ContentRepository> content_repo_;

  std::shared_ptr<core::ProcessorNode> proc_node_;
  std::shared_ptr<minifi::RemoteProcessorGroupPort> rpg_;
  std::shared_ptr<io::StreamFactory> stream_factory_;
  std::string url_;
  std::shared_ptr<Configure> configure_;

  utils::ThreadPool<state::Update> listener_thread_pool_;
};

} /* namespace minifi */
} /* namespace nifi */
} /* namespace apache */
} /* namespace org */
#endif /* LIBMINIFI_INCLUDE_CAPI_INSTANCE_H_ */