[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1503: MINIFICPP-2039 Dust off minificontroller
lordgamez commented on code in PR #1503:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1503#discussion_r1165408681
##
controller/MiNiFiController.cpp:
##
@@ -75,33 +80,60 @@ int main(int argc, char **argv) {
secure_context =
std::make_shared("ControllerSocketProtocolSSL",
configuration);
secure_context->onEnable();
}
+ } else {
+secure_context->onEnable();
}
+ return secure_context;
+}
- std::string value;
+int main(int argc, char **argv) {
+ const auto logger =
minifi::core::logging::LoggerConfiguration::getConfiguration().getLogger("controller");
+ const std::string minifi_home = determineMinifiHome(logger);
+ if (minifi_home.empty()) {
+// determineMinifiHome already logged everything we need
+return -1;
+ }
+
+ const auto configuration = std::make_shared();
+ configuration->setHome(minifi_home);
+ configuration->loadConfigureFile(DEFAULT_NIFI_PROPERTIES_FILE);
+
+ const auto log_properties =
std::make_shared();
+ log_properties->setHome(minifi_home);
+ log_properties->loadConfigureFile(DEFAULT_LOG_PROPERTIES_FILE);
+
minifi::core::logging::LoggerConfiguration::getConfiguration().initialize(log_properties);
+
+ std::shared_ptr secure_context;
+ try {
+secure_context = getSSLContextService(configuration);
+ } catch(const minifi::Exception& ex) {
+logger->log_error(ex.what());
+exit(1);
+ }
auto stream_factory_ = minifi::io::StreamFactory::getInstance(configuration);
std::string host = "localhost";
- std::string portStr;
- std::string caCert;
+ std::string port_str;
+ std::string ca_cert;
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())
//NOLINT
- ("port", "Specifies connecting host port", cxxopts::value()) //NOLINT
- ("stop", "Shuts down the provided component",
cxxopts::value>()) //NOLINT
- ("start", "Starts provided component",
cxxopts::value>()) //NOLINT
- ("l,list", "Provides a list of connections or processors",
cxxopts::value()) //NOLINT
- ("c,clear", "Clears the associated connection queue",
cxxopts::value>()) //NOLINT
- ("getsize", "Reports the size of the associated connection queue",
cxxopts::value>()) //NOLINT
- ("updateflow", "Updates the flow of the agent using the provided flow file",
cxxopts::value()) //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
+ options.add_options()
+ ("h,help", "Shows Help")
+ ("host", "Specifies connecting host name", cxxopts::value())
+ ("port", "Specifies connecting host port", cxxopts::value())
+ ("stop", "Shuts down the provided component",
cxxopts::value>())
+ ("start", "Starts provided component",
cxxopts::value>())
+ ("l,list", "Provides a list of connections or processors",
cxxopts::value())
+ ("c,clear", "Clears the associated connection queue",
cxxopts::value>())
+ ("getsize", "Reports the size of the associated connection queue",
cxxopts::value>())
+ ("updateflow", "Updates the flow of the agent using the provided flow file",
cxxopts::value())
+ ("getfull", "Reports a list of full connections")
+ ("jstack", "Returns backtraces from the agent")
+ ("manifest", "Generates a manifest for the current binary")
Review Comment:
Updated in 03f64d39caf87072f2cb0063cb4ea164f0ffe29e
##
controller/tests/ControllerTests.cpp:
##
@@ -0,0 +1,545 @@
+/**
+ *
+ * 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
+#include
+#include
+#include
+#include
+#include "range/v3/algorithm/find.hpp"
+
+#include "TestBase.h"
+#include "Catch.h"
+#include "io/ClientSocket.h"
+#include "core/Processor.h"
+#include "Controller.h"
+#include "c2/ControllerSocketProtocol.h"
+#include "utils/IntegrationTestUtils.h"
+#include "c2/ControllerSocketMetricsPublisher.h"
+#include "core/controller/ControllerServiceProvider.h"
+#include
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1503: MINIFICPP-2039 Dust off minificontroller
lordgamez commented on code in PR #1503:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1503#discussion_r1159633968
##
controller/Controller.cpp:
##
@@ -0,0 +1,214 @@
+/**
+ * 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 "Controller.h"
+
+#include
+
+#include "io/BufferStream.h"
+#include "c2/C2Payload.h"
+
+namespace org::apache::nifi::minifi::controller {
+
+bool sendSingleCommand(std::unique_ptr socket, uint8_t op, const
std::string& value) {
+ if (socket->initialize() < 0) {
+return false;
+ }
+ io::BufferStream stream;
+ stream.write(, 1);
+ stream.write(value);
+ return socket->write(stream.getBuffer()) == stream.size();
+}
+
+bool stopComponent(std::unique_ptr socket, const std::string&
component) {
+ return sendSingleCommand(std::move(socket), c2::Operation::STOP, component);
+}
+
+bool startComponent(std::unique_ptr socket, const std::string&
component) {
+ return sendSingleCommand(std::move(socket), c2::Operation::START, component);
+}
+
+bool clearConnection(std::unique_ptr socket, const std::string&
connection) {
+ return sendSingleCommand(std::move(socket), c2::Operation::CLEAR,
connection);
+}
+
+int updateFlow(std::unique_ptr socket, std::ostream , const
std::string& file) {
+ if (socket->initialize() < 0) {
+return -1;
+ }
+ uint8_t op = c2::Operation::UPDATE;
+ io::BufferStream stream;
+ stream.write(, 1);
+ stream.write("flow");
+ stream.write(file);
+ if (io::isError(socket->write(stream.getBuffer( {
+return -1;
+ }
+ // read the response
+ uint8_t resp = 0;
+ socket->read(resp);
+ if (resp == c2::Operation::DESCRIBE) {
+uint16_t connections = 0;
+socket->read(connections);
+out << connections << " are full" << std::endl;
+for (int i = 0; i < connections; i++) {
+ std::string fullcomponent;
+ socket->read(fullcomponent);
+ out << fullcomponent << " is full" << std::endl;
+}
+ }
+ return 0;
+}
+
+int getFullConnections(std::unique_ptr socket, std::ostream ) {
+ if (socket->initialize() < 0) {
+return -1;
+ }
+ uint8_t op = c2::Operation::DESCRIBE;
+ io::BufferStream stream;
+ stream.write(, 1);
+ stream.write("getfull");
+ if (io::isError(socket->write(stream.getBuffer( {
+return -1;
+ }
+ // read the response
+ uint8_t resp = 0;
+ socket->read(resp);
+ if (resp == c2::Operation::DESCRIBE) {
+uint16_t connections = 0;
+socket->read(connections);
+out << connections << " are full" << std::endl;
+for (int i = 0; i < connections; i++) {
+ std::string fullcomponent;
+ socket->read(fullcomponent);
+ out << fullcomponent << " is full" << std::endl;
+}
+ }
+ return 0;
+}
+
+int getConnectionSize(std::unique_ptr socket, std::ostream ,
const std::string& connection) {
+ if (socket->initialize() < 0) {
+return -1;
+ }
+ uint8_t op = c2::Operation::DESCRIBE;
+ io::BufferStream stream;
+ stream.write(, 1);
+ stream.write("queue");
+ stream.write(connection);
+ if (io::isError(socket->write(stream.getBuffer( {
+return -1;
+ }
+ // read the response
+ uint8_t resp = 0;
+ socket->read(resp);
+ if (resp == c2::Operation::DESCRIBE) {
+std::string size;
+socket->read(size);
+out << "Size/Max of " << connection << " " << size << std::endl;
+ }
+ return 0;
+}
+
+int listComponents(std::unique_ptr socket, std::ostream , bool
show_header) {
+ if (socket->initialize() < 0) {
+return -1;
+ }
+ io::BufferStream stream;
+ uint8_t op = c2::Operation::DESCRIBE;
+ stream.write(, 1);
+ stream.write("components");
+ if (io::isError(socket->write(stream.getBuffer( {
+return -1;
+ }
+ uint16_t responses = 0;
+ socket->read(op);
+ socket->read(responses);
+ if (show_header)
+out << "Components:" << std::endl;
+
+ for (int i = 0; i < responses; i++) {
+std::string name;
+socket->read(name, false);
+std::string status;
+socket->read(status, false);
+out << name << ", running: " << status << std::endl;
+ }
+ return 0;
+}
+
+int listConnections(std::unique_ptr socket, std::ostream ,
bool show_header) {
+ if (socket->initialize() < 0) {
+return -1;
+ }
+ io::BufferStream stream;
+ uint8_t op =
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1503: MINIFICPP-2039 Dust off minificontroller
lordgamez commented on code in PR #1503:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1503#discussion_r1152024653
##
controller/Controller.h:
##
@@ -18,346 +18,61 @@
#pragma once
#include
+#include
-#include "core/RepositoryFactory.h"
-#include "core/ConfigurationFactory.h"
-#include "core/extension/ExtensionManager.h"
#include "io/ClientSocket.h"
-#include "c2/ControllerSocketProtocol.h"
-#include "utils/gsl.h"
-#include "Exception.h"
-#include "FlowController.h"
-#include "core/repository/VolatileContentRepository.h"
-#include "core/repository/VolatileFlowFileRepository.h"
-#include "core/state/MetricsPublisherFactory.h"
-#include "core/state/MetricsPublisherStore.h"
+
+namespace org::apache::nifi::minifi::controller {
/**
* Sends a single argument comment
* @param socket socket unique ptr.
* @param op operation to perform
* @param value value to send
*/
-bool sendSingleCommand(std::unique_ptr
socket, uint8_t op, const std::string& value) {
- socket->initialize();
- std::vector data;
- org::apache::nifi::minifi::io::BufferStream stream;
- stream.write(, 1);
- stream.write(value);
- return socket->write(stream.getBuffer()) == stream.size();
-}
+bool sendSingleCommand(std::unique_ptr socket, uint8_t op, const
std::string& value);
/**
* Stops a stopped component
* @param socket socket unique ptr.
* @param op operation to perform
*/
-bool stopComponent(std::unique_ptr
socket, const std::string& component) {
- return sendSingleCommand(std::move(socket),
org::apache::nifi::minifi::c2::Operation::STOP, component);
-}
+bool stopComponent(std::unique_ptr socket, const std::string&
component);
/**
* Starts a previously stopped component.
* @param socket socket unique ptr.
* @param op operation to perform
*/
-bool startComponent(std::unique_ptr
socket, const std::string& component) {
- return sendSingleCommand(std::move(socket),
org::apache::nifi::minifi::c2::Operation::START, component);
-}
+bool startComponent(std::unique_ptr socket, const std::string&
component);
/**
* Clears a connection queue.
* @param socket socket unique ptr.
* @param op operation to perform
*/
-bool clearConnection(std::unique_ptr
socket, const std::string& connection) {
- return sendSingleCommand(std::move(socket),
org::apache::nifi::minifi::c2::Operation::CLEAR, connection);
-}
+bool clearConnection(std::unique_ptr socket, const std::string&
connection);
/**
* Updates the flow to the provided file
*/
-int updateFlow(std::unique_ptr socket,
std::ostream , const std::string& file) {
- socket->initialize();
- std::vector data;
- uint8_t op = org::apache::nifi::minifi::c2::Operation::UPDATE;
- org::apache::nifi::minifi::io::BufferStream stream;
- stream.write(, 1);
- stream.write("flow");
- stream.write(file);
- if
(org::apache::nifi::minifi::io::isError(socket->write(stream.getBuffer( {
-return -1;
- }
- // read the response
- uint8_t resp = 0;
- socket->read(resp);
- if (resp == org::apache::nifi::minifi::c2::Operation::DESCRIBE) {
-uint16_t connections = 0;
-socket->read(connections);
-out << connections << " are full" << std::endl;
-for (int i = 0; i < connections; i++) {
- std::string fullcomponent;
- socket->read(fullcomponent);
- out << fullcomponent << " is full" << std::endl;
-}
- }
- return 0;
-}
+int updateFlow(std::unique_ptr socket, std::ostream , const
std::string& file);
/**
* Lists connections which are full
* @param socket socket ptr
*/
-int getFullConnections(std::unique_ptr
socket, std::ostream ) {
- socket->initialize();
- std::vector data;
- uint8_t op = org::apache::nifi::minifi::c2::Operation::DESCRIBE;
- org::apache::nifi::minifi::io::BufferStream stream;
- stream.write(, 1);
- stream.write("getfull");
- if
(org::apache::nifi::minifi::io::isError(socket->write(stream.getBuffer( {
-return -1;
- }
- // read the response
- uint8_t resp = 0;
- socket->read(resp);
- if (resp == org::apache::nifi::minifi::c2::Operation::DESCRIBE) {
-uint16_t connections = 0;
-socket->read(connections);
-out << connections << " are full" << std::endl;
-for (int i = 0; i < connections; i++) {
- std::string fullcomponent;
- socket->read(fullcomponent);
- out << fullcomponent << " is full" << std::endl;
-}
- }
- return 0;
-}
-
-int getJstacks(std::unique_ptr socket,
std::ostream ) {
Review Comment:
Added comment in 8735105d4cd70f6bbe33fbe43bd17760382a9b1d
##
controller/MiNiFiController.cpp:
##
@@ -15,57 +15,52 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include
-#include
-#include
-#include
#include
-#include
-#include
#include
-#include "core/Core.h"
-
-#include "core/FlowConfiguration.h"
-#include "core/ConfigurationFactory.h"
-#include "core/RepositoryFactory.h"
-#include "FlowController.h"
#include
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1503: MINIFICPP-2039 Dust off minificontroller
lordgamez commented on code in PR #1503:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1503#discussion_r1151881367
##
controller/Controller.h:
##
@@ -18,346 +18,61 @@
#pragma once
#include
+#include
-#include "core/RepositoryFactory.h"
-#include "core/ConfigurationFactory.h"
-#include "core/extension/ExtensionManager.h"
#include "io/ClientSocket.h"
-#include "c2/ControllerSocketProtocol.h"
-#include "utils/gsl.h"
-#include "Exception.h"
-#include "FlowController.h"
-#include "core/repository/VolatileContentRepository.h"
-#include "core/repository/VolatileFlowFileRepository.h"
-#include "core/state/MetricsPublisherFactory.h"
-#include "core/state/MetricsPublisherStore.h"
+
+namespace org::apache::nifi::minifi::controller {
/**
* Sends a single argument comment
* @param socket socket unique ptr.
* @param op operation to perform
* @param value value to send
*/
-bool sendSingleCommand(std::unique_ptr
socket, uint8_t op, const std::string& value) {
- socket->initialize();
- std::vector data;
- org::apache::nifi::minifi::io::BufferStream stream;
- stream.write(, 1);
- stream.write(value);
- return socket->write(stream.getBuffer()) == stream.size();
-}
+bool sendSingleCommand(std::unique_ptr socket, uint8_t op, const
std::string& value);
/**
* Stops a stopped component
* @param socket socket unique ptr.
* @param op operation to perform
*/
-bool stopComponent(std::unique_ptr
socket, const std::string& component) {
- return sendSingleCommand(std::move(socket),
org::apache::nifi::minifi::c2::Operation::STOP, component);
-}
+bool stopComponent(std::unique_ptr socket, const std::string&
component);
/**
* Starts a previously stopped component.
* @param socket socket unique ptr.
* @param op operation to perform
*/
-bool startComponent(std::unique_ptr
socket, const std::string& component) {
- return sendSingleCommand(std::move(socket),
org::apache::nifi::minifi::c2::Operation::START, component);
-}
+bool startComponent(std::unique_ptr socket, const std::string&
component);
/**
* Clears a connection queue.
* @param socket socket unique ptr.
* @param op operation to perform
*/
-bool clearConnection(std::unique_ptr
socket, const std::string& connection) {
- return sendSingleCommand(std::move(socket),
org::apache::nifi::minifi::c2::Operation::CLEAR, connection);
-}
+bool clearConnection(std::unique_ptr socket, const std::string&
connection);
/**
* Updates the flow to the provided file
*/
-int updateFlow(std::unique_ptr socket,
std::ostream , const std::string& file) {
- socket->initialize();
- std::vector data;
- uint8_t op = org::apache::nifi::minifi::c2::Operation::UPDATE;
- org::apache::nifi::minifi::io::BufferStream stream;
- stream.write(, 1);
- stream.write("flow");
- stream.write(file);
- if
(org::apache::nifi::minifi::io::isError(socket->write(stream.getBuffer( {
-return -1;
- }
- // read the response
- uint8_t resp = 0;
- socket->read(resp);
- if (resp == org::apache::nifi::minifi::c2::Operation::DESCRIBE) {
-uint16_t connections = 0;
-socket->read(connections);
-out << connections << " are full" << std::endl;
-for (int i = 0; i < connections; i++) {
- std::string fullcomponent;
- socket->read(fullcomponent);
- out << fullcomponent << " is full" << std::endl;
-}
- }
- return 0;
-}
+int updateFlow(std::unique_ptr socket, std::ostream , const
std::string& file);
/**
* Lists connections which are full
* @param socket socket ptr
*/
-int getFullConnections(std::unique_ptr
socket, std::ostream ) {
- socket->initialize();
- std::vector data;
- uint8_t op = org::apache::nifi::minifi::c2::Operation::DESCRIBE;
- org::apache::nifi::minifi::io::BufferStream stream;
- stream.write(, 1);
- stream.write("getfull");
- if
(org::apache::nifi::minifi::io::isError(socket->write(stream.getBuffer( {
-return -1;
- }
- // read the response
- uint8_t resp = 0;
- socket->read(resp);
- if (resp == org::apache::nifi::minifi::c2::Operation::DESCRIBE) {
-uint16_t connections = 0;
-socket->read(connections);
-out << connections << " are full" << std::endl;
-for (int i = 0; i < connections; i++) {
- std::string fullcomponent;
- socket->read(fullcomponent);
- out << fullcomponent << " is full" << std::endl;
-}
- }
- return 0;
-}
-
-int getJstacks(std::unique_ptr socket,
std::ostream ) {
Review Comment:
I checked and the issue was with the `NoOpThreadedRepsitory` where the
thread was not running (because it does nothing) and it got stuck when waiting
for the thread to continue. I fixed the issue and implemented the jstack
command in aeee1f2be56608ee892e2a23ac93fa37
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@nifi.apache.org
For
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1503: MINIFICPP-2039 Dust off minificontroller
lordgamez commented on code in PR #1503:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1503#discussion_r1151879300
##
controller/MiNiFiController.cpp:
##
@@ -15,57 +15,52 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include
-#include
-#include
-#include
#include
-#include
-#include
#include
-#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 "core/controller/ControllerService.h"
+#include "core/extension/ExtensionManager.h"
+#include "io/StreamFactory.h"
+#include "core/ConfigurationFactory.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");
+std::shared_ptr
getControllerService(const std::shared_ptr ,
+const std::string _name, const std::string& minifi_home) {
+ std::string nifi_configuration_class_name = "yamlconfiguration";
Review Comment:
Good catch, updated in 74a1870e1bc46feb41bc8a88c67b683705aa7d0e
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@nifi.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1503: MINIFICPP-2039 Dust off minificontroller
lordgamez commented on code in PR #1503:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1503#discussion_r1150927120
##
controller/Controller.h:
##
@@ -18,346 +18,61 @@
#pragma once
#include
+#include
-#include "core/RepositoryFactory.h"
-#include "core/ConfigurationFactory.h"
-#include "core/extension/ExtensionManager.h"
#include "io/ClientSocket.h"
-#include "c2/ControllerSocketProtocol.h"
-#include "utils/gsl.h"
-#include "Exception.h"
-#include "FlowController.h"
-#include "core/repository/VolatileContentRepository.h"
-#include "core/repository/VolatileFlowFileRepository.h"
-#include "core/state/MetricsPublisherFactory.h"
-#include "core/state/MetricsPublisherStore.h"
+
+namespace org::apache::nifi::minifi::controller {
/**
* Sends a single argument comment
* @param socket socket unique ptr.
* @param op operation to perform
* @param value value to send
*/
-bool sendSingleCommand(std::unique_ptr
socket, uint8_t op, const std::string& value) {
- socket->initialize();
- std::vector data;
- org::apache::nifi::minifi::io::BufferStream stream;
- stream.write(, 1);
- stream.write(value);
- return socket->write(stream.getBuffer()) == stream.size();
-}
+bool sendSingleCommand(std::unique_ptr socket, uint8_t op, const
std::string& value);
/**
* Stops a stopped component
* @param socket socket unique ptr.
* @param op operation to perform
*/
-bool stopComponent(std::unique_ptr
socket, const std::string& component) {
- return sendSingleCommand(std::move(socket),
org::apache::nifi::minifi::c2::Operation::STOP, component);
-}
+bool stopComponent(std::unique_ptr socket, const std::string&
component);
/**
* Starts a previously stopped component.
* @param socket socket unique ptr.
* @param op operation to perform
*/
-bool startComponent(std::unique_ptr
socket, const std::string& component) {
- return sendSingleCommand(std::move(socket),
org::apache::nifi::minifi::c2::Operation::START, component);
-}
+bool startComponent(std::unique_ptr socket, const std::string&
component);
/**
* Clears a connection queue.
* @param socket socket unique ptr.
* @param op operation to perform
*/
-bool clearConnection(std::unique_ptr
socket, const std::string& connection) {
- return sendSingleCommand(std::move(socket),
org::apache::nifi::minifi::c2::Operation::CLEAR, connection);
-}
+bool clearConnection(std::unique_ptr socket, const std::string&
connection);
/**
* Updates the flow to the provided file
*/
-int updateFlow(std::unique_ptr socket,
std::ostream , const std::string& file) {
- socket->initialize();
- std::vector data;
- uint8_t op = org::apache::nifi::minifi::c2::Operation::UPDATE;
- org::apache::nifi::minifi::io::BufferStream stream;
- stream.write(, 1);
- stream.write("flow");
- stream.write(file);
- if
(org::apache::nifi::minifi::io::isError(socket->write(stream.getBuffer( {
-return -1;
- }
- // read the response
- uint8_t resp = 0;
- socket->read(resp);
- if (resp == org::apache::nifi::minifi::c2::Operation::DESCRIBE) {
-uint16_t connections = 0;
-socket->read(connections);
-out << connections << " are full" << std::endl;
-for (int i = 0; i < connections; i++) {
- std::string fullcomponent;
- socket->read(fullcomponent);
- out << fullcomponent << " is full" << std::endl;
-}
- }
- return 0;
-}
+int updateFlow(std::unique_ptr socket, std::ostream , const
std::string& file);
/**
* Lists connections which are full
* @param socket socket ptr
*/
-int getFullConnections(std::unique_ptr
socket, std::ostream ) {
- socket->initialize();
- std::vector data;
- uint8_t op = org::apache::nifi::minifi::c2::Operation::DESCRIBE;
- org::apache::nifi::minifi::io::BufferStream stream;
- stream.write(, 1);
- stream.write("getfull");
- if
(org::apache::nifi::minifi::io::isError(socket->write(stream.getBuffer( {
-return -1;
- }
- // read the response
- uint8_t resp = 0;
- socket->read(resp);
- if (resp == org::apache::nifi::minifi::c2::Operation::DESCRIBE) {
-uint16_t connections = 0;
-socket->read(connections);
-out << connections << " are full" << std::endl;
-for (int i = 0; i < connections; i++) {
- std::string fullcomponent;
- socket->read(fullcomponent);
- out << fullcomponent << " is full" << std::endl;
-}
- }
- return 0;
-}
-
-int getJstacks(std::unique_ptr socket,
std::ostream ) {
Review Comment:
It was removed as it was not working, interrupted the process, but did not
collect the stacktrace or continue the process so it only freezed the agent. I
could revisit this if needed, to check what caused this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@nifi.apache.org
For queries about this service, please contact Infrastructure at:
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1503: MINIFICPP-2039 Dust off minificontroller
lordgamez commented on code in PR #1503:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1503#discussion_r1150927120
##
controller/Controller.h:
##
@@ -18,346 +18,61 @@
#pragma once
#include
+#include
-#include "core/RepositoryFactory.h"
-#include "core/ConfigurationFactory.h"
-#include "core/extension/ExtensionManager.h"
#include "io/ClientSocket.h"
-#include "c2/ControllerSocketProtocol.h"
-#include "utils/gsl.h"
-#include "Exception.h"
-#include "FlowController.h"
-#include "core/repository/VolatileContentRepository.h"
-#include "core/repository/VolatileFlowFileRepository.h"
-#include "core/state/MetricsPublisherFactory.h"
-#include "core/state/MetricsPublisherStore.h"
+
+namespace org::apache::nifi::minifi::controller {
/**
* Sends a single argument comment
* @param socket socket unique ptr.
* @param op operation to perform
* @param value value to send
*/
-bool sendSingleCommand(std::unique_ptr
socket, uint8_t op, const std::string& value) {
- socket->initialize();
- std::vector data;
- org::apache::nifi::minifi::io::BufferStream stream;
- stream.write(, 1);
- stream.write(value);
- return socket->write(stream.getBuffer()) == stream.size();
-}
+bool sendSingleCommand(std::unique_ptr socket, uint8_t op, const
std::string& value);
/**
* Stops a stopped component
* @param socket socket unique ptr.
* @param op operation to perform
*/
-bool stopComponent(std::unique_ptr
socket, const std::string& component) {
- return sendSingleCommand(std::move(socket),
org::apache::nifi::minifi::c2::Operation::STOP, component);
-}
+bool stopComponent(std::unique_ptr socket, const std::string&
component);
/**
* Starts a previously stopped component.
* @param socket socket unique ptr.
* @param op operation to perform
*/
-bool startComponent(std::unique_ptr
socket, const std::string& component) {
- return sendSingleCommand(std::move(socket),
org::apache::nifi::minifi::c2::Operation::START, component);
-}
+bool startComponent(std::unique_ptr socket, const std::string&
component);
/**
* Clears a connection queue.
* @param socket socket unique ptr.
* @param op operation to perform
*/
-bool clearConnection(std::unique_ptr
socket, const std::string& connection) {
- return sendSingleCommand(std::move(socket),
org::apache::nifi::minifi::c2::Operation::CLEAR, connection);
-}
+bool clearConnection(std::unique_ptr socket, const std::string&
connection);
/**
* Updates the flow to the provided file
*/
-int updateFlow(std::unique_ptr socket,
std::ostream , const std::string& file) {
- socket->initialize();
- std::vector data;
- uint8_t op = org::apache::nifi::minifi::c2::Operation::UPDATE;
- org::apache::nifi::minifi::io::BufferStream stream;
- stream.write(, 1);
- stream.write("flow");
- stream.write(file);
- if
(org::apache::nifi::minifi::io::isError(socket->write(stream.getBuffer( {
-return -1;
- }
- // read the response
- uint8_t resp = 0;
- socket->read(resp);
- if (resp == org::apache::nifi::minifi::c2::Operation::DESCRIBE) {
-uint16_t connections = 0;
-socket->read(connections);
-out << connections << " are full" << std::endl;
-for (int i = 0; i < connections; i++) {
- std::string fullcomponent;
- socket->read(fullcomponent);
- out << fullcomponent << " is full" << std::endl;
-}
- }
- return 0;
-}
+int updateFlow(std::unique_ptr socket, std::ostream , const
std::string& file);
/**
* Lists connections which are full
* @param socket socket ptr
*/
-int getFullConnections(std::unique_ptr
socket, std::ostream ) {
- socket->initialize();
- std::vector data;
- uint8_t op = org::apache::nifi::minifi::c2::Operation::DESCRIBE;
- org::apache::nifi::minifi::io::BufferStream stream;
- stream.write(, 1);
- stream.write("getfull");
- if
(org::apache::nifi::minifi::io::isError(socket->write(stream.getBuffer( {
-return -1;
- }
- // read the response
- uint8_t resp = 0;
- socket->read(resp);
- if (resp == org::apache::nifi::minifi::c2::Operation::DESCRIBE) {
-uint16_t connections = 0;
-socket->read(connections);
-out << connections << " are full" << std::endl;
-for (int i = 0; i < connections; i++) {
- std::string fullcomponent;
- socket->read(fullcomponent);
- out << fullcomponent << " is full" << std::endl;
-}
- }
- return 0;
-}
-
-int getJstacks(std::unique_ptr socket,
std::ostream ) {
Review Comment:
It was removed as it was not working, interrupted the process, but did not
collect the stacktrace or continue the process so it only freezed the agent. On
the other hand this option required C2 agent to be enabled while everything
else was implemented on a separate controller socket without the need for C2 to
be enabled, so I thought it would be better to remove this option altogether.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1503: MINIFICPP-2039 Dust off minificontroller
lordgamez commented on code in PR #1503:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1503#discussion_r1150923563
##
controller/tests/ControllerTests.cpp:
##
@@ -0,0 +1,484 @@
+/**
+ *
+ * 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
+#include
+#include
+#include
+#include
+#include "TestBase.h"
+#include "Catch.h"
+#include "io/ClientSocket.h"
+#include "core/Processor.h"
+#include "Controller.h"
+#include "c2/ControllerSocketProtocol.h"
+#include "utils/IntegrationTestUtils.h"
+#include "c2/ControllerSocketMetricsPublisher.h"
+#include "core/controller/ControllerServiceProvider.h"
+#include "controllers/SSLContextService.h"
+
+#include "state/UpdateController.h"
+
+using namespace std::literals::chrono_literals;
+
+namespace org::apache::nifi::minifi::test {
+
+class TestStateController : public minifi::state::StateController {
+ public:
+ TestStateController()
+: is_running(false) {
+ }
+
+ std::string getComponentName() const override {
+return "TestStateController";
+ }
+
+ minifi::utils::Identifier getComponentUUID() const override {
+static auto dummyUUID =
minifi::utils::Identifier::parse("12345678-1234-1234-1234-123456789abc").value();
+return dummyUUID;
+ }
+
+ int16_t start() override {
+is_running = true;
+return 0;
+ }
+
+ int16_t stop() override {
+is_running = false;
+return 0;
+ }
+
+ bool isRunning() const override {
+return is_running;
+ }
+
+ int16_t pause() override {
+return 0;
+ }
+
+ int16_t resume() override {
+return 0;
+ }
+
+ std::atomic is_running;
+};
+
+class TestUpdateSink : public minifi::state::StateMonitor {
+ public:
+ explicit TestUpdateSink(std::shared_ptr controller)
+: is_running(true),
+ clear_calls(0),
+ controller(std::move(controller)),
+ update_calls(0) {
+ }
+
+ void executeOnComponent(const std::string&,
std::function func) override {
+func(*controller);
+ }
+
+ void
executeOnAllComponents(std::function
func) override {
+func(*controller);
+ }
+
+ std::string getComponentName() const override {
+return "TestUpdateSink";
+ }
+
+ minifi::utils::Identifier getComponentUUID() const override {
+static auto dummyUUID =
minifi::utils::Identifier::parse("12345678-1234-1234-1234-123456789abc").value();
+return dummyUUID;
+ }
+
+ int16_t start() override {
+is_running = true;
+return 0;
+ }
+
+ int16_t stop() override {
+is_running = false;
+return 0;
+ }
+
+ bool isRunning() const override {
+return is_running;
+ }
+
+ int16_t pause() override {
+return 0;
+ }
+
+ int16_t resume() override {
+return 0;
+ }
+ std::vector getTraces() override {
+std::vector traces;
+return traces;
+ }
+
+ int16_t drainRepositories() override {
+return 0;
+ }
+
+ std::map>
getDebugInfo() override {
+return {};
+ }
+
+ int16_t clearConnection(const std::string& /*connection*/) override {
+clear_calls++;
+return 0;
+ }
+
+ std::vector getSupportedConfigurationFormats() const override {
+return {};
+ }
+
+ int16_t applyUpdate(const std::string& /*source*/, const std::string&
/*configuration*/, bool /*persist*/ = false, const std::optional&
/*flow_id*/ = std::nullopt) override {
+update_calls++;
+return 0;
+ }
+
+ int16_t applyUpdate(const std::string& /*source*/, const
std::shared_ptr& /*updateController*/) override {
+update_calls++;
+return 0;
+ }
+
+ uint64_t getUptime() override {
+return 8765309;
+ }
+
+ std::atomic is_running;
+ std::atomic clear_calls;
+ std::shared_ptr controller;
+ std::atomic update_calls;
+};
+
+class TestControllerSocketReporter : public c2::ControllerSocketReporter {
+ std::unordered_map
getQueueSizes() override {
+return {
+ {"con1", {1, 2}},
+ {"con2", {3, 3}}
+};
+ }
+
+ std::unordered_set getFullConnections() override {
+return {"con2"};
+ }
+
+ std::unordered_set getConnections() override {
+return {"con1", "con2"};
+ }
+
+ std::string getAgentManifest() override {
+return "testAgentManifest";
+ }
+};
+
+class TestControllerServiceProvider : public
core::controller::ControllerServiceProvider {
+ public:
+
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1503: MINIFICPP-2039 Dust off minificontroller
lordgamez commented on code in PR #1503:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1503#discussion_r1150924463
##
libminifi/src/c2/ControllerSocketMetricsPublisher.cpp:
##
@@ -0,0 +1,96 @@
+/**
+ * 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 "c2/ControllerSocketMetricsPublisher.h"
+#include "utils/gsl.h"
+#include "core/Resource.h"
+#include "c2/C2Agent.h"
+
+namespace org::apache::nifi::minifi::c2 {
+
+std::unordered_map
ControllerSocketMetricsPublisher::getQueueSizes() {
+ std::lock_guard guard(queue_metrics_node_mutex_);
+ std::unordered_map sizes;
+ if (!queue_metrics_node_) {
+return sizes;
+ }
+ for (const auto& metric : queue_metrics_node_->calculateMetrics()) {
+gsl_Expects(metric.labels.contains("connection_name"));
+if (metric.name == "queue_size") {
+ sizes[metric.labels.at("connection_name")].queue_size =
static_cast(metric.value);
+} else if (metric.name == "queue_size_max") {
+ sizes[metric.labels.at("connection_name")].queue_size_max =
static_cast(metric.value);
+}
+ }
+ return sizes;
+}
+
+std::unordered_set
ControllerSocketMetricsPublisher::getFullConnections() {
+ std::lock_guard guard(queue_metrics_node_mutex_);
+ std::unordered_set full_connections;
+ if (!queue_metrics_node_) {
+return full_connections;
+ }
Review Comment:
You are right it's not needed here, updated in
be90c45f9f3fe3d065a8470e5b86514d3dfc6039
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@nifi.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1503: MINIFICPP-2039 Dust off minificontroller
lordgamez commented on code in PR #1503:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1503#discussion_r1150922841
##
controller/tests/ControllerTests.cpp:
##
@@ -0,0 +1,484 @@
+/**
+ *
+ * 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
+#include
+#include
+#include
+#include
+#include "TestBase.h"
+#include "Catch.h"
+#include "io/ClientSocket.h"
+#include "core/Processor.h"
+#include "Controller.h"
+#include "c2/ControllerSocketProtocol.h"
+#include "utils/IntegrationTestUtils.h"
+#include "c2/ControllerSocketMetricsPublisher.h"
+#include "core/controller/ControllerServiceProvider.h"
+#include "controllers/SSLContextService.h"
+
+#include "state/UpdateController.h"
+
+using namespace std::literals::chrono_literals;
+
+namespace org::apache::nifi::minifi::test {
+
+class TestStateController : public minifi::state::StateController {
+ public:
+ TestStateController()
+: is_running(false) {
+ }
+
+ std::string getComponentName() const override {
+return "TestStateController";
+ }
+
+ minifi::utils::Identifier getComponentUUID() const override {
+static auto dummyUUID =
minifi::utils::Identifier::parse("12345678-1234-1234-1234-123456789abc").value();
+return dummyUUID;
+ }
+
+ int16_t start() override {
+is_running = true;
+return 0;
+ }
+
+ int16_t stop() override {
+is_running = false;
+return 0;
+ }
+
+ bool isRunning() const override {
+return is_running;
+ }
+
+ int16_t pause() override {
+return 0;
+ }
+
+ int16_t resume() override {
+return 0;
+ }
+
+ std::atomic is_running;
+};
+
+class TestUpdateSink : public minifi::state::StateMonitor {
+ public:
+ explicit TestUpdateSink(std::shared_ptr controller)
+: is_running(true),
+ clear_calls(0),
+ controller(std::move(controller)),
+ update_calls(0) {
+ }
+
+ void executeOnComponent(const std::string&,
std::function func) override {
+func(*controller);
+ }
+
+ void
executeOnAllComponents(std::function
func) override {
+func(*controller);
+ }
+
+ std::string getComponentName() const override {
+return "TestUpdateSink";
+ }
+
+ minifi::utils::Identifier getComponentUUID() const override {
+static auto dummyUUID =
minifi::utils::Identifier::parse("12345678-1234-1234-1234-123456789abc").value();
+return dummyUUID;
+ }
+
+ int16_t start() override {
+is_running = true;
+return 0;
+ }
+
+ int16_t stop() override {
+is_running = false;
+return 0;
+ }
+
+ bool isRunning() const override {
+return is_running;
+ }
+
+ int16_t pause() override {
+return 0;
+ }
+
+ int16_t resume() override {
+return 0;
+ }
+ std::vector getTraces() override {
+std::vector traces;
+return traces;
+ }
+
+ int16_t drainRepositories() override {
+return 0;
+ }
+
+ std::map>
getDebugInfo() override {
+return {};
+ }
+
+ int16_t clearConnection(const std::string& /*connection*/) override {
+clear_calls++;
+return 0;
+ }
+
+ std::vector getSupportedConfigurationFormats() const override {
+return {};
+ }
+
+ int16_t applyUpdate(const std::string& /*source*/, const std::string&
/*configuration*/, bool /*persist*/ = false, const std::optional&
/*flow_id*/ = std::nullopt) override {
+update_calls++;
+return 0;
+ }
+
+ int16_t applyUpdate(const std::string& /*source*/, const
std::shared_ptr& /*updateController*/) override {
+update_calls++;
+return 0;
+ }
+
+ uint64_t getUptime() override {
+return 8765309;
+ }
+
+ std::atomic is_running;
+ std::atomic clear_calls;
+ std::shared_ptr controller;
+ std::atomic update_calls;
+};
+
+class TestControllerSocketReporter : public c2::ControllerSocketReporter {
+ std::unordered_map
getQueueSizes() override {
+return {
+ {"con1", {1, 2}},
+ {"con2", {3, 3}}
+};
+ }
+
+ std::unordered_set getFullConnections() override {
+return {"con2"};
+ }
+
+ std::unordered_set getConnections() override {
+return {"con1", "con2"};
+ }
+
+ std::string getAgentManifest() override {
+return "testAgentManifest";
+ }
+};
+
+class TestControllerServiceProvider : public
core::controller::ControllerServiceProvider {
+ public:
+
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1503: MINIFICPP-2039 Dust off minificontroller
lordgamez commented on code in PR #1503:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1503#discussion_r1150922408
##
controller/MiNiFiController.cpp:
##
@@ -15,57 +15,52 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include
-#include
-#include
-#include
#include
-#include
-#include
#include
-#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 "core/controller/ControllerService.h"
+#include "core/extension/ExtensionManager.h"
+#include "io/StreamFactory.h"
+#include "core/ConfigurationFactory.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");
+std::shared_ptr
getControllerService(const std::shared_ptr ,
+const std::string _name, const std::string& minifi_home) {
+ std::string nifi_configuration_class_name = "yamlconfiguration";
- const std::string minifiHome = determineMinifiHome(logger);
- if (minifiHome.empty()) {
-// determineMinifiHome already logged everything we need
-return -1;
- }
+ minifi::core::extension::ExtensionManager::get().initialize(configuration);
- const auto configuration = std::make_shared();
- configuration->setHome(minifiHome);
- configuration->loadConfigureFile(DEFAULT_NIFI_PROPERTIES_FILE);
+ configuration->get(minifi::Configure::nifi_configuration_class_name,
nifi_configuration_class_name);
+ const auto stream_factory =
minifi::io::StreamFactory::getInstance(configuration);
+ auto flow_configuration = minifi::core::createFlowConfiguration(
+minifi::core::ConfigurationContext{nullptr, nullptr, stream_factory,
configuration, std::filesystem::path(minifi_home) / "conf" / "config.yml"},
nifi_configuration_class_name);
- const auto log_properties =
std::make_shared();
- log_properties->setHome(minifiHome);
- log_properties->loadConfigureFile(DEFAULT_LOG_PROPERTIES_FILE);
-
minifi::core::logging::LoggerConfiguration::getConfiguration().initialize(log_properties);
+ auto root = flow_configuration->getRoot();
+ if (!root) {
+return nullptr;
+ }
+ auto controller = root->findControllerService(service_name);
+ if (!controller) {
+return nullptr;
+ }
+ return controller->getControllerServiceImplementation();
+}
+std::shared_ptr
getSSLContextService(const std::shared_ptr& configuration,
const std::string& minifi_home) {
+ std::shared_ptr secure_context;
std::string context_name;
-
- std::shared_ptr 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 (configuration->get(minifi::Configure::controller_ssl_context_service,
context_name)) {
+const auto service = getControllerService(configuration, context_name,
minifi_home);
if (nullptr != service) {
- secure_context =
std::static_pointer_cast(service);
+ secure_context =
std::dynamic_pointer_cast(service);
Review Comment:
You are right, it's better that way updated in
a582bd90b8f188e1cc20bc2e9e81d358250cb230
##
controller/MiNiFiController.cpp:
##
@@ -15,57 +15,52 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include
-#include
-#include
-#include
#include
-#include
-#include
#include
-#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 "core/controller/ControllerService.h"
+#include "core/extension/ExtensionManager.h"
+#include "io/StreamFactory.h"
+#include "core/ConfigurationFactory.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");
+std::shared_ptr
getControllerService(const std::shared_ptr ,
+const std::string _name, const std::string& minifi_home) {
+ std::string nifi_configuration_class_name = "yamlconfiguration";
- const std::string minifiHome = determineMinifiHome(logger);
- if (minifiHome.empty()) {
-// determineMinifiHome already logged everything we need
-return -1;
- }
+ minifi::core::extension::ExtensionManager::get().initialize(configuration);
-
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1503: MINIFICPP-2039 Dust off minificontroller
lordgamez commented on code in PR #1503:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1503#discussion_r1150921974
##
controller/MiNiFiController.cpp:
##
@@ -15,57 +15,52 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include
-#include
-#include
-#include
#include
-#include
-#include
#include
-#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 "core/controller/ControllerService.h"
+#include "core/extension/ExtensionManager.h"
+#include "io/StreamFactory.h"
+#include "core/ConfigurationFactory.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");
+std::shared_ptr
getControllerService(const std::shared_ptr ,
+const std::string _name, const std::string& minifi_home) {
+ std::string nifi_configuration_class_name = "yamlconfiguration";
Review Comment:
Good point, did not change it after rebasing, updated in
a582bd90b8f188e1cc20bc2e9e81d358250cb230
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@nifi.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1503: MINIFICPP-2039 Dust off minificontroller
lordgamez commented on code in PR #1503:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1503#discussion_r1108569955
##
controller/tests/ControllerTests.cpp:
##
@@ -0,0 +1,479 @@
+/**
+ *
+ * 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
+#include
+#include
+#include
+#include
+#include "TestBase.h"
+#include "Catch.h"
+#include "io/ClientSocket.h"
+#include "core/Processor.h"
+#include "Controller.h"
+#include "c2/ControllerSocketProtocol.h"
+#include "utils/IntegrationTestUtils.h"
+#include "c2/ControllerSocketMetricsPublisher.h"
+#include "core/controller/ControllerServiceProvider.h"
+#include "controllers/SSLContextService.h"
+
+#include "state/UpdateController.h"
+
+using namespace std::literals::chrono_literals;
+
+namespace org::apache::nifi::minifi::test {
+
+class TestStateController : public minifi::state::StateController {
+ public:
+ TestStateController()
+: is_running(false) {
+ }
+
+ std::string getComponentName() const override {
+return "TestStateController";
+ }
+
+ minifi::utils::Identifier getComponentUUID() const override {
+static auto dummyUUID =
minifi::utils::Identifier::parse("12345678-1234-1234-1234-123456789abc").value();
+return dummyUUID;
+ }
+
+ int16_t start() override {
+is_running = true;
+return 0;
+ }
+
+ int16_t stop() override {
+is_running = false;
+return 0;
+ }
+
+ bool isRunning() const override {
+return is_running;
+ }
+
+ int16_t pause() override {
+return 0;
+ }
+
+ int16_t resume() override {
+return 0;
+ }
+
+ std::atomic is_running;
+};
+
+class TestUpdateSink : public minifi::state::StateMonitor {
+ public:
+ explicit TestUpdateSink(std::shared_ptr controller)
+: is_running(true),
+ clear_calls(0),
+ controller(std::move(controller)),
+ update_calls(0) {
+ }
+
+ void executeOnComponent(const std::string&,
std::function func) override {
+func(*controller);
+ }
+
+ void
executeOnAllComponents(std::function
func) override {
+func(*controller);
+ }
+
+ std::string getComponentName() const override {
+return "TestUpdateSink";
+ }
+
+ minifi::utils::Identifier getComponentUUID() const override {
+static auto dummyUUID =
minifi::utils::Identifier::parse("12345678-1234-1234-1234-123456789abc").value();
+return dummyUUID;
+ }
+
+ int16_t start() override {
+is_running = true;
+return 0;
+ }
+
+ int16_t stop() override {
+is_running = false;
+return 0;
+ }
+
+ bool isRunning() const override {
+return is_running;
+ }
+
+ int16_t pause() override {
+return 0;
+ }
+
+ int16_t resume() override {
+return 0;
+ }
+ std::vector getTraces() override {
+std::vector traces;
+return traces;
+ }
+
+ int16_t drainRepositories() override {
+return 0;
+ }
+
+ std::map>
getDebugInfo() override {
+return {};
+ }
+
+ int16_t clearConnection(const std::string& /*connection*/) override {
+clear_calls++;
+return 0;
+ }
+
+ int16_t applyUpdate(const std::string& /*source*/, const std::string&
/*configuration*/, bool /*persist*/ = false, const std::optional&
/*flow_id*/ = std::nullopt) override {
+update_calls++;
+return 0;
+ }
+
+ int16_t applyUpdate(const std::string& /*source*/, const
std::shared_ptr& /*updateController*/) override {
+return 0;
Review Comment:
I think we should, I suppose this overload was unused in the tests, so the
stub was not implemented for it, updated in
2a82c349ac8f334e2a3f49640f136b26e0b7ce23
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@nifi.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
