Commit de326fd9 authored by Dominik Charousset's avatar Dominik Charousset

fixed actor_ostream

parent 8d1b0361
......@@ -136,6 +136,7 @@ set(LIBCPPA_SRC
src/actor_namespace.cpp
src/actor_proxy.cpp
src/actor_registry.cpp
src/actor_ostream.cpp
src/algorithm.cpp
src/any_tuple.cpp
src/atom.cpp
......
......@@ -338,3 +338,5 @@ src/blocking_untyped_actor.cpp
cppa/policy/policies.hpp
cppa/detail/response_future_util.hpp
cppa/system_messages.hpp
cppa/actor_ostream.hpp
src/actor_ostream.cpp
......@@ -69,6 +69,25 @@ class actor : util::comparable<actor> {
public:
// common_actor_ops does not provide a virtual destructor -> no new members
class ops : public common_actor_ops {
friend class actor;
typedef common_actor_ops super;
public:
ops() = default;
void enqueue(const message_header& hdr, any_tuple msg) const;
private:
inline ops(abstract_actor_ptr ptr) : super(std::move(ptr)) { }
};
actor() = default;
template<typename T>
......@@ -95,13 +114,11 @@ class actor : util::comparable<actor> {
inline bool operator!() const;
void enqueue(const message_header& hdr, any_tuple msg) const;
inline common_actor_ops* operator->() const {
inline ops* operator->() const {
// this const cast is safe, because common_actor_ops cannot be
// modified anyways and the offered operations are intended to
// be called on const elements
return const_cast<common_actor_ops*>(&m_ops);
return const_cast<ops*>(&m_ops);
}
intptr_t compare(const actor& other) const;
......@@ -110,7 +127,7 @@ class actor : util::comparable<actor> {
actor(abstract_actor*);
common_actor_ops m_ops;
ops m_ops;
};
......
......@@ -78,6 +78,10 @@ class actor_addr : util::comparable<actor_addr>
explicit operator bool() const;
bool operator!() const;
inline bool valid() const {
return static_cast<bool>(*this);
}
intptr_t compare(const actor& other) const;
intptr_t compare(const actor_addr& other) const;
intptr_t compare(const abstract_actor* other) const;
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_ACTOR_OSTREAM_HPP
#define CPPA_ACTOR_OSTREAM_HPP
#include "cppa/actor.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/to_string.hpp"
namespace cppa {
class local_actor;
class scoped_actor;
class actor_ostream {
public:
typedef actor_ostream& (*fun_type)(actor_ostream&);
actor_ostream(actor_ostream&&) = default;
actor_ostream(const actor_ostream&) = default;
actor_ostream& operator=(actor_ostream&&) = default;
actor_ostream& operator=(const actor_ostream&) = default;
explicit actor_ostream(local_actor* self);
actor_ostream& write(std::string arg);
actor_ostream& flush();
inline actor_ostream& operator<<(std::string arg) {
return write(move(arg));
}
inline actor_ostream& operator<<(const any_tuple& arg) {
return write(cppa::to_string(arg));
}
// disambiguate between conversion to string and to any_tuple
inline actor_ostream& operator<<(const char* arg) {
return *this << std::string{arg};
}
template<typename T>
inline typename std::enable_if<
!std::is_convertible<T, std::string>::value
&& !std::is_convertible<T, any_tuple>::value,
actor_ostream&
>::type
operator<<(T&& arg) {
return write(std::to_string(std::forward<T>(arg)));
}
inline actor_ostream& operator<<(actor_ostream::fun_type f) {
return f(*this);
}
private:
local_actor* m_self;
actor m_printer;
};
inline actor_ostream aout(local_actor* self) {
return actor_ostream{self};
}
actor_ostream aout(scoped_actor& self);
} // namespace cppa
namespace std {
// provide convenience overlaods for aout; implemented in logging.cpp
cppa::actor_ostream& endl(cppa::actor_ostream& o);
cppa::actor_ostream& flush(cppa::actor_ostream& o);
} // namespace std
#endif // CPPA_ACTOR_OSTREAM_HPP
......@@ -81,9 +81,9 @@ class common_actor_ops {
*/
bool is_remote() const;
private:
protected:
common_actor_ops(abstract_actor_ptr ptr) : m_ptr(std::move(ptr)) { }
inline common_actor_ops(abstract_actor_ptr ptr) : m_ptr(std::move(ptr)) { }
abstract_actor_ptr m_ptr;
......
......@@ -57,6 +57,7 @@
#include "cppa/local_actor.hpp"
#include "cppa/scoped_actor.hpp"
#include "cppa/spawn_options.hpp"
#include "cppa/actor_ostream.hpp"
#include "cppa/untyped_actor.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/system_messages.hpp"
......@@ -588,66 +589,22 @@ actor spawn_io_server(F fun, uint16_t port, Ts&&... args) {
*/
void shutdown(); // note: implemented in singleton_manager.cpp
struct actor_ostream {
typedef const actor_ostream& (*fun_type)(const actor_ostream&);
constexpr actor_ostream() { }
inline const actor_ostream& write(std::string arg) const {
send_as(invalid_actor, get_scheduler()->printer(), atom("add"), move(arg));
return *this;
}
inline const actor_ostream& flush() const {
send_as(invalid_actor, get_scheduler()->printer(), atom("flush"));
return *this;
}
};
namespace { constexpr actor_ostream aout; }
inline const actor_ostream& operator<<(const actor_ostream& o, std::string arg) {
return o.write(move(arg));
}
inline const actor_ostream& operator<<(const actor_ostream& o, const any_tuple& arg) {
return o.write(cppa::to_string(arg));
}
// disambiguate between conversion to string and to any_tuple
inline const actor_ostream& operator<<(const actor_ostream& o, const char* arg) {
return o << std::string{arg};
}
template<typename T>
inline typename std::enable_if<
!std::is_convertible<T, std::string>::value
&& !std::is_convertible<T, any_tuple>::value,
const actor_ostream&
>::type
operator<<(const actor_ostream& o, T&& arg) {
return o.write(std::to_string(std::forward<T>(arg)));
}
inline const actor_ostream& operator<<(const actor_ostream& o, actor_ostream::fun_type f) {
return f(o);
}
} // namespace cppa
namespace std {
// allow actor_ptr to be used in hash maps
// allow actor and actor_addr to be used in hash maps
template<>
struct hash<cppa::actor> {
inline size_t operator()(const cppa::actor& ref) const {
return static_cast<size_t>(ref->id());
}
};
// provide convenience overlaods for aout; implemented in logging.cpp
const cppa::actor_ostream& endl(const cppa::actor_ostream& o);
const cppa::actor_ostream& flush(const cppa::actor_ostream& o);
template<>
struct hash<cppa::actor_addr> {
inline size_t operator()(const cppa::actor_addr& ref) const {
return static_cast<size_t>(ref->id());
}
};
} // namespace std
#endif // CPPA_HPP
......@@ -96,7 +96,7 @@ class scheduler {
util::duration{rel_time},
std::move(hdr),
std::move(data));
delayed_send_helper().enqueue(message_header{}, std::move(tup));
delayed_send_helper()->enqueue(message_header{}, std::move(tup));
}
template<typename Duration, typename... Data>
......@@ -108,7 +108,7 @@ class scheduler {
util::duration{rel_time},
std::move(hdr),
std::move(data));
delayed_send_helper().enqueue(message_header{}, std::move(tup));
delayed_send_helper()->enqueue(message_header{}, std::move(tup));
}
private:
......
......@@ -109,10 +109,6 @@ constexpr size_t num_curl_workers = 10;
constexpr int min_req_interval = 10;
constexpr int max_req_interval = 300;
const actor_ostream& print(const char* color_code, const char* actor_name) {
return aout << color_code << actor_name;// << " (id = " << self->id() << "): ";
}
} // namespace <anonymous>
// provides print utility and each base_actor has a parent
......@@ -123,10 +119,11 @@ class base_actor : public untyped_actor {
base_actor(actor parent, std::string name, std::string color_code)
: m_parent(std::move(parent))
, m_name(std::move(name))
, m_color(std::move(color_code)) { }
, m_color(std::move(color_code))
, m_out(this) { }
inline const actor_ostream& print() const {
return ::print(m_color.c_str(), m_name.c_str());
inline actor_ostream& print() {
return m_out << m_color << m_name << " (id = " << id() << "): ";
}
void on_exit() override {
......@@ -139,6 +136,7 @@ class base_actor : public untyped_actor {
std::string m_name;
std::string m_color;
actor_ostream m_out;
};
......@@ -388,21 +386,25 @@ int main() {
set_sighandler();
// initialize CURL
curl_global_init(CURL_GLOBAL_DEFAULT);
{ // lifetime scope of self
scoped_actor self;
// spawn client and curl_master
auto master = spawn<curl_master, detached>();
spawn<client, detached>(master);
auto master = self->spawn<curl_master, detached>();
self->spawn<client, detached>(master);
// poll CTRL+C flag every second
while (!shutdown_flag) { sleep(1); }
aout << color::cyan << "received CTRL+C" << color::reset_endl;
aout(self) << color::cyan << "received CTRL+C" << color::reset_endl;
// shutdown actors
anon_send_exit(master, exit_reason::user_shutdown);
// await actors
act.sa_handler = [](int) { abort(); };
set_sighandler();
aout << color::cyan
<< "await CURL; this may take a while (press CTRL+C again to abort)"
aout(self) << color::cyan
<< "await CURL; this may take a while "
"(press CTRL+C again to abort)"
<< color::reset_endl;
await_all_actors_done();
self->await_all_other_actors_done();
}
// shutdown libcppa
shutdown();
// shutdown CURL
......
......@@ -11,7 +11,7 @@ behavior mirror(untyped_actor* self) {
// invoke this lambda expression if we receive a string
on_arg_match >> [=](const string& what) -> string {
// prints "Hello World!" via aout (thread-safe cout wrapper)
aout << what << endl;
aout(self) << what << endl;
// terminates this actor ('become' otherwise loops forever)
self->quit();
// reply "!dlroW olleH"
......@@ -26,7 +26,7 @@ void hello_world(untyped_actor* self, const actor& buddy) {
// ... and wait for a response
on_arg_match >> [=](const string& what) {
// prints "!dlroW olleH"
aout << what << endl;
aout(self) << what << endl;
}
);
}
......
......@@ -54,7 +54,7 @@ void tester(untyped_actor* self, const actor& testee) {
self->link_to(testee);
// will be invoked if we receive an unexpected response message
self->on_sync_failure([=] {
aout << "AUT (actor under test) failed" << endl;
aout(self) << "AUT (actor under test) failed" << endl;
self->quit(exit_reason::user_shutdown);
});
// first test: 2 + 1 = 3
......@@ -64,7 +64,8 @@ void tester(untyped_actor* self, const actor& testee) {
self->sync_send(testee, atom("minus"), 2, 1).then(
on(atom("result"), 1) >> [=] {
// both tests succeeded
aout << "AUT (actor under test) seems to be ok" << endl;
aout(self) << "AUT (actor under test) seems to be ok"
<< endl;
self->send(testee, atom("quit"));
}
);
......
......@@ -88,7 +88,7 @@ struct philosopher : untyped_actor {
behavior waiting_for(const actor& what) {
return (
on(atom("taken"), what) >> [=] {
aout << name
aout(this) << name
<< " has picked up chopsticks with IDs "
<< left->id()
<< " and "
......@@ -146,7 +146,7 @@ struct philosopher : untyped_actor {
send(left, atom("put"), this);
send(right, atom("put"), this);
delayed_send(this, seconds(5), atom("eat"));
aout << name
aout(this) << name
<< " puts down his chopsticks and starts to think\n";
become(thinking);
}
......@@ -159,7 +159,7 @@ struct philosopher : untyped_actor {
// philosophers start to think after receiving {think}
return (
on(atom("think")) >> [=] {
aout << name << " starts to think\n";
aout(this) << name << " starts to think\n";
delayed_send(this, seconds(5), atom("eat"));
become(thinking);
}
......@@ -168,21 +168,26 @@ struct philosopher : untyped_actor {
};
int main(int, char**) {
void dining_philosophers() {
scoped_actor self;
// create five chopsticks
aout << "chopstick ids are:";
aout(self) << "chopstick ids are:";
std::vector<actor> chopsticks;
for (size_t i = 0; i < 5; ++i) {
chopsticks.push_back(spawn(chopstick));
aout << " " << chopsticks.back()->id();
aout(self) << " " << chopsticks.back()->id();
}
aout << endl;
aout(self) << endl;
// spawn five philosophers
std::vector<std::string> names { "Plato", "Hume", "Kant",
"Nietzsche", "Descartes" };
for (size_t i = 0; i < 5; ++i) {
spawn<philosopher>(names[i], chopsticks[i], chopsticks[(i+1)%5]);
}
}
int main(int, char**) {
dining_philosophers();
// real philosophers are never done
await_all_actors_done();
shutdown();
......
......@@ -53,22 +53,23 @@ void client_bhvr(untyped_actor* self, const string& host, uint16_t port, const a
// recover from sync failures by trying to reconnect to server
if (!self->has_sync_failure_handler()) {
self->on_sync_failure([=] {
aout << "*** lost connection to " << host << ":" << port << endl;
aout(self) << "*** lost connection to " << host
<< ":" << port << endl;
client_bhvr(self, host, port, nullptr);
});
}
// connect to server if needed
if (!server) {
aout << "*** try to connect to " << host << ":" << port << endl;
aout(self) << "*** try to connect to " << host << ":" << port << endl;
try {
auto new_serv = remote_actor(host, port);
self->monitor(new_serv);
aout << "reconnection succeeded" << endl;
aout(self) << "reconnection succeeded" << endl;
client_bhvr(self, host, port, new_serv);
return;
}
catch (exception&) {
aout << "connection failed, try again in 3s" << endl;
aout(self) << "connection failed, try again in 3s" << endl;
self->delayed_send(self, chrono::seconds(3), atom("reconnect"));
}
}
......@@ -76,7 +77,7 @@ void client_bhvr(untyped_actor* self, const string& host, uint16_t port, const a
on_arg_match.when(_x1.in({atom("plus"), atom("minus")}) && gval(server) != nullptr) >> [=](atom_value op, int lhs, int rhs) {
self->sync_send_tuple(server, self->last_dequeued()).then(
on(atom("result"), arg_match) >> [=](int result) {
aout << lhs << " "
aout(self) << lhs << " "
<< to_string(op) << " "
<< rhs << " = "
<< result << endl;
......@@ -84,11 +85,12 @@ void client_bhvr(untyped_actor* self, const string& host, uint16_t port, const a
);
},
on_arg_match >> [=](const down_msg&) {
aout << "*** server down, try to reconnect ..." << endl;
aout(self) << "*** server down, try to reconnect ..." << endl;
client_bhvr(self, host, port, nullptr);
},
on(atom("rebind"), arg_match) >> [=](const string& host, uint16_t port) {
aout << "*** rebind to new server: " << host << ":" << port << endl;
aout(self) << "*** rebind to new server: "
<< host << ":" << port << endl;
client_bhvr(self, host, port, nullptr);
},
on(atom("reconnect")) >> [=] {
......@@ -99,7 +101,7 @@ void client_bhvr(untyped_actor* self, const string& host, uint16_t port, const a
void client_repl(const string& host, uint16_t port) {
// keeps track of requests and tries to reconnect on server failures
aout << "Usage:\n"
cout << "Usage:\n"
"quit Quit the program\n"
"<x> + <y> Calculate <x>+<y> and print result\n"
"<x> - <y> Calculate <x>-<y> and print result\n"
......@@ -126,16 +128,16 @@ void client_repl(const string& host, uint16_t port) {
anon_send(client, atom("rebind"), move(host), port);
}
else {
aout << lport << " is not a valid port" << endl;
cout << lport << " is not a valid port" << endl;
}
}
catch (std::exception& e) {
aout << "\"" << sport << "\" is not an unsigned integer"
cout << "\"" << sport << "\" is not an unsigned integer"
<< endl;
}
},
others() >> [] {
aout << "*** usage: connect <host> <port>" << endl;
cout << "*** usage: connect <host> <port>" << endl;
}
);
}
......@@ -143,7 +145,7 @@ void client_repl(const string& host, uint16_t port) {
auto toint = [](const string& str) -> optional<int> {
try { return {std::stoi(str)}; }
catch (std::exception&) {
aout << "\"" << str << "\" is not an integer" << endl;
cout << "\"" << str << "\" is not an integer" << endl;
return none;
}
};
......@@ -163,7 +165,7 @@ void client_repl(const string& host, uint16_t port) {
}
}
else if (!success) {
aout << "*** invalid format; usage: <x> [+|-] <y>" << endl;
cout << "*** invalid format; usage: <x> [+|-] <y>" << endl;
}
}
}
......
......@@ -42,7 +42,7 @@ bool operator==(const foo& lhs, const foo& rhs) {
void testee(untyped_actor* self) {
self->become (
on<foo>() >> [=](const foo& val) {
aout << "foo("
aout(self) << "foo("
<< val.a() << ", "
<< val.b() << ")"
<< endl;
......
......@@ -49,7 +49,7 @@ typedef void (foo::*foo_setter)(int);
void testee(untyped_actor* self) {
self->become (
on<foo>() >> [=](const foo& val) {
aout << "foo("
aout(self) << "foo("
<< val.a() << ", "
<< val.b() << ")"
<< endl;
......
......@@ -86,7 +86,7 @@ void testee(untyped_actor* self, size_t remaining) {
};
self->become (
on<bar>() >> [=](const bar& val) {
aout << "bar(foo("
aout(self) << "bar(foo("
<< val.f.a() << ", "
<< val.f.b() << "), "
<< val.i << ")"
......@@ -95,7 +95,7 @@ void testee(untyped_actor* self, size_t remaining) {
},
on<baz>() >> [=](const baz& val) {
// prints: baz ( foo ( 1, 2 ), bar ( foo ( 3, 4 ), 5 ) )
aout << to_string(object::from(val)) << endl;
aout(self) << to_string(object::from(val)) << endl;
set_next_behavior();
}
);
......
......@@ -39,18 +39,18 @@ struct tree_node {
void print() const {
// format is: value { children0, children1, ..., childrenN }
// e.g., 10 { 20 { 21, 22 }, 30 }
aout << value;
cout << value;
if (children.empty() == false) {
aout << " { ";
cout << " { ";
auto begin = children.begin();
auto end = children.end();
for (auto i = begin; i != end; ++i) {
if (i != begin) {
aout << ", ";
cout << ", ";
}
i->print();
}
aout << " } ";
cout << " } ";
}
}
......@@ -62,9 +62,9 @@ struct tree {
// print tree to stdout
void print() const {
aout << "tree::print: ";
cout << "tree::print: ";
root.print();
aout << endl;
cout << endl;
}
};
......@@ -140,7 +140,7 @@ void testee(untyped_actor* self, size_t remaining) {
on_arg_match >> [=](const tree& tmsg) {
// prints the tree in its serialized format:
// @<> ( { tree ( 0, { 10, { 11, { }, 12, { }, 13, { } }, 20, { 21, { }, 22, { } } } ) } )
aout << "to_string(self->last_dequeued()): "
cout << "to_string(self->last_dequeued()): "
<< to_string(self->last_dequeued())
<< endl;
// prints the tree using the print member function:
......@@ -150,7 +150,7 @@ void testee(untyped_actor* self, size_t remaining) {
},
on_arg_match >> [=](const tree_vector& trees) {
// prints "received 2 trees"
aout << "received " << trees.size() << " trees" << endl;
cout << "received " << trees.size() << " trees" << endl;
// prints:
// @<> ( {
// std::vector<tree, std::allocator<tree>> ( {
......@@ -158,7 +158,7 @@ void testee(untyped_actor* self, size_t remaining) {
// tree ( 0, { 10, { 11, { }, 12, { }, 13, { } }, 20, { 21, { }, 22, { } } } )
// )
// } )
aout << "to_string: " << to_string(self->last_dequeued()) << endl;
cout << "to_string: " << to_string(self->last_dequeued()) << endl;
set_next_behavior();
}
);
......
......@@ -155,7 +155,7 @@ bool abstract_actor::establish_backlink(const actor_addr& other) {
// send exit message without lock
if (reason != exit_reason::not_exited) {
auto ptr = detail::raw_access::unsafe_cast(other);
ptr.enqueue({address(), ptr},
ptr->enqueue({address(), ptr},
make_any_tuple(exit_msg{address(), exit_reason()}));
}
return false;
......
......@@ -52,8 +52,8 @@ actor::actor(abstract_actor* ptr) : m_ops(ptr) { }
actor::actor(const invalid_actor_t&) : m_ops(nullptr) { }
void actor::enqueue(const message_header& hdr, any_tuple msg) const {
if (m_ops.m_ptr) m_ops.m_ptr->enqueue(hdr, std::move(msg));
void actor::ops::enqueue(const message_header& hdr, any_tuple msg) const {
if (m_ptr) m_ptr->enqueue(hdr, std::move(msg));
}
intptr_t actor::compare(const actor& other) const {
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include "cppa/scheduler.hpp"
#include "cppa/singletons.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/scoped_actor.hpp"
#include "cppa/actor_ostream.hpp"
namespace cppa {
actor_ostream::actor_ostream(local_actor* self) : m_self(self) {
m_printer = get_scheduler()->printer();
}
actor_ostream& actor_ostream::write(std::string arg) {
m_self->send(m_printer, atom("add"), move(arg));
return *this;
}
actor_ostream& actor_ostream::flush() {
m_self->send(m_printer, atom("flush"));
return *this;
}
actor_ostream aout(scoped_actor& self) {
return actor_ostream{self.get()};
}
} // namespace cppa
namespace std {
cppa::actor_ostream& endl(cppa::actor_ostream& o) {
return o.write("\n");
}
cppa::actor_ostream& flush(cppa::actor_ostream& o) {
return o.flush();
}
} // namespace std
......@@ -49,7 +49,7 @@ void blocking_untyped_actor::await_all_other_actors_done() {
blocking_untyped_actor::response_future
blocking_untyped_actor::sync_send_tuple(const actor& dest, any_tuple what) {
auto nri = new_request_id();
dest.enqueue({address(), dest, nri}, std::move(what));
dest->enqueue({address(), dest, nri}, std::move(what));
return {nri.response_id(), this};
}
......
......@@ -106,7 +106,7 @@ void publish_local_groups_at(std::uint16_t port, const char* addr) {
publish(gn, port, addr);
}
catch (std::exception&) {
gn.enqueue({invalid_actor_addr, nullptr}, make_any_tuple(atom("SHUTDOWN")));
gn->enqueue({invalid_actor_addr, nullptr}, make_any_tuple(atom("SHUTDOWN")));
throw;
}
}
......
......@@ -82,7 +82,7 @@ class local_group : public group {
CPPA_LOG_TRACE(CPPA_TARG(hdr, to_string) << ", "
<< CPPA_TARG(msg, to_string));
send_all_subscribers(hdr, msg);
m_broker.enqueue(hdr, msg);
m_broker->enqueue(hdr, msg);
}
pair<bool, size_t> add_subscriber(const channel& who) {
......@@ -194,7 +194,7 @@ class local_broker : public untyped_actor {
<< " acquaintances; " << CPPA_TSARG(sender)
<< ", " << CPPA_TSARG(what));
for (auto& acquaintance : m_acquaintances) {
acquaintance.enqueue({sender, acquaintance}, what);
acquaintance->enqueue({sender, acquaintance}, what);
}
}
......@@ -250,7 +250,7 @@ class local_group_proxy : public local_group {
void enqueue(const message_header& hdr, any_tuple msg) override {
// forward message to the broker
m_broker.enqueue(hdr, make_any_tuple(atom("FORWARD"), move(msg)));
m_broker->enqueue(hdr, make_any_tuple(atom("FORWARD"), move(msg)));
}
private:
......
......@@ -187,8 +187,8 @@ void local_actor::quit(std::uint32_t reason) {
// actors breaks any assumption the user has about his code,
// in particular, receive_loop() is a deadlock when not throwing
// an exception here
aout << "*** warning: event-based actor killed because it tried to "
"use receive()\n";
aout(this) << "*** warning: event-based actor killed because it tried "
"to use receive()\n";
throw actor_exited(reason);
}
planned_exit_reason(reason);
......@@ -200,7 +200,7 @@ message_id local_actor::timed_sync_send_tuple_impl(message_priority mp,
any_tuple&& what) {
auto nri = new_request_id();
if (mp == message_priority::high) nri = nri.with_high_priority();
dest.enqueue({address(), dest, nri}, std::move(what));
dest->enqueue({address(), dest, nri}, std::move(what));
auto rri = nri.response_id();
get_scheduler()->delayed_send({address(), this, rri}, rtime,
make_any_tuple(sync_timeout_msg{}));
......
......@@ -171,15 +171,3 @@ logging::~logging() { }
logging* logging::create_singleton() { return new logging_impl; }
} // namespace cppa
namespace std {
const cppa::actor_ostream& endl(const cppa::actor_ostream& o) {
return o.write("\n");
}
const cppa::actor_ostream& flush(const cppa::actor_ostream& o) {
return o.flush();
}
} // namespace std
......@@ -224,7 +224,7 @@ void scheduler_helper::printer_loop(blocking_untyped_actor* self) {
self->receive_while (gref(running)) (
on(atom("add"), arg_match) >> [&](std::string& str) {
auto s = self->last_sender();
if (!str.empty() && s != nullptr) {
if (!str.empty() && s.valid()) {
auto i = out.find(s);
if (i == out.end()) {
i = out.insert(make_pair(s, move(str))).first;
......@@ -242,18 +242,17 @@ void scheduler_helper::printer_loop(blocking_untyped_actor* self) {
on(atom("flush")) >> [&] {
flush_output(self->last_sender());
},
on_arg_match >> [&](const down_msg&) {
auto s = self->last_sender();
flush_output(s);
out.erase(s);
on_arg_match >> [&](const down_msg& dm) {
flush_output(dm.source);
out.erase(dm.source);
},
on(atom("DIE")) >> [&] {
running = false;
},
others() >> [] {
//cout << "*** unexpected: "
// << to_string(self->last_dequeued())
// << endl;
others() >> [self] {
std::cerr << "*** unexpected: "
<< to_string(self->last_dequeued())
<< std::endl;
}
);
}
......
......@@ -52,7 +52,7 @@ void untyped_actor::forward_to(const actor& whom) {
untyped_actor::response_future untyped_actor::sync_send_tuple(const actor& dest,
any_tuple what) {
auto nri = new_request_id();
dest.enqueue({address(), dest, nri}, std::move(what));
dest->enqueue({address(), dest, nri}, std::move(what));
return {nri.response_id(), this};
}
......
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