ReactivePlusPlus
ReactiveX implementation for C++20
Loading...
Searching...
No Matches
last.hpp
1// ReactivePlusPlus library
2//
3// Copyright Aleksey Loginov 2023 - present.
4// Distributed under the Boost Software License, Version 1.0.
5// (See accompanying file LICENSE_1_0.txt or copy at
6// https://www.boost.org/LICENSE_1_0.txt)
7//
8// Project home: https://github.com/victimsnino/ReactivePlusPlus
9//
10
11#pragma once
12
13#include <rpp/operators/fwd.hpp>
14
15#include <rpp/defs.hpp>
16#include <rpp/operators/details/strategy.hpp>
17
18#include <optional>
19
20namespace rpp::operators::details
21{
22 template<rpp::constraint::decayed_type Type, rpp::constraint::observer TObserver>
24 {
25 static constexpr auto preferred_disposables_mode = rpp::details::observers::disposables_mode::None;
26
27 RPP_NO_UNIQUE_ADDRESS TObserver observer;
28 mutable std::optional<Type> value{};
29
30 template<typename T>
31 void on_next(T&& v) const
32 {
33 value.emplace(std::forward<T>(v));
34 }
35
36 void on_completed() const
37 {
38 if (value.has_value())
39 {
40 observer.on_next(std::move(value).value());
42 }
43 else
44 observer.on_error(std::make_exception_ptr(utils::not_enough_emissions{"last() operator expects at least one emission from observable before completion"}));
45 }
46
47 void on_error(const std::exception_ptr& err) const { observer.on_error(err); }
48
49 void set_upstream(const disposable_wrapper& d) { observer.set_upstream(d); }
50
51 bool is_disposed() const { return observer.is_disposed(); }
52 };
53
54 struct last_t : lift_operator<last_t>
55 {
56 using lift_operator<last_t>::lift_operator;
57
58 template<rpp::constraint::decayed_type T>
60 {
61 using result_type = T;
62
63 template<rpp::constraint::observer_of_type<result_type> TObserver>
65 };
66
67 template<rpp::details::observables::constraint::disposables_strategy Prev>
68 using updated_optimal_disposables_strategy = Prev;
69 };
70} // namespace rpp::operators::details
71
72namespace rpp::operators
73{
99 inline auto last()
100 {
101 return details::last_t{};
102 }
103} // namespace rpp::operators
void on_next(const Type &v) const noexcept
Observable calls this method to notify observer about new value.
Definition observer.hpp:84
void set_upstream(const disposable_wrapper &d) noexcept
Observable calls this method to pass disposable. Observer disposes this disposable WHEN observer want...
Definition observer.hpp:49
void on_completed() const noexcept
Observable calls this method to notify observer about completion of emissions.
Definition observer.hpp:135
void on_error(const std::exception_ptr &err) const noexcept
Observable calls this method to notify observer about some error during generation next data.
Definition observer.hpp:120
bool is_disposed() const noexcept
Observable calls this method to check if observer interested or not in emissions.
Definition observer.hpp:74
Base class for any observer used in RPP. It handles core callbacks of observers. Objects of this clas...
Definition observer.hpp:172
Definition strategy.hpp:28
auto last()
Emit only the last item provided before on_completed.
Definition last.hpp:95
Definition last.hpp:55
Definition exceptions.hpp:18