Unverified Commit 29879209 authored by Shariar Azad Riday's avatar Shariar Azad Riday Committed by GitHub

Implement take_last operator

parent cc7f3496
...@@ -135,6 +135,7 @@ caf_add_component( ...@@ -135,6 +135,7 @@ caf_add_component(
caf/detail/private_thread_pool.cpp caf/detail/private_thread_pool.cpp
caf/detail/rfc3629.cpp caf/detail/rfc3629.cpp
caf/detail/rfc3629.test.cpp caf/detail/rfc3629.test.cpp
caf/detail/ring_buffer.test.cpp
caf/detail/set_thread_name.cpp caf/detail/set_thread_name.cpp
caf/detail/stream_bridge.cpp caf/detail/stream_bridge.cpp
caf/detail/stringification_inspector.cpp caf/detail/stringification_inspector.cpp
...@@ -152,6 +153,7 @@ caf_add_component( ...@@ -152,6 +153,7 @@ caf_add_component(
caf/flow/observable_builder.cpp caf/flow/observable_builder.cpp
caf/flow/op/interval.cpp caf/flow/op/interval.cpp
caf/flow/scoped_coordinator.cpp caf/flow/scoped_coordinator.cpp
caf/flow/step/take_last.test.cpp
caf/flow/subscription.cpp caf/flow/subscription.cpp
caf/forwarding_actor_proxy.cpp caf/forwarding_actor_proxy.cpp
caf/group.cpp caf/group.cpp
......
// 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/config.hpp"
#include <algorithm>
#include <memory>
namespace caf::detail {
/// A simple ring buffer implementation.
template <class T>
class ring_buffer {
public:
explicit ring_buffer(size_t max_size) : max_size_(max_size) {
if (max_size > 0)
buf_ = std::make_unique<T[]>(max_size);
}
ring_buffer(ring_buffer&& other) noexcept = default;
ring_buffer& operator=(ring_buffer&& other) noexcept = default;
ring_buffer(const ring_buffer& other) {
max_size_ = other.max_size_;
write_pos_ = other.write_pos_;
size_ = other.size_;
if (max_size_ > 0) {
buf_ = std::make_unique<T[]>(max_size_);
std::copy(other.buf_.get(), other.buf_.get() + other.max_size_,
buf_.get());
}
}
ring_buffer& operator=(const ring_buffer& other) {
ring_buffer tmp{other};
swap(*this, tmp);
return *this;
}
T& front() {
return buf_[(write_pos_ + max_size_ - size_) % max_size_];
}
void pop_front() {
CAF_ASSERT(!empty());
--size_;
}
void push_back(const T& x) {
if (max_size_ == 0)
return;
buf_[write_pos_] = x;
write_pos_ = (write_pos_ + 1) % max_size_;
if (!full())
++size_;
}
bool full() const noexcept {
return size_ == max_size_;
}
bool empty() const noexcept {
return size_ == 0;
}
size_t size() const noexcept {
return size_;
}
friend void swap(ring_buffer& first, ring_buffer& second) noexcept {
using std::swap;
swap(first.buf_, second.buf_);
swap(first.size_, second.size_);
swap(first.max_size_, second.max_size_);
swap(first.write_pos_, second.write_pos_);
}
private:
/// The index for writing new elements.
size_t write_pos_ = 0;
/// Maximum size of the buffer.
size_t max_size_;
/// The number of elements in the buffer currently.
size_t size_ = 0;
/// Stores events in a circular ringbuffer.
std::unique_ptr<T[]> buf_;
};
} // namespace caf::detail
// 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.
#include "caf/detail/ring_buffer.hpp"
#include "caf/test/caf_test_main.hpp"
#include "caf/test/test.hpp"
using namespace caf;
using detail::ring_buffer;
int pop(ring_buffer<int>& buf) {
auto result = buf.front();
buf.pop_front();
return result;
}
TEST("push_back adds element") {
ring_buffer<int> buf{3};
info("full capacity of ring buffer");
for (int i = 1; i <= 3; ++i) {
buf.push_back(i);
}
check_eq(buf.full(), true);
check_eq(buf.empty(), false);
check_eq(buf.front(), 1);
info("additional element after full capacity");
buf.push_back(4);
check_eq(buf.full(), true);
check_eq(buf.empty(), false);
check_eq(buf.front(), 2);
buf.push_back(5);
check_eq(buf.full(), true);
check_eq(buf.empty(), false);
check_eq(buf.front(), 3);
buf.push_back(6);
check_eq(buf.full(), true);
check_eq(buf.empty(), false);
check_eq(buf.front(), 4);
}
TEST("pop_front removes the oldest element") {
ring_buffer<int> buf{3};
info("full capacity of ring buffer");
for (int i = 1; i <= 3; ++i) {
buf.push_back(i);
}
check_eq(buf.full(), true);
check_eq(buf.empty(), false);
check_eq(buf.front(), 1);
info("remove element from buffer");
buf.pop_front();
check_eq(buf.full(), false);
check_eq(buf.empty(), false);
check_eq(buf.front(), 2);
buf.pop_front();
check_eq(buf.full(), false);
check_eq(buf.empty(), false);
check_eq(buf.front(), 3);
buf.pop_front();
check_eq(buf.empty(), true);
}
TEST("circular buffer overwrites oldest element after it is full") {
ring_buffer<int> buf{5};
info("full capacity of ring buffer");
for (int i = 1; i <= 5; ++i) {
buf.push_back(i);
}
check_eq(buf.full(), true);
check_eq(buf.empty(), false);
check_eq(buf.front(), 1);
info("add some elements into buffer");
buf.push_back(6);
buf.push_back(7);
check_eq(buf.full(), true);
check_eq(buf.empty(), false);
check_eq(buf.front(), 3);
buf.push_back(8);
check_eq(buf.full(), true);
check_eq(buf.empty(), false);
check_eq(buf.front(), 4);
buf.push_back(9);
check_eq(buf.full(), true);
check_eq(buf.empty(), false);
check_eq(buf.front(), 5);
}
TEST("pop_front removes the oldest element from the buffer") {
ring_buffer<int> buf{5};
info("full capacity of ring buffer");
for (int i = 1; i <= 5; ++i) {
buf.push_back(i);
}
check_eq(buf.full(), true);
check_eq(buf.empty(), false);
check_eq(buf.front(), 1);
info("remove some element from buffer");
buf.pop_front();
check_eq(buf.full(), false);
check_eq(buf.empty(), false);
check_eq(buf.front(), 2);
buf.pop_front();
check_eq(buf.full(), false);
check_eq(buf.front(), 3);
info("add some elements into buffer");
buf.push_back(6);
buf.push_back(7);
check_eq(buf.full(), true);
check_eq(buf.empty(), false);
check_eq(buf.front(), 3);
buf.push_back(8);
check_eq(buf.full(), true);
check_eq(buf.empty(), false);
check_eq(buf.front(), 4);
buf.push_back(9);
check_eq(buf.full(), true);
check_eq(buf.empty(), false);
check_eq(buf.front(), 5);
}
TEST("push_back does nothing for ring buffer with a capacity of 0") {
ring_buffer<int> buf{0};
info("empty buffer is initialized");
check_eq(buf.size(), 0u);
for (int i = 1; i <= 3; ++i) {
buf.push_back(i);
}
info("buffer size after adding some elements");
check_eq(buf.size(), 0u);
}
TEST("size() returns the number of elements in a buffer") {
ring_buffer<int> buf{5};
info("empty buffer is initialized");
check_eq(buf.size(), 0u);
for (int i = 1; i <= 3; ++i) {
buf.push_back(i);
}
info("buffer size after adding some elements");
check_eq(buf.size(), 3u);
}
TEST("ring-buffers are copiable") {
ring_buffer<int> buf{5};
info("empty buffer is initialized");
check_eq(buf.size(), 0u);
for (int i = 1; i <= 3; ++i) {
buf.push_back(i);
}
check_eq(buf.size(), 3u);
SECTION("copy-assignment") {
ring_buffer<int> new_buf{0};
new_buf = buf;
info("check size and elements of new_buf after copy-assignment");
check_eq(new_buf.size(), 3u);
check_eq(pop(new_buf), 1);
check_eq(pop(new_buf), 2);
check_eq(pop(new_buf), 3);
check_eq(new_buf.empty(), true);
}
SECTION("copy constructor") {
ring_buffer<int> new_buf{buf};
info("check size and elements of new_buf after copy constructor");
check_eq(new_buf.size(), 3u);
check_eq(pop(new_buf), 1);
check_eq(pop(new_buf), 2);
check_eq(pop(new_buf), 3);
check_eq(new_buf.empty(), true);
}
info("check size and elements of buf after copy");
check_eq(buf.size(), 3u);
check_eq(pop(buf), 1);
check_eq(pop(buf), 2);
check_eq(pop(buf), 3);
check_eq(buf.empty(), true);
}
CAF_TEST_MAIN()
...@@ -223,6 +223,11 @@ public: ...@@ -223,6 +223,11 @@ public:
return add_step(step::take<output_type>{n}); return add_step(step::take<output_type>{n});
} }
/// @copydoc observable::take_last
auto take_last(size_t n) && {
return add_step(step::take_last<output_type>{n});
}
/// @copydoc observable::buffer /// @copydoc observable::buffer
auto buffer(size_t count) && { auto buffer(size_t count) && {
return materialize().buffer(count); return materialize().buffer(count);
...@@ -588,6 +593,11 @@ transformation<step::take<T>> observable<T>::take(size_t n) { ...@@ -588,6 +593,11 @@ transformation<step::take<T>> observable<T>::take(size_t n) {
return transform(step::take<T>{n}); return transform(step::take<T>{n});
} }
template <class T>
transformation<step::take_last<T>> observable<T>::take_last(size_t n) {
return transform(step::take_last<T>{n});
}
template <class T> template <class T>
template <class Predicate> template <class Predicate>
transformation<step::take_while<Predicate>> transformation<step::take_while<Predicate>>
......
...@@ -121,6 +121,9 @@ public: ...@@ -121,6 +121,9 @@ public:
/// Returns a transformation that selects only the first `n` items. /// Returns a transformation that selects only the first `n` items.
transformation<step::take<T>> take(size_t n); transformation<step::take<T>> take(size_t n);
/// Returns a transformation that selects only the last `n` items.
transformation<step::take_last<T>> take_last(size_t n);
/// Returns a transformation that selects all value until the `predicate` /// Returns a transformation that selects all value until the `predicate`
/// returns false. /// returns false.
template <class Predicate> template <class Predicate>
......
...@@ -9,4 +9,5 @@ ...@@ -9,4 +9,5 @@
#include "caf/flow/step/reduce.hpp" #include "caf/flow/step/reduce.hpp"
#include "caf/flow/step/skip.hpp" #include "caf/flow/step/skip.hpp"
#include "caf/flow/step/take.hpp" #include "caf/flow/step/take.hpp"
#include "caf/flow/step/take_last.hpp"
#include "caf/flow/step/take_while.hpp" #include "caf/flow/step/take_while.hpp"
...@@ -33,6 +33,9 @@ class skip; ...@@ -33,6 +33,9 @@ class skip;
template <class> template <class>
class take; class take;
template <class>
class take_last;
template <class> template <class>
class take_while; class take_while;
......
// 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/detail/ring_buffer.hpp"
#include "caf/fwd.hpp"
#include <cstddef>
namespace caf::flow::step {
template <class T>
class take_last {
public:
using input_type = T;
using output_type = T;
explicit take_last(size_t num) : elements_(num) {
// nop
}
take_last(take_last&&) = default;
take_last(const take_last&) = default;
take_last& operator=(take_last&&) = default;
take_last& operator=(const take_last&) = default;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
elements_.push_back(item);
return true;
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
while (!elements_.empty()) {
if (!next.on_next(elements_.front(), steps...))
return;
elements_.pop_front();
};
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
private:
detail::ring_buffer<output_type> elements_;
};
} // namespace caf::flow::step
// 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.
#include "caf/flow/step/take_last.hpp"
#include "caf/test/caf_test_main.hpp"
#include "caf/test/test.hpp"
#include "caf/flow/scoped_coordinator.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <vector>
using namespace caf;
using caf::flow::make_observer;
struct fixture {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
template <class... Ts>
static auto ls(Ts... xs) {
return std::vector<int>{xs...};
}
};
WITH_FIXTURE(fixture) {
TEST("calling take_last(5) on range(1, 100) produces[96, 97, 98, 99, 100]") {
auto result = std::vector<int>{};
SECTION("blueprint") {
ctx->make_observable().range(1, 100).take_last(5).for_each(
[&](const int& result_observable) {
result.emplace_back(result_observable);
});
}
SECTION("observable") {
ctx->make_observable().range(1, 100).as_observable().take_last(5).for_each(
[&](const int& result_observable) {
result.emplace_back(result_observable);
});
}
ctx->run();
check_eq(result.size(), 5u);
check_eq(result, ls(96, 97, 98, 99, 100));
}
TEST("calling take_last(5) on range(1, 5) produces[1, 2, 3, 4, 5]") {
auto result = std::vector<int>{};
SECTION("blueprint") {
ctx->make_observable().range(1, 5).take_last(5).for_each(
[&](const int& result_observable) {
result.emplace_back(result_observable);
});
}
SECTION("observable") {
ctx->make_observable().range(1, 5).as_observable().take_last(5).for_each(
[&](const int& result_observable) {
result.emplace_back(result_observable);
});
}
ctx->run();
check_eq(result.size(), 5u);
check_eq(result, ls(1, 2, 3, 4, 5));
}
TEST("calling take_last(5) on range(1, 3) produces[1, 2, 3]") {
auto result = std::vector<int>{};
SECTION("blueprint") {
ctx->make_observable().range(1, 3).take_last(5).for_each(
[&](const int& result_observable) {
result.emplace_back(result_observable);
});
}
SECTION("observable") {
ctx->make_observable().range(1, 3).as_observable().take_last(5).for_each(
[&](const int& result_observable) {
result.emplace_back(result_observable);
});
}
ctx->run();
check_eq(result.size(), 3u);
check_eq(result, ls(1, 2, 3));
}
TEST("calling take(3) on take_last(5) ignores the last two items") {
auto result = std::vector<int>{};
SECTION("blueprint") {
ctx->make_observable().range(1, 100).take_last(5).take(3).for_each(
[&](const int& result_observable) {
result.emplace_back(result_observable);
});
}
SECTION("observable") {
ctx->make_observable()
.range(1, 100)
.as_observable()
.take_last(5)
.take(3)
.for_each([&](const int& result_observable) {
result.emplace_back(result_observable);
});
}
ctx->run();
check_eq(result.size(), 3u);
check_eq(result, ls(96, 97, 98));
}
TEST("take_last operator forwards errors") {
caf::error result;
auto outputs = std::vector<int>{};
SECTION("blueprint") {
ctx->make_observable()
.fail<int>(make_error(sec::runtime_error))
.take_last(5)
.do_on_error([&result](const error& err) { result = err; })
.for_each([&](const int& result_observable) {
outputs.emplace_back(result_observable);
});
}
SECTION("observable") {
ctx->make_observable()
.fail<int>(make_error(sec::runtime_error))
.as_observable()
.take_last(5)
.do_on_error([&result](const error& err) { result = err; })
.for_each([&](const int& result_observable) {
outputs.emplace_back(result_observable);
});
}
ctx->run();
check_eq(result, caf::sec::runtime_error);
check_eq(outputs.size(), 0u);
}
} // WITH_FIXTURE(fixture)
CAF_TEST_MAIN()
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