input/mms: move blocking I/O to thread

NEWS

@@ -21,6 +21,7 @@ ver 0.19 (not yet released)
   - read tags from songs in an archive
 * input
   - alsa: new input plugin
+  - mms: non-blocking I/O
   - nfs: new input plugin
   - smbclient: new input plugin
 * filter

src/input/plugins/MmsInputPlugin.cxx

@@ -19,40 +19,56 @@
 
 #include "config.h"
 #include "MmsInputPlugin.hxx"
-#include "../InputStream.hxx"
+#include "input/ThreadInputStream.hxx"
-#include "../InputPlugin.hxx"
+#include "input/InputPlugin.hxx"
 #include "util/StringUtil.hxx"
 #include "util/Error.hxx"
 #include "util/Domain.hxx"
 
 #include <libmms/mmsx.h>
 
-struct MmsInputStream {
+static constexpr size_t MMS_BUFFER_SIZE = 256 * 1024;
-	InputStream base;
 
+class MmsInputStream final : public ThreadInputStream {
 	mmsx_t *mms;
 
-	bool eof;
+public:
-
+	MmsInputStream(const char *uri, Mutex &mutex, Cond &cond)
-	MmsInputStream(const char *uri,
+		:ThreadInputStream(input_plugin_mms, uri, mutex, cond,
-		       Mutex &mutex, Cond &cond,
+				   MMS_BUFFER_SIZE) {
-		       mmsx_t *_mms)
-		:base(input_plugin_mms, uri, mutex, cond),
-		 mms(_mms), eof(false) {
-		/* XX is this correct?  at least this selects the ffmpeg
-		   decoder, which seems to work fine*/
-		base.mime = "audio/x-ms-wma";
-
-		base.ready = true;
 	}
 
-	~MmsInputStream() {
+protected:
+	virtual bool Open(gcc_unused Error &error) override;
+	virtual size_t Read(void *ptr, size_t size, Error &error) override;
+
+	virtual void Close() {
 		mmsx_close(mms);
 	}
 };
 
 static constexpr Domain mms_domain("mms");
 
+bool
+MmsInputStream::Open(Error &error)
+{
+	Unlock();
+
+	mms = mmsx_connect(nullptr, nullptr, GetURI(), 128 * 1024);
+	if (mms == nullptr) {
+		Lock();
+		error.Set(mms_domain, "mmsx_connect() failed");
+		return false;
+	}
+
+	Lock();
+
+	/* TODO: is this correct?  at least this selects the ffmpeg
+	   decoder, which seems to work fine */
+	SetMimeType("audio/x-ms-wma");
+	return true;
+}
+
 static InputStream *
 input_mms_open(const char *url,
 	       Mutex &mutex, Cond &cond,
@@ -64,51 +80,25 @@ input_mms_open(const char *url,
 	    !StringStartsWith(url, "mmsu://"))
 		return nullptr;
 
-	const auto mms = mmsx_connect(nullptr, nullptr, url, 128 * 1024);
+	auto m = new MmsInputStream(url, mutex, cond);
-	if (mms == nullptr) {
+	auto is = m->Start(error);
-		error.Set(mms_domain, "mmsx_connect() failed");
+	if (is == nullptr)
-		return nullptr;
+		delete m;
-	}
 
-	auto m = new MmsInputStream(url, mutex, cond, mms);
+	return is;
-	return &m->base;
 }
 
-static size_t
+size_t
-input_mms_read(InputStream *is, void *ptr, size_t size,
+MmsInputStream::Read(void *ptr, size_t size, Error &error)
-	       Error &error)
 {
-	MmsInputStream *m = (MmsInputStream *)is;
+	int nbytes = mmsx_read(nullptr, mms, (char *)ptr, size);
-	int ret;
+	if (nbytes <= 0) {
-
+		if (nbytes < 0)
-	ret = mmsx_read(nullptr, m->mms, (char *)ptr, size);
-	if (ret <= 0) {
-		if (ret < 0)
 			error.SetErrno("mmsx_read() failed");
-
+		return 0;
-		m->eof = true;
-		return false;
 	}
 
-	is->offset += ret;
+	return (size_t)nbytes;
-
-	return (size_t)ret;
-}
-
-static void
-input_mms_close(InputStream *is)
-{
-	MmsInputStream *m = (MmsInputStream *)is;
-
-	delete m;
-}
-
-static bool
-input_mms_eof(InputStream *is)
-{
-	MmsInputStream *m = (MmsInputStream *)is;
-
-	return m->eof;
 }
 
 const InputPlugin input_plugin_mms = {
@@ -116,12 +106,12 @@ const InputPlugin input_plugin_mms = {
 	nullptr,
 	nullptr,
 	input_mms_open,
-	input_mms_close,
+	ThreadInputStream::Close,
+	ThreadInputStream::Check,
 	nullptr,
 	nullptr,
-	nullptr,
+	ThreadInputStream::Available,
-	nullptr,
+	ThreadInputStream::Read,
-	input_mms_read,
+	ThreadInputStream::IsEOF,
-	input_mms_eof,
 	nullptr,
 };