/*
Copyright (C) 2007, 2008, 2009, 2010, 2011, 2012
Her Majesty the Queen in Right of Canada (Communications Research
Center Canada)
Copyright (C) 2019
Matthias P. Braendli, matthias.braendli@mpb.li
http://www.opendigitalradio.org
*/
/*
This file is part of ODR-DabMux.
ODR-DabMux is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as
published by the Free Software Foundation, either version 3 of the
License, or (at your option) any later version.
ODR-DabMux is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with ODR-DabMux. If not, see .
*/
#include
#include
#include
#include
#include
#include
#include "RemoteControl.h"
using boost::asio::ip::tcp;
using namespace std;
RemoteControllers rcs;
RemoteControllerTelnet::~RemoteControllerTelnet()
{
m_active = false;
m_io_service.stop();
if (m_restarter_thread.joinable()) {
m_restarter_thread.join();
}
if (m_child_thread.joinable()) {
m_child_thread.join();
}
}
void RemoteControllerTelnet::restart()
{
if (m_restarter_thread.joinable()) {
m_restarter_thread.join();
}
m_restarter_thread = std::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;
}
void RemoteControllers::add_controller(std::shared_ptr rc) {
m_controllers.push_back(rc);
}
void RemoteControllers::enrol(RemoteControllable *rc) {
controllables.push_back(rc);
}
void RemoteControllers::remove_controllable(RemoteControllable *rc) {
controllables.remove(rc);
}
std::list< std::vector > RemoteControllers::get_param_list_values(const 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);
try {
item.push_back(controllable->get_parameter(param));
}
catch (const ParameterError &e) {
item.push_back(std::string("error: ") + e.what());
}
allparams.push_back(item);
}
return allparams;
}
std::string RemoteControllers::get_param(const std::string& name, const std::string& param) {
RemoteControllable* controllable = get_controllable_(name);
return controllable->get_parameter(param);
}
void RemoteControllers::check_faults() {
for (auto &controller : m_controllers) {
if (controller->fault_detected()) {
etiLog.level(warn) <<
"Detected Remote Control fault, restarting it";
controller->restart();
}
}
}
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;
}
}
void RemoteControllers::set_param(
const std::string& name,
const std::string& param,
const std::string& value)
{
etiLog.level(info) << "RC: Setting " << name << " " << param
<< " to " << value;
RemoteControllable* controllable = get_controllable_(name);
try {
return controllable->set_parameter(param, value);
}
catch (const ios_base::failure& e) {
etiLog.level(info) << "RC: Failed to set " << name << " " << param
<< " to " << value << ": " << e.what();
throw ParameterError("Cannot understand value");
}
}
// 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_active = false;
m_io_service.stop();
if (m_child_thread.joinable()) {
m_child_thread.join();
}
m_child_thread = std::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)
{
const std::string welcome = "ODR-DabMux Remote Control CLI\n"
"Write 'help' for help.\n"
"**********\n";
const std::string 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_active && in_message != "quit") {
boost::asio::write(*socket, boost::asio::buffer(prompt),
boost::asio::transfer_all(),
ignored_error);
in_message = "";
boost::asio::streambuf buffer;
length = boost::asio::read_until(*socket, buffer, "\n", ignored_error);
std::istream str(&buffer);
std::getline(str, in_message);
if (length == 0) {
etiLog.level(info) << "RC: Connection terminated";
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);
}
if (in_message.length() == 0) {
continue;
}
etiLog.level(info) << "RC: Got message '" << in_message << "'";
dispatch_command(*socket, in_message);
}
etiLog.level(info) << "RC: Closing socket";
socket->close();
}
catch (const std::exception& e)
{
etiLog.level(error) << "Remote control caught exception: " << e.what();
}
}
void RemoteControllerTelnet::process(long)
{
m_active = true;
while (m_active) {
m_io_service.reset();
tcp::acceptor acceptor(m_io_service, tcp::endpoint(
boost::asio::ip::address::from_string("127.0.0.1"), m_port) );
// Add a job to start accepting connections.
boost::shared_ptr socket(
new tcp::socket(acceptor.get_io_service()));
// 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)));
// Process event loop.
m_io_service.run();
}
etiLog.level(info) << "RC: Leaving";
m_fault = true;
}
static std::vector tokenise(const std::string& message) {
stringstream ss(message);
std::vector all_tokens;
std::string item;
while (std::getline(ss, item, ' ')) {
all_tokens.push_back(move(item));
}
return all_tokens;
}
void RemoteControllerTelnet::dispatch_command(tcp::socket& socket, string command)
{
vector cmd = tokenise(command);
if (cmd[0] == "help") {
reply(socket,
"The following commands are supported:\n"
" list\n"
" * Lists the modules that are loaded and their parameters\n"
" show MODULE\n"
" * Lists all parameters and their values from module MODULE\n"
" get MODULE PARAMETER\n"
" * Gets the value for the specified PARAMETER from module MODULE\n"
" set MODULE PARAMETER VALUE\n"
" * Sets the value for the PARAMETER ofr module MODULE\n"
" quit\n"
" * Terminate this session\n"
"\n");
}
else if (cmd[0] == "list") {
stringstream ss;
if (cmd.size() == 1) {
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;
}
}
}
else {
reply(socket, "Too many arguments for command 'list'");
}
reply(socket, ss.str());
}
else if (cmd[0] == "show") {
if (cmd.size() == 2) {
try {
stringstream ss;
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());
}
catch (const ParameterError &e) {
reply(socket, e.what());
}
}
else {
reply(socket, "Incorrect parameters for command 'show'");
}
}
else if (cmd[0] == "get") {
if (cmd.size() == 3) {
try {
string r = rcs.get_param(cmd[1], cmd[2]);
reply(socket, r);
}
catch (const ParameterError &e) {
reply(socket, e.what());
}
}
else {
reply(socket, "Incorrect parameters for command 'get'");
}
}
else if (cmd[0] == "set") {
if (cmd.size() >= 4) {
try {
stringstream new_param_value;
for (size_t i = 3; i < cmd.size(); i++) {
new_param_value << cmd[i];
if (i+1 < cmd.size()) {
new_param_value << " ";
}
}
rcs.set_param(cmd[1], cmd[2], new_param_value.str());
reply(socket, "ok");
}
catch (const ParameterError &e) {
reply(socket, e.what());
}
catch (const exception &e) {
reply(socket, "Error: Invalid parameter value. ");
}
}
else {
reply(socket, "Incorrect parameters for command 'set'");
}
}
else if (cmd[0] == "quit") {
reply(socket, "Goodbye");
}
else {
reply(socket, "Message not understood");
}
}
void RemoteControllerTelnet::reply(tcp::socket& socket, string message)
{
boost::system::error_code ignored_error;
stringstream ss;
ss << message << "\r\n";
boost::asio::write(socket, boost::asio::buffer(ss.str()),
boost::asio::transfer_all(),
ignored_error);
}
#if defined(HAVE_RC_ZEROMQ)
RemoteControllerZmq::~RemoteControllerZmq() {
m_active = false;
m_fault = false;
if (m_restarter_thread.joinable()) {
m_restarter_thread.join();
}
if (m_child_thread.joinable()) {
m_child_thread.join();
}
}
void RemoteControllerZmq::restart()
{
if (m_restarter_thread.joinable()) {
m_restarter_thread.join();
}
m_restarter_thread = std::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_active = false;
if (m_child_thread.joinable()) {
m_child_thread.join();
}
m_child_thread = std::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()
{
m_fault = false;
// create zmq reply socket for receiving ctrl parameters
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} };
while (m_active) {
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 = rcs.controllables.size();
for (auto &controllable : rcs.controllables) {
std::stringstream ss;
ss << "{ \"name\": \"" << controllable->get_rc_name() << "\"," <<
" \"params\": { ";
list< vector > params = controllable->get_parameter_descriptions();
size_t i = 0;
for (auto ¶m : params) {
if (i > 0) {
ss << ", ";
}
ss << "\"" << param[0] << "\": " <<
"\"" << param[1] << "\"";
i++;
}
ss << " } }";
std::string msg_s = ss.str();
zmq::message_t zmsg(ss.str().size());
memcpy ((void*) zmsg.data(), msg_s.data(), msg_s.size());
int flag = (--cohort_size > 0) ? ZMQ_SNDMORE : 0;
repSocket.send(zmsg, flag);
}
}
else if (msg.size() == 2 && command == "show") {
std::string module((char*) msg[1].data(), msg[1].size());
try {
list< vector > r = rcs.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 zmsg(ss.str().size());
memcpy(zmsg.data(), ss.str().data(), ss.str().size());
int flag = (--r_size > 0) ? ZMQ_SNDMORE : 0;
repSocket.send(zmsg, flag);
}
}
catch (const ParameterError &err) {
send_fail_reply(repSocket, err.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 = rcs.get_param(module, parameter);
zmq::message_t zmsg(value.size());
memcpy ((void*) zmsg.data(), value.data(), value.size());
repSocket.send(zmsg, 0);
}
catch (const 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 {
rcs.set_param(module, parameter, value);
send_ok_reply(repSocket);
}
catch (const ParameterError &err) {
send_fail_reply(repSocket, err.what());
}
}
else {
send_fail_reply(repSocket,
"Unsupported command. commands: list, show, get, set");
}
}
}
repSocket.close();
}
catch (const zmq::error_t &e) {
etiLog.level(error) << "ZMQ RC error: " << std::string(e.what());
}
catch (const std::exception& e) {
etiLog.level(error) << "ZMQ RC caught exception: " << e.what();
m_fault = true;
}
}
#endif