From b0f2bade7a34aaff6573c81d9875d321dd889370 Mon Sep 17 00:00:00 2001 From: "Matthias P. Braendli" Date: Fri, 7 Oct 2016 16:00:38 +0200 Subject: Rework remotecontrol --- src/RemoteControl.cpp | 369 +++++++++++++++++++++++++++++++++++--------------- 1 file changed, 258 insertions(+), 111 deletions(-) (limited to 'src/RemoteControl.cpp') diff --git a/src/RemoteControl.cpp b/src/RemoteControl.cpp index 9ecb018..bca0b41 100644 --- a/src/RemoteControl.cpp +++ b/src/RemoteControl.cpp @@ -27,145 +27,131 @@ #include #include #include -#include #include -#include "Log.h" #include "RemoteControl.h" using boost::asio::ip::tcp; using namespace std; -RemoteControllerTelnet::~RemoteControllerTelnet() -{ - m_running = false; - m_io_service.stop(); - m_child_thread.join(); -} +RemoteControllers rcs; void RemoteControllerTelnet::restart() { - m_restarter_thread = boost::thread(&RemoteControllerTelnet::restart_thread, + m_restarter_thread = boost::thread( + &RemoteControllerTelnet::restart_thread, this, 0); } +RemoteControllable::~RemoteControllable() { + rcs.remove_controllable(this); +} + +std::list RemoteControllable::get_supported_parameters() const { + std::list parameterlist; + for (const auto& param : m_parameters) { + parameterlist.push_back(param[0]); + } + return parameterlist; +} + +RemoteControllable* RemoteControllers::get_controllable_(const std::string& name) { + auto rc = std::find_if(controllables.begin(), controllables.end(), + [&](RemoteControllable* r) { return r->get_rc_name() == name; }); + + if (rc == controllables.end()) { + throw ParameterError("Module name unknown"); + } + else { + return *rc; + } +} + // This runs in a separate thread, because // it would take too long to be done in the main loop // thread. void RemoteControllerTelnet::restart_thread(long) { m_running = false; - m_io_service.stop(); - m_child_thread.join(); + if (m_port) { + m_child_thread.interrupt(); + m_child_thread.join(); + } m_child_thread = boost::thread(&RemoteControllerTelnet::process, this, 0); } -void RemoteControllerTelnet::handle_accept( - const boost::system::error_code& boost_error, - boost::shared_ptr< boost::asio::ip::tcp::socket > socket, - boost::asio::ip::tcp::acceptor& acceptor) +void RemoteControllerTelnet::process(long) { - - const std::string welcome = "ODR-DabMux Remote Control CLI\n" - "Write 'help' for help.\n" - "**********\n"; - const std::string prompt = "> "; + std::string m_welcome = "ODR-DabMux Remote Control CLI\n" + "Write 'help' for help.\n" + "**********\n"; + std::string m_prompt = "> "; std::string in_message; size_t length; - if (boost_error) - { - etiLog.level(error) << "RC: Error accepting connection"; - return; - } - try { - etiLog.level(info) << "RC: Accepted"; - - boost::system::error_code ignored_error; - - boost::asio::write(*socket, boost::asio::buffer(welcome), - boost::asio::transfer_all(), - ignored_error); - - while (m_running && in_message != "quit") { - boost::asio::write(*socket, boost::asio::buffer(prompt), - boost::asio::transfer_all(), - ignored_error); + boost::asio::io_service io_service; + tcp::acceptor acceptor(io_service, tcp::endpoint( + boost::asio::ip::address::from_string("127.0.0.1"), m_port) ); + while (m_running) { in_message = ""; - boost::asio::streambuf buffer; - length = boost::asio::read_until(*socket, buffer, "\n", ignored_error); + tcp::socket socket(io_service); - std::istream str(&buffer); - std::getline(str, in_message); + acceptor.accept(socket); - if (length == 0) { - etiLog.level(info) << "RC: Connection terminated"; - break; - } + boost::system::error_code ignored_error; - while (in_message.length() > 0 && - (in_message[in_message.length()-1] == '\r' || - in_message[in_message.length()-1] == '\n')) { - in_message.erase(in_message.length()-1, 1); - } + boost::asio::write(socket, boost::asio::buffer(m_welcome), + boost::asio::transfer_all(), + ignored_error); - if (in_message.length() == 0) { - continue; - } + while (m_running && in_message != "quit") { + boost::asio::write(socket, boost::asio::buffer(m_prompt), + boost::asio::transfer_all(), + ignored_error); - etiLog.level(info) << "RC: Got message '" << in_message << "'"; + in_message = ""; - dispatch_command(*socket, in_message); - } - etiLog.level(info) << "RC: Closing socket"; - socket->close(); - } - catch (std::exception& e) - { - etiLog.level(error) << "Remote control caught exception: " << e.what(); - } -} + boost::asio::streambuf buffer; + length = boost::asio::read_until( socket, buffer, "\n", ignored_error); -void RemoteControllerTelnet::process(long) -{ - m_running = true; - - while (m_running) { - m_io_service.reset(); + std::istream str(&buffer); + std::getline(str, in_message); - tcp::acceptor acceptor(m_io_service, tcp::endpoint( - boost::asio::ip::address::from_string("127.0.0.1"), m_port) ); + if (length == 0) { + std::cerr << "RC: Connection terminated" << std::endl; + break; + } + while (in_message.length() > 0 && + (in_message[in_message.length()-1] == '\r' || + in_message[in_message.length()-1] == '\n')) { + in_message.erase(in_message.length()-1, 1); + } - // Add a job to start accepting connections. - boost::shared_ptr socket( - new tcp::socket(acceptor.get_io_service())); + if (in_message.length() == 0) { + continue; + } - // Add an accept call to the service. This will prevent io_service::run() - // from returning. - etiLog.level(info) << "RC: Waiting for connection on port " << m_port; - acceptor.async_accept(*socket, - boost::bind(&RemoteControllerTelnet::handle_accept, - this, - boost::asio::placeholders::error, - socket, - boost::ref(acceptor))); + std::cerr << "RC: Got message '" << in_message << "'" << std::endl; - // Process event loop. - m_io_service.run(); + dispatch_command(socket, in_message); + } + std::cerr << "RC: Closing socket" << std::endl; + socket.close(); + } + } + catch (std::exception& e) { + std::cerr << "Remote control caught exception: " << e.what() << std::endl; + m_fault = true; } - - etiLog.level(info) << "RC: Leaving"; - m_fault = true; } - void RemoteControllerTelnet::dispatch_command(tcp::socket& socket, string command) { vector cmd = tokenise_(command); @@ -189,16 +175,12 @@ void RemoteControllerTelnet::dispatch_command(tcp::socket& socket, string comman stringstream ss; if (cmd.size() == 1) { - for (list::iterator it = m_cohort.begin(); - it != m_cohort.end(); ++it) { - ss << (*it)->get_rc_name() << endl; - - list< vector >::iterator param; - list< vector > params = (*it)->get_parameter_descriptions(); - for (param = params.begin(); - param != params.end(); - ++param) { - ss << "\t" << (*param)[0] << " : " << (*param)[1] << endl; + for (auto &controllable : rcs.controllables) { + ss << controllable->get_rc_name() << endl; + + list< vector > params = controllable->get_parameter_descriptions(); + for (auto ¶m : params) { + ss << "\t" << param[0] << " : " << param[1] << endl; } } } @@ -212,10 +194,9 @@ void RemoteControllerTelnet::dispatch_command(tcp::socket& socket, string comman if (cmd.size() == 2) { try { stringstream ss; - list< vector > r = get_param_list_values_(cmd[1]); - for (list< vector >::iterator it = r.begin(); - it != r.end(); ++it) { - ss << (*it)[0] << ": " << (*it)[1] << endl; + list< vector > r = rcs.get_param_list_values(cmd[1]); + for (auto ¶m_val : r) { + ss << param_val[0] << ": " << param_val[1] << endl; } reply(socket, ss.str()); @@ -224,23 +205,21 @@ void RemoteControllerTelnet::dispatch_command(tcp::socket& socket, string comman reply(socket, e.what()); } } - else - { + else { reply(socket, "Incorrect parameters for command 'show'"); } } else if (cmd[0] == "get") { if (cmd.size() == 3) { try { - string r = get_param_(cmd[1], cmd[2]); + string r = rcs.get_param(cmd[1], cmd[2]); reply(socket, r); } catch (ParameterError &e) { reply(socket, e.what()); } } - else - { + else { reply(socket, "Incorrect parameters for command 'get'"); } } @@ -256,7 +235,7 @@ void RemoteControllerTelnet::dispatch_command(tcp::socket& socket, string comman } } - set_param_(cmd[1], cmd[2], new_param_value.str()); + rcs.set_param(cmd[1], cmd[2], new_param_value.str()); reply(socket, "ok"); } catch (ParameterError &e) { @@ -288,3 +267,171 @@ void RemoteControllerTelnet::reply(tcp::socket& socket, string message) ignored_error); } + +#if 0 // #if defined(HAVE_ZEROMQ) + +void RemoteControllerZmq::restart() +{ + m_restarter_thread = boost::thread(&RemoteControllerZmq::restart_thread, this); +} + +// This runs in a separate thread, because +// it would take too long to be done in the main loop +// thread. +void RemoteControllerZmq::restart_thread() +{ + m_running = false; + + if (!m_endpoint.empty()) { + m_child_thread.interrupt(); + m_child_thread.join(); + } + + m_child_thread = boost::thread(&RemoteControllerZmq::process, this); +} + +void RemoteControllerZmq::recv_all(zmq::socket_t& pSocket, std::vector &message) +{ + bool more = true; + do { + zmq::message_t msg; + pSocket.recv(&msg); + std::string incoming((char*)msg.data(), msg.size()); + message.push_back(incoming); + more = msg.more(); + } while (more); +} + +void RemoteControllerZmq::send_ok_reply(zmq::socket_t &pSocket) +{ + zmq::message_t msg(2); + char repCode[2] = {'o', 'k'}; + memcpy ((void*) msg.data(), repCode, 2); + pSocket.send(msg, 0); +} + +void RemoteControllerZmq::send_fail_reply(zmq::socket_t &pSocket, const std::string &error) +{ + zmq::message_t msg1(4); + char repCode[4] = {'f', 'a', 'i', 'l'}; + memcpy ((void*) msg1.data(), repCode, 4); + pSocket.send(msg1, ZMQ_SNDMORE); + + zmq::message_t msg2(error.length()); + memcpy ((void*) msg2.data(), error.c_str(), error.length()); + pSocket.send(msg2, 0); +} + +void RemoteControllerZmq::process() +{ + // create zmq reply socket for receiving ctrl parameters + etiLog.level(info) << "Starting zmq remote control thread"; + try { + zmq::socket_t repSocket(m_zmqContext, ZMQ_REP); + + // connect the socket + int hwm = 100; + int linger = 0; + repSocket.setsockopt(ZMQ_RCVHWM, &hwm, sizeof(hwm)); + repSocket.setsockopt(ZMQ_SNDHWM, &hwm, sizeof(hwm)); + repSocket.setsockopt(ZMQ_LINGER, &linger, sizeof(linger)); + repSocket.bind(m_endpoint.c_str()); + + // create pollitem that polls the ZMQ sockets + zmq::pollitem_t pollItems[] = { {repSocket, 0, ZMQ_POLLIN, 0} }; + for (;;) { + zmq::poll(pollItems, 1, 100); + std::vector msg; + + if (pollItems[0].revents & ZMQ_POLLIN) { + recv_all(repSocket, msg); + + std::string command((char*)msg[0].data(), msg[0].size()); + + if (msg.size() == 1 && command == "ping") { + send_ok_reply(repSocket); + } + else if (msg.size() == 1 && command == "list") { + size_t cohort_size = m_cohort.size(); + for (auto &controllable : m_cohort) { + std::stringstream ss; + ss << controllable->get_rc_name(); + + std::string msg_s = ss.str(); + + zmq::message_t msg(ss.str().size()); + memcpy ((void*) msg.data(), msg_s.data(), msg_s.size()); + + int flag = (--cohort_size > 0) ? ZMQ_SNDMORE : 0; + repSocket.send(msg, flag); + } + } + else if (msg.size() == 2 && command == "show") { + std::string module((char*) msg[1].data(), msg[1].size()); + try { + list< vector > r = get_param_list_values_(module); + size_t r_size = r.size(); + for (auto ¶m_val : r) { + std::stringstream ss; + ss << param_val[0] << ": " << param_val[1] << endl; + zmq::message_t msg(ss.str().size()); + memcpy(msg.data(), ss.str().data(), ss.str().size()); + + int flag = (--r_size > 0) ? ZMQ_SNDMORE : 0; + repSocket.send(msg, flag); + } + } + catch (ParameterError &e) { + send_fail_reply(repSocket, e.what()); + } + } + else if (msg.size() == 3 && command == "get") { + std::string module((char*) msg[1].data(), msg[1].size()); + std::string parameter((char*) msg[2].data(), msg[2].size()); + + try { + std::string value = get_param_(module, parameter); + zmq::message_t msg(value.size()); + memcpy ((void*) msg.data(), value.data(), value.size()); + repSocket.send(msg, 0); + } + catch (ParameterError &err) { + send_fail_reply(repSocket, err.what()); + } + } + else if (msg.size() == 4 && command == "set") { + std::string module((char*) msg[1].data(), msg[1].size()); + std::string parameter((char*) msg[2].data(), msg[2].size()); + std::string value((char*) msg[3].data(), msg[3].size()); + + try { + set_param_(module, parameter, value); + send_ok_reply(repSocket); + } + catch (ParameterError &err) { + send_fail_reply(repSocket, err.what()); + } + } + else { + send_fail_reply(repSocket, + "Unsupported command. commands: list, show, get, set"); + } + } + + // check if thread is interrupted + boost::this_thread::interruption_point(); + } + repSocket.close(); + } + catch (boost::thread_interrupted&) {} + catch (zmq::error_t &e) { + etiLog.level(error) << "ZMQ RC error: " << std::string(e.what()); + } + catch (std::exception& e) { + etiLog.level(error) << "ZMQ RC caught exception: " << e.what(); + m_fault = true; + } +} + +#endif + -- cgit v1.2.3 From 17e6a246149c11bac667a233fff1a33a1d06a1fb Mon Sep 17 00:00:00 2001 From: "Matthias P. Braendli" Date: Fri, 7 Oct 2016 16:30:08 +0200 Subject: Add ZeroMQ RC --- configure.ac | 1 + doc/example.mux | 10 +++++++ doc/zmq_remote.py | 75 +++++++++++++++++++++++++++++++++++++++++++++++++++ src/DabMux.cpp | 7 +++-- src/RemoteControl.cpp | 14 +++++----- src/RemoteControl.h | 57 +++------------------------------------ 6 files changed, 101 insertions(+), 63 deletions(-) create mode 100755 doc/zmq_remote.py (limited to 'src/RemoteControl.cpp') diff --git a/configure.ac b/configure.ac index c04492c..09ca8c1 100644 --- a/configure.ac +++ b/configure.ac @@ -187,6 +187,7 @@ AC_CHECK_LIB(zmq, zmq_init, [] , AC_MSG_ERROR(ZeroMQ libzmq is required)) AC_DEFINE([HAVE_INPUT_ZEROMQ], [1], [Define if ZeroMQ input is enabled]) AC_DEFINE([HAVE_OUTPUT_ZEROMQ], [1], [Define if ZeroMQ output is enabled]) +AC_DEFINE([HAVE_RC_ZEROMQ], [1], [Define if ZeroMQ enabled for rc]) # Link against cURL AM_CONDITIONAL([HAVE_CURL_TEST], diff --git a/doc/example.mux b/doc/example.mux index 5a65829..3f2d6f4 100644 --- a/doc/example.mux +++ b/doc/example.mux @@ -50,6 +50,16 @@ remotecontrol { ; Set the port to 0 to disable the server telnetport 12721 + ; The remote control is also accessible through a ZMQ REQ/REP socket, + ; and is useful for machine-triggered interactions. It supports the + ; same commands as the telnet RC. + ; The example code in doc/zmq_remote.py illustrates how to use this rc. + ; To disable the zeromq endpoint, remove the zmqendpoint line. + ; By specifying "lo" in the URL, we make the server only accessible + ; from localhost. You can write tcp://*:12722 to make it accessible + ; on all interfaces. + zmqendpoint tcp://lo:12722 + ; the remote control server makes use of the unique identifiers ; for the subchannels, services and components. Make sure you ; chose them so that you can identify them. diff --git a/doc/zmq_remote.py b/doc/zmq_remote.py new file mode 100755 index 0000000..bc9dd5d --- /dev/null +++ b/doc/zmq_remote.py @@ -0,0 +1,75 @@ +#!/usr/bin/env python +# +# This is an example program that illustrates +# how to interact with the zeromq remote control +# +# LICENSE: see bottom of file + +import sys +import zmq + +context = zmq.Context() + +sock = context.socket(zmq.REQ) + +if len(sys.argv) < 2: + print("Usage: program url cmd [args...]") + sys.exit(1) + +sock.connect(sys.argv[1]) + +message_parts = sys.argv[2:] + +# first do a ping test + +print("ping") +sock.send("ping") +data = sock.recv_multipart() +print("Received: {}".format(len(data))) +for i,part in enumerate(data): + print(" {}".format(part)) + +for i, part in enumerate(message_parts): + if i == len(message_parts) - 1: + f = 0 + else: + f = zmq.SNDMORE + + print("Send {}({}): '{}'".format(i, f, part)) + + sock.send(part, flags=f) + +data = sock.recv_multipart() + +print("Received: {}".format(len(data))) +for i,part in enumerate(data): + print(" RX {}: {}".format(i, part)) + + + +# This is free and unencumbered software released into the public domain. +# +# Anyone is free to copy, modify, publish, use, compile, sell, or +# distribute this software, either in source code form or as a compiled +# binary, for any purpose, commercial or non-commercial, and by any +# means. +# +# In jurisdictions that recognize copyright laws, the author or authors +# of this software dedicate any and all copyright interest in the +# software to the public domain. We make this dedication for the benefit +# of the public at large and to the detriment of our heirs and +# successors. We intend this dedication to be an overt act of +# relinquishment in perpetuity of all present and future rights to this +# software under copyright law. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. +# IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR +# OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, +# ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR +# OTHER DEALINGS IN THE SOFTWARE. +# +# For more information, please refer to + + diff --git a/src/DabMux.cpp b/src/DabMux.cpp index cc6c327..aefa701 100644 --- a/src/DabMux.cpp +++ b/src/DabMux.cpp @@ -273,11 +273,14 @@ int main(int argc, char *argv[]) /************** READ REMOTE CONTROL PARAMETERS *************/ int telnetport = pt.get("remotecontrol.telnetport", 0); - - if (telnetport != 0) { auto rc = std::make_shared(telnetport); + rcs.add_controller(rc); + } + auto zmqendpoint = pt.get("remotecontrol.zmqendpoint", ""); + if (not zmqendpoint.empty()) { + auto rc = std::make_shared(zmqendpoint); rcs.add_controller(rc); } diff --git a/src/RemoteControl.cpp b/src/RemoteControl.cpp index bca0b41..12ab84e 100644 --- a/src/RemoteControl.cpp +++ b/src/RemoteControl.cpp @@ -120,7 +120,7 @@ void RemoteControllerTelnet::process(long) boost::asio::streambuf buffer; length = boost::asio::read_until( socket, buffer, "\n", ignored_error); - std::istream str(&buffer); + std::istream str(&buffer); std::getline(str, in_message); if (length == 0) { @@ -268,7 +268,7 @@ void RemoteControllerTelnet::reply(tcp::socket& socket, string message) } -#if 0 // #if defined(HAVE_ZEROMQ) +#if defined(HAVE_RC_ZEROMQ) void RemoteControllerZmq::restart() { @@ -352,8 +352,8 @@ void RemoteControllerZmq::process() send_ok_reply(repSocket); } else if (msg.size() == 1 && command == "list") { - size_t cohort_size = m_cohort.size(); - for (auto &controllable : m_cohort) { + size_t cohort_size = rcs.controllables.size(); + for (auto &controllable : rcs.controllables) { std::stringstream ss; ss << controllable->get_rc_name(); @@ -369,7 +369,7 @@ void RemoteControllerZmq::process() else if (msg.size() == 2 && command == "show") { std::string module((char*) msg[1].data(), msg[1].size()); try { - list< vector > r = get_param_list_values_(module); + list< vector > r = rcs.get_param_list_values(module); size_t r_size = r.size(); for (auto ¶m_val : r) { std::stringstream ss; @@ -390,7 +390,7 @@ void RemoteControllerZmq::process() std::string parameter((char*) msg[2].data(), msg[2].size()); try { - std::string value = get_param_(module, parameter); + std::string value = rcs.get_param(module, parameter); zmq::message_t msg(value.size()); memcpy ((void*) msg.data(), value.data(), value.size()); repSocket.send(msg, 0); @@ -405,7 +405,7 @@ void RemoteControllerZmq::process() std::string value((char*) msg[3].data(), msg[3].size()); try { - set_param_(module, parameter, value); + rcs.set_param(module, parameter, value); send_ok_reply(repSocket); } catch (ParameterError &err) { diff --git a/src/RemoteControl.h b/src/RemoteControl.h index df99386..c682826 100644 --- a/src/RemoteControl.h +++ b/src/RemoteControl.h @@ -32,7 +32,7 @@ # include "config.h" #endif -#if defined(HAVE_ZEROMQ) +#if defined(HAVE_RC_ZEROMQ) # include "zmq.hpp" #endif @@ -254,7 +254,7 @@ class RemoteControllerTelnet : public BaseRemoteController { int m_port; }; -#if 0 // #if defined(HAVE_ZEROMQ) +#if defined(HAVE_RC_ZEROMQ) /* Implements a Remote controller using zmq transportlayer * that listens on localhost */ @@ -265,7 +265,7 @@ class RemoteControllerZmq : public BaseRemoteController { m_zmqContext(1), m_endpoint("") { } - RemoteControllerZmq(std::string endpoint) + RemoteControllerZmq(const std::string& endpoint) : m_running(true), m_fault(false), m_zmqContext(1), m_endpoint(endpoint), @@ -283,14 +283,6 @@ class RemoteControllerZmq : public BaseRemoteController { } } - void enrol(RemoteControllable* controllable) { - m_cohort.push_back(controllable); - } - - void disengage(RemoteControllable* controllable) { - m_cohort.remove(controllable); - } - virtual bool fault_detected() { return m_fault; } virtual void restart(); @@ -303,46 +295,6 @@ class RemoteControllerZmq : public BaseRemoteController { void send_fail_reply(zmq::socket_t &pSocket, const std::string &error); void process(); - - RemoteControllable* get_controllable_(std::string name) { - for (std::list::iterator it = m_cohort.begin(); - it != m_cohort.end(); ++it) { - if ((*it)->get_rc_name() == name) - { - return *it; - } - } - throw ParameterError("Module name unknown"); - } - - std::string get_param_(std::string name, std::string param) { - RemoteControllable* controllable = get_controllable_(name); - return controllable->get_parameter(param); - } - - void set_param_(std::string name, std::string param, std::string value) { - RemoteControllable* controllable = get_controllable_(name); - return controllable->set_parameter(param, value); - } - - std::list< std::vector > - get_param_list_values_(std::string name) { - RemoteControllable* controllable = get_controllable_(name); - - std::list< std::vector > allparams; - - for (auto ¶m : controllable->get_supported_parameters()) { - std::vector item; - item.push_back(param); - item.push_back(controllable->get_parameter(param)); - - allparams.push_back(item); - } - - return allparams; - } - - std::atomic m_running; /* This is set to true if a fault occurred */ @@ -351,9 +303,6 @@ class RemoteControllerZmq : public BaseRemoteController { zmq::context_t m_zmqContext; - /* This controller commands the controllables in the cohort */ - std::list m_cohort; - std::string m_endpoint; boost::thread m_child_thread; }; -- cgit v1.2.3