Fix reading from freed memory in rpl::take().

Telegram/SourceFiles/rpl/consumer.h

@@ -329,6 +329,11 @@ public:
 	Type *make_state(Args&& ...args) const;
 
 	void terminate() const;
+	auto terminator() const {
+		return [self = *this] {
+			self.terminate();
+		};
+	}
 
 	const details::type_erased_handlers<Value, Error> *comparable() const {
 		return _handlers.get();

Telegram/SourceFiles/rpl/flatten_latest.h

@@ -44,7 +44,7 @@ public:
 			[consumer, state](producer<Value, Error> &&inner) {
 				state->finished = false;
 				state->alive = lifetime();
-				auto started = std::move(inner).start(
+				std::move(inner).start(
 				[consumer](auto &&value) {
 					consumer.put_next_forward(std::forward<decltype(value)>(value));
 				}, [consumer](auto &&error) {
@@ -55,10 +55,7 @@ public:
 					} else {
 						state->finished = true;
 					}
-				});
+				}, state->alive);
-				if (started) {
-					state->alive = std::move(started);
-				}
 			}, [consumer](auto &&error) {
 				consumer.put_error_forward(std::forward<decltype(error)>(error));
 			}, [consumer, state] {

Telegram/SourceFiles/rpl/operators_tests.cpp

@@ -406,6 +406,24 @@ TEST_CASE("basic operators tests", "[rpl::operators]") {
 					*sum += "done";
 				}, lifetime);
 		}
-		REQUIRE(*sum == "012done");
+		{
+			rpl::lifetime lifetime;
+			rpl::ints(3) | take(3)
+				| start_with_next_done([=](int value) {
+					*sum += std::to_string(value);
+				}, [=] {
+					*sum += "done";
+				}, lifetime);
+		}
+		{
+			rpl::lifetime lifetime;
+			rpl::ints(3) | take(10)
+				| start_with_next_done([=](int value) {
+					*sum += std::to_string(value);
+				}, [=] {
+					*sum += "done";
+				}, lifetime);
+		}
+		REQUIRE(*sum == "012done012done012done");
 	}
 }

Telegram/SourceFiles/rpl/producer.h

@@ -346,7 +346,7 @@ template <typename Handlers>
 inline void producer_base<Value, Error, Generator>::start_existing(
 		const consumer_type<Handlers> &consumer,
 		lifetime &alive_while) && {
-	alive_while.add([consumer] { consumer.terminate(); });
+	alive_while.add(consumer.terminator());
 	consumer.add_lifetime(std::move(_generator)(consumer));
 }
 

Telegram/SourceFiles/rpl/take.h

@@ -38,7 +38,7 @@ public:
 			limit = _count
 		](const auto &consumer) mutable {
 			auto count = consumer.template make_state<int>(limit);
-			return std::move(initial).start(
+			auto initial_consumer = make_consumer<Value, Error>(
 			[consumer, count](auto &&value) {
 				auto left = (*count)--;
 				if (left) {
@@ -55,6 +55,8 @@ public:
 			}, [consumer] {
 				consumer.put_done();
 			});
+			consumer.add_lifetime(initial_consumer.terminator());
+			return std::move(initial).start_existing(initial_consumer);
 		});
 	}