| 254 | } |
| 255 | |
| 256 | std::unique_ptr<core::ProcessGroup> FlowController::loadInitialFlow() { |
| 257 | std::unique_ptr<core::ProcessGroup> root = flow_configuration_->getRoot(); |
| 258 | if (root) { |
| 259 | return root; |
| 260 | } |
| 261 | logger_->log_error("Couldn't load flow configuration file, trying to fetch it from C2 server"); |
| 262 | auto opt_flow_url = configuration_->get(Configure::nifi_c2_flow_url); |
| 263 | if (!opt_flow_url) { |
| 264 | logger_->log_error("No flow configuration url found"); |
| 265 | return nullptr; |
| 266 | } |
| 267 | // ensure that C2 connection is up and running |
| 268 | // since we don't have access to the flow definition, the C2 communication |
| 269 | // won't be able to use the services defined there, e.g. SSLContextService |
| 270 | controller_service_provider_impl_ = flow_configuration_->getControllerServiceProvider(); |
| 271 | C2Client::initialize(this, shared_from_this()); |
| 272 | auto opt_source = fetchFlow(*opt_flow_url); |
| 273 | if (!opt_source) { |
| 274 | logger_->log_error("Couldn't fetch flow configuration from C2 server"); |
| 275 | return nullptr; |
| 276 | } |
| 277 | root = flow_configuration_->updateFromPayload(*opt_flow_url, *opt_source); |
| 278 | if (root) { |
| 279 | logger_->log_info("Successfully fetched valid flow configuration"); |
| 280 | if (!flow_configuration_->persist(*opt_source)) { |
| 281 | logger_->log_info("Failed to write the fetched flow to disk"); |
| 282 | } |
| 283 | } |
| 284 | return root; |
| 285 | } |
| 286 | |
| 287 | void FlowController::load(const std::shared_ptr<core::ProcessGroup> &root, bool reload) { |
| 288 | std::lock_guard<std::recursive_mutex> flow_lock(mutex_); |