diff --git a/include/fc/interprocess/signals.hpp b/include/fc/interprocess/signals.hpp index 876c684..dd791d1 100644 --- a/include/fc/interprocess/signals.hpp +++ b/include/fc/interprocess/signals.hpp @@ -1,8 +1,11 @@ #pragma once #include +#include namespace fc { - /// handler will be called from ASIO thread - void set_signal_handler( std::function handler, int signal_num ); + /// Set a handler to process an IPC (inter process communication) signal. + /// Handler will be called from ASIO thread. + /// @return shared pointer to the signal_set that holds the handler + std::shared_ptr set_signal_handler( std::function handler, int signal_num ); } diff --git a/include/fc/rpc/cli.hpp b/include/fc/rpc/cli.hpp index 1acfb42..038159a 100644 --- a/include/fc/rpc/cli.hpp +++ b/include/fc/rpc/cli.hpp @@ -25,6 +25,7 @@ namespace fc { namespace rpc { void start(); void stop(); + void cancel(); void wait(); void format_result( const string& method, std::function formatter); @@ -40,5 +41,6 @@ namespace fc { namespace rpc { std::string _prompt = ">>>"; std::map > _result_formatters; fc::future _run_complete; + fc::thread* _getline_thread = nullptr; ///< Wait for user input in this thread }; } } diff --git a/include/fc/rpc/websocket_api.hpp b/include/fc/rpc/websocket_api.hpp index ebb6fe7..8f06a19 100644 --- a/include/fc/rpc/websocket_api.hpp +++ b/include/fc/rpc/websocket_api.hpp @@ -10,7 +10,8 @@ namespace fc { namespace rpc { class websocket_api_connection : public api_connection { public: - websocket_api_connection( fc::http::websocket_connection& c, uint32_t max_conversion_depth ); + websocket_api_connection( const std::shared_ptr &c, + uint32_t max_conversion_depth ); ~websocket_api_connection(); virtual variant send_call( @@ -29,8 +30,8 @@ namespace fc { namespace rpc { const std::string& message, bool send_message = true ); - fc::http::websocket_connection& _connection; - fc::rpc::state _rpc_state; + std::shared_ptr _connection; + fc::rpc::state _rpc_state; }; } } // namespace fc::rpc diff --git a/include/fc/thread/thread.hpp b/include/fc/thread/thread.hpp index 7a886fa..b71c066 100644 --- a/include/fc/thread/thread.hpp +++ b/include/fc/thread/thread.hpp @@ -129,6 +129,11 @@ namespace fc { * @todo make quit non-blocking of the calling thread by eliminating the call to boost::thread::join */ void quit(); + + /** + * Send signal to underlying native thread. Only for Linux and macOS + */ + void signal(int); /** * @return true unless quit() has been called. diff --git a/src/interprocess/signals.cpp b/src/interprocess/signals.cpp index f65f122..2a4652c 100644 --- a/src/interprocess/signals.cpp +++ b/src/interprocess/signals.cpp @@ -3,15 +3,16 @@ namespace fc { - void set_signal_handler( std::function handler, int signal_num ) - { - std::shared_ptr sig_set(new boost::asio::signal_set(fc::asio::default_io_service(), signal_num)); - sig_set->async_wait( - [sig_set,handler]( const boost::system::error_code& err, int num ) - { - handler( num ); - sig_set->cancel(); - // set_signal_handler( handler, signal_num ); - } ); - } + std::shared_ptr set_signal_handler( std::function handler, int signal_num ) + { + std::shared_ptr sig_set( new boost::asio::signal_set( fc::asio::default_io_service(), + signal_num) ); + sig_set->async_wait( [sig_set,handler]( const boost::system::error_code& err, int num ) + { + if( err != boost::asio::error::operation_aborted ) + handler( num ); + sig_set->cancel(); + } ); + return sig_set; + } } diff --git a/src/rpc/cli.cpp b/src/rpc/cli.cpp index 658a968..ec6cfe6 100644 --- a/src/rpc/cli.cpp +++ b/src/rpc/cli.cpp @@ -9,6 +9,7 @@ #ifdef HAVE_EDITLINE # include "editline.h" +# include # ifdef WIN32 # include # endif @@ -53,18 +54,24 @@ void cli::send_notice( uint64_t callback_id, variants args /* = variants() */ ) FC_ASSERT(false); } -void cli::start() -{ - cli_commands() = get_method_names(0); - _run_complete = fc::async( [&](){ run(); } ); -} - void cli::stop() { - _run_complete.cancel(); + cancel(); _run_complete.wait(); } +void cli::cancel() +{ + _run_complete.cancel(); +#ifdef HAVE_EDITLINE + if( _getline_thread ) + { + _getline_thread->signal(SIGINT); + _getline_thread = nullptr; + } +#endif +} + void cli::wait() { _run_complete.wait(); @@ -98,6 +105,12 @@ void cli::run() } catch ( const fc::eof_exception& e ) { + _getline_thread = nullptr; + break; + } + catch ( const fc::canceled_exception& e ) + { + _getline_thread = nullptr; break; } @@ -119,12 +132,12 @@ void cli::run() } catch ( const fc::exception& e ) { - std::cout << e.to_detail_string() << "\n"; - if (e.code() == fc::canceled_exception_code) { + _getline_thread = nullptr; break; } + std::cout << e.to_detail_string() << "\n"; } } } @@ -137,36 +150,52 @@ void cli::run() */ static char *my_rl_complete(char *token, int *match) { - bool have_one = false; - std::string method_name; - - auto& cmd = cli_commands(); + const auto& cmds = cli_commands(); const size_t partlen = strlen (token); /* Part of token */ - for (const std::string& it : cmd) + std::vector> matched_cmds; + for( const std::string& it : cmds ) { - if (it.compare(0, partlen, token) == 0) + if( it.compare(0, partlen, token) == 0 ) { - if (have_one) { - // we can only have 1, but we found a second - return NULL; - } - else - { - method_name = it; - have_one = true; - } + matched_cmds.push_back( it ); } } - if (have_one) + if( matched_cmds.size() == 0 ) + return NULL; + + const std::string& first_matched_cmd = matched_cmds[0]; + if( matched_cmds.size() == 1 ) { *match = 1; - method_name += " "; - return strdup (method_name.c_str() + partlen); + std::string matched_cmd = first_matched_cmd + " "; + return strdup( matched_cmd.c_str() + partlen ); } - return NULL; + size_t first_cmd_len = first_matched_cmd.size(); + size_t matched_len = partlen; + for( ; matched_len < first_cmd_len; ++matched_len ) + { + char next_char = first_matched_cmd[matched_len]; + bool end = false; + for( const std::string& s : matched_cmds ) + { + if( s.size() <= matched_len || s[matched_len] != next_char ) + { + end = true; + break; + } + } + if( end ) + break; + } + + if( matched_len == partlen ) + return NULL; + + std::string matched_cmd_part = first_matched_cmd.substr( partlen, matched_len - partlen ); + return strdup( matched_cmd_part.c_str() ); } /*** @@ -216,6 +245,53 @@ static int cli_check_secret(const char *source) return 0; } +/*** + * Indicates whether CLI is quitting after got a SIGINT signal. + * In order to be used by editline which is C-style, this is a global variable. + */ +static int cli_quitting = false; + +/** + * Get next character from stdin, or EOF if got a SIGINT signal + */ +static int interruptible_getc(void) +{ + if( cli_quitting ) + return EOF; + + int r; + char c; + + r = read(0, &c, 1); // read from stdin, will return -1 on SIGINT + + if( r == -1 && errno == EINTR ) + cli_quitting = true; + + return r == 1 ? c : EOF; +} + +void cli::start() +{ + +#ifdef HAVE_EDITLINE + el_hist_size = 256; + + rl_set_complete_func(my_rl_complete); + rl_set_list_possib_func(cli_completion); + //rl_set_check_secret_func(cli_check_secret); + rl_set_getc_func(interruptible_getc); + + static fc::thread getline_thread("getline"); + _getline_thread = &getline_thread; + + cli_quitting = false; + + cli_commands() = get_method_names(0); +#endif + + _run_complete = fc::async( [this](){ run(); } ); +} + /*** * @brief Read input from the user * @param prompt the prompt to display @@ -237,21 +313,19 @@ void cli::getline( const std::string& prompt, std::string& line) if( _isatty( _fileno( stdin ) ) ) #endif { - rl_set_complete_func(my_rl_complete); - rl_set_list_possib_func(cli_completion); - rl_set_check_secret_func(cli_check_secret); - - static fc::thread getline_thread("getline"); - getline_thread.async( [&](){ - char* line_read = nullptr; - std::cout.flush(); //readline doesn't use cin, so we must manually flush _out - line_read = readline(prompt.c_str()); - if( line_read == nullptr ) - FC_THROW_EXCEPTION( fc::eof_exception, "" ); - line = line_read; - // we don't need here to add line in editline's history, cause it will be doubled - free(line_read); - }).wait(); + if( _getline_thread ) + { + _getline_thread->async( [&prompt,&line](){ + char* line_read = nullptr; + std::cout.flush(); //readline doesn't use cin, so we must manually flush _out + line_read = readline(prompt.c_str()); + if( line_read == nullptr ) + FC_THROW_EXCEPTION( fc::eof_exception, "" ); + line = line_read; + // we don't need here to add line in editline's history, cause it will be doubled + free(line_read); + }).wait(); + } } else #endif @@ -259,7 +333,6 @@ void cli::getline( const std::string& prompt, std::string& line) std::cout << prompt; // sync_call( cin_thread, [&](){ std::getline( *input_stream, line ); }, "getline"); fc::getline( fc::cin, line ); - return; } } diff --git a/src/rpc/websocket_api.cpp b/src/rpc/websocket_api.cpp index ae4e26e..d1dd997 100644 --- a/src/rpc/websocket_api.cpp +++ b/src/rpc/websocket_api.cpp @@ -7,9 +7,11 @@ websocket_api_connection::~websocket_api_connection() { } -websocket_api_connection::websocket_api_connection( fc::http::websocket_connection& c, uint32_t max_depth ) +websocket_api_connection::websocket_api_connection( const std::shared_ptr& c, + uint32_t max_depth ) : api_connection(max_depth),_connection(c) { + FC_ASSERT( _connection, "A valid websocket connection is required" ); _rpc_state.add_method( "call", [this]( const variants& args ) -> variant { FC_ASSERT( args.size() == 3 && args[2].is_array() ); @@ -47,9 +49,9 @@ websocket_api_connection::websocket_api_connection( fc::http::websocket_connecti return this->receive_call( 0, method_name, args ); } ); - _connection.on_message_handler( [&]( const std::string& msg ){ on_message(msg,true); } ); - _connection.on_http_handler( [&]( const std::string& msg ){ return on_message(msg,false); } ); - _connection.closed.connect( [this](){ closed(); } ); + _connection->on_message_handler( [this]( const std::string& msg ){ on_message(msg,true); } ); + _connection->on_http_handler( [this]( const std::string& msg ){ return on_message(msg,false); } ); + _connection->closed.connect( [this](){ closed(); } ); } variant websocket_api_connection::send_call( @@ -57,9 +59,13 @@ variant websocket_api_connection::send_call( string method_name, variants args /* = variants() */ ) { - auto request = _rpc_state.start_remote_call( "call", {api_id, std::move(method_name), std::move(args) } ); - _connection.send_message( fc::json::to_string(fc::variant(request, _max_conversion_depth), - fc::json::stringify_large_ints_and_doubles, _max_conversion_depth ) ); + if( !_connection ) // defensive check + return variant(); // TODO return an error? + + auto request = _rpc_state.start_remote_call( "call", { api_id, std::move(method_name), std::move(args) } ); + _connection->send_message( fc::json::to_string( fc::variant( request, _max_conversion_depth ), + fc::json::stringify_large_ints_and_doubles, + _max_conversion_depth ) ); return _rpc_state.wait_for_response( *request.id ); } @@ -67,9 +73,13 @@ variant websocket_api_connection::send_callback( uint64_t callback_id, variants args /* = variants() */ ) { - auto request = _rpc_state.start_remote_call( "callback", {callback_id, std::move(args) } ); - _connection.send_message( fc::json::to_string(fc::variant(request, _max_conversion_depth), - fc::json::stringify_large_ints_and_doubles, _max_conversion_depth ) ); + if( !_connection ) // defensive check + return variant(); // TODO return an error? + + auto request = _rpc_state.start_remote_call( "callback", { callback_id, std::move(args) } ); + _connection->send_message( fc::json::to_string( fc::variant( request, _max_conversion_depth ), + fc::json::stringify_large_ints_and_doubles, + _max_conversion_depth ) ); return _rpc_state.wait_for_response( *request.id ); } @@ -77,9 +87,13 @@ void websocket_api_connection::send_notice( uint64_t callback_id, variants args /* = variants() */ ) { - fc::rpc::request req{ optional(), "notice", {callback_id, std::move(args)}}; - _connection.send_message( fc::json::to_string(fc::variant(req, _max_conversion_depth), - fc::json::stringify_large_ints_and_doubles, _max_conversion_depth ) ); + if( !_connection ) // defensive check + return; + + fc::rpc::request req{ optional(), "notice", { callback_id, std::move(args) } }; + _connection->send_message( fc::json::to_string( fc::variant( req, _max_conversion_depth ), + fc::json::stringify_large_ints_and_doubles, + _max_conversion_depth ) ); } std::string websocket_api_connection::on_message( @@ -109,16 +123,20 @@ std::string websocket_api_connection::on_message( auto end = time_point::now(); if( end - start > fc::milliseconds( LOG_LONG_API_MAX_MS ) ) - elog( "API call execution time limit exceeded. method: ${m} params: ${p} time: ${t}", ("m",call.method)("p",call.params)("t", end - start) ); + elog( "API call execution time limit exceeded. method: ${m} params: ${p} time: ${t}", + ("m",call.method)("p",call.params)("t", end - start) ); else if( end - start > fc::milliseconds( LOG_LONG_API_WARN_MS ) ) - wlog( "API call execution time nearing limit. method: ${m} params: ${p} time: ${t}", ("m",call.method)("p",call.params)("t", end - start) ); + wlog( "API call execution time nearing limit. method: ${m} params: ${p} time: ${t}", + ("m",call.method)("p",call.params)("t", end - start) ); #endif if( call.id ) { - auto reply = fc::json::to_string( response( *call.id, result, "2.0" ), fc::json::stringify_large_ints_and_doubles, _max_conversion_depth ); - if( send_message ) - _connection.send_message( reply ); + auto reply = fc::json::to_string( response( *call.id, result, "2.0" ), + fc::json::stringify_large_ints_and_doubles, + _max_conversion_depth ); + if( send_message && _connection ) + _connection->send_message( reply ); return reply; } } @@ -135,8 +153,8 @@ std::string websocket_api_connection::on_message( auto reply = fc::json::to_string( variant(response( *call.id, error_object{ 1, optexcept->to_string(), fc::variant(*optexcept, _max_conversion_depth)}, "2.0" ), _max_conversion_depth ), fc::json::stringify_large_ints_and_doubles, _max_conversion_depth ); - if( send_message ) - _connection.send_message( reply ); + if( send_message && _connection ) + _connection->send_message( reply ); return reply; } diff --git a/src/thread/thread.cpp b/src/thread/thread.cpp index 0d1a9aa..775eb51 100644 --- a/src/thread/thread.cpp +++ b/src/thread/thread.cpp @@ -155,6 +155,17 @@ namespace fc { void thread::debug( const std::string& d ) { /*my->debug(d);*/ } +#if defined(__linux__) || defined(__APPLE__) +#include +#endif + + void thread::signal(int sig) + { +#if defined(__linux__) || defined(__APPLE__) + pthread_kill( my->boost_thread->native_handle(), sig ); +#endif + } + void thread::quit() { //if quitting from a different thread, start quit task on thread. diff --git a/tests/api.cpp b/tests/api.cpp index e0514c3..fbe42f2 100644 --- a/tests/api.cpp +++ b/tests/api.cpp @@ -61,7 +61,7 @@ int main( int argc, char** argv ) fc::http::websocket_server server; server.on_connection([&]( const websocket_connection_ptr& c ){ - auto wsc = std::make_shared(*c, MAX_DEPTH); + auto wsc = std::make_shared(c, MAX_DEPTH); auto login = std::make_shared(); login->calc = calc_api; wsc->register_api(fc::api(login)); @@ -76,7 +76,7 @@ int main( int argc, char** argv ) try { fc::http::websocket_client client; auto con = client.connect( "ws://localhost:8090" ); - auto apic = std::make_shared(*con, MAX_DEPTH); + auto apic = std::make_shared(con, MAX_DEPTH); auto remote_login_api = apic->get_remote_api(); auto remote_calc = remote_login_api->get_calc(); remote_calc->on_result( []( uint32_t r ) { elog( "callback result ${r}", ("r",r) ); } ); diff --git a/vendor/editline b/vendor/editline index fbb1f88..13f8d5f 160000 --- a/vendor/editline +++ b/vendor/editline @@ -1 +1 @@ -Subproject commit fbb1f8800adbb70264fa3893dc221f524e25708c +Subproject commit 13f8d5f69c3a048e0b6bdb979d617940eed32ef0