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
59070cf4
Commit
59070cf4
authored
Oct 04, 2022
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Re-implement prefix-and-tail without prefetching
parent
56d43a75
Changes
6
Show whitespace changes
Inline
Side-by-side
Showing
6 changed files
with
556 additions
and
254 deletions
+556
-254
libcaf_core/caf/flow/observable.hpp
libcaf_core/caf/flow/observable.hpp
+10
-24
libcaf_core/caf/flow/op/mcast.hpp
libcaf_core/caf/flow/op/mcast.hpp
+2
-108
libcaf_core/caf/flow/op/prefetch.hpp
libcaf_core/caf/flow/op/prefetch.hpp
+0
-122
libcaf_core/caf/flow/op/prefix_and_tail.hpp
libcaf_core/caf/flow/op/prefix_and_tail.hpp
+237
-0
libcaf_core/caf/flow/op/ucast.hpp
libcaf_core/caf/flow/op/ucast.hpp
+274
-0
libcaf_core/test/flow/prefix_and_tail.cpp
libcaf_core/test/flow/prefix_and_tail.cpp
+33
-0
No files found.
libcaf_core/caf/flow/observable.hpp
View file @
59070cf4
...
...
@@ -26,7 +26,7 @@
#include "caf/flow/op/interval.hpp"
#include "caf/flow/op/merge.hpp"
#include "caf/flow/op/never.hpp"
#include "caf/flow/op/pref
etch
.hpp"
#include "caf/flow/op/pref
ix_and_tail
.hpp"
#include "caf/flow/op/publish.hpp"
#include "caf/flow/step/all.hpp"
#include "caf/flow/subscription.hpp"
...
...
@@ -681,33 +681,19 @@ auto observable<T>::concat_map(F f) {
template
<
class
T
>
observable
<
cow_tuple
<
cow_vector
<
T
>
,
observable
<
T
>>>
observable
<
T
>::
prefix_and_tail
(
size_t
n
)
{
using
vector_t
=
cow_vector
<
T
>
;
CAF_ASSERT
(
n
>
0
);
auto
do_prefetch
=
[](
auto
in
)
{
auto
ptr
=
op
::
prefetch
<
T
>::
apply
(
std
::
move
(
in
).
as_observable
().
pimpl
());
return
observable
<
T
>
{
std
::
move
(
ptr
)};
};
auto
split
=
share
(
2
);
auto
tail
=
split
.
skip
(
n
).
compose
(
do_prefetch
);
return
split
//
.
take
(
n
)
.
to_vector
()
.
filter
([
n
](
const
vector_t
&
xs
)
{
return
xs
.
size
()
==
n
;
})
.
map
([
tail
](
const
vector_t
&
xs
)
{
return
make_cow_tuple
(
xs
,
tail
);
})
.
as_observable
();
using
impl_t
=
op
::
prefix_and_tail
<
T
>
;
return
make_observable
<
impl_t
>
(
ctx
(),
as_observable
(),
n
);
}
template
<
class
T
>
observable
<
cow_tuple
<
T
,
observable
<
T
>>>
observable
<
T
>::
head_and_tail
()
{
auto
do_prefetch
=
[](
auto
in
)
{
auto
ptr
=
op
::
prefetch
<
T
>::
apply
(
std
::
move
(
in
).
as_observable
().
pimpl
());
return
observable
<
T
>
{
std
::
move
(
ptr
)};
};
auto
split
=
share
(
2
);
auto
tail
=
split
.
skip
(
1
).
compose
(
do_prefetch
);
return
split
//
.
take
(
1
)
.
map
([
tail
](
const
T
&
x
)
{
return
make_cow_tuple
(
x
,
tail
);
})
using
prefix_tuple_t
=
cow_tuple
<
cow_vector
<
T
>
,
observable
<
T
>>
;
return
prefix_and_tail
(
1
)
.
map
([](
const
prefix_tuple_t
&
tup
)
{
auto
&
[
vec
,
obs
]
=
tup
.
data
();
CAF_ASSERT
(
vec
.
size
()
==
1
);
return
make_cow_tuple
(
vec
.
front
(),
obs
);
})
.
as_observable
();
}
...
...
libcaf_core/caf/flow/op/mcast.hpp
View file @
59070cf4
...
...
@@ -8,6 +8,7 @@
#include "caf/flow/observer.hpp"
#include "caf/flow/op/empty.hpp"
#include "caf/flow/op/hot.hpp"
#include "caf/flow/op/ucast.hpp"
#include "caf/flow/subscription.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/make_counted.hpp"
...
...
@@ -20,114 +21,7 @@ namespace caf::flow::op {
/// State shared between one multicast operator and one subscribed observer.
template
<
class
T
>
class
mcast_sub_state
:
public
detail
::
plain_ref_counted
{
public:
friend
void
intrusive_ptr_add_ref
(
const
mcast_sub_state
*
ptr
)
noexcept
{
ptr
->
ref
();
}
friend
void
intrusive_ptr_release
(
const
mcast_sub_state
*
ptr
)
noexcept
{
ptr
->
deref
();
}
mcast_sub_state
(
coordinator
*
ctx
,
observer
<
T
>
out
)
:
ctx
(
ctx
),
out
(
std
::
move
(
out
))
{
// nop
}
coordinator
*
ctx
;
std
::
deque
<
T
>
buf
;
size_t
demand
=
0
;
observer
<
T
>
out
;
bool
disposed
=
false
;
bool
closed
=
false
;
bool
running
=
false
;
error
err
;
action
when_disposed
;
action
when_consumed_some
;
void
push
(
const
T
&
item
)
{
if
(
disposed
)
{
// nop
}
else
if
(
demand
>
0
&&
!
running
)
{
CAF_ASSERT
(
out
);
CAF_ASSERT
(
buf
.
empty
());
--
demand
;
out
.
on_next
(
item
);
if
(
when_consumed_some
)
ctx
->
delay
(
when_consumed_some
);
}
else
{
buf
.
push_back
(
item
);
}
}
void
close
()
{
if
(
!
disposed
)
{
closed
=
true
;
if
(
!
running
&&
buf
.
empty
())
{
disposed
=
true
;
out
.
on_complete
();
out
=
nullptr
;
when_disposed
=
nullptr
;
when_consumed_some
=
nullptr
;
}
}
}
void
abort
(
const
error
&
reason
)
{
if
(
!
disposed
&&
!
err
)
{
closed
=
true
;
err
=
reason
;
if
(
!
running
&&
buf
.
empty
())
{
disposed
=
true
;
out
.
on_error
(
reason
);
out
=
nullptr
;
when_disposed
=
nullptr
;
when_consumed_some
=
nullptr
;
}
}
}
void
do_dispose
()
{
if
(
out
)
{
out
.
on_complete
();
out
=
nullptr
;
}
if
(
when_disposed
)
{
ctx
->
delay
(
std
::
move
(
when_disposed
));
}
if
(
when_consumed_some
)
{
when_consumed_some
.
dispose
();
when_consumed_some
=
nullptr
;
}
buf
.
clear
();
demand
=
0
;
disposed
=
true
;
}
void
do_run
()
{
auto
guard
=
detail
::
make_scope_guard
([
this
]
{
running
=
false
;
});
if
(
!
disposed
)
{
auto
got_some
=
demand
>
0
&&
!
buf
.
empty
();
for
(
bool
run
=
got_some
;
run
;
run
=
demand
>
0
&&
!
buf
.
empty
())
{
out
.
on_next
(
buf
.
front
());
buf
.
pop_front
();
--
demand
;
}
if
(
buf
.
empty
()
&&
closed
)
{
if
(
err
)
out
.
on_error
(
err
);
else
out
.
on_complete
();
out
=
nullptr
;
do_dispose
();
}
else
if
(
got_some
&&
when_consumed_some
)
{
ctx
->
delay
(
when_consumed_some
);
}
}
}
};
using
mcast_sub_state
=
ucast_sub_state
<
T
>
;
template
<
class
T
>
using
mcast_sub_state_ptr
=
intrusive_ptr
<
mcast_sub_state
<
T
>>
;
...
...
libcaf_core/caf/flow/op/prefetch.hpp
deleted
100644 → 0
View file @
56d43a75
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/error.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/hot.hpp"
#include "caf/flow/subscription.hpp"
#include "caf/sec.hpp"
#include <utility>
namespace
caf
::
flow
::
op
{
/// Allows operators to subscribe to an observable immediately to force an eager
/// subscription while the observable that actually consumes the items
/// subscribes later. May only be subscribed once.
template
<
class
T
>
class
prefetch
:
public
hot
<
T
>
,
public
observer_impl
<
T
>
{
public:
// -- member types -----------------------------------------------------------
using
super
=
hot
<
T
>
;
using
output_type
=
T
;
// -- constructors, destructors, and assignment operators --------------------
explicit
prefetch
(
coordinator
*
ctx
)
:
super
(
ctx
)
{
// nop
}
// -- ref counting (and disambiguation due to multiple base types) -----------
void
ref_coordinated
()
const
noexcept
override
{
this
->
ref
();
}
void
deref_coordinated
()
const
noexcept
override
{
this
->
deref
();
}
friend
void
intrusive_ptr_add_ref
(
const
prefetch
*
ptr
)
noexcept
{
ptr
->
ref_coordinated
();
}
friend
void
intrusive_ptr_release
(
const
prefetch
*
ptr
)
noexcept
{
ptr
->
deref_coordinated
();
}
// -- implementation of observable_impl<T> -----------------------------------
disposable
subscribe
(
observer
<
T
>
out
)
override
{
if
(
completed_
)
{
if
(
err_
)
out
.
on_error
(
err_
);
else
out
.
on_complete
();
return
{};
}
else
if
(
!
out_
&&
sub_
)
{
out_
=
std
::
move
(
out
);
out_
.
on_subscribe
(
sub_
);
return
sub_
.
as_disposable
();
}
else
{
auto
err
=
make_error
(
sec
::
invalid_observable
,
"prefetch cannot add more than one subscriber"
);
out
.
on_error
(
err
);
return
{};
}
}
// -- implementation of observer_impl<T> -------------------------------------
void
on_next
(
const
T
&
item
)
override
{
out_
.
on_next
(
item
);
}
void
on_complete
()
override
{
completed_
=
true
;
if
(
out_
)
{
out_
.
on_complete
();
out_
=
nullptr
;
sub_
=
nullptr
;
}
}
void
on_error
(
const
error
&
what
)
override
{
completed_
=
true
;
err_
=
what
;
if
(
out_
)
{
out_
.
on_error
(
what
);
out_
=
nullptr
;
sub_
=
nullptr
;
}
}
void
on_subscribe
(
subscription
sub
)
override
{
if
(
!
sub_
)
{
sub_
=
sub
;
}
else
{
sub
.
dispose
();
}
}
// -- convenience functions --------------------------------------------------
static
intrusive_ptr
<
base
<
T
>>
apply
(
intrusive_ptr
<
base
<
T
>>
src
)
{
auto
ptr
=
make_counted
<
prefetch
>
(
src
->
ctx
());
src
->
subscribe
(
observer
<
T
>
{
ptr
});
return
ptr
;
}
private:
bool
completed_
=
false
;
error
err_
;
observer
<
T
>
out_
;
subscription
sub_
;
};
}
// namespace caf::flow::op
libcaf_core/caf/flow/op/prefix_and_tail.hpp
0 → 100644
View file @
59070cf4
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/cow_tuple.hpp"
#include "caf/cow_vector.hpp"
#include "caf/flow/coordinator.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/cold.hpp"
#include "caf/flow/op/ucast.hpp"
#include "caf/flow/subscription.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/make_counted.hpp"
#include <memory>
#include <vector>
namespace
caf
::
flow
::
op
{
/// @relates prefix_and_tail
template
<
class
T
>
class
prefix_and_tail_sub
:
public
detail
::
plain_ref_counted
,
public
observer_impl
<
T
>
,
public
subscription_impl
{
public:
using
tuple_t
=
cow_tuple
<
cow_vector
<
T
>
,
observable
<
T
>>
;
prefix_and_tail_sub
(
coordinator
*
ctx
,
observer
<
tuple_t
>
out
,
size_t
prefix_size
)
:
ctx_
(
ctx
),
out_
(
std
::
move
(
out
)),
prefix_size_
(
prefix_size
)
{
prefix_buf_
.
reserve
(
prefix_size
);
}
void
ref_coordinated
()
const
noexcept
override
{
ref
();
}
void
deref_coordinated
()
const
noexcept
override
{
deref
();
}
void
on_next
(
const
T
&
item
)
override
{
if
(
sink_
)
{
// We're in tail mode: push to sink.
CAF_ASSERT
(
in_flight_
>
0
);
--
in_flight_
;
sink_
->
push
(
item
);
}
else
if
(
out_
)
{
// We're waiting for the prefix.
prefix_buf_
.
push_back
(
item
);
if
(
prefix_buf_
.
size
()
==
prefix_size_
)
{
// Create the sink to deliver to tail lazily and deliver the prefix.
sink_
=
make_counted
<
ucast
<
T
>>
(
ctx_
);
set_callbacks
();
// Force member to be null before calling on_next / on_complete.
auto
out
=
std
::
move
(
out_
);
auto
tup
=
make_cow_tuple
(
cow_vector
<
T
>
{
std
::
move
(
prefix_buf_
)},
observable
<
T
>
{
sink_
});
out
.
on_next
(
tup
);
out
.
on_complete
();
}
}
}
void
on_error
(
const
error
&
reason
)
override
{
if
(
sink_
)
{
sink_
->
state
().
when_demand_changed
=
nullptr
;
sink_
->
abort
(
reason
);
sub_
=
nullptr
;
}
else
if
(
out_
)
{
out_
.
on_error
(
reason
);
out_
=
nullptr
;
}
}
void
on_complete
()
override
{
if
(
sink_
)
{
sink_
->
state
().
when_demand_changed
=
nullptr
;
sink_
->
close
();
sub_
=
nullptr
;
}
else
if
(
out_
)
{
out_
.
on_complete
();
out_
=
nullptr
;
}
}
void
on_subscribe
(
flow
::
subscription
sub
)
override
{
if
(
!
sub_
&&
out_
)
{
sub_
=
std
::
move
(
sub
);
if
(
prefix_demand_
>
0
&&
!
requested_prefix_
)
{
sub_
.
request
(
prefix_size_
);
requested_prefix_
=
true
;
}
}
else
{
sub
.
dispose
();
}
}
void
dispose
()
override
{
if
(
out_
)
{
out_
=
nullptr
;
if
(
sub_
)
{
sub_
.
dispose
();
sub_
=
nullptr
;
}
}
}
bool
disposed
()
const
noexcept
override
{
return
!
out_
&&
!
sink_
;
}
void
ref_disposable
()
const
noexcept
override
{
ref
();
}
void
deref_disposable
()
const
noexcept
override
{
deref
();
}
void
request
(
size_t
demand
)
override
{
// Only called by the out_, never by the sink_. The latter triggers
prefix_demand_
+=
demand
;
if
(
sub_
&&
!
requested_prefix_
)
{
sub_
.
request
(
prefix_size_
);
requested_prefix_
=
true
;
}
}
private:
intrusive_ptr
<
prefix_and_tail_sub
>
strong_this
()
{
return
{
this
};
}
void
set_callbacks
()
{
auto
sptr
=
strong_this
();
auto
demand_cb
=
[
sptr
]
{
sptr
->
on_sink_demand_change
();
};
sink_
->
state
().
when_demand_changed
=
make_action
(
std
::
move
(
demand_cb
));
auto
disposed_cb
=
[
sptr
]
{
sptr
->
on_sink_dispose
();
};
sink_
->
state
().
when_disposed
=
make_action
(
std
::
move
(
disposed_cb
));
}
void
on_sink_demand_change
()
{
if
(
sink_
&&
sub_
)
{
auto
&
st
=
sink_
->
state
();
auto
pending
=
in_flight_
+
st
.
buf
.
size
();
if
(
st
.
demand
>
pending
)
{
auto
delta
=
st
.
demand
-
pending
;
in_flight_
+=
delta
;
sub_
.
request
(
delta
);
}
}
}
void
on_sink_dispose
()
{
sink_
=
nullptr
;
if
(
sub_
)
{
sub_
.
dispose
();
sub_
=
nullptr
;
}
}
/// Our scheduling context.
coordinator
*
ctx_
;
/// The observer for the initial prefix-and-tail tuple.
observer
<
tuple_t
>
out_
;
/// Caches items for the prefix until we can emit them.
std
::
vector
<
T
>
prefix_buf_
;
/// Allows us to push to the "tail" observable after emitting the prefix.
ucast_ptr
<
T
>
sink_
;
/// Pulls data from the decorated observable.
flow
::
subscription
sub_
;
/// Stores how much items are currently in-flight while receiving the tail.
size_t
in_flight_
=
0
;
/// Stores whether we have asked the decorated observable for data yet.
bool
requested_prefix_
=
false
;
/// Keeps track of demand of @p out_ while we receive the prefix.
size_t
prefix_demand_
=
0
;
/// Stores how many items we need to buffer for the prefix.
size_t
prefix_size_
;
};
/// @relates prefix_and_tail_sub
template
<
class
T
>
void
intrusive_ptr_add_ref
(
prefix_and_tail_sub
<
T
>*
ptr
)
{
ptr
->
ref
();
}
/// @relates prefix_and_tail_sub
template
<
class
T
>
void
intrusive_ptr_release
(
prefix_and_tail_sub
<
T
>*
ptr
)
{
ptr
->
deref
();
}
/// Decorates an observable to split its output into a prefix of fixed size plus
/// an observable remainder.
template
<
class
T
>
class
prefix_and_tail
:
public
cold
<
cow_tuple
<
cow_vector
<
T
>
,
observable
<
T
>>>
{
public:
// -- member types -----------------------------------------------------------
using
tuple_t
=
cow_tuple
<
cow_vector
<
T
>
,
observable
<
T
>>
;
using
super
=
cold
<
tuple_t
>
;
// -- constructors, destructors, and assignment operators --------------------
explicit
prefix_and_tail
(
coordinator
*
ctx
,
observable
<
T
>
decorated
,
size_t
prefix_size
)
:
super
(
ctx
),
decorated_
(
std
::
move
(
decorated
)),
prefix_size_
(
prefix_size
)
{
// nop
}
disposable
subscribe
(
observer
<
tuple_t
>
out
)
override
{
using
impl_t
=
prefix_and_tail_sub
<
T
>
;
auto
obs
=
make_counted
<
impl_t
>
(
super
::
ctx
(),
out
,
prefix_size_
);
decorated_
.
subscribe
(
observer
<
T
>
{
obs
});
out
.
on_subscribe
(
subscription
{
obs
});
return
obs
->
as_disposable
();
}
private:
observable
<
T
>
decorated_
;
size_t
prefix_size_
;
};
}
// namespace caf::flow::op
libcaf_core/caf/flow/op/ucast.hpp
0 → 100644
View file @
59070cf4
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/flow/coordinator.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/empty.hpp"
#include "caf/flow/op/hot.hpp"
#include "caf/flow/subscription.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/make_counted.hpp"
#include <deque>
#include <memory>
namespace
caf
::
flow
::
op
{
/// State shared between one multicast operator and one subscribed observer.
template
<
class
T
>
class
ucast_sub_state
:
public
detail
::
plain_ref_counted
{
public:
friend
void
intrusive_ptr_add_ref
(
const
ucast_sub_state
*
ptr
)
noexcept
{
ptr
->
ref
();
}
friend
void
intrusive_ptr_release
(
const
ucast_sub_state
*
ptr
)
noexcept
{
ptr
->
deref
();
}
explicit
ucast_sub_state
(
coordinator
*
ptr
)
:
ctx
(
ptr
)
{
// nop
}
ucast_sub_state
(
coordinator
*
ctx
,
observer
<
T
>
out
)
:
ctx
(
ctx
),
out
(
std
::
move
(
out
))
{
// nop
}
coordinator
*
ctx
;
std
::
deque
<
T
>
buf
;
size_t
demand
=
0
;
observer
<
T
>
out
;
bool
disposed
=
false
;
bool
closed
=
false
;
bool
running
=
false
;
error
err
;
action
when_disposed
;
action
when_consumed_some
;
action
when_demand_changed
;
void
push
(
const
T
&
item
)
{
if
(
disposed
)
{
// nop
}
else
if
(
demand
>
0
&&
!
running
)
{
CAF_ASSERT
(
out
);
CAF_ASSERT
(
buf
.
empty
());
--
demand
;
out
.
on_next
(
item
);
if
(
when_consumed_some
)
ctx
->
delay
(
when_consumed_some
);
}
else
{
buf
.
push_back
(
item
);
}
}
void
close
()
{
if
(
!
disposed
)
{
closed
=
true
;
if
(
!
running
&&
buf
.
empty
())
{
disposed
=
true
;
if
(
out
)
{
out
.
on_complete
();
out
=
nullptr
;
}
when_disposed
=
nullptr
;
when_consumed_some
=
nullptr
;
when_demand_changed
=
nullptr
;
}
}
}
void
abort
(
const
error
&
reason
)
{
if
(
!
disposed
&&
!
err
)
{
closed
=
true
;
err
=
reason
;
if
(
!
running
&&
buf
.
empty
())
{
disposed
=
true
;
if
(
out
)
{
out
.
on_error
(
reason
);
out
=
nullptr
;
}
when_disposed
=
nullptr
;
when_consumed_some
=
nullptr
;
when_demand_changed
=
nullptr
;
}
}
}
void
do_dispose
()
{
if
(
out
)
{
out
.
on_complete
();
out
=
nullptr
;
}
if
(
when_disposed
)
{
ctx
->
delay
(
std
::
move
(
when_disposed
));
}
if
(
when_consumed_some
)
{
when_consumed_some
.
dispose
();
when_consumed_some
=
nullptr
;
}
when_demand_changed
=
nullptr
;
buf
.
clear
();
demand
=
0
;
disposed
=
true
;
}
void
do_run
()
{
auto
guard
=
detail
::
make_scope_guard
([
this
]
{
running
=
false
;
});
if
(
!
disposed
)
{
auto
got_some
=
demand
>
0
&&
!
buf
.
empty
();
for
(
bool
run
=
got_some
;
run
;
run
=
demand
>
0
&&
!
buf
.
empty
())
{
out
.
on_next
(
buf
.
front
());
buf
.
pop_front
();
--
demand
;
}
if
(
buf
.
empty
()
&&
closed
)
{
if
(
err
)
out
.
on_error
(
err
);
else
out
.
on_complete
();
out
=
nullptr
;
do_dispose
();
}
else
if
(
got_some
&&
when_consumed_some
)
{
ctx
->
delay
(
when_consumed_some
);
}
}
}
};
template
<
class
T
>
using
ucast_sub_state_ptr
=
intrusive_ptr
<
ucast_sub_state
<
T
>>
;
template
<
class
T
>
class
ucast_sub
:
public
subscription
::
impl_base
{
public:
// -- constructors, destructors, and assignment operators --------------------
ucast_sub
(
coordinator
*
ctx
,
ucast_sub_state_ptr
<
T
>
state
)
:
ctx_
(
ctx
),
state_
(
std
::
move
(
state
))
{
// nop
}
// -- implementation of subscription -----------------------------------------
bool
disposed
()
const
noexcept
override
{
return
!
state_
;
}
void
dispose
()
override
{
if
(
state_
)
{
ctx_
->
delay_fn
([
state
=
std
::
move
(
state_
)]()
{
state
->
do_dispose
();
});
}
}
void
request
(
size_t
n
)
override
{
state_
->
demand
+=
n
;
if
(
state_
->
when_demand_changed
)
state_
->
when_demand_changed
.
run
();
if
(
!
state_
->
running
)
{
state_
->
running
=
true
;
ctx_
->
delay_fn
([
state
=
state_
]
{
state
->
do_run
();
});
}
}
private:
/// Stores the context (coordinator) that runs this flow.
coordinator
*
ctx_
;
/// Stores a handle to the state.
ucast_sub_state_ptr
<
T
>
state_
;
};
// Base type for *hot* operators that "unicast" data to a subscribed observer.
template
<
class
T
>
class
ucast
:
public
hot
<
T
>
{
public:
// -- member types -----------------------------------------------------------
using
super
=
hot
<
T
>
;
using
state_type
=
ucast_sub_state
<
T
>
;
using
state_ptr_type
=
ucast_sub_state_ptr
<
T
>
;
using
observer_type
=
observer
<
T
>
;
// -- constructors, destructors, and assignment operators --------------------
explicit
ucast
(
coordinator
*
ctx
)
:
super
(
ctx
)
{
state_
=
make_counted
<
ucast_sub_state
<
T
>>
(
ctx
);
}
/// Pushes @p item to the subscriber or buffers them until subscribed.
void
push
(
const
T
&
item
)
{
state_
->
push
(
item
);
}
/// Closes the operator, eventually emitting on_complete on all observers.
void
close
()
{
state_
->
close
();
}
/// Closes the operator, eventually emitting on_error on all observers.
void
abort
(
const
error
&
reason
)
{
state_
->
abort
(
reason
);
}
size_t
demand
()
const
noexcept
{
return
state_
->
demand
;
}
size_t
buffered
()
const
noexcept
{
return
state_
->
buf
.
size
();
}
/// Queries the current number of subscribed observers.
size_t
has_observer
()
const
noexcept
{
return
state_
.
out
.
valid
();
}
bool
disposed
()
const
noexcept
{
return
state_
->
disposed
;
}
state_type
&
state
()
{
return
*
state_
;
}
state_ptr_type
state_ptr
()
{
return
state_
;
}
disposable
subscribe
(
observer
<
T
>
out
)
override
{
if
(
state_
->
closed
)
{
if
(
state_
->
err
)
{
out
.
on_error
(
state_
->
err
);
return
disposable
{};
}
return
make_counted
<
op
::
empty
<
T
>>
(
super
::
ctx_
)
->
subscribe
(
out
);
}
if
(
state_
->
out
)
{
auto
err
=
make_error
(
sec
::
too_many_observers
,
"may only subscribe once to an unicast operator"
);
out
.
on_error
(
err
);
return
disposable
{};
}
state_
->
out
=
out
;
auto
sub_ptr
=
make_counted
<
ucast_sub
<
T
>>
(
super
::
ctx
(),
state_
);
out
.
on_subscribe
(
subscription
{
sub_ptr
});
return
disposable
{
sub_ptr
};
}
private:
state_ptr_type
state_
;
};
/// @relates ucast
template
<
class
T
>
using
ucast_ptr
=
intrusive_ptr
<
ucast
<
T
>>
;
}
// namespace caf::flow::op
libcaf_core/test/flow/prefix_and_tail.cpp
View file @
59070cf4
...
...
@@ -29,6 +29,15 @@ auto ls(T x, Ts... xs) {
return
std
::
vector
<
T
>
{
x
,
xs
...};
}
// Note: last is inclusive.
template
<
class
T
>
auto
ls_range
(
T
first
,
T
last
)
{
auto
result
=
std
::
vector
<
T
>
{};
for
(;
first
<=
last
;
++
first
)
result
.
push_back
(
first
);
return
result
;
}
}
// namespace
BEGIN_FIXTURE_SCOPE
(
fixture
)
...
...
@@ -113,6 +122,30 @@ SCENARIO("prefix_and_tail splits off initial elements") {
}
}
}
GIVEN
(
"a generation with 256 values"
)
{
WHEN
(
"calling prefix_and_tail(7)"
)
{
THEN
(
"the observer receives the first 7 elements plus remainder"
)
{
auto
snk
=
flow
::
make_auto_observer
<
int
>
();
auto
flat_map_calls
=
0
;
ctx
->
make_observable
()
.
iota
(
1
)
.
take
(
256
)
.
prefix_and_tail
(
7
)
.
flat_map
([
&
](
const
tuple_t
&
x
)
{
++
flat_map_calls
;
auto
&
[
prefix
,
tail
]
=
x
.
data
();
CHECK_EQ
(
prefix
,
ls
(
1
,
2
,
3
,
4
,
5
,
6
,
7
));
return
tail
;
})
.
subscribe
(
snk
->
as_observer
());
ctx
->
run
();
CHECK_EQ
(
flat_map_calls
,
1
);
CHECK_EQ
(
snk
->
buf
,
ls_range
(
8
,
256
));
CHECK_EQ
(
snk
->
state
,
flow
::
observer_state
::
completed
);
CHECK_EQ
(
snk
->
err
,
error
{});
}
}
}
}
SCENARIO
(
"head_and_tail splits off the first element"
)
{
...
...
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