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 <memory>
+#include <string>
-#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<org::apache::nifi::minifi::io::Socket>
socket, uint8_t op, const std::string& value) {
- socket->initialize();
- std::vector<uint8_t> data;
- org::apache::nifi::minifi::io::BufferStream stream;
- stream.write(&op, 1);
- stream.write(value);
- return socket->write(stream.getBuffer()) == stream.size();
-}
+bool sendSingleCommand(std::unique_ptr<io::Socket> 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<org::apache::nifi::minifi::io::Socket>
socket, const std::string& component) {
- return sendSingleCommand(std::move(socket),
org::apache::nifi::minifi::c2::Operation::STOP, component);
-}
+bool stopComponent(std::unique_ptr<io::Socket> 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<org::apache::nifi::minifi::io::Socket>
socket, const std::string& component) {
- return sendSingleCommand(std::move(socket),
org::apache::nifi::minifi::c2::Operation::START, component);
-}
+bool startComponent(std::unique_ptr<io::Socket> socket, const std::string&
component);
/**
* Clears a connection queue.
* @param socket socket unique ptr.
* @param op operation to perform
*/
-bool clearConnection(std::unique_ptr<org::apache::nifi::minifi::io::Socket>
socket, const std::string& connection) {
- return sendSingleCommand(std::move(socket),
org::apache::nifi::minifi::c2::Operation::CLEAR, connection);
-}
+bool clearConnection(std::unique_ptr<io::Socket> socket, const std::string&
connection);
/**
* Updates the flow to the provided file
*/
-int updateFlow(std::unique_ptr<org::apache::nifi::minifi::io::Socket> socket,
std::ostream &out, const std::string& file) {
- socket->initialize();
- std::vector<uint8_t> data;
- uint8_t op = org::apache::nifi::minifi::c2::Operation::UPDATE;
- org::apache::nifi::minifi::io::BufferStream stream;
- stream.write(&op, 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<io::Socket> socket, std::ostream &out, const
std::string& file);
/**
* Lists connections which are full
* @param socket socket ptr
*/
-int getFullConnections(std::unique_ptr<org::apache::nifi::minifi::io::Socket>
socket, std::ostream &out) {
- socket->initialize();
- std::vector<uint8_t> data;
- uint8_t op = org::apache::nifi::minifi::c2::Operation::DESCRIBE;
- org::apache::nifi::minifi::io::BufferStream stream;
- stream.write(&op, 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<org::apache::nifi::minifi::io::Socket> socket,
std::ostream &out) {
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 <fcntl.h>
-#include <cstdio>
-#include <semaphore.h>
-#include <csignal>
#include <vector>
-#include <queue>
-#include <map>
#include <iostream>
-#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<minifi::core::controller::ControllerService>
getControllerService(const std::shared_ptr<minifi::Configure> &configuration,
+ const std::string &service_name, const std::string& minifi_home) {
+ std::string nifi_configuration_class_name = "yamlconfiguration";
Review Comment:
Although I think it should not be used in this case, it may be better to
stick with the default value, updated in
8735105d4cd70f6bbe33fbe43bd17760382a9b1d
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
