Commit 26a0016e authored by Dominik Charousset's avatar Dominik Charousset

Fix formatting

parent a40a675c
...@@ -72,21 +72,22 @@ attach_stream_stage(scheduled_actor* self, const stream<In>& in, ...@@ -72,21 +72,22 @@ attach_stream_stage(scheduled_actor* self, const stream<In>& in,
CAF_IGNORE_UNUSED(token); CAF_IGNORE_UNUSED(token);
using output_type = typename stream_stage_trait_t<Fun>::output; using output_type = typename stream_stage_trait_t<Fun>::output;
using state_type = typename stream_stage_trait_t<Fun>::state; using state_type = typename stream_stage_trait_t<Fun>::state;
static_assert(std::is_same< static_assert(
void(state_type&), std::is_same<void(state_type&),
typename detail::get_callable_trait<Init>::fun_sig>::value, typename detail::get_callable_trait<Init>::fun_sig>::value,
"Expected signature `void (State&)` for init function"); "Expected signature `void (State&)` for init function");
using consume_one = void(state_type&, downstream<output_type>&, In); using consume_one = void(state_type&, downstream<output_type>&, In);
using consume_all = void(state_type&, downstream<output_type>&, using consume_all
std::vector<In>&); = void(state_type&, downstream<output_type>&, std::vector<In>&);
using fun_sig = typename detail::get_callable_trait<Fun>::fun_sig; using fun_sig = typename detail::get_callable_trait<Fun>::fun_sig;
static_assert(std::is_same<fun_sig, consume_one>::value static_assert(std::is_same<fun_sig, consume_one>::value
|| std::is_same<fun_sig, consume_all>::value, || std::is_same<fun_sig, consume_all>::value,
"Expected signature `void (State&, downstream<Out>&, In)` " "Expected signature `void (State&, downstream<Out>&, In)` "
"or `void (State&, downstream<Out>&, std::vector<In>&)` " "or `void (State&, downstream<Out>&, std::vector<In>&)` "
"for consume function"); "for consume function");
using driver = detail::stream_stage_driver_impl< using driver
typename Trait::input, DownstreamManager, Fun, Finalize>; = detail::stream_stage_driver_impl<typename Trait::input, DownstreamManager,
Fun, Finalize>;
return attach_stream_stage<driver>(self, in, std::move(xs), std::move(init), return attach_stream_stage<driver>(self, in, std::move(xs), std::move(init),
std::move(fun), std::move(fin)); std::move(fun), std::move(fin));
} }
......
...@@ -132,10 +132,9 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) { ...@@ -132,10 +132,9 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) {
stream_aborter::source_aborter); stream_aborter::source_aborter);
// Send message. // Send message.
unsafe_send_as(self, hdl, unsafe_send_as(self, hdl,
make<upstream_msg::ack_open>(slots.invert(), self->address(), make<upstream_msg::ack_open>(
std::move(rebind_from), slots.invert(), self->address(), std::move(rebind_from),
self->ctrl(), assigned_credit, self->ctrl(), assigned_credit, desired_batch_size));
desired_batch_size));
last_credit_decision = self->clock().now(); last_credit_decision = self->clock().now();
} }
......
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