Unverified Commit 86801cb1 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #1132

Remove obsolete files, fix serial_reply unit test suite
parents ecffc163 1ae0863d
......@@ -303,6 +303,7 @@ caf_add_test_suites(caf-core-test
request_timeout
result
selective_streaming
serial_reply
serialization
settings
simple_timeout
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* 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/decorator/splitter.hpp"
#include "caf/actor_system.hpp"
#include "caf/default_attachable.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/response_promise.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/detail/disposer.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
namespace caf::decorator {
namespace {
struct splitter_state {
response_promise rp;
message result;
size_t pending;
};
behavior fan_out_fan_in(stateful_actor<splitter_state>* self,
const std::vector<strong_actor_ptr>& workers) {
auto f = [=](local_actor*, message& msg) -> result<message> {
self->state.rp = self->make_response_promise();
self->state.pending = workers.size();
// request().await() has LIFO ordering
for (auto i = workers.rbegin(); i != workers.rend(); ++i)
// TODO: maybe infer some useful timeout or use config parameter?
self->request(actor_cast<actor>(*i), infinite, msg)
.await(
[=]() {
// nop
},
[=](error& err) mutable {
if (err == sec::unexpected_response) {
self->state.result += std::move(err.context());
if (--self->state.pending == 0)
self->state.rp.deliver(std::move(self->state.result));
} else {
self->state.rp.deliver(err);
self->quit();
}
});
return delegated<message>{};
};
self->set_default_handler(f);
return [] {
// nop
};
}
} // namespace
splitter::splitter(std::vector<strong_actor_ptr> workers,
message_types_set msg_types)
: monitorable_actor(actor_config{}.add_flag(is_actor_dot_decorator_flag)),
num_workers(workers.size()),
workers_(std::move(workers)),
msg_types_(std::move(msg_types)) {
// composed actor has dependency on constituent actors by default;
// if either constituent actor is already dead upon establishing
// the dependency, the actor is spawned dead
auto addr = address();
for (auto& worker : workers_)
worker->get()->attach(
default_attachable::make_monitor(actor_cast<actor_addr>(worker), addr));
}
void splitter::enqueue(mailbox_element_ptr what, execution_unit* context) {
auto down_msg_handler = [&](down_msg& dm) {
// quit if any worker fails
cleanup(std::move(dm.reason), context);
};
if (handle_system_message(*what, context, false, down_msg_handler))
return;
std::vector<strong_actor_ptr> workers;
workers.reserve(num_workers);
error fail_state;
shared_critical_section([&] {
workers = workers_;
fail_state = fail_state_;
});
if (workers.empty()) {
bounce(what, fail_state);
return;
}
auto helper = context->system().spawn(fan_out_fan_in, std::move(workers));
helper->enqueue(std::move(what), context);
}
splitter::message_types_set splitter::message_types() const {
return msg_types_;
}
} // namespace caf::decorator
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* 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/type_id.hpp"
namespace caf {
type_id::type_id() {
// nop
}
type_id::~type_id() {
// nop
}
} // namespace caf
......@@ -274,6 +274,13 @@ CAF_BEGIN_TYPE_ID_BLOCK(core_test, caf::first_custom_type_id)
ADD_ATOM(abc_atom)
ADD_ATOM(get_state_atom)
ADD_ATOM(name_atom)
ADD_ATOM(sub0_atom)
ADD_ATOM(sub1_atom)
ADD_ATOM(sub2_atom)
ADD_ATOM(sub3_atom)
ADD_ATOM(sub4_atom)
ADD_ATOM(hi_atom)
ADD_ATOM(ho_atom)
CAF_END_TYPE_ID_BLOCK(core_test)
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#define CAF_SUITE detail.message_builder_element
#include "caf/detail/message_builder_element.hpp"
#include "caf/test/dsl.hpp"
using namespace caf;
namespace {
struct fixture {};
} // namespace
CAF_TEST_FIXTURE_SCOPE(message_builder_element_tests, fixture)
CAF_TEST(todo) {
// implement me
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -29,11 +29,13 @@ CAF_TEST(test_serial_reply) {
actor_system system{cfg};
auto mirror_behavior = [=](event_based_actor* self) -> behavior {
self->set_default_handler(reflect);
return {[] {
// nop
}};
return {
[] {
// nop
},
};
};
auto master = system.spawn([=](event_based_actor* self) {
auto master = system.spawn([=](event_based_actor* self) -> behavior {
CAF_MESSAGE("ID of master: " << self->id());
// spawn 5 mirror actors
auto c0 = self->spawn<linked>(mirror_behavior);
......@@ -41,36 +43,36 @@ CAF_TEST(test_serial_reply) {
auto c2 = self->spawn<linked>(mirror_behavior);
auto c3 = self->spawn<linked>(mirror_behavior);
auto c4 = self->spawn<linked>(mirror_behavior);
self->become([=](int) mutable {
auto rp = self->make_response_promise();
CAF_MESSAGE("received 'hi there'");
self->request(c0, infinite, sub0_atom::value)
.then([=](sub0_atom) mutable {
return {
[=](hi_atom) mutable {
auto rp = self->make_response_promise();
CAF_MESSAGE("received 'hi there'");
self->request(c0, infinite, sub0_atom_v).then([=](sub0_atom) mutable {
CAF_MESSAGE("received 'sub0'");
self->request(c1, infinite, sub1_atom::value)
.then([=](sub1_atom) mutable {
CAF_MESSAGE("received 'sub1'");
self->request(c2, infinite, sub2_atom::value)
.then([=](sub2_atom) mutable {
CAF_MESSAGE("received 'sub2'");
self->request(c3, infinite, sub3_atom::value)
.then([=](sub3_atom) mutable {
CAF_MESSAGE("received 'sub3'");
self->request(c4, infinite, sub4_atom::value)
.then([=](sub4_atom) mutable {
CAF_MESSAGE("received 'sub4'");
rp.deliver(ho_atom::value);
});
});
});
});
self->request(c1, infinite, sub1_atom_v).then([=](sub1_atom) mutable {
CAF_MESSAGE("received 'sub1'");
self->request(c2, infinite, sub2_atom_v)
.then([=](sub2_atom) mutable {
CAF_MESSAGE("received 'sub2'");
self->request(c3, infinite, sub3_atom_v)
.then([=](sub3_atom) mutable {
CAF_MESSAGE("received 'sub3'");
self->request(c4, infinite, sub4_atom_v)
.then([=](sub4_atom) mutable {
CAF_MESSAGE("received 'sub4'");
rp.deliver(ho_atom_v);
});
});
});
});
});
});
},
};
});
scoped_actor self{system};
CAF_MESSAGE("ID of main: " << self->id());
self->request(master, infinite, hi_atom::value)
self->request(master, infinite, hi_atom_v)
.receive([](ho_atom) { CAF_MESSAGE("received 'ho'"); },
[&](const error& err) { CAF_ERROR("Error: " << to_string(err)); });
[&](const error& err) { CAF_ERROR("Error: " << err); });
CAF_REQUIRE(self->mailbox().empty());
}
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