input/cache: synchronize lease notifications with lifetime

src/input/cache/Lease.hxx

@@ -36,8 +36,7 @@ public:
 	}
 
 	~InputCacheLease() noexcept {
-		if (item != nullptr)
-			item->RemoveLease(*this);
+		Reset();
 	}
 
 	InputCacheLease &operator=(InputCacheLease &&src) noexcept {
@@ -68,6 +67,13 @@ public:
 		return *item;
 	}
 
+	void Reset() noexcept {
+		if (item != nullptr) {
+			item->RemoveLease(*this);
+			item = nullptr;
+		}
+	}
+
 	/**
 	 * Caller locks #InputCacheItem::mutex.
 	 */

src/input/cache/Stream.cxx

@@ -7,20 +7,29 @@
 CacheInputStream::CacheInputStream(InputCacheLease _lease,
 				   Mutex &_mutex) noexcept
 	:InputStream(_lease->GetUri().c_str(), _mutex),
-	 InputCacheLease(std::move(_lease))
+	 lease(*this)
 {
-	const auto &i = GetCacheItem();
+	const auto &i = _lease.GetCacheItem();
 	size = i.size();
 	seekable = true;
 	SetReady();
+
+	/* Register only after this object has been fully initialized. */
+	lease.Set(std::move(_lease));
+}
+
+CacheInputStream::~CacheInputStream() noexcept
+{
+	/* Wait for a running notification before destructing this object. */
+	lease.Reset();
 }
 
 void
 CacheInputStream::Check()
 {
 	const ScopeUnlock unlock(mutex);
 
-	auto &i = GetCacheItem();
+	auto &i = lease.GetCacheItem();
 	const std::lock_guard protect{i.mutex};
 
 	i.Check();
@@ -44,7 +53,7 @@ CacheInputStream::IsAvailable() const noexcept
 	const auto _offset = offset;
 	const ScopeUnlock unlock(mutex);
 
-	auto &i = GetCacheItem();
+	auto &i = lease.GetCacheItem();
 	const std::lock_guard protect{i.mutex};
 
 	return i.IsAvailable(_offset);
@@ -55,7 +64,7 @@ CacheInputStream::Read(std::unique_lock<Mutex> &lock,
 		       std::span<std::byte> dest)
 {
 	const auto _offset = offset;
-	auto &i = GetCacheItem();
+	auto &i = lease.GetCacheItem();
 
 	size_t nbytes;
 
@@ -76,9 +85,7 @@ CacheInputStream::Read(std::unique_lock<Mutex> &lock,
 void
 CacheInputStream::OnInputCacheAvailable(std::unique_lock<Mutex> &lock) noexcept
 {
-	assert(lock.mutex() == &GetCacheItem().mutex);
-
-	const ScopeUnlock unlock{lock};
+	assert(lock.mutex() == &lease.GetCacheItem().mutex);
 
 	const std::lock_guard protect{mutex};
 	InvokeOnAvailable();

src/input/cache/Stream.hxx

@@ -10,9 +10,29 @@
  * An #InputStream implementation which reads data from an
  * #InputCacheItem.
  */
-class CacheInputStream final : public InputStream, InputCacheLease {
+class CacheInputStream final : public InputStream {
+	class Lease final : public InputCacheLease {
+		CacheInputStream &stream;
+
+	public:
+		explicit Lease(CacheInputStream &_stream) noexcept
+			:stream(_stream) {}
+
+		void Set(InputCacheLease &&src) noexcept {
+			InputCacheLease::operator=(std::move(src));
+		}
+
+	private:
+		void OnInputCacheAvailable(std::unique_lock<Mutex> &lock) noexcept override {
+			stream.OnInputCacheAvailable(lock);
+		}
+	};
+
+	Lease lease;
+
 public:
 	CacheInputStream(InputCacheLease _lease, Mutex &_mutex) noexcept;
+	~CacheInputStream() noexcept override;
 
 	/* virtual methods from class InputStream */
 	void Check() override;
@@ -28,6 +48,5 @@ public:
 		    std::span<std::byte> dest) override;
 
 private:
-	/* virtual methods from class InputCacheLease */
-	void OnInputCacheAvailable(std::unique_lock<Mutex> &lock) noexcept override;
+	void OnInputCacheAvailable(std::unique_lock<Mutex> &lock) noexcept;
 };