Commit 5f302fe9 authored by Dominik Charousset's avatar Dominik Charousset

Add `set_middleman`, remote separate ASIO lib

Rather than having separate versions of `libcaf_io`, users can now pick a
network packend via `set_middleman`. The ASIO multiplexer is available via two
headers (one implementation header) that users can include to get the
functionality. This change also removes Boost dependencies from CAF.
parent 5ff3ad6d
...@@ -28,9 +28,6 @@ endif() ...@@ -28,9 +28,6 @@ endif()
if(NOT CAF_NO_SUMMARY) if(NOT CAF_NO_SUMMARY)
set(CAF_NO_SUMMARY no) set(CAF_NO_SUMMARY no)
endif() endif()
if(NOT CAF_USE_ASIO)
set(CAF_USE_ASIO no)
endif()
################################################################################ ################################################################################
...@@ -247,10 +244,6 @@ if(CAF_NO_MEM_MANAGEMENT) ...@@ -247,10 +244,6 @@ if(CAF_NO_MEM_MANAGEMENT)
add_definitions(-DCAF_NO_MEM_MANAGEMENT) add_definitions(-DCAF_NO_MEM_MANAGEMENT)
endif() endif()
if (CAF_USE_ASIO)
add_definitions(-DCAF_USE_ASIO)
endif()
################################################################################ ################################################################################
# check for biicode build pipeline # # check for biicode build pipeline #
################################################################################ ################################################################################
...@@ -331,20 +324,6 @@ endif() ...@@ -331,20 +324,6 @@ endif()
# all projects need the headers of the core components # all projects need the headers of the core components
include_directories("${LIBCAF_INCLUDE_DIRS}") include_directories("${LIBCAF_INCLUDE_DIRS}")
# Find boost asio if the asio multiplexer should be used
if (CAF_USE_ASIO)
if (CAF_BUILD_STATIC_ONLY)
set(Boost_USE_STATIC_LIBS ON) # only find static libs
endif()
set(Boost_USE_MULTITHREADED ON)
find_package(Boost REQUIRED COMPONENTS system)
if (Boost_FOUND)
set(INCLUDE_DIRS "${INCLUDE_DIRS} ${Boost_INCLUDE_DIRS}")
set(LD_DIRS ${LD_DIRS} ${Boost_LIBRARIES})
set(LD_FLAGS ${LD_FLAGS} ${Boost_SYSTEM_LIBRARY})
endif()
endif()
################################################################################ ################################################################################
# add subprojects # # add subprojects #
...@@ -466,6 +445,20 @@ endif() ...@@ -466,6 +445,20 @@ endif()
################################################################################ ################################################################################
if(NOT CAF_NO_UNIT_TESTS) if(NOT CAF_NO_UNIT_TESTS)
# find boost asio if the asio multiplexer should be used for testing
if(CAF_USE_ASIO)
set(Boost_USE_MULTITHREADED ON)
find_package(Boost COMPONENTS system)
if (Boost_FOUND)
add_definitions(-DCAF_USE_ASIO)
include_directories(${Boost_INCLUDE_DIRS})
set(LD_DIRS ${LD_DIRS} ${Boost_LIBRARIES})
set(LD_FLAGS ${LD_FLAGS} ${Boost_SYSTEM_LIBRARY})
else()
set(CAF_USE_ASIO no)
endif()
endif()
# setup unit test binary
add_executable(caf-test add_executable(caf-test
libcaf_test/src/caf-test.cpp libcaf_test/src/caf-test.cpp
libcaf_test/caf/test/unit_test.hpp libcaf_test/caf/test/unit_test.hpp
...@@ -493,6 +486,11 @@ if(NOT CAF_NO_UNIT_TESTS) ...@@ -493,6 +486,11 @@ if(NOT CAF_NO_UNIT_TESTS)
string(REPLACE " " "_" test_name ${suite}) string(REPLACE " " "_" test_name ${suite})
set(caf_test ${EXECUTABLE_OUTPUT_PATH}/caf-test) set(caf_test ${EXECUTABLE_OUTPUT_PATH}/caf-test)
add_test(${test_name} ${caf_test} -n -v 5 -s "${suite}" ${ARGN}) add_test(${test_name} ${caf_test} -n -v 5 -s "${suite}" ${ARGN})
# add some extra unit testing options when testing ASIO as well
if(CAF_USE_ASIO AND "${suite}" MATCHES "^io_.+$")
add_test(${test_name}_asio ${caf_test} -n -v 5 -s
"${suite}" ${ARGN} -- --use-asio)
endif()
endmacro () endmacro ()
list(LENGTH suites num_suites) list(LENGTH suites num_suites)
message(STATUS "Found ${num_suites} test suites") message(STATUS "Found ${num_suites} test suites")
...@@ -588,7 +586,6 @@ if(NOT CAF_NO_SUMMARY) ...@@ -588,7 +586,6 @@ if(NOT CAF_NO_SUMMARY)
"\nRuntime checks: ${CAF_ENABLE_RUNTIME_CHECKS}" "\nRuntime checks: ${CAF_ENABLE_RUNTIME_CHECKS}"
"\nLog level: ${LOG_LEVEL_STR}" "\nLog level: ${LOG_LEVEL_STR}"
"\nWith mem. mgmt.: ${CAF_BUILD_MEM_MANAGEMENT}" "\nWith mem. mgmt.: ${CAF_BUILD_MEM_MANAGEMENT}"
"\nUse Boost ASIO: ${CAF_USE_ASIO}"
"\n" "\n"
"\nBuild examples: ${CAF_BUILD_EXAMPLES}" "\nBuild examples: ${CAF_BUILD_EXAMPLES}"
"\nBuild unit tests: ${CAF_BUILD_UNIT_TESTS}" "\nBuild unit tests: ${CAF_BUILD_UNIT_TESTS}"
......
...@@ -49,7 +49,6 @@ Usage: $0 [OPTION]... [VAR=VALUE]... ...@@ -49,7 +49,6 @@ Usage: $0 [OPTION]... [VAR=VALUE]...
--no-compiler-check disable compiler version check --no-compiler-check disable compiler version check
--no-auto-libc++ do not automatically enable libc++ for Clang --no-auto-libc++ do not automatically enable libc++ for Clang
--warnings-as-errors enables -Werror --warnings-as-errors enables -Werror
--with-asio enable ASIO multiplexer
Installation Directories: Installation Directories:
--prefix=PREFIX installation directory [/usr/local] --prefix=PREFIX installation directory [/usr/local]
...@@ -68,6 +67,9 @@ Usage: $0 [OPTION]... [VAR=VALUE]... ...@@ -68,6 +67,9 @@ Usage: $0 [OPTION]... [VAR=VALUE]...
--no-riac build without riac --no-riac build without riac
--no-summary do not print configuration before building --no-summary do not print configuration before building
Testing:
--with-asio use ASIO multiplexer in unit tests
Debugging: Debugging:
--with-runtime-checks build with requirement checks at runtime --with-runtime-checks build with requirement checks at runtime
--with-log-level=LVL build with debugging output, possible values: --with-log-level=LVL build with debugging output, possible values:
......
...@@ -10,6 +10,7 @@ set (LIBCAF_IO_SRCS ...@@ -10,6 +10,7 @@ set (LIBCAF_IO_SRCS
src/basp_broker.cpp src/basp_broker.cpp
src/abstract_broker.cpp src/abstract_broker.cpp
src/broker.cpp src/broker.cpp
src/default_multiplexer.cpp
src/max_msg_size.cpp src/max_msg_size.cpp
src/middleman.cpp src/middleman.cpp
src/hook.cpp src/hook.cpp
...@@ -19,6 +20,7 @@ set (LIBCAF_IO_SRCS ...@@ -19,6 +20,7 @@ set (LIBCAF_IO_SRCS
src/remote_actor.cpp src/remote_actor.cpp
src/remote_group.cpp src/remote_group.cpp
src/manager.cpp src/manager.cpp
src/set_middleman.cpp
src/stream_manager.cpp src/stream_manager.cpp
src/unpublish.cpp src/unpublish.cpp
src/acceptor_manager.cpp src/acceptor_manager.cpp
...@@ -26,12 +28,6 @@ set (LIBCAF_IO_SRCS ...@@ -26,12 +28,6 @@ set (LIBCAF_IO_SRCS
add_custom_target(libcaf_io) add_custom_target(libcaf_io)
if (CAF_USE_ASIO)
set(LIBCAF_IO_SRCS ${LIBCAF_IO_SRCS} src/asio_multiplexer.cpp)
else()
set(LIBCAF_IO_SRCS ${LIBCAF_IO_SRCS} src/default_multiplexer.cpp)
endif()
# build shared library if not compiling static only # build shared library if not compiling static only
if (NOT CAF_BUILD_STATIC_ONLY) if (NOT CAF_BUILD_STATIC_ONLY)
add_library(libcaf_io_shared SHARED ${LIBCAF_IO_SRCS} ${LIBCAF_IO_HDRS}) add_library(libcaf_io_shared SHARED ${LIBCAF_IO_SRCS} ${LIBCAF_IO_HDRS})
...@@ -65,3 +61,4 @@ if(NOT WIN32) ...@@ -65,3 +61,4 @@ if(NOT WIN32)
install(DIRECTORY caf/ DESTINATION include/caf FILES_MATCHING PATTERN "*.hpp") install(DIRECTORY caf/ DESTINATION include/caf FILES_MATCHING PATTERN "*.hpp")
install(DIRECTORY cppa/ DESTINATION include/cppa FILES_MATCHING PATTERN "*.hpp") install(DIRECTORY cppa/ DESTINATION include/cppa FILES_MATCHING PATTERN "*.hpp")
endif() endif()
...@@ -29,6 +29,7 @@ ...@@ -29,6 +29,7 @@
#include "caf/io/max_msg_size.hpp" #include "caf/io/max_msg_size.hpp"
#include "caf/io/remote_actor.hpp" #include "caf/io/remote_actor.hpp"
#include "caf/io/remote_group.hpp" #include "caf/io/remote_group.hpp"
#include "caf/io/set_middleman.hpp"
#include "caf/io/receive_policy.hpp" #include "caf/io/receive_policy.hpp"
#include "caf/io/system_messages.hpp" #include "caf/io/system_messages.hpp"
#include "caf/io/publish_local_groups.hpp" #include "caf/io/publish_local_groups.hpp"
......
...@@ -119,6 +119,10 @@ public: ...@@ -119,6 +119,10 @@ public:
/// @cond PRIVATE /// @cond PRIVATE
using backend_pointer = std::unique_ptr<network::multiplexer>;
using backend_factory = std::function<backend_pointer()>;
// stops the singleton // stops the singleton
void stop() override; void stop() override;
...@@ -128,13 +132,13 @@ public: ...@@ -128,13 +132,13 @@ public:
// initializes the singleton // initializes the singleton
void initialize() override; void initialize() override;
middleman(const backend_factory&);
/// @endcond /// @endcond
private: private:
// guarded by singleton-getter `instance`
middleman();
// networking backend // networking backend
std::unique_ptr<network::multiplexer> backend_; backend_pointer backend_;
// prevents backend from shutting down unless explicitly requested // prevents backend from shutting down unless explicitly requested
network::multiplexer::supervisor_ptr backend_supervisor_; network::multiplexer::supervisor_ptr backend_supervisor_;
// runs the backend // runs the backend
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_SET_MIDDLEMAN_HPP
#define CAF_IO_SET_MIDDLEMAN_HPP
#include "caf/io/middleman.hpp"
namespace caf {
namespace io {
/// Sets a user-defined middleman using given network backend.
/// @note This function must be used before actor is spawned. Dynamically
/// changing the middleman at runtime is not supported.
/// @throws std::logic_error if a scheduler is already defined
void set_middleman(network::multiplexer* ptr);
/// Sets a user-defined middleman using given network backend.
/// @note This function must be used before actor is spawned. Dynamically
/// changing the middleman at runtime is not supported.
/// @throws std::logic_error if a scheduler is already defined
template <class Multiplexer>
void set_middleman() {
set_middleman(new Multiplexer);
}
} // namespace io
} // namespace caf
#endif // CAF_IO_SET_MIDDLEMAN_HPP
...@@ -329,8 +329,12 @@ middleman_actor_impl::~middleman_actor_impl() { ...@@ -329,8 +329,12 @@ middleman_actor_impl::~middleman_actor_impl() {
middleman* middleman::instance() { middleman* middleman::instance() {
CAF_LOGF_TRACE(""); CAF_LOGF_TRACE("");
// store lambda in a plain old function pointer to make sure
// std::function has minimal overhead
using funptr = backend_pointer (*)();
funptr backend_fac = [] { return network::multiplexer::make(); };
auto fac = [&] { return new middleman(backend_fac); };
auto sid = detail::singletons::middleman_plugin_id; auto sid = detail::singletons::middleman_plugin_id;
auto fac = [] { return new middleman; };
auto res = detail::singletons::get_plugin_singleton(sid, fac); auto res = detail::singletons::get_plugin_singleton(sid, fac);
return static_cast<middleman*>(res); return static_cast<middleman*>(res);
} }
...@@ -342,7 +346,6 @@ void middleman::add_broker(broker_ptr bptr) { ...@@ -342,7 +346,6 @@ void middleman::add_broker(broker_ptr bptr) {
void middleman::initialize() { void middleman::initialize() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
backend_ = network::multiplexer::make();
backend_supervisor_ = backend_->make_supervisor(); backend_supervisor_ = backend_->make_supervisor();
thread_ = std::thread{[this] { thread_ = std::thread{[this] {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
...@@ -396,7 +399,7 @@ void middleman::dispose() { ...@@ -396,7 +399,7 @@ void middleman::dispose() {
delete this; delete this;
} }
middleman::middleman() { middleman::middleman(const backend_factory& factory) : backend_(factory()) {
// nop // nop
} }
......
...@@ -18,14 +18,7 @@ ...@@ -18,14 +18,7 @@
******************************************************************************/ ******************************************************************************/
#include "caf/io/network/multiplexer.hpp" #include "caf/io/network/multiplexer.hpp"
#include "caf/io/network/default_multiplexer.hpp"
#ifdef CAF_USE_ASIO
# include "caf/io/network/asio_multiplexer.hpp"
using caf_multiplexer_impl = caf::io::network::asio_multiplexer;
#else
# include "caf/io/network/default_multiplexer.hpp"
using caf_multiplexer_impl = caf::io::network::default_multiplexer;
#endif
namespace caf { namespace caf {
namespace io { namespace io {
...@@ -41,7 +34,7 @@ boost::asio::io_service* pimpl() { ...@@ -41,7 +34,7 @@ boost::asio::io_service* pimpl() {
multiplexer_ptr multiplexer::make() { multiplexer_ptr multiplexer::make() {
CAF_LOGF_TRACE(""); CAF_LOGF_TRACE("");
return multiplexer_ptr{new caf_multiplexer_impl}; return multiplexer_ptr{new default_multiplexer};
} }
boost::asio::io_service* multiplexer::pimpl() { boost::asio::io_service* multiplexer::pimpl() {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/io/set_middleman.hpp"
namespace caf {
namespace io {
void set_middleman(network::multiplexer* multiplexer_ptr) {
middleman::backend_pointer uptr;
uptr.reset(multiplexer_ptr);
auto fac = [&uptr] { return std::move(uptr); };
auto mm = new middleman(fac);
auto getter = [mm] { return mm; };
auto sid = detail::singletons::middleman_plugin_id;
auto res = detail::singletons::get_plugin_singleton(sid, getter);
if (res != mm) {
delete mm;
throw std::logic_error("middleman already defined");
}
}
} // namespace io
} // namespace caf
...@@ -17,7 +17,7 @@ ...@@ -17,7 +17,7 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#define CAF_SUITE dynamic_broker #define CAF_SUITE io_dynamic_broker
#include "caf/test/unit_test.hpp" #include "caf/test/unit_test.hpp"
#include <memory> #include <memory>
...@@ -30,6 +30,10 @@ ...@@ -30,6 +30,10 @@
#include "caf/detail/run_program.hpp" #include "caf/detail/run_program.hpp"
#ifdef CAF_USE_ASIO
#include "caf/io/network/asio_multiplexer.hpp"
#endif // CAF_USE_ASIO
using namespace std; using namespace std;
using namespace caf; using namespace caf;
using namespace caf::io; using namespace caf::io;
...@@ -173,7 +177,7 @@ behavior peer_acceptor_fun(broker* self, const actor& buddy) { ...@@ -173,7 +177,7 @@ behavior peer_acceptor_fun(broker* self, const actor& buddy) {
}; };
} }
void run_server(bool spawn_client, const char* bin_path) { void run_server(bool spawn_client, const char* bin_path, bool use_asio) {
scoped_actor self; scoped_actor self;
auto serv = io::spawn_io(peer_acceptor_fun, spawn(pong)); auto serv = io::spawn_io(peer_acceptor_fun, spawn(pong));
self->sync_send(serv, publish_atom::value).await( self->sync_send(serv, publish_atom::value).await(
...@@ -181,7 +185,8 @@ void run_server(bool spawn_client, const char* bin_path) { ...@@ -181,7 +185,8 @@ void run_server(bool spawn_client, const char* bin_path) {
CAF_MESSAGE("server is running on port " << port); CAF_MESSAGE("server is running on port " << port);
if (spawn_client) { if (spawn_client) {
auto child = detail::run_program(self, bin_path, "-n", "-s", auto child = detail::run_program(self, bin_path, "-n", "-s",
"dynamic_broker", "--", "-c", port); CAF_XSTR(CAF_SUITE), "--", "-c", port,
(use_asio ? "--use-asio" : ""));
CAF_MESSAGE("block till child process has finished"); CAF_MESSAGE("block till child process has finished");
child.join(); child.join();
} }
...@@ -201,29 +206,35 @@ void run_server(bool spawn_client, const char* bin_path) { ...@@ -201,29 +206,35 @@ void run_server(bool spawn_client, const char* bin_path) {
CAF_TEST(test_broker) { CAF_TEST(test_broker) {
auto argv = caf::test::engine::argv(); auto argv = caf::test::engine::argv();
auto argc = caf::test::engine::argc(); auto argc = caf::test::engine::argc();
if (argc > 0) { uint16_t port = 0;
uint16_t port = 0; auto r = message_builder(argv, argv + argc).extract_opts({
auto r = message_builder(argv, argv + argc).extract_opts({ {"client-port,c", "set port for IO client", port},
{"client-port,c", "set port for IO client", port}, {"server,s", "run in server mode"},
{"server,s", "run in server mode"} {"use-asio", "use ASIO network backend (if available)"}
}); });
if (! r.error.empty() || r.opts.count("help") > 0 || ! r.remainder.empty()) { if (! r.error.empty() || r.opts.count("help") > 0 || ! r.remainder.empty()) {
cout << r.error << endl << endl << r.helptext << endl; cout << r.error << endl << endl << r.helptext << endl;
return; return;
} }
if (r.opts.count("client-port") > 0) { auto use_asio = r.opts.count("use-asio") > 0;
auto p = spawn(ping, 10); if (use_asio) {
CAF_MESSAGE("spawn_io_client..."); # ifdef CAF_USE_ASIO
auto cl = spawn_io_client(peer_fun, "localhost", port, p); CAF_MESSAGE("enable ASIO backend");
CAF_MESSAGE("spawn_io_client finished"); io::set_middleman<io::network::asio_multiplexer>();
anon_send(p, kickoff_atom::value, cl); # endif // CAF_USE_ASIO
CAF_MESSAGE("`kickoff_atom` has been send"); }
} else { if (r.opts.count("client-port") > 0) {
// run in server mode auto p = spawn(ping, 10);
run_server(false, argv[0]); CAF_MESSAGE("spawn_io_client...");
} auto cl = spawn_io_client(peer_fun, "localhost", port, p);
CAF_MESSAGE("spawn_io_client finished");
anon_send(p, kickoff_atom::value, cl);
CAF_MESSAGE("`kickoff_atom` has been send");
} else if (r.opts.count("server") > 0) {
// run in server mode
run_server(false, argv[0], use_asio);
} else { } else {
run_server(true, caf::test::engine::path()); run_server(true, caf::test::engine::path(), use_asio);
} }
CAF_MESSAGE("block on `await_all_actors_done`"); CAF_MESSAGE("block on `await_all_actors_done`");
await_all_actors_done(); await_all_actors_done();
......
...@@ -17,7 +17,7 @@ ...@@ -17,7 +17,7 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#define CAF_SUITE remote_actor #define CAF_SUITE io_dynamic_remote_actor
#include "caf/test/unit_test.hpp" #include "caf/test/unit_test.hpp"
#include <thread> #include <thread>
...@@ -34,6 +34,10 @@ ...@@ -34,6 +34,10 @@
#include "caf/detail/singletons.hpp" #include "caf/detail/singletons.hpp"
#include "caf/detail/run_program.hpp" #include "caf/detail/run_program.hpp"
#ifdef CAF_USE_ASIO
#include "caf/io/network/asio_multiplexer.hpp"
#endif // CAF_USE_ASIO
using namespace std; using namespace std;
using namespace caf; using namespace caf;
...@@ -438,9 +442,9 @@ private: ...@@ -438,9 +442,9 @@ private:
bool run_in_loop_; bool run_in_loop_;
}; };
void test_remote_actor(const char* app_path, bool run_remote_actor) { void test_remote_actor(const char* path, bool run_remote, bool use_asio) {
scoped_actor self; scoped_actor self;
auto serv = self->spawn<server, monitored>(! run_remote_actor); auto serv = self->spawn<server, monitored>(! run_remote);
// publish on two distinct ports and use the latter one afterwards // publish on two distinct ports and use the latter one afterwards
auto port1 = io::publish(serv, 0, "127.0.0.1"); auto port1 = io::publish(serv, 0, "127.0.0.1");
CAF_CHECK(port1 > 0); CAF_CHECK(port1 > 0);
...@@ -457,9 +461,10 @@ void test_remote_actor(const char* app_path, bool run_remote_actor) { ...@@ -457,9 +461,10 @@ void test_remote_actor(const char* app_path, bool run_remote_actor) {
CAF_CHECK(serv2 != invalid_actor && ! serv2->is_remote()); CAF_CHECK(serv2 != invalid_actor && ! serv2->is_remote());
CAF_CHECK(serv == serv2); CAF_CHECK(serv == serv2);
thread child; thread child;
if (run_remote_actor) { if (run_remote) {
child = detail::run_program(self, app_path, "-n", "-s", "remote_actor", child = detail::run_program(self, path, "-n", "-s", CAF_XSTR(CAF_SUITE),
"--", "-c", port2, "-c", port1, "-g", gport); "--", "-c", port2, "-c", port1, "-g", gport,
(use_asio ? "--use-asio" : ""));
} else { } else {
CAF_MESSAGE("please run client with: " CAF_MESSAGE("please run client with: "
<< "-c " << port2 << " -c " << port1 << " -g " << gport); << "-c " << port2 << " -c " << port1 << " -g " << gport);
...@@ -472,7 +477,7 @@ void test_remote_actor(const char* app_path, bool run_remote_actor) { ...@@ -472,7 +477,7 @@ void test_remote_actor(const char* app_path, bool run_remote_actor) {
); );
// wait until separate process (in sep. thread) finished execution // wait until separate process (in sep. thread) finished execution
self->await_all_other_actors_done(); self->await_all_other_actors_done();
if (run_remote_actor) { if (run_remote) {
child.join(); child.join();
self->receive( self->receive(
[](const std::string& output) { [](const std::string& output) {
...@@ -491,50 +496,55 @@ CAF_TEST(test_remote_actor) { ...@@ -491,50 +496,55 @@ CAF_TEST(test_remote_actor) {
announce<actor_vector>("actor_vector"); announce<actor_vector>("actor_vector");
cout << "this node is: " << to_string(caf::detail::singletons::get_node_id()) cout << "this node is: " << to_string(caf::detail::singletons::get_node_id())
<< endl; << endl;
if (argc > 0) { std::vector<uint16_t> ports;
std::vector<uint16_t> ports; uint16_t gport = 0;
uint16_t gport = 0; auto r = message_builder(argv, argv + argc).extract_opts({
auto r = message_builder(argv, argv + argc).extract_opts({ {"server,s", "run in server mode"},
{"server,s", "run in server mode"}, {"client-port,c", "add client port (two needed)", ports},
{"client-port,c", "add client port (two needed)", ports}, {"group-port,g", "set group port", gport},
{"group-port,g", "set group port", gport} {"use-asio", "use ASIO network backend (if available)"}
}); });
if (! r.error.empty() || r.opts.count("help") > 0 || ! r.remainder.empty()) { if (! r.error.empty() || r.opts.count("help") > 0 || ! r.remainder.empty()) {
cout << r.error << endl << endl << r.helptext << endl; cout << r.error << endl << endl << r.helptext << endl;
return;
}
auto use_asio = r.opts.count("use-asio") > 0;
if (use_asio) {
# ifdef CAF_USE_ASIO
CAF_MESSAGE("enable ASIO backend");
io::set_middleman<io::network::asio_multiplexer>();
# endif // CAF_USE_ASIO
}
if (r.opts.count("server") > 0) {
CAF_MESSAGE("don't run remote actor (server mode)");
test_remote_actor(argv[0], false, use_asio);
} else if (r.opts.count("client-port") > 0) {
if (ports.size() != 2 || r.opts.count("group-port") == 0) {
cerr << "*** expected exactly two ports and one group port" << endl
<< endl << r.helptext << endl;
return; return;
} }
if (r.opts.count("server") > 0) { scoped_actor self;
CAF_MESSAGE("don't run remote actor (server mode)"); auto serv = io::remote_actor("localhost", ports.front());
test_remote_actor(argv[0], false); auto serv2 = io::remote_actor("localhost", ports.back());
} else { // remote_actor is supposed to return the same server
if (ports.size() != 2 || r.opts.count("group-port") == 0) { // when connecting to the same host again
cerr << "*** expected exactly two ports and one group port" << endl {
<< endl << r.helptext << endl; CAF_CHECK(serv == io::remote_actor("localhost", ports.front()));
return; CAF_CHECK(serv2 == io::remote_actor("127.0.0.1", ports.back()));
}
scoped_actor self;
auto serv = io::remote_actor("localhost", ports.front());
auto serv2 = io::remote_actor("localhost", ports.back());
// remote_actor is supposed to return the same server
// when connecting to the same host again
{
CAF_CHECK(serv == io::remote_actor("localhost", ports.front()));
CAF_CHECK(serv2 == io::remote_actor("127.0.0.1", ports.back()));
}
// connect to published groups
auto grp = io::remote_group("whatever", "127.0.0.1", gport);
auto c = self->spawn<client, monitored>(serv);
self->receive(
[&](const down_msg& dm) {
CAF_CHECK_EQUAL(dm.source, c);
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
}
);
grp->stop();
} }
} // connect to published groups
else { auto grp = io::remote_group("whatever", "127.0.0.1", gport);
test_remote_actor(caf::test::engine::path(), true); auto c = self->spawn<client, monitored>(serv);
self->receive(
[&](const down_msg& dm) {
CAF_CHECK_EQUAL(dm.source, c);
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
}
);
grp->stop();
} else {
test_remote_actor(caf::test::engine::path(), true, use_asio);
} }
await_all_actors_done(); await_all_actors_done();
shutdown(); shutdown();
......
...@@ -17,7 +17,7 @@ ...@@ -17,7 +17,7 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#define CAF_SUITE typed_broker #define CAF_SUITE io_typed_broker
#include "caf/test/unit_test.hpp" #include "caf/test/unit_test.hpp"
#include <memory> #include <memory>
...@@ -32,6 +32,10 @@ ...@@ -32,6 +32,10 @@
#include "caf/detail/run_program.hpp" #include "caf/detail/run_program.hpp"
#ifdef CAF_USE_ASIO
#include "caf/io/network/asio_multiplexer.hpp"
#endif // CAF_USE_ASIO
using namespace std; using namespace std;
using namespace caf; using namespace caf;
using namespace caf::io; using namespace caf::io;
...@@ -177,7 +181,7 @@ acceptor::behavior_type acceptor_fun(acceptor::broker_pointer self, ...@@ -177,7 +181,7 @@ acceptor::behavior_type acceptor_fun(acceptor::broker_pointer self,
}; };
} }
void run_server(bool spawn_client, const char* bin_path) { void run_server(bool spawn_client, const char* bin_path, bool use_asio) {
scoped_actor self; scoped_actor self;
auto serv = spawn_io_typed(acceptor_fun, spawn(pong)); auto serv = spawn_io_typed(acceptor_fun, spawn(pong));
self->sync_send(serv, publish_atom::value).await( self->sync_send(serv, publish_atom::value).await(
...@@ -185,7 +189,8 @@ void run_server(bool spawn_client, const char* bin_path) { ...@@ -185,7 +189,8 @@ void run_server(bool spawn_client, const char* bin_path) {
CAF_MESSAGE("server is running on port " << port); CAF_MESSAGE("server is running on port " << port);
if (spawn_client) { if (spawn_client) {
auto child = detail::run_program(self, bin_path, "-n", "-s", auto child = detail::run_program(self, bin_path, "-n", "-s",
"typed_broker", "--", "-c", port); CAF_XSTR(CAF_SUITE), "--", "-c", port,
(use_asio ? "--use-asio" : ""));
CAF_MESSAGE("block till child process has finished"); CAF_MESSAGE("block till child process has finished");
child.join(); child.join();
} }
...@@ -205,30 +210,35 @@ void run_server(bool spawn_client, const char* bin_path) { ...@@ -205,30 +210,35 @@ void run_server(bool spawn_client, const char* bin_path) {
CAF_TEST(test_typed_broker) { CAF_TEST(test_typed_broker) {
auto argv = caf::test::engine::argv(); auto argv = caf::test::engine::argv();
auto argc = caf::test::engine::argc(); auto argc = caf::test::engine::argc();
if (argc > 0) { uint16_t port = 0;
uint16_t port = 0; auto r = message_builder(argv, argv + argc).extract_opts({
auto r = message_builder(argv, argv + argc).extract_opts({ {"client-port,c", "set port for IO client", port},
{"client-port,c", "set port for IO client", port}, {"server,s", "run in server mode"},
{"server,s", "run in server mode"} {"use-asio", "use ASIO network backend (if available)"}
}); });
if (! r.error.empty() || r.opts.count("help") > 0 || ! r.remainder.empty()) { if (! r.error.empty() || r.opts.count("help") > 0 || ! r.remainder.empty()) {
cout << r.error << endl << endl << r.helptext << endl; cout << r.error << endl << endl << r.helptext << endl;
return; return;
} }
if (r.opts.count("client-port") > 0) { auto use_asio = r.opts.count("use-asio") > 0;
auto p = spawn(ping, 10); if (use_asio) {
CAF_MESSAGE("spawn_io_client_typed..."); # ifdef CAF_USE_ASIO
auto cl = spawn_io_client_typed(peer_fun, "localhost", port, p); CAF_MESSAGE("enable ASIO backend");
CAF_MESSAGE("spawn_io_client_typed finished"); io::set_middleman<io::network::asio_multiplexer>();
anon_send(p, kickoff_atom::value, cl); # endif // CAF_USE_ASIO
CAF_MESSAGE("`kickoff_atom` has been send");
} else {
// run in server mode
run_server(false, argv[0]);
}
} }
else { if (r.opts.count("client-port") > 0) {
run_server(true, caf::test::engine::path()); auto p = spawn(ping, 10);
CAF_MESSAGE("spawn_io_client_typed...");
auto cl = spawn_io_client_typed(peer_fun, "localhost", port, p);
CAF_MESSAGE("spawn_io_client_typed finished");
anon_send(p, kickoff_atom::value, cl);
CAF_MESSAGE("`kickoff_atom` has been send");
} else if (r.opts.count("server") > 0) {
// run in server mode
run_server(false, argv[0], use_asio);
} else {
run_server(true, caf::test::engine::path(), use_asio);
} }
CAF_MESSAGE("block on `await_all_actors_done`"); CAF_MESSAGE("block on `await_all_actors_done`");
await_all_actors_done(); await_all_actors_done();
......
...@@ -17,7 +17,7 @@ ...@@ -17,7 +17,7 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#define CAF_SUITE typed_remote_actor #define CAF_SUITE io_typed_remote_actor
#include "caf/test/unit_test.hpp" #include "caf/test/unit_test.hpp"
#include <thread> #include <thread>
...@@ -32,6 +32,10 @@ ...@@ -32,6 +32,10 @@
#include "caf/detail/run_program.hpp" #include "caf/detail/run_program.hpp"
#ifdef CAF_USE_ASIO
#include "caf/io/network/asio_multiplexer.hpp"
#endif // CAF_USE_ASIO
using namespace std; using namespace std;
using namespace caf; using namespace caf;
...@@ -99,32 +103,38 @@ CAF_TEST(test_typed_remote_actor) { ...@@ -99,32 +103,38 @@ CAF_TEST(test_typed_remote_actor) {
auto argc = caf::test::engine::argc(); auto argc = caf::test::engine::argc();
announce<ping>("ping", &ping::value); announce<ping>("ping", &ping::value);
announce<pong>("pong", &pong::value); announce<pong>("pong", &pong::value);
if (argc > 0) { uint16_t port = 0;
uint16_t port = 0; auto r = message_builder(argv, argv + argc).extract_opts({
auto r = message_builder(argv, argv + argc).extract_opts({ {"client-port,c", "set port for client", port},
{"client-port,c", "set port for client", port}, {"server,s", "run in server mode"},
{"server,s", "run in server mode"} {"use-asio", "use ASIO network backend (if available)"}
}); });
if (! r.error.empty() || r.opts.count("help") > 0 || ! r.remainder.empty()) { if (! r.error.empty() || r.opts.count("help") > 0 || ! r.remainder.empty()) {
cout << r.error << endl << endl << r.helptext << endl; cout << r.error << endl << endl << r.helptext << endl;
return; return;
} }
if (r.opts.count("client-port") > 0) { auto use_asio = r.opts.count("use-asio") > 0;
CAF_MESSAGE("run in client mode"); if (use_asio) {
run_client("localhost", port); # ifdef CAF_USE_ASIO
} else { CAF_MESSAGE("enable ASIO backend");
CAF_MESSAGE("run in server mode"); io::set_middleman<io::network::asio_multiplexer>();
run_server(); # endif // CAF_USE_ASIO
}
} }
else { if (r.opts.count("client-port") > 0) {
CAF_MESSAGE("run in client mode");
run_client("localhost", port);
} else if (r.opts.count("server") > 0) {
CAF_MESSAGE("run in server mode");
run_server();
} else {
auto port = run_server(); auto port = run_server();
// execute client_part() in a separate process, // execute client_part() in a separate process,
// connected via localhost socket // connected via localhost socket
scoped_actor self; scoped_actor self;
auto child = detail::run_program(self, caf::test::engine::path(), "-n", auto child = detail::run_program(self, caf::test::engine::path(), "-n",
"-s", "typed_remote_actor", "-s", CAF_XSTR(CAF_SUITE),
"--", "-c", port); "--", "-c", port,
(use_asio ? "--use-asio" : ""));
CAF_MESSAGE("block till child process has finished"); CAF_MESSAGE("block till child process has finished");
child.join(); child.join();
self->await_all_other_actors_done(); self->await_all_other_actors_done();
......
...@@ -17,7 +17,7 @@ ...@@ -17,7 +17,7 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#define CAF_SUITE unpublish #define CAF_SUITE io_unpublish
#include "caf/test/unit_test.hpp" #include "caf/test/unit_test.hpp"
#include <thread> #include <thread>
...@@ -26,6 +26,10 @@ ...@@ -26,6 +26,10 @@
#include "caf/all.hpp" #include "caf/all.hpp"
#include "caf/io/all.hpp" #include "caf/io/all.hpp"
#ifdef CAF_USE_ASIO
#include "caf/io/network/asio_multiplexer.hpp"
#endif // CAF_USE_ASIO
using namespace caf; using namespace caf;
namespace { namespace {
...@@ -56,6 +60,14 @@ void test_invalid_unpublish(const actor& published, uint16_t port) { ...@@ -56,6 +60,14 @@ void test_invalid_unpublish(const actor& published, uint16_t port) {
} }
CAF_TEST(unpublishing) { CAF_TEST(unpublishing) {
auto argv = caf::test::engine::argv();
auto argc = caf::test::engine::argc();
if (argc == 1 && strcmp(argv[0], "--use-asio") == 0) {
# ifdef CAF_USE_ASIO
CAF_MESSAGE("enable ASIO backend");
io::set_middleman<io::network::asio_multiplexer>();
# endif // CAF_USE_ASIO
}
{ // scope for local variables { // scope for local variables
auto d = spawn<dummy>(); auto d = spawn<dummy>();
auto port = io::publish(d, 0); auto port = io::publish(d, 0);
......
...@@ -18,3 +18,7 @@ ...@@ -18,3 +18,7 @@
******************************************************************************/ ******************************************************************************/
#include "caf/test/unit_test_impl.hpp" #include "caf/test/unit_test_impl.hpp"
#ifdef CAF_USE_ASIO
#include "caf/io/network/asio_multiplexer_impl.hpp"
#endif // CAF_USE_ASIO
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment