Line data Source code
1 : /**
2 : * @file RestCommandFacility.cpp
3 : *
4 : * This is part of the DUNE DAQ Application Framework, copyright 2020.
5 : * Licensing/copyright details are in the COPYING file that you should have
6 : * received with this code.
7 : */
8 : #include "restcmd/RestEndpoint.hpp"
9 :
10 : #include "cmdlib/CommandFacility.hpp"
11 : #include "cmdlib/Issues.hpp"
12 :
13 : #include "confmodel/ConnectivityService.hpp"
14 : #include "iomanager/network/ConfigClient.hpp"
15 : #include "iomanager/network/ConfigClientStructs.hpp"
16 : #include "utilities/get_ips.hpp"
17 :
18 : #include "logging/Logging.hpp"
19 : #include <cetlib/BasicPluginFactory.h>
20 : // #include <tbb/concurrent_queue.h>
21 : #include <folly/Uri.h>
22 :
23 : #include <chrono>
24 : #include <climits>
25 : #include <fstream>
26 : #include <functional>
27 : #include <map>
28 : #include <memory>
29 : #include <string>
30 : #include <unistd.h>
31 :
32 : using namespace dunedaq::cmdlib;
33 : using namespace dunedaq::restcmd;
34 :
35 : class restCommandFacility : public CommandFacility
36 : {
37 :
38 : public:
39 : // friend backend for calling functions on CF
40 : friend class RestEndpoint;
41 :
42 0 : explicit restCommandFacility(std::string uri,
43 : std::string session_name,
44 : const dunedaq::confmodel::ConnectivityService* connectivity_service)
45 0 : : CommandFacility(uri)
46 0 : , m_session_name(session_name)
47 : {
48 :
49 0 : folly::Uri furi(uri);
50 :
51 0 : std::string hostname = furi.hostname();
52 0 : int port = furi.port();
53 :
54 0 : if (connectivity_service != nullptr) {
55 0 : auto connectivity_service_port = std::to_string(connectivity_service->get_service()->get_port());
56 0 : m_connectivity_client = std::make_unique<dunedaq::iomanager::ConfigClient>(
57 0 : connectivity_service->get_host(),
58 : connectivity_service_port,
59 : m_session_name,
60 0 : std::chrono::milliseconds(connectivity_service->get_interval_ms()));
61 0 : }
62 :
63 0 : if (port == 0 && connectivity_service == nullptr) {
64 0 : throw dunedaq::cmdlib::MalformedUri(
65 0 : ERS_HERE, "Can't bind the REST API to port 0 without connectivity service", std::to_string(port));
66 : }
67 :
68 0 : try { // to setup backend
69 0 : command_executor_ = std::bind(&inherited::execute_command, this, std::placeholders::_1, std::placeholders::_2);
70 0 : rest_endpoint_ = std::make_unique<dunedaq::restcmd::RestEndpoint>(hostname, port, command_executor_);
71 0 : rest_endpoint_->init(1); // 1 thread
72 0 : TLOG() << std::format("Endpoint open on host: {} port: {}", hostname, port);
73 :
74 0 : } catch (const std::exception& ex) {
75 0 : ers::error(dunedaq::cmdlib::CommandFacilityInitialization(ERS_HERE, ex.what()));
76 0 : }
77 :
78 : // Store hostname for connectivity service registration
79 0 : m_hostname = hostname;
80 0 : }
81 :
82 0 : void run(std::atomic<bool>& end_marker)
83 : {
84 :
85 0 : char* app_name_c = std::getenv("DUNEDAQ_APPLICATION_NAME");
86 0 : if (!app_name_c)
87 0 : throw dunedaq::restcmd::EnvVarNotFound(ERS_HERE, "DUNEDAQ_APPLICATION_NAME");
88 0 : std::string app_name = std::string(app_name_c);
89 :
90 : // Start endpoint
91 0 : try {
92 0 : rest_endpoint_->start();
93 :
94 0 : if (m_connectivity_client) {
95 0 : int port = rest_endpoint_->getPort();
96 :
97 0 : auto ips = dunedaq::utilities::get_hostname_ips(m_hostname);
98 :
99 0 : if (ips.size() == 0)
100 0 : throw dunedaq::cmdlib::CommandFacilityInitialization(ERS_HERE, "Could not resolve hostname to IP address");
101 :
102 0 : TLOG() << "Registering the control endpoint (" << app_name
103 0 : << "_control) on the connectivity service: " << ips[0] << ":" << port;
104 0 : dunedaq::iomanager::ConnectionRegistration cr;
105 0 : cr.uid = app_name + "_control";
106 0 : cr.data_type = "RunControlMessage";
107 0 : cr.uri = "rest://" + ips[0] + ":" + std::to_string(port);
108 0 : cr.connection_type = dunedaq::iomanager::ConnectionType::kSendRecv;
109 0 : m_connectivity_client->publish(cr);
110 0 : }
111 0 : } catch (const std::exception& ex) {
112 0 : ers::fatal(dunedaq::cmdlib::RunLoopTerminated(ERS_HERE, ex.what()));
113 0 : }
114 :
115 : // Wait until marked
116 0 : while (end_marker) {
117 0 : std::this_thread::sleep_for(std::chrono::seconds(1));
118 : }
119 :
120 : // Shutdown
121 0 : rest_endpoint_->shutdown();
122 0 : if (m_connectivity_client) {
123 0 : dunedaq::iomanager::ConnectionId ci{ app_name + "_control", "RunControlMessage", m_session_name };
124 0 : m_connectivity_client->retract(ci);
125 0 : }
126 0 : }
127 :
128 : protected:
129 : typedef CommandFacility inherited;
130 :
131 : // Implementation of completionHandler interface
132 0 : void completion_callback(const cmdobj_t& cmd, cmd::CommandReply& meta)
133 : {
134 0 : rest_endpoint_->handleResponseCommand(cmd, meta);
135 0 : }
136 :
137 : private:
138 : // Manager, HTTP REST Endpoint and backend resources
139 : mutable std::unique_ptr<RestEndpoint> rest_endpoint_;
140 :
141 : typedef std::function<void(const cmdobj_t&, cmd::CommandReply)> RequestCallback;
142 : RequestCallback command_executor_;
143 :
144 : std::string m_session_name;
145 : std::string m_hostname;
146 : std::unique_ptr<dunedaq::iomanager::ConfigClient> m_connectivity_client;
147 : };
148 :
149 : // NOLINTBEGIN
150 : extern "C"
151 : {
152 : std::shared_ptr<dunedaq::cmdlib::CommandFacility>
153 0 : make(std::string uri, std::string session_name, const dunedaq::confmodel::ConnectivityService* connectivity_service)
154 : {
155 :
156 0 : return std::shared_ptr<dunedaq::cmdlib::CommandFacility>(
157 0 : new restCommandFacility(uri, session_name, connectivity_service));
158 : }
159 : }
160 : // NOLINTEND
|
