input/Stream: remove attribute "cond", replace with handler interface

This adds a bit of overhead, but also adds flexibility to the API,
because arbitrary triggers may be invoked from that virtual method
implementation, not just Cond::signal().

The motivation for this is to make the handlers more dynamic, for the
upcoming buffering class utilizing ProxyInputStream.

src/input/plugins/AlsaInputPlugin.cxx

@@ -75,10 +75,10 @@ class AlsaInputStream final
 
 public:
 	AlsaInputStream(EventLoop &_loop,
-			const char *_uri, Mutex &_mutex, Cond &_cond,
+			const char *_uri, Mutex &_mutex,
 			const char *_device,
 			snd_pcm_t *_handle, int _frame_size)
-		:AsyncInputStream(_loop, _uri, _mutex, _cond,
+		:AsyncInputStream(_loop, _uri, _mutex,
 				  ALSA_MAX_BUFFERED, ALSA_RESUME_AT),
 		 MultiSocketMonitor(_loop),
 		 device(_device),
@@ -111,7 +111,7 @@ public:
 	}
 
 	static InputStreamPtr Create(EventLoop &event_loop, const char *uri,
-				     Mutex &mutex, Cond &cond);
+				     Mutex &mutex);
 
 protected:
 	/* virtual methods from AsyncInputStream */
@@ -148,7 +148,7 @@ private:
 
 inline InputStreamPtr
 AlsaInputStream::Create(EventLoop &event_loop, const char *uri,
-			Mutex &mutex, Cond &cond)
+			Mutex &mutex)
 {
 	const char *device = StringAfterPrefix(uri, "alsa://");
 	if (device == nullptr)
@@ -168,7 +168,7 @@ AlsaInputStream::Create(EventLoop &event_loop, const char *uri,
 
 	int frame_size = snd_pcm_format_width(format) / 8 * channels;
 	return std::make_unique<AlsaInputStream>(event_loop,
-						 uri, mutex, cond,
+						 uri, mutex,
 						 device, handle, frame_size);
 }
 
@@ -204,7 +204,7 @@ AlsaInputStream::DispatchSockets() noexcept
 
 		if (Recover(n_frames) < 0) {
 			postponed_exception = std::make_exception_ptr(std::runtime_error("PCM error - stream aborted"));
-			cond.broadcast();
+			InvokeOnAvailable();
 			return;
 		}
 	}
@@ -403,10 +403,10 @@ alsa_input_init(EventLoop &event_loop, const ConfigBlock &)
 }
 
 static InputStreamPtr
-alsa_input_open(const char *uri, Mutex &mutex, Cond &cond)
+alsa_input_open(const char *uri, Mutex &mutex)
 {
 	return AlsaInputStream::Create(*alsa_input_event_loop, uri,
-				       mutex, cond);
+				       mutex);
 }
 
 const struct InputPlugin input_plugin_alsa = {

src/input/plugins/ArchiveInputPlugin.cxx

@@ -35,7 +35,7 @@
 #include <stdlib.h>
 
 InputStreamPtr
-OpenArchiveInputStream(Path path, Mutex &mutex, Cond &cond)
+OpenArchiveInputStream(Path path, Mutex &mutex)
 {
 	const ArchivePlugin *arplug;
 
@@ -61,5 +61,5 @@ OpenArchiveInputStream(Path path, Mutex &mutex, Cond &cond)
 	}
 
 	return archive_file_open(arplug, Path::FromFS(archive))
-		->OpenStream(filename, mutex, cond);
+		->OpenStream(filename, mutex);
 }

src/input/plugins/ArchiveInputPlugin.hxx

@@ -24,9 +24,8 @@
 
 class Path;
 class Mutex;
-class Cond;
 
 InputStreamPtr
-OpenArchiveInputStream(Path path, Mutex &mutex, Cond &cond);
+OpenArchiveInputStream(Path path, Mutex &mutex);
 
 #endif

src/input/plugins/CdioParanoiaInputPlugin.cxx

@@ -61,11 +61,11 @@ class CdioParanoiaInputStream final : public InputStream {
 	int buffer_lsn;
 
  public:
-	CdioParanoiaInputStream(const char *_uri, Mutex &_mutex, Cond &_cond,
+	CdioParanoiaInputStream(const char *_uri, Mutex &_mutex,
 				cdrom_drive_t *_drv, CdIo_t *_cdio,
 				bool reverse_endian,
 				lsn_t _lsn_from, lsn_t _lsn_to)
-		:InputStream(_uri, _mutex, _cond),
+		:InputStream(_uri, _mutex),
 		 drv(_drv), cdio(_cdio), para(cdio_paranoia_init(drv)),
 		 lsn_from(_lsn_from), lsn_to(_lsn_to),
 		 lsn_relofs(0),
@@ -184,7 +184,7 @@ cdio_detect_device(void)
 
 static InputStreamPtr
 input_cdio_open(const char *uri,
-		Mutex &mutex, Cond &cond)
+		Mutex &mutex)
 {
 	struct cdio_uri parsed_uri;
 	if (!parse_cdio_uri(&parsed_uri, uri))
@@ -250,7 +250,7 @@ input_cdio_open(const char *uri,
 		lsn_to = cdio_get_disc_last_lsn(cdio);
 	}
 
-	return std::make_unique<CdioParanoiaInputStream>(uri, mutex, cond,
+	return std::make_unique<CdioParanoiaInputStream>(uri, mutex,
 							 drv, cdio,
 							 reverse_endian,
 							 lsn_from, lsn_to);

src/input/plugins/CurlInputPlugin.cxx

@@ -35,7 +35,6 @@
 #include "tag/Tag.hxx"
 #include "event/Call.hxx"
 #include "event/Loop.hxx"
-#include "thread/Cond.hxx"
 #include "util/ASCII.hxx"
 #include "util/StringUtil.hxx"
 #include "util/StringFormat.hxx"
@@ -83,7 +82,7 @@ public:
 	CurlInputStream(EventLoop &event_loop, const char *_url,
 			const std::multimap<std::string, std::string> &headers,
 			I &&_icy,
-			Mutex &_mutex, Cond &_cond);
+			Mutex &_mutex);
 
 	~CurlInputStream() noexcept;
 
@@ -92,7 +91,7 @@ public:
 
 	static InputStreamPtr Open(const char *url,
 				   const std::multimap<std::string, std::string> &headers,
-				   Mutex &mutex, Cond &cond);
+				   Mutex &mutex);
 
 private:
 	/**
@@ -274,7 +273,7 @@ void
 CurlInputStream::OnEnd()
 {
 	const std::lock_guard<Mutex> protect(mutex);
-	cond.broadcast();
+	InvokeOnAvailable();
 
 	AsyncInputStream::SetClosed();
 }
@@ -290,7 +289,7 @@ CurlInputStream::OnError(std::exception_ptr e) noexcept
 	else if (!IsReady())
 		SetReady();
 	else
-		cond.broadcast();
+		InvokeOnAvailable();
 
 	AsyncInputStream::SetClosed();
 }
@@ -352,8 +351,8 @@ inline
 CurlInputStream::CurlInputStream(EventLoop &event_loop, const char *_url,
 				 const std::multimap<std::string, std::string> &headers,
 				 I &&_icy,
-				 Mutex &_mutex, Cond &_cond)
-	:AsyncInputStream(event_loop, _url, _mutex, _cond,
+				 Mutex &_mutex)
+	:AsyncInputStream(event_loop, _url, _mutex,
 			  CURL_MAX_BUFFERED,
 			  CURL_RESUME_AT),
 	 icy(std::forward<I>(_icy))
@@ -445,14 +444,14 @@ CurlInputStream::DoSeek(offset_type new_offset)
 inline InputStreamPtr
 CurlInputStream::Open(const char *url,
 		      const std::multimap<std::string, std::string> &headers,
-		      Mutex &mutex, Cond &cond)
+		      Mutex &mutex)
 {
 	auto icy = std::make_shared<IcyMetaDataParser>();
 
 	auto c = std::make_unique<CurlInputStream>((*curl_init)->GetEventLoop(),
 						   url, headers,
 						   icy,
-						   mutex, cond);
+						   mutex);
 
 	BlockingCall(c->GetEventLoop(), [&c](){
 			c->InitEasy();
@@ -465,19 +464,19 @@ CurlInputStream::Open(const char *url,
 InputStreamPtr
 OpenCurlInputStream(const char *uri,
 		    const std::multimap<std::string, std::string> &headers,
-		    Mutex &mutex, Cond &cond)
+		    Mutex &mutex)
 {
-	return CurlInputStream::Open(uri, headers, mutex, cond);
+	return CurlInputStream::Open(uri, headers, mutex);
 }
 
 static InputStreamPtr
-input_curl_open(const char *url, Mutex &mutex, Cond &cond)
+input_curl_open(const char *url, Mutex &mutex)
 {
 	if (strncmp(url, "http://", 7) != 0 &&
 	    strncmp(url, "https://", 8) != 0)
 		return nullptr;
 
-	return CurlInputStream::Open(url, {}, mutex, cond);
+	return CurlInputStream::Open(url, {}, mutex);
 }
 
 const struct InputPlugin input_plugin_curl = {

src/input/plugins/CurlInputPlugin.hxx

@@ -26,7 +26,6 @@
 #include <map>
 
 class Mutex;
-class Cond;
 
 extern const struct InputPlugin input_plugin_curl;
 
@@ -40,6 +39,6 @@ extern const struct InputPlugin input_plugin_curl;
 InputStreamPtr
 OpenCurlInputStream(const char *uri,
 		    const std::multimap<std::string, std::string> &headers,
-		    Mutex &mutex, Cond &cond);
+		    Mutex &mutex);
 
 #endif

src/input/plugins/FfmpegInputPlugin.cxx

@@ -39,9 +39,9 @@ struct FfmpegInputStream final : public InputStream {
 
 	bool eof;
 
-	FfmpegInputStream(const char *_uri, Mutex &_mutex, Cond &_cond,
+	FfmpegInputStream(const char *_uri, Mutex &_mutex,
 			  AVIOContext *_h)
-		:InputStream(_uri, _mutex, _cond),
+		:InputStream(_uri, _mutex),
 		 h(_h), eof(false) {
 		seekable = (h->seekable & AVIO_SEEKABLE_NORMAL) != 0;
 		size = avio_size(h);
@@ -83,7 +83,7 @@ input_ffmpeg_init(EventLoop &, const ConfigBlock &)
 
 static InputStreamPtr
 input_ffmpeg_open(const char *uri,
-		  Mutex &mutex, Cond &cond)
+		  Mutex &mutex)
 {
 	if (!StringStartsWith(uri, "gopher://") &&
 	    !StringStartsWith(uri, "rtp://") &&
@@ -98,7 +98,7 @@ input_ffmpeg_open(const char *uri,
 	if (result != 0)
 		throw MakeFfmpegError(result);
 
-	return std::make_unique<FfmpegInputStream>(uri, mutex, cond, h);
+	return std::make_unique<FfmpegInputStream>(uri, mutex, h);
 }
 
 size_t

src/input/plugins/FileInputPlugin.cxx

@@ -35,8 +35,8 @@ class FileInputStream final : public InputStream {
 
 public:
 	FileInputStream(const char *path, FileReader &&_reader, off_t _size,
-			Mutex &_mutex, Cond &_cond)
-		:InputStream(path, _mutex, _cond),
+			Mutex &_mutex)
+		:InputStream(path, _mutex),
 		 reader(std::move(_reader)) {
 		size = _size;
 		seekable = true;
@@ -54,8 +54,7 @@ public:
 };
 
 InputStreamPtr
-OpenFileInputStream(Path path,
-		    Mutex &mutex, Cond &cond)
+OpenFileInputStream(Path path, Mutex &mutex)
 {
 	FileReader reader(path);
 
@@ -75,7 +74,7 @@ OpenFileInputStream(Path path,
 
 	return std::make_unique<FileInputStream>(path.ToUTF8().c_str(),
 						 std::move(reader), info.GetSize(),
-						 mutex, cond);
+						 mutex);
 }
 
 void

src/input/plugins/FileInputPlugin.hxx

@@ -24,10 +24,8 @@
 
 class Path;
 class Mutex;
-class Cond;
 
 InputStreamPtr
-OpenFileInputStream(Path path,
-		    Mutex &mutex, Cond &cond);
+OpenFileInputStream(Path path, Mutex &mutex);
 
 #endif

src/input/plugins/MmsInputPlugin.cxx

@@ -34,8 +34,8 @@ class MmsInputStream final : public ThreadInputStream {
 	mmsx_t *mms;
 
 public:
-	MmsInputStream(const char *_uri, Mutex &_mutex, Cond &_cond)
-		:ThreadInputStream(input_plugin_mms.name, _uri, _mutex, _cond,
+	MmsInputStream(const char *_uri, Mutex &_mutex)
+		:ThreadInputStream(input_plugin_mms.name, _uri, _mutex,
 				   MMS_BUFFER_SIZE) {
 	}
 
@@ -70,7 +70,7 @@ MmsInputStream::Open()
 
 static InputStreamPtr
 input_mms_open(const char *url,
-	       Mutex &mutex, Cond &cond)
+	       Mutex &mutex)
 {
 	if (!StringStartsWith(url, "mms://") &&
 	    !StringStartsWith(url, "mmsh://") &&
@@ -78,7 +78,7 @@ input_mms_open(const char *url,
 	    !StringStartsWith(url, "mmsu://"))
 		return nullptr;
 
-	auto m = std::make_unique<MmsInputStream>(url, mutex, cond);
+	auto m = std::make_unique<MmsInputStream>(url, mutex);
 	m->Start();
 	return m;
 }

src/input/plugins/NfsInputPlugin.cxx

@@ -23,7 +23,6 @@
 #include "../InputPlugin.hxx"
 #include "lib/nfs/Glue.hxx"
 #include "lib/nfs/FileReader.hxx"
-#include "thread/Cond.hxx"
 #include "util/StringCompare.hxx"
 
 #include <string.h>
@@ -46,9 +45,9 @@ class NfsInputStream final : NfsFileReader, public AsyncInputStream {
 	bool reconnect_on_resume = false, reconnecting = false;
 
 public:
-	NfsInputStream(const char *_uri, Mutex &_mutex, Cond &_cond)
+	NfsInputStream(const char *_uri, Mutex &_mutex)
 		:AsyncInputStream(NfsFileReader::GetEventLoop(),
-				  _uri, _mutex, _cond,
+				  _uri, _mutex,
 				  NFS_MAX_BUFFERED,
 				  NFS_RESUME_AT) {}
 
@@ -100,7 +99,7 @@ NfsInputStream::DoRead()
 		NfsFileReader::Read(next_offset, nbytes);
 	} catch (...) {
 		postponed_exception = std::current_exception();
-		cond.broadcast();
+		InvokeOnAvailable();
 	}
 }
 
@@ -196,7 +195,7 @@ NfsInputStream::OnNfsFileError(std::exception_ptr &&e) noexcept
 	else if (!IsReady())
 		SetReady();
 	else
-		cond.broadcast();
+		InvokeOnAvailable();
 }
 
 /*
@@ -218,12 +217,12 @@ input_nfs_finish() noexcept
 
 static InputStreamPtr
 input_nfs_open(const char *uri,
-	       Mutex &mutex, Cond &cond)
+	       Mutex &mutex)
 {
 	if (!StringStartsWith(uri, "nfs://"))
 		return nullptr;
 
-	auto is = std::make_unique<NfsInputStream>(uri, mutex, cond);
+	auto is = std::make_unique<NfsInputStream>(uri, mutex);
 	is->Open();
 	return is;
 }

src/input/plugins/QobuzInputPlugin.cxx

@@ -49,8 +49,8 @@ class QobuzInputStream final
 
 public:
 	QobuzInputStream(const char *_uri, const char *_track_id,
-			 Mutex &_mutex, Cond &_cond) noexcept
-		:ProxyInputStream(_uri, _mutex, _cond),
+			 Mutex &_mutex) noexcept
+		:ProxyInputStream(_uri, _mutex),
 		 track_id(_track_id)
 	{
 		qobuz_client->AddLoginHandler(*this);
@@ -70,7 +70,7 @@ public:
 private:
 	void Failed(std::exception_ptr e) {
 		SetInput(std::make_unique<FailingInputStream>(GetURI(), e,
-							      mutex, cond));
+							      mutex));
 	}
 
 	/* virtual methods from QobuzSessionHandler */
@@ -89,11 +89,11 @@ QobuzInputStream::OnQobuzSession() noexcept
 	try {
 		const auto session = qobuz_client->GetSession();
 
-		QobuzTrackHandler &handler = *this;
+		QobuzTrackHandler &h = *this;
 		track_request = std::make_unique<QobuzTrackRequest>(*qobuz_client,
 								    session,
 								    track_id.c_str(),
-								    handler);
+								    h);
 		track_request->Start();
 	} catch (...) {
 		Failed(std::current_exception());
@@ -108,7 +108,7 @@ QobuzInputStream::OnQobuzTrackSuccess(std::string url) noexcept
 
 	try {
 		SetInput(OpenCurlInputStream(url.c_str(), {},
-					     mutex, cond));
+					     mutex));
 	} catch (...) {
 		Failed(std::current_exception());
 	}
@@ -180,7 +180,7 @@ ExtractQobuzTrackId(const char *uri)
 }
 
 static InputStreamPtr
-OpenQobuzInput(const char *uri, Mutex &mutex, Cond &cond)
+OpenQobuzInput(const char *uri, Mutex &mutex)
 {
 	assert(qobuz_client != nullptr);
 
@@ -190,7 +190,7 @@ OpenQobuzInput(const char *uri, Mutex &mutex, Cond &cond)
 
 	// TODO: validate track_id
 
-	return std::make_unique<QobuzInputStream>(uri, track_id, mutex, cond);
+	return std::make_unique<QobuzInputStream>(uri, track_id, mutex);
 }
 
 static std::unique_ptr<RemoteTagScanner>

src/input/plugins/SmbclientInputPlugin.cxx

@@ -37,9 +37,9 @@ class SmbclientInputStream final : public InputStream {
 
 public:
 	SmbclientInputStream(const char *_uri,
-			     Mutex &_mutex, Cond &_cond,
+			     Mutex &_mutex,
 			     SMBCCTX *_ctx, int _fd, const struct stat &st)
-		:InputStream(_uri, _mutex, _cond),
+		:InputStream(_uri, _mutex),
 		 ctx(_ctx), fd(_fd) {
 		seekable = true;
 		size = st.st_size;
@@ -85,7 +85,7 @@ input_smbclient_init(EventLoop &, const ConfigBlock &)
 
 static InputStreamPtr
 input_smbclient_open(const char *uri,
-		     Mutex &mutex, Cond &cond)
+		     Mutex &mutex)
 {
 	if (!StringStartsWith(uri, "smb://"))
 		return nullptr;
@@ -119,7 +119,7 @@ input_smbclient_open(const char *uri,
 		throw MakeErrno(e, "smbc_fstat() failed");
 	}
 
-	return std::make_unique<SmbclientInputStream>(uri, mutex, cond,
+	return std::make_unique<SmbclientInputStream>(uri, mutex,
 						      ctx, fd, st);
 }
 

src/input/plugins/TidalInputPlugin.cxx

@@ -60,8 +60,8 @@ class TidalInputStream final
 
 public:
 	TidalInputStream(const char *_uri, const char *_track_id,
-			 Mutex &_mutex, Cond &_cond) noexcept
-		:ProxyInputStream(_uri, _mutex, _cond),
+			 Mutex &_mutex) noexcept
+		:ProxyInputStream(_uri, _mutex),
 		 track_id(_track_id)
 	{
 		tidal_session->AddLoginHandler(*this);
@@ -81,7 +81,7 @@ public:
 private:
 	void Failed(std::exception_ptr e) {
 		SetInput(std::make_unique<FailingInputStream>(GetURI(), e,
-							      mutex, cond));
+							      mutex));
 	}
 
 	/* virtual methods from TidalSessionHandler */
@@ -98,14 +98,14 @@ TidalInputStream::OnTidalSession() noexcept
 	const std::lock_guard<Mutex> protect(mutex);
 
 	try {
-		TidalTrackHandler &handler = *this;
+		TidalTrackHandler &h = *this;
 		track_request = std::make_unique<TidalTrackRequest>(tidal_session->GetCurl(),
 								    tidal_session->GetBaseUrl(),
 								    tidal_session->GetToken(),
 								    tidal_session->GetSession().c_str(),
 								    track_id.c_str(),
 								    tidal_audioquality,
-								    handler);
+								    h);
 		track_request->Start();
 	} catch (...) {
 		Failed(std::current_exception());
@@ -124,7 +124,7 @@ TidalInputStream::OnTidalTrackSuccess(std::string url) noexcept
 
 	try {
 		SetInput(OpenCurlInputStream(url.c_str(), {},
-					     mutex, cond));
+					     mutex));
 	} catch (...) {
 		Failed(std::current_exception());
 	}
@@ -211,7 +211,7 @@ ExtractTidalTrackId(const char *uri)
 }
 
 static InputStreamPtr
-OpenTidalInput(const char *uri, Mutex &mutex, Cond &cond)
+OpenTidalInput(const char *uri, Mutex &mutex)
 {
 	assert(tidal_session != nullptr);
 
@@ -221,7 +221,7 @@ OpenTidalInput(const char *uri, Mutex &mutex, Cond &cond)
 
 	// TODO: validate track_id
 
-	return std::make_unique<TidalInputStream>(uri, track_id, mutex, cond);
+	return std::make_unique<TidalInputStream>(uri, track_id, mutex);
 }
 
 static std::unique_ptr<RemoteTagScanner>