Commit 6c50824f authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Move stream impl classes to detail namespace

parent 3b70a5bd
......@@ -16,8 +16,8 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_SINK_IMPL_HPP
#define CAF_STREAM_SINK_IMPL_HPP
#ifndef CAF_DETAIL_STREAM_SINK_IMPL_HPP
#define CAF_DETAIL_STREAM_SINK_IMPL_HPP
#include "caf/config.hpp"
#include "caf/logger.hpp"
......@@ -31,6 +31,7 @@
#include "caf/policy/arg.hpp"
namespace caf {
namespace detail {
template <class Fun, class Finalize>
class stream_sink_impl : public stream_manager {
......@@ -105,14 +106,15 @@ private:
};
template <class Init, class Fun, class Finalize>
stream_manager_ptr make_receiving_manager(local_actor* self, Init init, Fun f,
Finalize fin) {
stream_manager_ptr make_stream_sink(local_actor* self, Init init, Fun f,
Finalize fin) {
using impl = stream_sink_impl<Fun, Finalize>;
auto ptr = make_counted<impl>(self, std::move(f), std::move(fin));
init(ptr->state());
return ptr;
}
} // namespace detail
} // namespace caf
#endif // CAF_STREAM_SINK_IMPL_HPP
#endif // CAF_DETAIL_STREAM_SINK_IMPL_HPP
......@@ -16,8 +16,8 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_SOURCE_IMPL_HPP
#define CAF_STREAM_SOURCE_IMPL_HPP
#ifndef CAF_DETAIL_STREAM_SOURCE_IMPL_HPP
#define CAF_DETAIL_STREAM_SOURCE_IMPL_HPP
#include "caf/downstream.hpp"
#include "caf/logger.hpp"
......@@ -29,6 +29,7 @@
#include "caf/policy/arg.hpp"
namespace caf {
namespace detail {
template <class Fun, class Predicate, class DownstreamPolicy>
class stream_source_impl : public stream_manager {
......@@ -96,6 +97,7 @@ stream_manager_ptr make_stream_source(local_actor* self, Init init, Fun f,
return ptr;
}
} // namespace detail
} // namespace caf
#endif // CAF_STREAM_SOURCE_IMPL_HPP
#endif // CAF_DETAIL_STREAM_SOURCE_IMPL_HPP
......@@ -16,8 +16,8 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_STAGE_IMPL_HPP
#define CAF_STREAM_STAGE_IMPL_HPP
#ifndef CAF_DETAIL_STREAM_STAGE_IMPL_HPP
#define CAF_DETAIL_STREAM_STAGE_IMPL_HPP
#include "caf/downstream.hpp"
#include "caf/logger.hpp"
......@@ -30,6 +30,7 @@
#include "caf/policy/arg.hpp"
namespace caf {
namespace detail {
template <class Fun, class Cleanup, class DownstreamPolicy>
class stream_stage_impl : public stream_manager {
......@@ -89,7 +90,7 @@ public:
return sec::unexpected_message;
}
message make_output_token(const stream_id& x) const override {
message make_output_token(const stream_id&) const override {
// TODO: return make_message(stream<output_type>{x});
return make_message();
}
......@@ -115,6 +116,7 @@ stream_manager_ptr make_stream_stage(local_actor* self, Init init, Fun f,
return ptr;
}
} // namespace detail
} // namespace caf
#endif // CAF_STREAM_STAGE_IMPL_HPP
#endif // CAF_DETAIL_STREAM_STAGE_IMPL_HPP
......@@ -45,10 +45,7 @@
#include "caf/send.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_scatterer.hpp"
#include "caf/stream_sink_impl.hpp"
#include "caf/stream_slot.hpp"
#include "caf/stream_source_impl.hpp"
#include "caf/stream_stage_impl.hpp"
#include "caf/system_messages.hpp"
#include "caf/upstream_msg.hpp"
#include "caf/variant.hpp"
......@@ -66,6 +63,9 @@
#include "caf/intrusive/wdrr_fixed_multiplexed_queue.hpp"
#include "caf/detail/overload.hpp"
#include "caf/detail/stream_sink_impl.hpp"
#include "caf/detail/stream_source_impl.hpp"
#include "caf/detail/stream_stage_impl.hpp"
using std::vector;
......@@ -300,8 +300,8 @@ public:
auto fin = [=](const int& x) {
return x == num_messages;
};
auto ptr = make_stream_source(this, init, f, fin,
policy::arg<broadcast_scatterer<int>>::value);
policy::arg<broadcast_scatterer<int>> token;
auto ptr = detail::make_stream_source(this, init, f, fin, token);
ptr->generate_messages();
pending_managers_.emplace(slot, std::move(ptr));
}
......@@ -324,8 +324,8 @@ public:
auto cleanup = [](log_ptr&) {
// nop
};
forwarder = make_stream_stage(this, init, f, cleanup,
policy::arg<broadcast_scatterer<int>>::value);
policy::arg<broadcast_scatterer<int>> token;
forwarder = detail::make_stream_stage(this, init, f, cleanup, token);
pending_managers_.emplace(slot, forwarder);
}
......@@ -349,8 +349,8 @@ public:
auto fin = [](log_ptr&) {
// nop
};
mgr = make_receiving_manager(this, std::move(init), std::move(f),
std::move(fin));
mgr = detail::make_stream_sink(this, std::move(init), std::move(f),
std::move(fin));
}
// mgr->out().add_path(id, hs.prev_stage);
managers_.emplace(id, mgr);
......
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