Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
A
Actor Framework
Project overview
Project overview
Details
Activity
Releases
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Operations
Operations
Metrics
Environments
Analytics
Analytics
CI / CD
Repository
Value Stream
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
cpp-libs
Actor Framework
Commits
9053a3af
Commit
9053a3af
authored
Sep 03, 2019
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Fix composition of typed actors using streams
Close #908.
parent
5616bec9
Changes
21
Hide whitespace changes
Inline
Side-by-side
Showing
21 changed files
with
146 additions
and
263 deletions
+146
-263
libcaf_core/caf/composable_behavior.hpp
libcaf_core/caf/composable_behavior.hpp
+7
-11
libcaf_core/caf/composed_type.hpp
libcaf_core/caf/composed_type.hpp
+23
-70
libcaf_core/caf/deduce_mpi.hpp
libcaf_core/caf/deduce_mpi.hpp
+0
-8
libcaf_core/caf/detail/invoke_result_visitor.hpp
libcaf_core/caf/detail/invoke_result_visitor.hpp
+0
-6
libcaf_core/caf/detail/type_traits.hpp
libcaf_core/caf/detail/type_traits.hpp
+13
-0
libcaf_core/caf/fwd.hpp
libcaf_core/caf/fwd.hpp
+0
-4
libcaf_core/caf/make_sink_result.hpp
libcaf_core/caf/make_sink_result.hpp
+5
-7
libcaf_core/caf/make_source_result.hpp
libcaf_core/caf/make_source_result.hpp
+16
-14
libcaf_core/caf/make_stage_result.hpp
libcaf_core/caf/make_stage_result.hpp
+15
-15
libcaf_core/caf/output_stream.hpp
libcaf_core/caf/output_stream.hpp
+0
-31
libcaf_core/caf/replies_to.hpp
libcaf_core/caf/replies_to.hpp
+0
-12
libcaf_core/caf/result.hpp
libcaf_core/caf/result.hpp
+22
-23
libcaf_core/caf/scheduled_actor.hpp
libcaf_core/caf/scheduled_actor.hpp
+0
-1
libcaf_core/caf/stream_manager.hpp
libcaf_core/caf/stream_manager.hpp
+0
-1
libcaf_core/caf/stream_slot.hpp
libcaf_core/caf/stream_slot.hpp
+12
-13
libcaf_core/test/composable_behavior.cpp
libcaf_core/test/composable_behavior.cpp
+27
-13
libcaf_core/test/continuous_streaming.cpp
libcaf_core/test/continuous_streaming.cpp
+1
-1
libcaf_core/test/fused_downstream_manager.cpp
libcaf_core/test/fused_downstream_manager.cpp
+2
-2
libcaf_core/test/metaprogramming.cpp
libcaf_core/test/metaprogramming.cpp
+0
-28
libcaf_core/test/pipeline_streaming.cpp
libcaf_core/test/pipeline_streaming.cpp
+2
-2
libcaf_core/test/selective_streaming.cpp
libcaf_core/test/selective_streaming.cpp
+1
-1
No files found.
libcaf_core/caf/composable_behavior.hpp
View file @
9053a3af
...
...
@@ -34,8 +34,8 @@ template <class MPI>
class
composable_behavior_base
;
template
<
class
...
Xs
,
class
...
Ys
>
class
composable_behavior_base
<
typed_mpi
<
detail
::
type_list
<
Xs
...
>
,
output_tuple
<
Ys
...
>>>
{
class
composable_behavior_base
<
typed_mpi
<
detail
::
type_list
<
Xs
...
>
,
output_tuple
<
Ys
...
>>>
{
public:
virtual
~
composable_behavior_base
()
noexcept
{
// nop
...
...
@@ -44,16 +44,16 @@ public:
virtual
result
<
Ys
...
>
operator
()(
param_t
<
Xs
>
...)
=
0
;
// C++14 and later
#
if __cplusplus > 201103L
#if __cplusplus > 201103L
auto
make_callback
()
{
return
[
=
](
param_t
<
Xs
>
...
xs
)
{
return
(
*
this
)(
std
::
move
(
xs
)...);
};
}
#
else
#else
// C++11
std
::
function
<
result
<
Ys
...
>
(
param_t
<
Xs
>
...)
>
make_callback
()
{
std
::
function
<
result
<
Ys
...
>
(
param_t
<
Xs
>
...)
>
make_callback
()
{
return
[
=
](
param_t
<
Xs
>
...
xs
)
{
return
(
*
this
)(
std
::
move
(
xs
)...);
};
}
#
endif
#endif
};
/// Base type for composable actor states.
...
...
@@ -67,11 +67,7 @@ class composable_behavior<typed_actor<Clauses...>>
public:
using
signatures
=
detail
::
type_list
<
Clauses
...
>
;
using
handle_type
=
typename
detail
::
tl_apply
<
signatures
,
typed_actor
>::
type
;
using
handle_type
=
typename
detail
::
tl_apply
<
signatures
,
typed_actor
>::
type
;
using
actor_base
=
typename
handle_type
::
base
;
...
...
libcaf_core/caf/composed_type.hpp
View file @
9053a3af
...
...
@@ -18,11 +18,10 @@
#pragma once
#include "caf/detail/type_list.hpp"
#include "caf/fwd.hpp"
#include "caf/replies_to.hpp"
#include "caf/detail/type_list.hpp"
namespace
caf
{
/// Computes the type for f*g (actor composition).
...
...
@@ -58,82 +57,36 @@ struct composed_type<detail::type_list<>, Ys, Zs, detail::type_list<Rs...>> {
// end of inner loop Ys (Zs)
template
<
class
X
,
class
...
Xs
,
class
Ys
,
class
Rs
>
struct
composed_type
<
detail
::
type_list
<
X
,
Xs
...
>
,
Ys
,
detail
::
type_list
<>
,
Rs
>
:
composed_type
<
detail
::
type_list
<
Xs
...
>
,
Ys
,
Ys
,
Rs
>
{};
:
composed_type
<
detail
::
type_list
<
Xs
...
>
,
Ys
,
Ys
,
Rs
>
{};
// case #1
template
<
class
...
In
,
class
...
Out
,
class
...
Xs
,
class
Ys
,
class
...
MapsTo
,
class
...
Zs
,
class
...
Rs
>
struct
composed_type
<
detail
::
type_list
<
typed_mpi
<
detail
::
type_list
<
In
...
>
,
output_tuple
<
Out
...
>>
,
Xs
...
>
,
Ys
,
detail
::
type_list
<
typed_mpi
<
detail
::
type_list
<
Out
...
>
,
output_tuple
<
MapsTo
...
>>
,
Zs
...
>
,
detail
::
type_list
<
Rs
...
>>
:
composed_type
<
detail
::
type_list
<
Xs
...
>
,
Ys
,
Ys
,
detail
::
type_list
<
Rs
...,
typed_mpi
<
detail
::
type_list
<
In
...
>
,
output_tuple
<
MapsTo
...
>>>>
{};
// case #2
template
<
class
...
In
,
class
...
Out
,
class
...
Xs
,
class
Ys
,
class
M
,
class
...
Ms
,
class
...
Zs
,
class
...
Rs
>
struct
composed_type
<
detail
::
type_list
<
typed_mpi
<
detail
::
type_list
<
In
...
>
,
output_tuple
<
Out
...
>>
,
Xs
...
>
,
Ys
,
detail
::
type_list
<
typed_mpi
<
detail
::
type_list
<
Out
...
>
,
output_stream
<
M
,
Ms
...
>>
,
Zs
...
>
,
detail
::
type_list
<
Rs
...
>>
:
composed_type
<
detail
::
type_list
<
Xs
...
>
,
Ys
,
Ys
,
detail
::
type_list
<
Rs
...,
typed_mpi
<
detail
::
type_list
<
In
...
>
,
output_stream
<
M
,
Ms
...
>>>>
{
};
// case #3
template
<
class
...
In
,
class
O
,
class
...
Out
,
class
...
Xs
,
class
Ys
,
class
M
,
class
...
Ms
,
class
...
Zs
,
class
...
Rs
>
struct
composed_type
<
detail
::
type_list
<
typed_mpi
<
detail
::
type_list
<
In
...
>
,
output_stream
<
O
,
Out
...
>>
,
Xs
...
>
,
Ys
,
detail
::
type_list
<
typed_mpi
<
detail
::
type_list
<
O
,
Out
...
>
,
output_tuple
<
M
,
Ms
...
>>
,
Zs
...
>
,
detail
::
type_list
<
Rs
...
>>
:
composed_type
<
detail
::
type_list
<
Xs
...
>
,
Ys
,
Ys
,
detail
::
type_list
<
Rs
...,
typed_mpi
<
detail
::
type_list
<
In
...
>
,
output_stream
<
M
,
Ms
...
>>>>
{
};
// case #4
template
<
class
...
In
,
class
O
,
class
...
Out
,
class
...
Xs
,
class
Ys
,
class
M
,
class
...
Ms
,
class
...
Zs
,
class
...
Rs
>
struct
composed_type
<
detail
::
type_list
<
typed_mpi
<
detail
::
type_list
<
In
...
>
,
output_stream
<
O
,
Out
...
>>
,
Xs
...
>
,
Ys
,
detail
::
type_list
<
typed_mpi
<
detail
::
type_list
<
O
,
Out
...
>
,
output_stream
<
M
,
Ms
...
>>
,
Zs
...
>
,
detail
::
type_list
<
Rs
...
>>
:
composed_type
<
detail
::
type_list
<
Xs
...
>
,
Ys
,
Ys
,
detail
::
type_list
<
Rs
...,
typed_mpi
<
detail
::
type_list
<
In
...
>
,
output_stream
<
M
,
Ms
...
>>>>
{
template
<
class
...
In
,
class
...
Out
,
class
...
Xs
,
class
Ys
,
class
...
MapsTo
,
class
...
Zs
,
class
...
Rs
>
struct
composed_type
<
detail
::
type_list
<
typed_mpi
<
detail
::
type_list
<
In
...
>
,
output_tuple
<
Out
...
>>
,
Xs
...
>
,
Ys
,
detail
::
type_list
<
typed_mpi
<
detail
::
type_list
<
Out
...
>
,
output_tuple
<
MapsTo
...
>>
,
Zs
...
>
,
detail
::
type_list
<
Rs
...
>>
:
composed_type
<
detail
::
type_list
<
Xs
...
>
,
Ys
,
Ys
,
detail
::
type_list
<
Rs
...,
typed_mpi
<
detail
::
type_list
<
In
...
>
,
output_tuple
<
MapsTo
...
>>>>
{
};
// default case (recurse over Zs)
template
<
class
In
,
class
Out
,
class
...
Xs
,
class
Ys
,
class
Unrelated
,
class
MapsTo
,
class
...
Zs
,
class
Rs
>
struct
composed_type
<
detail
::
type_list
<
typed_mpi
<
In
,
Out
>
,
Xs
...
>
,
Ys
,
detail
::
type_list
<
typed_mpi
<
Unrelated
,
MapsTo
>
,
Zs
...
>
,
Rs
>
:
composed_type
<
detail
::
type_list
<
typed_mpi
<
In
,
Out
>
,
Xs
...
>
,
Ys
,
detail
::
type_list
<
Zs
...
>
,
Rs
>
{};
template
<
class
In
,
class
Out
,
class
...
Xs
,
class
Ys
,
class
Unrelated
,
class
MapsTo
,
class
...
Zs
,
class
Rs
>
struct
composed_type
<
detail
::
type_list
<
typed_mpi
<
In
,
Out
>
,
Xs
...
>
,
Ys
,
detail
::
type_list
<
typed_mpi
<
Unrelated
,
MapsTo
>
,
Zs
...
>
,
Rs
>
:
composed_type
<
detail
::
type_list
<
typed_mpi
<
In
,
Out
>
,
Xs
...
>
,
Ys
,
detail
::
type_list
<
Zs
...
>
,
Rs
>
{};
/// Convenience type alias.
/// @relates composed_type
template
<
class
F
,
class
G
>
using
composed_type_t
=
typename
composed_type
<
G
,
F
,
F
,
detail
::
type_list
<>>::
type
;
using
composed_type_t
=
typename
composed_type
<
G
,
F
,
F
,
detail
::
type_list
<>>::
type
;
}
// namespace caf
libcaf_core/caf/deduce_mpi.hpp
View file @
9053a3af
...
...
@@ -79,14 +79,6 @@ struct dmi<optional<Y> (Xs...)> : dmi<Y (Xs...)> {};
template
<
class
Y
,
class
...
Xs
>
struct
dmi
<
expected
<
Y
>
(
Xs
...)
>
:
dmi
<
Y
(
Xs
...)
>
{};
// case #5: function returning an output_stream<>
template
<
class
Y
,
class
...
Ys
,
class
P
,
class
...
Xs
>
struct
dmi
<
output_stream
<
Y
,
std
::
tuple
<
Ys
...
>
,
P
>
(
Xs
...)
>
:
dmi
<
Y
(
Xs
...)
>
{
using
type
=
typed_mpi
<
type_list
<
typename
param_decay
<
Xs
>::
type
...
>
,
output_tuple
<
stream
<
Y
>
,
strip_and_convert_t
<
Ys
>
...
>>
;
};
// -- dmfou = deduce_mpi_function_object_unboxing
template
<
class
T
,
bool
isClass
=
std
::
is_class
<
T
>
::
value
>
...
...
libcaf_core/caf/detail/invoke_result_visitor.hpp
View file @
9053a3af
...
...
@@ -140,12 +140,6 @@ public:
(
*
this
)();
}
/// Calls `(*this)()`.
template
<
class
Out
,
class
...
Ts
>
void
operator
()(
output_stream
<
Out
,
Ts
...
>&
)
{
(
*
this
)();
}
/// Calls `(*this)()`.
template
<
class
Out
,
class
...
Ts
>
void
operator
()(
outbound_stream_slot
<
Out
,
Ts
...
>&
)
{
...
...
libcaf_core/caf/detail/type_traits.hpp
View file @
9053a3af
...
...
@@ -769,6 +769,19 @@ CAF_HAS_ALIAS_TRAIT(mapped_type);
// -- constexpr functions for use in enable_if & friends -----------------------
template
<
class
List1
,
class
List2
>
struct
all_constructible
:
std
::
false_type
{};
template
<
>
struct
all_constructible
<
type_list
<>
,
type_list
<>>
:
std
::
true_type
{};
template
<
class
T
,
class
...
Ts
,
class
U
,
class
...
Us
>
struct
all_constructible
<
type_list
<
T
,
Ts
...
>
,
type_list
<
U
,
Us
...
>>
{
static
constexpr
bool
value
=
std
::
is_constructible
<
T
,
U
>::
value
&&
all_constructible
<
type_list
<
Ts
...
>
,
type_list
<
Us
...
>>::
value
;
};
/// Checks whether T behaves like `std::map`.
template
<
class
T
>
struct
is_map_like
{
...
...
libcaf_core/caf/fwd.hpp
View file @
9053a3af
...
...
@@ -73,10 +73,6 @@ template <class...> class typed_event_based_actor;
template
<
class
...
>
class
typed_response_promise
;
template
<
class
...
>
class
variant
;
// -- variadic templates with fixed arguments ----------------------------------
//
template
<
class
,
class
...>
class
output_stream
;
// clang-format on
// -- classes ------------------------------------------------------------------
...
...
libcaf_core/caf/make_sink_result.hpp
View file @
9053a3af
...
...
@@ -18,6 +18,7 @@
#pragma once
#include "caf/delegated.hpp"
#include "caf/fwd.hpp"
#include "caf/stream_sink.hpp"
#include "caf/stream_slot.hpp"
...
...
@@ -26,7 +27,7 @@ namespace caf {
/// Returns a stream sink with the slot ID of its first inbound path.
template
<
class
In
>
class
make_sink_result
{
class
make_sink_result
:
public
delegated
<
void
>
{
public:
// -- member types -----------------------------------------------------------
...
...
@@ -39,9 +40,6 @@ public:
/// Pointer to a fully typed stream manager.
using
sink_ptr_type
=
intrusive_ptr
<
sink_type
>
;
/// The return type for `scheduled_actor::make_sink`.
using
output_stream_type
=
stream
<
input_type
>
;
// -- constructors, destructors, and assignment operators --------------------
make_sink_result
()
noexcept
:
slot_
(
0
)
{
...
...
@@ -61,15 +59,15 @@ public:
// -- properties -------------------------------------------------------------
inline
stream_slot
inbound_slot
()
const
noexcept
{
stream_slot
inbound_slot
()
const
noexcept
{
return
slot_
;
}
inline
sink_ptr_type
&
ptr
()
noexcept
{
sink_ptr_type
&
ptr
()
noexcept
{
return
ptr_
;
}
inline
const
sink_ptr_type
&
ptr
()
const
noexcept
{
const
sink_ptr_type
&
ptr
()
const
noexcept
{
return
ptr_
;
}
...
...
libcaf_core/caf/make_source_result.hpp
View file @
9053a3af
...
...
@@ -18,17 +18,22 @@
#pragma once
#include <tuple>
#include "caf/delegated.hpp"
#include "caf/detail/implicit_conversions.hpp"
#include "caf/fwd.hpp"
#include "caf/stream.hpp"
#include "caf/stream_slot.hpp"
#include "caf/stream_source.hpp"
#include "caf/detail/implicit_conversions.hpp"
namespace
caf
{
/// Returns a stream source with the slot ID of its first outbound path.
template
<
class
DownstreamManager
,
class
...
Ts
>
struct
make_source_result
{
class
make_source_result
:
public
delegated
<
stream
<
typename
DownstreamManager
::
output_type
>
,
Ts
...
>
{
public:
// -- member types -----------------------------------------------------------
/// Type of a single element.
...
...
@@ -40,8 +45,11 @@ struct make_source_result {
/// Pointer to a fully typed stream manager.
using
source_ptr_type
=
intrusive_ptr
<
source_type
>
;
/// The return type for `scheduled_actor::make_source`.
using
output_stream_type
=
output_stream
<
output_type
,
Ts
...
>
;
/// The return type for `scheduled_actor::make_stage`.
using
stream_type
=
stream
<
output_type
>
;
/// Type of user-defined handshake arguments.
using
handshake_arguments
=
std
::
tuple
<
Ts
...
>
;
// -- constructors, destructors, and assignment operators --------------------
...
...
@@ -60,23 +68,17 @@ struct make_source_result {
make_source_result
&
operator
=
(
make_source_result
&&
)
=
default
;
make_source_result
&
operator
=
(
const
make_source_result
&
)
=
default
;
// -- conversion operators ---------------------------------------------------
inline
operator
output_stream_type
()
const
noexcept
{
return
{};
}
// -- properties -------------------------------------------------------------
inline
stream_slot
outbound_slot
()
const
noexcept
{
stream_slot
outbound_slot
()
const
noexcept
{
return
slot_
;
}
inline
source_ptr_type
&
ptr
()
noexcept
{
source_ptr_type
&
ptr
()
noexcept
{
return
ptr_
;
}
inline
const
source_ptr_type
&
ptr
()
const
noexcept
{
const
source_ptr_type
&
ptr
()
const
noexcept
{
return
ptr_
;
}
...
...
libcaf_core/caf/make_stage_result.hpp
View file @
9053a3af
...
...
@@ -18,19 +18,22 @@
#pragma once
#include <tuple>
#include "caf/delegated.hpp"
#include "caf/detail/implicit_conversions.hpp"
#include "caf/fwd.hpp"
#include "caf/
output_
stream.hpp"
#include "caf/stream.hpp"
#include "caf/stream_slot.hpp"
#include "caf/stream_stage.hpp"
#include "caf/detail/implicit_conversions.hpp"
namespace
caf
{
/// Returns a stream stage with the slot IDs of its first in- and outbound
/// paths.
template
<
class
In
,
class
DownstreamManager
,
class
...
Ts
>
class
make_stage_result
{
class
make_stage_result
:
public
delegated
<
stream
<
typename
DownstreamManager
::
output_type
>
,
Ts
...
>
{
public:
// -- member types -----------------------------------------------------------
...
...
@@ -47,7 +50,10 @@ public:
using
stage_ptr_type
=
intrusive_ptr
<
stage_type
>
;
/// The return type for `scheduled_actor::make_stage`.
using
output_stream_type
=
output_stream
<
output_type
,
Ts
...
>
;
using
stream_type
=
stream
<
output_type
>
;
/// Type of user-defined handshake arguments.
using
handshake_arguments
=
std
::
tuple
<
Ts
...
>
;
// -- constructors, destructors, and assignment operators --------------------
...
...
@@ -67,27 +73,21 @@ public:
make_stage_result
&
operator
=
(
make_stage_result
&&
)
=
default
;
make_stage_result
&
operator
=
(
const
make_stage_result
&
)
=
default
;
// -- conversion operators ---------------------------------------------------
inline
operator
output_stream_type
()
const
noexcept
{
return
{};
}
// -- properties -------------------------------------------------------------
inline
stream_slot
inbound_slot
()
const
noexcept
{
stream_slot
inbound_slot
()
const
noexcept
{
return
inbound_slot_
;
}
inline
stream_slot
outbound_slot
()
const
noexcept
{
stream_slot
outbound_slot
()
const
noexcept
{
return
outbound_slot_
;
}
inline
stage_ptr_type
&
ptr
()
noexcept
{
stage_ptr_type
&
ptr
()
noexcept
{
return
ptr_
;
}
inline
const
stage_ptr_type
&
ptr
()
const
noexcept
{
const
stage_ptr_type
&
ptr
()
const
noexcept
{
return
ptr_
;
}
...
...
libcaf_core/caf/output_stream.hpp
deleted
100644 → 0
View file @
5616bec9
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#pragma once
namespace
caf
{
/// Empty marker type for type-checking of stream sources and stages.
template
<
class
T
,
class
...
Ts
>
class
output_stream
{
public:
using
value_type
=
T
;
};
}
// namespace caf
libcaf_core/caf/replies_to.hpp
View file @
9053a3af
...
...
@@ -20,13 +20,7 @@
#include <string>
#include "caf/illegal_message_element.hpp"
#include "caf/output_stream.hpp"
#include "caf/stream.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_pair.hpp"
#include "caf/detail/type_traits.hpp"
namespace
caf
{
...
...
@@ -47,15 +41,9 @@ template <class... Is>
struct
replies_to
{
template
<
class
...
Os
>
using
with
=
typed_mpi
<
detail
::
type_list
<
Is
...
>
,
output_tuple
<
Os
...
>>
;
/// @private
template
<
class
O
,
class
...
Os
>
using
with_stream
=
typed_mpi
<
detail
::
type_list
<
Is
...
>
,
output_stream
<
O
,
Os
...
>>
;
};
template
<
class
...
Is
>
using
reacts_to
=
typed_mpi
<
detail
::
type_list
<
Is
...
>
,
output_tuple
<
void
>>
;
}
// namespace caf
libcaf_core/caf/result.hpp
View file @
9053a3af
...
...
@@ -18,15 +18,17 @@
#pragma once
#include "caf/fwd.hpp"
#include "caf/none.hpp"
#include "caf/skip.hpp"
#include <type_traits>
#include "caf/delegated.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/fwd.hpp"
#include "caf/message.hpp"
#include "caf/delegated.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/none.hpp"
#include "caf/skip.hpp"
namespace
caf
{
...
...
@@ -40,13 +42,15 @@ enum result_runtime_type {
template
<
class
...
Ts
>
class
result
{
public:
result
(
Ts
...
xs
)
:
flag
(
rt_value
),
value
(
make_message
(
std
::
move
(
xs
)...))
{
// nop
}
template
<
class
U
,
class
...
Us
>
result
(
U
x
,
Us
...
xs
)
:
flag
(
rt_value
)
{
init
(
std
::
move
(
x
),
std
::
move
(
xs
)...);
// clang-format off
template
<
class
...
Us
,
class
=
detail
::
enable_if_tt
<
detail
::
all_constructible
<
detail
::
type_list
<
Ts
...>,
detail
::
type_list
<
detail
::
decay_t
<
Us
>
...
>>>>
// clang-format on
result
(
Us
&&
...
xs
)
:
flag
(
rt_value
)
{
value
=
make_message
(
Ts
{
std
::
forward
<
Us
>
(
xs
)}...);
}
template
<
class
E
,
class
=
enable_if_has_make_error_t
<
E
>
>
...
...
@@ -58,16 +62,11 @@ public:
// nop
}
template
<
class
T
,
class
=
typename
std
::
enable_if
<
sizeof
...(
Ts
)
==
1
&&
std
::
is_convertible
<
T
,
detail
::
tl_head_t
<
detail
::
type_list
<
Ts
...>
>
>::
value
>::
type
>
template
<
class
T
,
class
=
typename
std
::
enable_if
<
sizeof
...(
Ts
)
==
1
&&
std
::
is_convertible
<
T
,
detail
::
tl_head_t
<
detail
::
type_list
<
Ts
...>
>>::
value
>::
type
>
result
(
expected
<
T
>
x
)
{
if
(
x
)
{
flag
=
rt_value
;
...
...
libcaf_core/caf/scheduled_actor.hpp
View file @
9053a3af
...
...
@@ -44,7 +44,6 @@
#include "caf/make_source_result.hpp"
#include "caf/make_stage_result.hpp"
#include "caf/no_stages.hpp"
#include "caf/output_stream.hpp"
#include "caf/response_handle.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/sec.hpp"
...
...
libcaf_core/caf/stream_manager.hpp
View file @
9053a3af
...
...
@@ -31,7 +31,6 @@
#include "caf/mailbox_element.hpp"
#include "caf/make_message.hpp"
#include "caf/message_builder.hpp"
#include "caf/output_stream.hpp"
#include "caf/ref_counted.hpp"
#include "caf/stream.hpp"
#include "caf/stream_slot.hpp"
...
...
libcaf_core/caf/stream_slot.hpp
View file @
9053a3af
...
...
@@ -20,8 +20,6 @@
#include <cstdint>
#include "caf/output_stream.hpp"
#include "caf/detail/comparable.hpp"
namespace
caf
{
...
...
@@ -35,7 +33,7 @@ constexpr stream_slot invalid_stream_slot = 0;
/// Maps two `stream_slot` values into a pair for storing sender and receiver
/// slot information.
struct
stream_slots
:
detail
::
comparable
<
stream_slots
>
{
struct
stream_slots
:
detail
::
comparable
<
stream_slots
>
{
stream_slot
sender
;
stream_slot
receiver
;
...
...
@@ -46,8 +44,7 @@ struct stream_slots : detail::comparable<stream_slots>{
}
constexpr
stream_slots
(
stream_slot
sender_slot
,
stream_slot
receiver_slot
)
:
sender
(
sender_slot
),
receiver
(
receiver_slot
)
{
:
sender
(
sender_slot
),
receiver
(
receiver_slot
)
{
// nop
}
...
...
@@ -82,7 +79,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
constexpr
inbound_stream_slot
(
stream_slot
value
=
0
)
:
value_
(
value
)
{
constexpr
inbound_stream_slot
(
stream_slot
value
=
0
)
:
value_
(
value
)
{
// nop
}
...
...
@@ -117,26 +114,28 @@ public:
/// Type of a single element.
using
output_type
=
OutputType
;
/// The return type for `scheduled_actor::make_source`.
using
output_stream_type
=
output_stream
<
output_type
,
HandshakeArgs
...
>
;
/// Type of a stream over the elements.
using
stream_type
=
stream
<
output_type
>
;
/// Type of user-defined handshake arguments.
using
handshake_arguments
=
std
::
tuple
<
HandshakeArgs
...
>
;
// -- constructors, destructors, and assignment operators --------------------
constexpr
outbound_stream_slot
(
stream_slot
value
=
0
)
:
value_
(
value
)
{
constexpr
outbound_stream_slot
(
stream_slot
value
=
0
)
:
value_
(
value
)
{
// nop
}
outbound_stream_slot
(
outbound_stream_slot
&&
)
=
default
;
outbound_stream_slot
(
const
outbound_stream_slot
&
)
=
default
;
outbound_stream_slot
&
operator
=
(
outbound_stream_slot
&&
)
=
default
;
outbound_stream_slot
&
operator
=
(
const
outbound_stream_slot
&
)
=
default
;
// -- conversion operators ---------------------------------------------------
constexpr
operator
output_stream_type
()
const
noexcept
{
return
{};
}
constexpr
operator
stream_slot
()
const
noexcept
{
return
value_
;
}
...
...
libcaf_core/test/composable_behavior.cpp
View file @
9053a3af
...
...
@@ -40,11 +40,9 @@ using i3_actor = typed_actor<replies_to<int, int, int>::with<int>>;
using
d_actor
=
typed_actor
<
replies_to
<
double
>::
with
<
double
,
double
>>
;
using
source_actor
=
typed_actor
<
replies_to
<
open_atom
>::
with
<
output_stream
<
int
>>>
;
using
source_actor
=
typed_actor
<
replies_to
<
open_atom
>::
with
<
stream
<
int
>>>
;
using
stage_actor
=
typed_actor
<
replies_to
<
stream
<
int
>>::
with
<
output_stream
<
int
>>>
;
using
stage_actor
=
typed_actor
<
replies_to
<
stream
<
int
>>::
with
<
stream
<
int
>>>
;
using
sink_actor
=
typed_actor
<
reacts_to
<
stream
<
int
>>>
;
...
...
@@ -92,7 +90,8 @@ struct foo_actor_state2
};
class
source_actor_state
:
public
composable_behavior
<
source_actor
>
{
result
<
output_stream
<
int
>>
operator
()(
open_atom
)
override
{
public:
result
<
stream
<
int
>>
operator
()(
open_atom
)
override
{
return
attach_stream_source
(
self
,
[](
size_t
&
counter
)
{
counter
=
0
;
},
[](
size_t
&
counter
,
downstream
<
int
>&
out
,
size_t
hint
)
{
...
...
@@ -105,7 +104,8 @@ class source_actor_state : public composable_behavior<source_actor> {
};
class
stage_actor_state
:
public
composable_behavior
<
stage_actor
>
{
result
<
output_stream
<
int
>>
operator
()(
stream
<
int
>
in
)
override
{
public:
result
<
stream
<
int
>>
operator
()(
stream
<
int
>
in
)
override
{
return
attach_stream_stage
(
self
,
in
,
[](
unit_t
&
)
{
...
...
@@ -119,6 +119,7 @@ class stage_actor_state : public composable_behavior<stage_actor> {
};
class
sink_actor_state
:
public
composable_behavior
<
sink_actor
>
{
public:
std
::
vector
<
int
>
buf
;
result
<
void
>
operator
()(
stream
<
int
>
in
)
override
{
...
...
@@ -399,13 +400,26 @@ CAF_TEST(dynamic_spawning) {
}
CAF_TEST
(
streaming
)
{
// TODO: currently broken
// auto src = sys.spawn<source_actor_state>();
// auto stg = sys.spawn<stage_actor_state>();
// auto snk = sys.spawn<sink_actor_state>();
// auto pipeline = snk * stg * src;
// self->send(pipeline, open_atom::value);
// sched.run();
auto
src
=
sys
.
spawn
<
source_actor_state
>
();
auto
stg
=
sys
.
spawn
<
stage_actor_state
>
();
auto
snk
=
sys
.
spawn
<
sink_actor_state
>
();
using
src_to_stg
=
typed_actor
<
replies_to
<
open_atom
>::
with
<
stream
<
int
>>>
;
using
stg_to_snk
=
typed_actor
<
reacts_to
<
stream
<
int
>>>
;
static_assert
(
std
::
is_same
<
decltype
(
stg
*
src
),
src_to_stg
>::
value
,
"stg * src produces the wrong type"
);
static_assert
(
std
::
is_same
<
decltype
(
snk
*
stg
),
stg_to_snk
>::
value
,
"stg * src produces the wrong type"
);
auto
pipeline
=
snk
*
stg
*
src
;
self
->
send
(
pipeline
,
open_atom
::
value
);
run
();
using
sink_actor
=
composable_behavior_based_actor
<
sink_actor_state
>
;
auto
&
st
=
deref
<
sink_actor
>
(
snk
).
state
;
CAF_CHECK_EQUAL
(
st
.
buf
.
size
(),
50u
);
auto
is_even
=
[](
int
x
)
{
return
x
%
2
==
0
;
};
CAF_CHECK
(
std
::
all_of
(
st
.
buf
.
begin
(),
st
.
buf
.
end
(),
is_even
));
anon_send_exit
(
src
,
exit_reason
::
user_shutdown
);
anon_send_exit
(
stg
,
exit_reason
::
user_shutdown
);
anon_send_exit
(
snk
,
exit_reason
::
user_shutdown
);
}
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_core/test/continuous_streaming.cpp
View file @
9053a3af
...
...
@@ -48,7 +48,7 @@ TESTEE_STATE(file_reader) {
};
VARARGS_TESTEE
(
file_reader
,
size_t
buf_size
)
{
return
{[
=
](
string
&
fname
)
->
output_stream
<
int
,
string
>
{
return
{[
=
](
string
&
fname
)
->
result
<
stream
<
int
>
,
string
>
{
CAF_CHECK_EQUAL
(
fname
,
"numbers.txt"
);
CAF_CHECK_EQUAL
(
self
->
mailbox
().
empty
(),
true
);
return
attach_stream_source
(
...
...
libcaf_core/test/fused_downstream_manager.cpp
View file @
9053a3af
...
...
@@ -58,7 +58,7 @@ void push(std::deque<T>& xs, downstream<T>& out, size_t num) {
VARARGS_TESTEE
(
int_file_reader
,
size_t
buf_size
)
{
using
buf
=
std
::
deque
<
int32_t
>
;
return
{[
=
](
string
&
fname
)
->
output_stream
<
int32_t
>
{
return
{[
=
](
string
&
fname
)
->
result
<
stream
<
int32_t
>
>
{
CAF_CHECK_EQUAL
(
fname
,
"numbers.txt"
);
return
attach_stream_source
(
self
,
...
...
@@ -76,7 +76,7 @@ VARARGS_TESTEE(int_file_reader, size_t buf_size) {
VARARGS_TESTEE
(
string_file_reader
,
size_t
buf_size
)
{
using
buf
=
std
::
deque
<
string
>
;
return
{[
=
](
string
&
fname
)
->
output_stream
<
string
>
{
return
{[
=
](
string
&
fname
)
->
result
<
stream
<
string
>
>
{
CAF_CHECK_EQUAL
(
fname
,
"strings.txt"
);
return
attach_stream_source
(
self
,
...
...
libcaf_core/test/metaprogramming.cpp
View file @
9053a3af
...
...
@@ -253,34 +253,6 @@ CAF_TEST(typed_behavior_assignment) {
h5
,
h6
,
h7
,
h8
));
}
CAF_TEST
(
composed_types
)
{
// message type for test message #1
auto
msg_1
=
tk
<
type_list
<
int
>>
();
// message type for test message #1
auto
msg_2
=
tk
<
type_list
<
double
>>
();
// interface type a
auto
if_a
=
tk
<
type_list
<
replies_to
<
int
>::
with
<
double
>
,
replies_to
<
double
,
double
>::
with
<
int
,
int
>>>
();
// interface type b
auto
if_b
=
tk
<
type_list
<
replies_to
<
double
>::
with
<
std
::
string
>>>
();
// interface type c
auto
if_c
=
tk
<
type_list
<
replies_to
<
int
>::
with_stream
<
double
>>>
();
// interface type b . a
auto
if_ba
=
tk
<
typed_actor
<
replies_to
<
int
>::
with
<
std
::
string
>>>
();
// interface type b . c
auto
if_bc
=
tk
<
typed_actor
<
replies_to
<
int
>::
with_stream
<
std
::
string
>>>
();
CAF_MESSAGE
(
"check whether actors return the correct types"
);
auto
nil
=
tk
<
none_t
>
();
auto
dbl
=
tk
<
type_list
<
double
>>
();
//auto dbl_stream = tk<output_stream<double>>();
CAF_CHECK_EQUAL
(
res
(
if_a
,
msg_1
),
dbl
);
CAF_CHECK_EQUAL
(
res
(
if_a
,
msg_2
),
nil
);
//CAF_CHECK_EQUAL(res(if_c, msg_1), dbl_stream);
CAF_MESSAGE
(
"check types of actor compositions"
);
CAF_CHECK_EQUAL
(
dot_op
(
if_b
,
if_a
),
if_ba
);
CAF_CHECK_EQUAL
(
dot_op
(
if_b
,
if_c
),
if_bc
);
}
struct
foo
{};
struct
bar
{};
bool
operator
==
(
const
bar
&
,
const
bar
&
);
...
...
libcaf_core/test/pipeline_streaming.cpp
View file @
9053a3af
...
...
@@ -77,7 +77,7 @@ std::function<void(T&, const error&)> fin(scheduled_actor* self) {
}
TESTEE
(
infinite_source
)
{
return
{[
=
](
string
&
fname
)
->
output_stream
<
int
>
{
return
{[
=
](
string
&
fname
)
->
result
<
stream
<
int
>
>
{
CAF_CHECK_EQUAL
(
fname
,
"numbers.txt"
);
CAF_CHECK_EQUAL
(
self
->
mailbox
().
empty
(),
true
);
return
attach_stream_source
(
...
...
@@ -91,7 +91,7 @@ TESTEE(infinite_source) {
}
VARARGS_TESTEE
(
file_reader
,
size_t
buf_size
)
{
return
{[
=
](
string
&
fname
)
->
output_stream
<
int
>
{
return
{[
=
](
string
&
fname
)
->
result
<
stream
<
int
>
>
{
CAF_CHECK_EQUAL
(
fname
,
"numbers.txt"
);
CAF_CHECK_EQUAL
(
self
->
mailbox
().
empty
(),
true
);
return
attach_stream_source
(
self
,
init
(
buf_size
),
push_from_buf
,
...
...
libcaf_core/test/selective_streaming.cpp
View file @
9053a3af
...
...
@@ -73,7 +73,7 @@ buf make_log(level lvl) {
TESTEE_SETUP
();
TESTEE
(
log_producer
)
{
return
{[
=
](
level
lvl
)
->
output_stream
<
value_type
>
{
return
{[
=
](
level
lvl
)
->
result
<
stream
<
value_type
>
>
{
auto
res
=
attach_stream_source
(
self
,
// initialize state
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment