Commit 1fcfec71 authored by Dominik Charousset's avatar Dominik Charousset

added IO actors

this patch adds a new kind of actors - IO actors - to libcppa as well
as some example code featuring Google Protobuf integration
parent 4b0a9051
...@@ -132,6 +132,9 @@ set(LIBCPPA_SRC ...@@ -132,6 +132,9 @@ set(LIBCPPA_SRC
src/get_mac_addresses.cpp src/get_mac_addresses.cpp
src/group.cpp src/group.cpp
src/group_manager.cpp src/group_manager.cpp
src/io_actor.cpp
src/io_actor_backend.cpp
src/io_service.cpp
src/ipv4_acceptor.cpp src/ipv4_acceptor.cpp
src/ipv4_io_stream.cpp src/ipv4_io_stream.cpp
src/local_actor.cpp src/local_actor.cpp
......
...@@ -100,6 +100,9 @@ cppa/network/default_peer.hpp ...@@ -100,6 +100,9 @@ cppa/network/default_peer.hpp
cppa/network/default_peer_acceptor.hpp cppa/network/default_peer_acceptor.hpp
cppa/network/default_protocol.hpp cppa/network/default_protocol.hpp
cppa/network/input_stream.hpp cppa/network/input_stream.hpp
cppa/network/io_actor.hpp
cppa/network/io_actor_backend.hpp
cppa/network/io_service.hpp
cppa/network/io_stream.hpp cppa/network/io_stream.hpp
cppa/network/ipv4_acceptor.hpp cppa/network/ipv4_acceptor.hpp
cppa/network/ipv4_io_stream.hpp cppa/network/ipv4_io_stream.hpp
...@@ -238,6 +241,9 @@ src/get_mac_addresses.cpp ...@@ -238,6 +241,9 @@ src/get_mac_addresses.cpp
src/get_root_uuid.cpp src/get_root_uuid.cpp
src/group.cpp src/group.cpp
src/group_manager.cpp src/group_manager.cpp
src/io_actor.cpp
src/io_actor_backend.cpp
src/io_service.cpp
src/ipv4_acceptor.cpp src/ipv4_acceptor.cpp
src/ipv4_io_stream.cpp src/ipv4_io_stream.cpp
src/local_actor.cpp src/local_actor.cpp
......
...@@ -68,6 +68,12 @@ ...@@ -68,6 +68,12 @@
#include "cppa/util/type_traits.hpp" #include "cppa/util/type_traits.hpp"
#include "cppa/network/acceptor.hpp" #include "cppa/network/acceptor.hpp"
#include "cppa/network/io_actor.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/io_service.hpp"
#include "cppa/network/input_stream.hpp"
#include "cppa/network/output_stream.hpp"
#include "cppa/network/io_actor_backend.hpp"
#include "cppa/detail/memory.hpp" #include "cppa/detail/memory.hpp"
#include "cppa/detail/get_behavior.hpp" #include "cppa/detail/get_behavior.hpp"
...@@ -606,6 +612,46 @@ inline actor_ptr remote_actor(const std::string& host, std::uint16_t port) { ...@@ -606,6 +612,46 @@ inline actor_ptr remote_actor(const std::string& host, std::uint16_t port) {
*/ */
actor_ptr remote_actor(network::io_stream_ptr_pair connection); actor_ptr remote_actor(network::io_stream_ptr_pair connection);
/**
* @brief Spawns an IO actor of type @p Impl.
* @param args Constructor arguments.
* @tparam Impl Subtype of {@link network::io_actor}.
* @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
*/
template<class Impl, spawn_options Options = no_spawn_options, typename... Ts>
actor_ptr spawn_io(network::input_stream_ptr in,
network::output_stream_ptr out,
Ts&&... args) {
using namespace network;
auto mm = get_middleman();
auto ptr = make_counted<Impl>(std::forward<Ts>(args)...);
auto backend = make_counted<io_actor_backend>(std::move(in), std::move(out), ptr);
backend->init();
mm->run_later([=] { mm->continue_reader(backend); });
return eval_sopts(Options, ptr);
}
/**
* @brief Spawns a new {@link actor} that evaluates given arguments.
* @param args A functor followed by its arguments.
* @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
*/
template<spawn_options Options = no_spawn_options, typename... Ts>
actor_ptr spawn_io(std::function<void (network::io_service*)> fun,
network::input_stream_ptr in,
network::output_stream_ptr out,
Ts&&... args) {
using namespace network;
auto mm = get_middleman();
auto ptr = io_actor::from(std::move(fun));
auto backend = make_counted<io_actor_backend>(std::move(in), std::move(out), ptr);
backend->init();
mm->run_later([=] { mm->continue_reader(backend); });
return eval_sopts(Options, ptr);
}
/** /**
* @brief Destroys all singletons, disconnects all peers and stops the * @brief Destroys all singletons, disconnects all peers and stops the
* scheduler. It is recommended to use this function as very last * scheduler. It is recommended to use this function as very last
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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 IO_ACTOR_HPP
#define IO_ACTOR_HPP
#include "cppa/stackless.hpp"
#include "cppa/threadless.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/mailbox_element.hpp"
#include "cppa/network/io_service.hpp"
namespace cppa { namespace network {
class io_actor_backend;
class io_actor_continuation;
class io_actor : public extend<local_actor>::with<threadless, stackless> {
typedef combined_type super;
friend class io_actor_backend;
friend class io_actor_continuation;
public:
void enqueue(const message_header& hdr, any_tuple msg);
bool initialized() const;
void quit(std::uint32_t reason);
static intrusive_ptr<io_actor> from(std::function<void (io_service*)> fun);
protected:
io_service& io_handle();
private:
void invoke_message(mailbox_element* elem);
void invoke_message(any_tuple msg);
intrusive_ptr<io_actor_backend> m_parent;
};
typedef intrusive_ptr<io_actor> io_actor_ptr;
} } // namespace cppa::network
#endif // IO_ACTOR_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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 IO_ACTOR_BACKEND_HPP
#define IO_ACTOR_BACKEND_HPP
#include "cppa/network/io_actor.hpp"
#include "cppa/network/io_service.hpp"
#include "cppa/network/input_stream.hpp"
#include "cppa/network/output_stream.hpp"
#include "cppa/network/buffered_writer.hpp"
namespace cppa { namespace network {
class io_actor_backend : public buffered_writer, public io_service {
typedef buffered_writer super; // io_service is merely an interface
// 65k is the maximum TCP package size
static constexpr size_t default_max_buffer_size = 65535;
public:
io_actor_backend(input_stream_ptr in, output_stream_ptr out, io_actor_ptr ptr);
~io_actor_backend();
void init();
void handle_disconnect();
void io_failed() override;
void receive_policy(policy_flag policy, size_t buffer_size) override;
continue_reading_result continue_reading() override;
void close() override;
void write(size_t num_bytes, const void* data) override;
private:
bool m_dirty;
policy_flag m_policy;
size_t m_policy_buffer_size;
input_stream_ptr m_in;
intrusive_ptr<io_actor> m_self;
cow_tuple<atom_value, util::buffer> m_read;
};
} } // namespace cppa::network
#endif // IO_ACTOR_BACKEND_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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 IO_SERVICE_HPP
#define IO_SERVICE_HPP
#include <cstddef>
namespace cppa { namespace network {
class io_service {
public:
virtual ~io_service();
/**
* @brief Denotes when an actor will receive a read buffer.
*/
enum policy_flag { at_least, at_most, exactly };
/**
* @brief Closes the network connection.
*/
virtual void close() = 0;
/**
* @brief Asynchronously sends @p size bytes of @p data.
*/
virtual void write(size_t size, const void* data) = 0;
/**
* @brief Adjusts the rule receiving 'IO_receive' messages.
* The default settings are <tt>policy = io_handle::at_least</tt>
* and <tt>buffer_size = 0</tt>.
*/
virtual void receive_policy(policy_flag policy, size_t buffer_size) = 0;
};
} } // namespace cppa::network
#endif // IO_SERVICE_HPP
...@@ -139,6 +139,11 @@ match_expr_convert(const T0& arg0, const T1& arg1, const Ts&... args) { ...@@ -139,6 +139,11 @@ match_expr_convert(const T0& arg0, const T1& arg1, const Ts&... args) {
return detail::match_expr_concat(arg0, arg1, args...); return detail::match_expr_concat(arg0, arg1, args...);
} }
template<typename... Cases>
partial_function operator,(const match_expr<Cases...>& mexpr,
const partial_function& pfun) {
return mexpr.as_behavior_impl()->or_else(pfun.as_behavior_impl);
}
/****************************************************************************** /******************************************************************************
* inline and template member function implementations * * inline and template member function implementations *
......
package org.libcppa;
message Ping {
required int32 id = 1;
}
message Pong {
required int32 id = 1;
}
message PingOrPong {
optional Ping ping = 1;
optional Pong pong = 2;
}
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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/>. *
\******************************************************************************/
#define CPPA_LOG_LEVEL 4
#include <vector>
#include <string>
#include <limits>
#include <iostream>
#include "cppa/cppa.hpp"
#include "cppa/logging.hpp"
#include "cppa/network/ipv4_acceptor.hpp"
#include "cppa/network/ipv4_io_stream.hpp"
#include "pingpong.pb.h"
#include "cppa/singletons.hpp"
#include "cppa/network/io_actor.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/io_service.hpp"
#include "cppa/network/io_actor_backend.hpp"
using namespace std;
using namespace cppa;
using namespace cppa::network;
void protobuf_io(io_service* ios) {
partial_function default_bhvr {
on(atom("IO_closed")) >> [=] {
cout << "IO_closed" << endl;
self->quit(exit_reason::normal);
},
others() >> [=] {
cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
}
};
ios->receive_policy(io_service::exactly, 4);
become(
partial_function{
on(atom("IO_read"), arg_match) >> [=](const util::buffer& buf) {
int num_bytes;
memcpy(&num_bytes, buf.data(), 4);
num_bytes = htonl(num_bytes);
if (num_bytes < 0 || num_bytes > (1024 * 1024)) {
self->quit(exit_reason::user_defined);
return;
}
ios->receive_policy(io_service::exactly, (size_t) num_bytes);
become(
partial_function{
on(atom("IO_read"), arg_match) >> [=](const util::buffer& buf) {
org::libcppa::PingOrPong p;
p.ParseFromArray(buf.data(), (int) buf.size());
auto print = [](const char* name, int value) {
cout << name << "{" << value << "}" << endl;
};
if (p.has_ping()) print("Ping", p.ping().id());
else if (p.has_pong()) print("Pong", p.pong().id());
else cerr << "neither Pong nor Ping!" << endl;
self->quit(exit_reason::normal);
}//,default_bhvr
}.or_else(default_bhvr)
);
}
}.or_else(default_bhvr)
);
}
/*
class protobuf_io : public io_actor {
enum { await_size, await_data } state;
public:
protobuf_io() : state(await_size) { }
void init() {
io_handle().receive_policy(io_service::exactly, 4);
become (
on(atom("IO_read"), arg_match) >> [=](const util::buffer& buf) {
switch (state) {
case await_size:
int num_bytes;
memcpy(&num_bytes, buf.data(), 4);
num_bytes = htonl(num_bytes);
if (num_bytes < 0 || num_bytes > (1024 * 1024)) {
quit(exit_reason::user_defined);
return;
}
io_handle().receive_policy(io_service::exactly,
static_cast<size_t>(num_bytes));
state = await_data;
break;
case await_data:
org::libcppa::PingOrPong p;
p.ParseFromArray(buf.data(), (int) buf.size());
auto print = [](const char* name, int value) {
cout << name << "{" << value << "}" << endl;
};
if (p.has_ping()) print("Ping", p.ping().id());
else if (p.has_pong()) print("Pong", p.pong().id());
else cerr << "neither Pong nor Ping!" << endl;
quit(exit_reason::normal);
}
},
on(atom("IO_closed")) >> [=] {
cout << "IO_closed" << endl;
quit(exit_reason::normal);
},
others() >> [=] {
cout << "unexpected: " << to_string(last_dequeued()) << endl;
}
);
}
};
*/
int main(int argc, char** argv) {
match(std::vector<string>{argv + 1, argv + argc}) (
on("-s") >> [] {
cout << "run in server mode" << endl;
auto ack = network::ipv4_acceptor::create(4242);
auto p = ack->accept_connection();
//spawn_io<protobuf_io>(p.first, p.second);
spawn_io(protobuf_io, std::move(p.first), std::move(p.second));
await_all_others_done();
},
on_arg_match >> [](const string& host, const string& port_str) {
auto port = static_cast<uint16_t>(stoul(port_str));
auto serv = network::ipv4_io_stream::connect_to(host.c_str(), port);
org::libcppa::PingOrPong p;
p.mutable_ping()->set_id(numeric_limits<int32_t>::max());
string buf = p.SerializeAsString();
int32_t s = htonl(static_cast<int32_t>(buf.size()));
serv->write(&s, 4);
serv->write(buf.data(), buf.size());
cout << "run in client mode" << endl;
}
);
}
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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/singletons.hpp"
#include "cppa/network/io_actor.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/io_actor_backend.hpp"
#include "cppa/detail/sync_request_bouncer.hpp"
namespace cppa { namespace network {
class io_actor_continuation {
public:
io_actor_continuation(io_actor_ptr ptr, mailbox_element* elem)
: m_self(std::move(ptr)), m_elem(elem) { }
void operator()() const { m_self->invoke_message(m_elem); }
private:
intrusive_ptr<io_actor> m_self;
mailbox_element* m_elem;
};
class default_io_actor_impl : public io_actor {
public:
typedef std::function<void (io_service*)> function_type;
default_io_actor_impl(function_type fun) : m_fun(std::move(fun)) { }
void init() override {
m_fun(&io_handle());
}
private:
function_type m_fun;
};
void io_actor::invoke_message(mailbox_element* elem) {
if (exit_reason() != exit_reason::not_exited) {
if (elem->mid.valid()) {
detail::sync_request_bouncer srb{exit_reason()};
srb(*elem);
detail::disposer d;
d(elem);
}
return;
}
try {
scoped_self_setter sss{this};
m_bhvr_stack.invoke(m_recv_policy, this, elem);
}
catch (std::exception& e) {
CPPA_LOG_ERROR("IO actor killed due to an unhandled exception: "
<< to_verbose_string(e));
// keep compiler happy in non-debug mode
static_cast<void>(e);
quit(exit_reason::unhandled_exception);
}
catch (...) {
CPPA_LOG_ERROR("IO actor killed due to an unhandled exception");
quit(exit_reason::unhandled_exception);
}
}
void io_actor::invoke_message(any_tuple msg) {
invoke_message(mailbox_element::create(this, std::move(msg)));
}
void io_actor::enqueue(const message_header& hdr, any_tuple msg) {
auto e = mailbox_element::create(hdr, std::move(msg));
get_middleman()->run_later(io_actor_continuation{this, e});
}
bool io_actor::initialized() const {
return true;
}
void io_actor::quit(std::uint32_t reason) {
cleanup(reason);
m_parent->handle_disconnect();
}
io_service& io_actor::io_handle() {
return *m_parent;
}
intrusive_ptr<io_actor> io_actor::from(std::function<void (io_service*)> fun) {
return make_counted<default_io_actor_impl>(std::move(fun));
}
} } // namespace cppa::network
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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/singletons.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/io_actor_backend.hpp"
#include "cppa/detail/actor_registry.hpp"
namespace cppa { namespace network {
io_actor_backend::io_actor_backend(input_stream_ptr in,
output_stream_ptr out,
io_actor_ptr ptr)
: super(get_middleman(), in->read_handle(), std::move(out))
, m_in(in), m_self(ptr) {
m_self->m_parent = this;
get_ref<0>(m_read) = atom("IO_read");
get_ref<1>(m_read).final_size(default_max_buffer_size);
m_dirty = false;
m_policy = at_least;
m_policy_buffer_size = 0;
}
io_actor_backend::~io_actor_backend() {
handle_disconnect();
get_actor_registry()->dec_running();
}
void io_actor_backend::init() {
get_actor_registry()->inc_running();
auto selfptr = m_self.get();
scoped_self_setter sss{selfptr};
selfptr->init();
}
void io_actor_backend::handle_disconnect() {
if (m_self) {
CPPA_LOG_DEBUG("became disconnected");
if (m_self->exit_reason() == exit_reason::not_exited) {
m_self->invoke_message(make_any_tuple(atom("IO_closed")));
}
m_self.reset();
}
}
void io_actor_backend::io_failed() {
handle_disconnect();
}
void io_actor_backend::receive_policy(policy_flag policy, size_t buffer_size) {
CPPA_LOG_TRACE(CPPA_ARG(policy) << ", " << CPPA_ARG(buffer_size));
m_dirty = true;
m_policy = policy;
m_policy_buffer_size = buffer_size;
}
continue_reading_result io_actor_backend::continue_reading() {
CPPA_LOG_TRACE("");
for (;;) {
auto& buf = get_ref<1>(m_read);
if (m_dirty) {
m_dirty = false;
if (m_policy == at_most || m_policy == exactly) {
buf.final_size(m_policy_buffer_size);
}
else buf.final_size(default_max_buffer_size);
}
auto before = buf.size();
try { buf.append_from(m_in.get()); }
catch (std::ios_base::failure&) {
handle_disconnect();
return read_failure;
}
CPPA_LOG_DEBUG("received " << (buf.size() - before) << " bytes");
if ( before == buf.size()
|| (m_policy == exactly && buf.size() != m_policy_buffer_size)) {
return read_continue_later;
}
if ( (m_policy == at_least && buf.size() >= m_policy_buffer_size)
|| m_policy == exactly
|| m_policy == at_most) {
CPPA_LOG_DEBUG("invoke io actor");
m_self->invoke_message(m_read);
CPPA_LOG_INFO_IF(!m_read.vals()->unique(), "buffer became detached");
if (m_self == nullptr) {
// io_actor::quit() calls handle_disconnect, which sets
// m_self to nullptr
return read_closed;
}
get_ref<1>(m_read).clear();
}
}
}
void io_actor_backend::close() {
CPPA_LOG_DEBUG("");
get_middleman()->stop_reader(this);
}
void io_actor_backend::write(size_t num_bytes, const void* data) {
super::write(num_bytes, data);
}
} } // namespace cppa::network
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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/network/io_service.hpp"
namespace cppa { namespace network {
io_service::~io_service() { }
} } // namespace cppa::network
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