Merge pull request #2044

0299cb77 Fix various oversights/bugs in ZMQ RPC server code (Thomas Winget)
77986023 json serialization for rpc-relevant monero types (Thomas Winget)
5c1e08fe Refactor some things into more composable (smaller) functions (Thomas Winget)
9ac2ad07 DRY refactoring (Thomas Winget)

src/daemon/CMakeLists.txt

@@ -92,12 +92,15 @@ target_link_libraries(daemon
     p2p
     cryptonote_protocol
     daemonizer
+    serialization
+    daemon_rpc_server
     ${Boost_CHRONO_LIBRARY}
     ${Boost_FILESYSTEM_LIBRARY}
     ${Boost_PROGRAM_OPTIONS_LIBRARY}
     ${Boost_REGEX_LIBRARY}
     ${Boost_SYSTEM_LIBRARY}
     ${CMAKE_THREAD_LIBS_INIT}
+    ${ZMQ_LIB}
     ${EXTRA_LIBRARIES})
 add_dependencies(daemon version)
 set_property(TARGET daemon

src/daemon/command_line_args.h

@@ -64,6 +64,25 @@ namespace daemon_args
   , "Max number of threads to use for a parallel job"
   , 0
   };
+
+  const command_line::arg_descriptor<std::string> arg_zmq_rpc_bind_ip   = {
+    "zmq-rpc-bind-ip"
+      , "IP for ZMQ RPC server to listen on"
+      , "127.0.0.1"
+  };
+
+  const command_line::arg_descriptor<std::string> arg_zmq_rpc_bind_port = {
+    "zmq-rpc-bind-port"
+      , "Port for ZMQ RPC server to listen on"
+      , std::to_string(config::ZMQ_RPC_DEFAULT_PORT)
+  };
+
+  const command_line::arg_descriptor<std::string> arg_zmq_testnet_rpc_bind_port = {
+    "zmq-testnet-rpc-bind-port"
+      , "Port for testnet ZMQ RPC server to listen on"
+      , std::to_string(config::testnet::ZMQ_RPC_DEFAULT_PORT)
+  };
+
 }  // namespace daemon_args
 
 #endif // DAEMON_COMMAND_LINE_ARGS_H

src/daemon/daemon.cpp

@@ -32,6 +32,8 @@
 #include <stdexcept>
 #include "misc_log_ex.h"
 #include "daemon/daemon.h"
+#include "rpc/daemon_handler.h"
+#include "rpc/zmq_server.h"
 
 #include "common/password.h"
 #include "common/util.h"
@@ -85,7 +87,18 @@ t_daemon::t_daemon(
     boost::program_options::variables_map const & vm
   )
   : mp_internals{new t_internals{vm}}
-{}
+{
+  bool testnet = command_line::get_arg(vm, command_line::arg_testnet_on);
+  if (testnet)
+  {
+    zmq_rpc_bind_port = command_line::get_arg(vm, daemon_args::arg_zmq_testnet_rpc_bind_port);
+  }
+  else
+  {
+    zmq_rpc_bind_port = command_line::get_arg(vm, daemon_args::arg_zmq_rpc_bind_port);
+  }
+  zmq_rpc_bind_address = command_line::get_arg(vm, daemon_args::arg_zmq_rpc_bind_ip);
+}
 
 t_daemon::~t_daemon() = default;
 
@@ -133,6 +146,30 @@ bool t_daemon::run(bool interactive)
       rpc_commands->start_handling(std::bind(&daemonize::t_daemon::stop_p2p, this));
     }
 
+    cryptonote::rpc::DaemonHandler rpc_daemon_handler(mp_internals->core.get(), mp_internals->p2p.get());
+    cryptonote::rpc::ZmqServer zmq_server(rpc_daemon_handler);
+
+    if (!zmq_server.addTCPSocket(zmq_rpc_bind_address, zmq_rpc_bind_port))
+    {
+      LOG_ERROR(std::string("Failed to add TCP Socket (") + zmq_rpc_bind_address
+          + ":" + zmq_rpc_bind_port + ") to ZMQ RPC Server");
+
+      if (interactive)
+      {
+        rpc_commands->stop_handling();
+      }
+
+      mp_internals->rpc.stop();
+
+      return false;
+    }
+
+    MINFO("Starting ZMQ server...");
+    zmq_server.run();
+
+    MINFO(std::string("ZMQ server started at ") + zmq_rpc_bind_address
+          + ":" + zmq_rpc_bind_port + ".");
+
     mp_internals->p2p.run(); // blocks until p2p goes down
 
     if (rpc_commands)
@@ -140,6 +177,8 @@ bool t_daemon::run(bool interactive)
       rpc_commands->stop_handling();
     }
 
+    zmq_server.stop();
+
     mp_internals->rpc.stop();
     mp_internals->core.get().get_miner().stop();
     MGINFO("Node stopped.");

src/daemon/daemon.h

@@ -43,6 +43,8 @@ private:
   void stop_p2p();
 private:
   std::unique_ptr<t_internals> mp_internals;
+  std::string zmq_rpc_bind_address;
+  std::string zmq_rpc_bind_port;
 public:
   t_daemon(
       boost::program_options::variables_map const & vm

src/daemon/main.cpp

@@ -89,6 +89,9 @@ int main(int argc, char const * argv[])
       command_line::add_arg(core_settings, daemon_args::arg_log_file, default_log.string());
       command_line::add_arg(core_settings, daemon_args::arg_log_level);
       command_line::add_arg(core_settings, daemon_args::arg_max_concurrency);
+      command_line::add_arg(core_settings, daemon_args::arg_zmq_rpc_bind_ip);
+      command_line::add_arg(core_settings, daemon_args::arg_zmq_rpc_bind_port);
+      command_line::add_arg(core_settings, daemon_args::arg_zmq_testnet_rpc_bind_port);
 
       daemonizer::init_options(hidden_options, visible_options);
       daemonize::t_executor::init_options(core_settings);