Adds take while and take until methods

CMakeLists.txt

@@ -41,6 +41,8 @@ set(SOURCES
 	${CMAKE_CURRENT_SOURCE_DIR}/include/recpp/processors/ObserveOn.h
 	${CMAKE_CURRENT_SOURCE_DIR}/include/recpp/processors/Reduce.h
 	${CMAKE_CURRENT_SOURCE_DIR}/include/recpp/processors/SubscribeOn.h
+	${CMAKE_CURRENT_SOURCE_DIR}/include/recpp/processors/Take.h
+	${CMAKE_CURRENT_SOURCE_DIR}/include/recpp/processors/TakeWhile.h
 	${CMAKE_CURRENT_SOURCE_DIR}/include/recpp/processors/Tap.h
 	${CMAKE_CURRENT_SOURCE_DIR}/include/recpp/processors/inl/AllOf.inl
 	${CMAKE_CURRENT_SOURCE_DIR}/include/recpp/processors/inl/AndThen.inl
@@ -61,6 +63,8 @@ set(SOURCES
 	${CMAKE_CURRENT_SOURCE_DIR}/include/recpp/processors/inl/ObserveOn.inl
 	${CMAKE_CURRENT_SOURCE_DIR}/include/recpp/processors/inl/Reduce.inl
 	${CMAKE_CURRENT_SOURCE_DIR}/include/recpp/processors/inl/SubscribeOn.inl
+	${CMAKE_CURRENT_SOURCE_DIR}/include/recpp/processors/inl/Take.inl
+	${CMAKE_CURRENT_SOURCE_DIR}/include/recpp/processors/inl/TakeWhile.inl
 	${CMAKE_CURRENT_SOURCE_DIR}/include/recpp/processors/inl/Tap.inl
 	${CMAKE_CURRENT_SOURCE_DIR}/include/recpp/publishers/CreatePublisher.h
 	${CMAKE_CURRENT_SOURCE_DIR}/include/recpp/publishers/DeferPublisher.h

include/recpp/processors/Take.h

@@ -4,9 +4,7 @@
 
 #include <recpp/subscriptions/FilterSubscription.h>
 
-#include <atomic>
 #include <functional>
-#include <mutex>
 #include <vector>
 
 namespace recpp::processors
@@ -40,7 +38,6 @@ namespace recpp::processors
 			rscpp::Publisher<T>									  m_publisher;
 			std::vector<recpp::subscriptions::FilterSubscription> m_subscriptions;
 			rscpp::Subscriber<T>								  m_subscriber;
-			std::mutex											  m_mutex;
 			std::size_t											  m_count = 0;
 		};
 

include/recpp/processors/TakeWhile.h

@@ -0,0 +1,55 @@
+#pragma once
+
+#include <rscpp/Processor.h>
+
+#include <recpp/subscriptions/FilterSubscription.h>
+
+#include <functional>
+#include <vector>
+
+namespace recpp::processors
+{
+	/**
+	 * @class TakeWhile TakeWhile.h <recpp/processors/TakeWhile.h>
+	 * @brief {@link rscpp::Processor} that will subscribe to a given {@link rscpp::Publisher} and emits items as long as they satisfy the given predicate.
+	 *
+	 * @tparam T The type of element signaled to the {@link rscpp::Subscriber} and signaled from the {@link rscpp::Publisher}.
+	 */
+	template <typename T>
+	class TakeWhile : public rscpp::Processor<T, T>
+	{
+		class Impl : public rscpp::Processor<T, T>
+		{
+		public:
+			explicit Impl(rscpp::Processor<T, T> &parent, const rscpp::Publisher<T> &publisher, const std::function<bool(const T & /* value */)> &predicate);
+
+			void onSubscribe(rscpp::Subscription &subscription) override;
+
+			void onNext(const T &value) override;
+
+			void onError(const std::exception_ptr &error) override;
+
+			void onComplete() override;
+
+			void subscribe(rscpp::Subscriber<T> &subscriber) override;
+
+		private:
+			rscpp::Processor<T, T>								 &m_parent;
+			rscpp::Publisher<T>									  m_publisher;
+			std::vector<recpp::subscriptions::FilterSubscription> m_subscriptions;
+			rscpp::Subscriber<T>								  m_subscriber;
+			std::function<bool(const T & /* value */)>			  m_predicate = 0;
+		};
+
+	public:
+		/**
+		 * @brief Construct a new {@link TakeWhile} instance.
+		 *
+		 * @param publisher The source {@link rscpp::Publisher} the {@link TakeWhile} {@link rscpp::Processor} subscribes to and filter the items.
+		 * @param predicate The maximum number of items to emit.
+		 */
+		explicit TakeWhile(const rscpp::Publisher<T> &publisher, const std::function<bool(const T & /* value */)> &predicate);
+	};
+} // namespace recpp::processors
+
+#include <recpp/processors/inl/TakeWhile.inl>

include/recpp/processors/inl/TakeWhile.inl

@@ -0,0 +1,59 @@
+#pragma once
+
+template <typename T>
+recpp::processors::TakeWhile<T>::TakeWhile(const rscpp::Publisher<T> &publisher, const std::function<bool(const T & /* value */)> &predicate)
+	: rscpp::Processor<T, T>(std::make_shared<Impl>(*this, publisher, predicate))
+{
+}
+
+template <typename T>
+recpp::processors::TakeWhile<T>::Impl::Impl(rscpp::Processor<T, T> &parent, const rscpp::Publisher<T> &publisher,
+											const std::function<bool(const T & /* value */)> &predicate)
+	: m_parent(parent)
+	, m_publisher(publisher)
+	, m_predicate(predicate)
+{
+}
+
+template <typename T>
+void recpp::processors::TakeWhile<T>::Impl::onSubscribe(rscpp::Subscription &subscription)
+{
+	auto filterSubscription = recpp::subscriptions::FilterSubscription(subscription);
+	m_subscriptions.push_back(filterSubscription);
+	m_subscriber.onSubscribe(filterSubscription);
+}
+
+template <typename T>
+void recpp::processors::TakeWhile<T>::Impl::onNext(const T &value)
+{
+	bool stop = !m_predicate(value);
+	if (!stop)
+		m_subscriber.onNext(value);
+	for (auto &subscription : m_subscriptions)
+		subscription.onNext(stop);
+	if (stop)
+	{
+		for (auto &subscription : m_subscriptions)
+			subscription.cancel();
+		m_subscriber.onComplete();
+	}
+}
+
+template <typename T>
+void recpp::processors::TakeWhile<T>::Impl::onError(const std::exception_ptr &error)
+{
+	m_subscriber.onError(error);
+}
+
+template <typename T>
+void recpp::processors::TakeWhile<T>::Impl::onComplete()
+{
+	m_subscriber.onComplete();
+}
+
+template <typename T>
+void recpp::processors::TakeWhile<T>::Impl::subscribe(rscpp::Subscriber<T> &subscriber)
+{
+	m_subscriber = subscriber;
+	m_publisher.subscribe(m_parent);
+}

include/recpp/rx/Observable.h

@@ -196,6 +196,22 @@ namespace recpp::rx
 		 */
 		Observable<T> take(std::size_t count);
 
+		/**
+		 * @brief Emits items as long as each item satisfies the given predicate.
+		 *
+		 * @param predicate The predicate the items must satisfy.
+		 * @return The new {@link Observable} instance.
+		 */
+		Observable<T> takeWhile(const std::function<bool(const T & /* value */)> &predicate);
+
+		/**
+		 * @brief Emits items until one item satisfies the given predicate.
+		 *
+		 * @param stopPredicate The stop predicate.
+		 * @return The new {@link Observable} instance.
+		 */
+		Observable<T> takeUntil(const std::function<bool(const T & /* value */)> &stopPredicate);
+
 		/**
 		 * Convert this {@link Observable} into a {@link Completable} by discarding all contained value.
 		 *

include/recpp/rx/inl/Observable.inl

@@ -20,6 +20,7 @@
 #include <recpp/processors/Reduce.h>
 #include <recpp/processors/SubscribeOn.h>
 #include <recpp/processors/Take.h>
+#include <recpp/processors/TakeWhile.h>
 #include <recpp/processors/Tap.h>
 #include <recpp/publishers/CreatePublisher.h>
 #include <recpp/publishers/DeferPublisher.h>
@@ -133,6 +134,22 @@ recpp::rx::Observable<T> recpp::rx::Observable<T>::take(std::size_t count)
 	return Observable<T>(std::make_shared<processors::Take<T>>(*this, count));
 }
 
+template <typename T>
+recpp::rx::Observable<T> recpp::rx::Observable<T>::takeWhile(const std::function<bool(const T & /* value */)> &predicate)
+{
+	return Observable<T>(std::make_shared<processors::TakeWhile<T>>(*this, predicate));
+}
+
+template <typename T>
+recpp::rx::Observable<T> recpp::rx::Observable<T>::takeUntil(const std::function<bool(const T & /* value */)> &stopPredicate)
+{
+	return Observable<T>(std::make_shared<processors::TakeWhile<T>>(*this,
+																	[stopPredicate](const auto &value)
+																	{
+																		return !stopPredicate(value);
+																	}));
+}
+
 template <typename T>
 recpp::rx::Completable recpp::rx::Observable<T>::ignoreElements()
 {