diff --git a/out/bi_web/include/fantasy.hpp b/out/bi_web/include/fantasy.hpp index 6f2c804..fe0f61f 100644 --- a/out/bi_web/include/fantasy.hpp +++ b/out/bi_web/include/fantasy.hpp @@ -1326,13 +1326,13 @@ class StreamClient final { [] {}); auto channel_ptr = std::make_shared>(m_pool_ptr->getIoContext(), + frpc::error_code, std::string)>>(m_pool_ptr->getIoContext(), m_config.channel_size); { std::function)> func = [channel_ptr, this](std::tuple tp) mutable { auto& [reply] = tp; - if (!channel_ptr->try_send(asio::error_code{}, + if (!channel_ptr->try_send(frpc::error_code{}, std::move(reply))) m_error(FRPC_ERROR_FORMAT( "Failed to store message to channel!!!")); @@ -1433,7 +1433,7 @@ class StreamClient final { struct StreamServerHandler { virtual void hello_world( std::shared_ptr>, + void(frpc::error_code, std::string)>>, std::shared_ptr>) noexcept = 0; }; @@ -1441,12 +1441,12 @@ struct CoroStreamServerHandler { #ifdef __cpp_impl_coroutine virtual asio::awaitable hello_world( std::shared_ptr>, + void(frpc::error_code, std::string)>>, std::shared_ptr>) noexcept = 0; #else virtual void hello_world( std::shared_ptr>, + void(frpc::error_code, std::string)>>, std::shared_ptr>) noexcept = 0; #endif }; @@ -1580,14 +1580,14 @@ class StreamServer final { recv_bufs[2].data(), recv_bufs[2].size()); auto& [bank_name] = tp; std::shared_ptr> + void(frpc::error_code, std::string)>> channel_ptr; { std::lock_guard lk(m_mtx); if (m_channel_mapping.contains(req_id)) { channel_ptr = std::any_cast( m_channel_mapping[req_id]); - if (!channel_ptr->try_send(asio::error_code{}, + if (!channel_ptr->try_send(frpc::error_code{}, std::move(bank_name))) m_error(FRPC_ERROR_FORMAT( "Failed to store message to channel!!!")); @@ -1595,11 +1595,11 @@ class StreamServer final { } channel_ptr = std::make_shared< asio::experimental::concurrent_channel< - void(asio::error_code, std::string)>>( + void(frpc::error_code, std::string)>>( m_pool_ptr->getIoContext(), m_config.channel_size); m_channel_mapping[req_id] = channel_ptr; } - if (!channel_ptr->try_send(asio::error_code{}, + if (!channel_ptr->try_send(frpc::error_code{}, std::move(bank_name))) m_error(FRPC_ERROR_FORMAT( "Failed to store message to channel!!!")); diff --git a/out/bi_web/include/impl/asio_context_pool.h b/out/bi_web/include/impl/asio_context_pool.h index 6f0082b..04548c3 100644 --- a/out/bi_web/include/impl/asio_context_pool.h +++ b/out/bi_web/include/impl/asio_context_pool.h @@ -3,9 +3,23 @@ #ifndef _FRPC_CONTEXT_POOL_H_ #define _FRPC_CONTEXT_POOL_H_ +#ifdef FRPC_USE_BOOST_ASIO +#include +#include +namespace asio = boost::asio; + +namespace frpc { +using error_code = boost::system::error_code; +} +#else #include #include +namespace frpc { +using error_code = asio::error_code; +} +#endif + namespace frpc { class ContextPool final { diff --git a/out/include/fantasy.hpp b/out/include/fantasy.hpp index 6f2c804..fe0f61f 100644 --- a/out/include/fantasy.hpp +++ b/out/include/fantasy.hpp @@ -1326,13 +1326,13 @@ class StreamClient final { [] {}); auto channel_ptr = std::make_shared>(m_pool_ptr->getIoContext(), + frpc::error_code, std::string)>>(m_pool_ptr->getIoContext(), m_config.channel_size); { std::function)> func = [channel_ptr, this](std::tuple tp) mutable { auto& [reply] = tp; - if (!channel_ptr->try_send(asio::error_code{}, + if (!channel_ptr->try_send(frpc::error_code{}, std::move(reply))) m_error(FRPC_ERROR_FORMAT( "Failed to store message to channel!!!")); @@ -1433,7 +1433,7 @@ class StreamClient final { struct StreamServerHandler { virtual void hello_world( std::shared_ptr>, + void(frpc::error_code, std::string)>>, std::shared_ptr>) noexcept = 0; }; @@ -1441,12 +1441,12 @@ struct CoroStreamServerHandler { #ifdef __cpp_impl_coroutine virtual asio::awaitable hello_world( std::shared_ptr>, + void(frpc::error_code, std::string)>>, std::shared_ptr>) noexcept = 0; #else virtual void hello_world( std::shared_ptr>, + void(frpc::error_code, std::string)>>, std::shared_ptr>) noexcept = 0; #endif }; @@ -1580,14 +1580,14 @@ class StreamServer final { recv_bufs[2].data(), recv_bufs[2].size()); auto& [bank_name] = tp; std::shared_ptr> + void(frpc::error_code, std::string)>> channel_ptr; { std::lock_guard lk(m_mtx); if (m_channel_mapping.contains(req_id)) { channel_ptr = std::any_cast( m_channel_mapping[req_id]); - if (!channel_ptr->try_send(asio::error_code{}, + if (!channel_ptr->try_send(frpc::error_code{}, std::move(bank_name))) m_error(FRPC_ERROR_FORMAT( "Failed to store message to channel!!!")); @@ -1595,11 +1595,11 @@ class StreamServer final { } channel_ptr = std::make_shared< asio::experimental::concurrent_channel< - void(asio::error_code, std::string)>>( + void(frpc::error_code, std::string)>>( m_pool_ptr->getIoContext(), m_config.channel_size); m_channel_mapping[req_id] = channel_ptr; } - if (!channel_ptr->try_send(asio::error_code{}, + if (!channel_ptr->try_send(frpc::error_code{}, std::move(bank_name))) m_error(FRPC_ERROR_FORMAT( "Failed to store message to channel!!!")); diff --git a/out/include/impl/asio_context_pool.h b/out/include/impl/asio_context_pool.h index 6f0082b..04548c3 100644 --- a/out/include/impl/asio_context_pool.h +++ b/out/include/impl/asio_context_pool.h @@ -3,9 +3,23 @@ #ifndef _FRPC_CONTEXT_POOL_H_ #define _FRPC_CONTEXT_POOL_H_ +#ifdef FRPC_USE_BOOST_ASIO +#include +#include +namespace asio = boost::asio; + +namespace frpc { +using error_code = boost::system::error_code; +} +#else #include #include +namespace frpc { +using error_code = asio::error_code; +} +#endif + namespace frpc { class ContextPool final { diff --git a/template/cpp/bi_stream.inja b/template/cpp/bi_stream.inja index cd63c8b..d3ba804 100644 --- a/template/cpp/bi_stream.inja +++ b/template/cpp/bi_stream.inja @@ -86,13 +86,13 @@ public: m_channel->send(std::move(snd_bufs)); }, [] {}); - auto channel_ptr = std::make_shared>( + auto channel_ptr = std::make_shared>( m_pool_ptr->getIoContext(), m_config.channel_size); { std::function)> func = [channel_ptr, this](std::tuple<{{_format_args_type(func.outputs)}}> tp) mutable { auto& [{{_format_args_name(func.outputs)}}] = tp; - if (!channel_ptr->try_send(asio::error_code{}, {{_format_args_name_and_move(func.outputs)}})) + if (!channel_ptr->try_send(frpc::error_code{}, {{_format_args_name_and_move(func.outputs)}})) m_error(FRPC_ERROR_FORMAT("Failed to store message to channel!!!")); }; std::lock_guard lk(m_mtx); @@ -179,7 +179,7 @@ private: struct {{value.callee}}Handler { {% for func in value.definitions %} - virtual void {{func.func_name}}(std::shared_ptr>, + virtual void {{func.func_name}}(std::shared_ptr>, std::shared_ptr>) noexcept = 0; {% endfor %} }; @@ -187,10 +187,10 @@ struct {{value.callee}}Handler { struct Coro{{value.callee}}Handler { {% for func in value.definitions %} #ifdef __cpp_impl_coroutine - virtual asio::awaitable {{func.func_name}}(std::shared_ptr>, + virtual asio::awaitable {{func.func_name}}(std::shared_ptr>, std::shared_ptr>) noexcept = 0; #else - virtual void {{func.func_name}}(std::shared_ptr>, + virtual void {{func.func_name}}(std::shared_ptr>, std::shared_ptr>) noexcept = 0; #endif {% endfor %} @@ -291,21 +291,21 @@ private: case {{value.caller}}{{value.callee}}::{{func.func_name}}: { auto tp = frpc::unpack>(recv_bufs[2].data(), recv_bufs[2].size()); auto& [{{_format_args_name(func.inputs)}}] = tp; - std::shared_ptr> channel_ptr; + std::shared_ptr> channel_ptr; { std::lock_guard lk(m_mtx); if (m_channel_mapping.contains(req_id)) { channel_ptr = std::any_cast(m_channel_mapping[req_id]); - if (!channel_ptr->try_send(asio::error_code{}, {{_format_args_name_and_move(func.inputs)}})) + if (!channel_ptr->try_send(frpc::error_code{}, {{_format_args_name_and_move(func.inputs)}})) m_error(FRPC_ERROR_FORMAT("Failed to store message to channel!!!")); return; } - channel_ptr = std::make_shared>( + channel_ptr = std::make_shared>( m_pool_ptr->getIoContext(), m_config.channel_size); m_channel_mapping[req_id] = channel_ptr; } - if (!channel_ptr->try_send(asio::error_code{}, {{_format_args_name_and_move(func.inputs)}})) + if (!channel_ptr->try_send(frpc::error_code{}, {{_format_args_name_and_move(func.inputs)}})) m_error(FRPC_ERROR_FORMAT("Failed to store message to channel!!!")); auto is_open = std::make_tuple(req_id, req_type, false); diff --git a/template/cpp/impl/asio_context_pool.inja b/template/cpp/impl/asio_context_pool.inja index 80e3fc7..5123f51 100644 --- a/template/cpp/impl/asio_context_pool.inja +++ b/template/cpp/impl/asio_context_pool.inja @@ -3,8 +3,20 @@ #ifndef _FRPC_CONTEXT_POOL_H_ #define _FRPC_CONTEXT_POOL_H_ +#ifdef FRPC_USE_BOOST_ASIO +#include +#include +namespace asio = boost::asio; +namespace frpc { +using error_code = boost::system::error_code; +} +#else #include #include +namespace frpc { +using error_code = asio::error_code; +} +#endif namespace frpc { diff --git a/test/cpp/bi_stream.cpp b/test/cpp/bi_stream.cpp index 9570e53..59803d8 100644 --- a/test/cpp/bi_stream.cpp +++ b/test/cpp/bi_stream.cpp @@ -13,7 +13,7 @@ void start(std::function func) { #ifdef __cpp_impl_coroutine struct CoroStreamServerHandler final : public fantasy::CoroStreamServerHandler { - virtual asio::awaitable hello_world(std::shared_ptr> ins, + virtual asio::awaitable hello_world(std::shared_ptr> ins, std::shared_ptr> outs) noexcept override { start([outs = std::move(outs)]() mutable { for (int i = 0; i < 5; i++) { diff --git a/test/cpp/coro_bi.cpp b/test/cpp/coro_bi.cpp index 5acc05e..3dec50e 100644 --- a/test/cpp/coro_bi.cpp +++ b/test/cpp/coro_bi.cpp @@ -4,6 +4,7 @@ #include +#define FRPC_USE_BOOST_ASIO 1 #include "fantasy.hpp" inline std::string addr{"tcp://127.0.0.1:5878"}; @@ -37,6 +38,13 @@ struct CoroHandler final : public fantasy::AsioCoroHelloWorldServerHandler { fantasy::toString(bank_info), bank_name, blance, date.has_value() ? date.value() : "nullopt", frpc::toString(date_time)); fantasy::Info info; info.name = "coro test"; + boost::asio::co_spawn( + co_await boost::asio::this_coro::executor, + []() -> boost::asio::awaitable { + spdlog::info("test boost"); + co_return; + }, + boost::asio::detached); cb("coro hello world", std::move(info), 556, std::nullopt); co_return; } @@ -76,7 +84,7 @@ void start_server() { server->monitor( [](std::tuple data) { auto& [event, point] = data; - spdlog::info("bi coro server monitor: {} {}",frpc::getEventName(event.event), point); + spdlog::info("bi coro server monitor: {} {}", frpc::getEventName(event.event), point); }, ZMQ_EVENT_ACCEPTED | ZMQ_EVENT_DISCONNECTED); server->start();