| 136 | } |
| 137 | |
| 138 | void RemoteProcessorGroupPort::onSchedule(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSessionFactory>& /*sessionFactory*/) { |
| 139 | std::string value; |
| 140 | if (context->getProperty(portUUID.getName(), value) && !value.empty()) { |
| 141 | protocol_uuid_ = value; |
| 142 | } |
| 143 | |
| 144 | std::string http_enabled_str; |
| 145 | if (configure_->get(Configure::nifi_remote_input_http, http_enabled_str)) { |
| 146 | if (utils::StringUtils::StringToBool(http_enabled_str, http_enabled_)) { |
| 147 | if (client_type_ == sitetosite::CLIENT_TYPE::RAW) { |
| 148 | logger_->log_trace("Remote Input HTTP Enabled, but raw has been suggested for %s", protocol_uuid_.to_string()); |
| 149 | } |
| 150 | } |
| 151 | } |
| 152 | |
| 153 | std::string context_name; |
| 154 | if (!context->getProperty(SSLContext.getName(), context_name) || IsNullOrEmpty(context_name)) { |
| 155 | context_name = RPG_SSL_CONTEXT_SERVICE_NAME; |
| 156 | } |
| 157 | std::shared_ptr<core::controller::ControllerService> service = context->getControllerService(context_name); |
| 158 | if (nullptr != service) { |
| 159 | ssl_service = std::static_pointer_cast<minifi::controllers::SSLContextService>(service); |
| 160 | } else { |
| 161 | std::string secureStr; |
| 162 | bool is_secure = false; |
| 163 | if (configure_->get(Configure::nifi_remote_input_secure, secureStr) && |
| 164 | org::apache::nifi::minifi::utils::StringUtils::StringToBool(secureStr, is_secure)) { |
| 165 | ssl_service = std::make_shared<minifi::controllers::SSLContextService>(RPG_SSL_CONTEXT_SERVICE_NAME, configure_); |
| 166 | ssl_service->onEnable(); |
| 167 | } |
| 168 | } |
| 169 | { |
| 170 | uint64_t idleTimeoutVal = 15000; |
| 171 | std::string idleTimeoutStr; |
| 172 | if (!context->getProperty(idleTimeout.getName(), idleTimeoutStr) |
| 173 | || !core::Property::getTimeMSFromString(idleTimeoutStr, idleTimeoutVal)) { |
| 174 | logger_->log_debug("%s attribute is invalid, so default value of %s will be used", idleTimeout.getName(), |
| 175 | idleTimeout.getValue()); |
| 176 | if (!core::Property::getTimeMSFromString(idleTimeout.getValue(), idleTimeoutVal)) { |
| 177 | assert(false); // Couldn't parse our default value |
| 178 | } |
| 179 | } |
| 180 | idle_timeout_ = std::chrono::milliseconds(idleTimeoutVal); |
| 181 | } |
| 182 | |
| 183 | std::lock_guard<std::mutex> lock(peer_mutex_); |
| 184 | if (!nifi_instances_.empty()) { |
| 185 | refreshPeerList(); |
| 186 | if (peers_.size() > 0) |
| 187 | peer_index_ = 0; |
| 188 | } |
| 189 | /** |
| 190 | * If at this point we have no peers and HTTP support is disabled this means |
| 191 | * we must rely on the configured host/port |
| 192 | */ |
| 193 | if (peers_.empty() && is_http_disabled()) { |
| 194 | std::string host, portStr; |
| 195 | int configured_port = -1; |
no test coverage detected