1 files changed, 104 insertions, 77 deletions

diff --git a/src/lib/upnp/Discovery.cxx b/src/lib/upnp/Discovery.cxx
index 121c01c57..0ce1d6c13 100644
--- a/src/lib/upnp/Discovery.cxx
+++ b/src/lib/upnp/Discovery.cxx
@@ -21,6 +21,9 @@
 #include "Discovery.hxx"
 #include "ContentDirectoryService.hxx"
 #include "Log.hxx"
+#include "lib/curl/Global.hxx"
+#include "event/Call.hxx"
+#include "util/DeleteDisposer.hxx"
 #include "util/ScopeExit.hxx"
 #include "util/RuntimeError.hxx"
 
@@ -29,6 +32,66 @@
 #include <stdlib.h>
 #include <string.h>
 
+UPnPDeviceDirectory::Downloader::Downloader(UPnPDeviceDirectory &_parent,
+					    const Upnp_Discovery &disco)
+	:defer_start_event(_parent.GetEventLoop(),
+			   BIND_THIS_METHOD(OnDeferredStart)),
+	 parent(_parent),
+	 id(disco.DeviceId), url(disco.Location),
+	 expires(std::chrono::seconds(disco.Expires)),
+	 request(*parent.curl, url.c_str(), *this)
+{
+	parent.downloaders.push_back(*this);
+}
+
+void
+UPnPDeviceDirectory::Downloader::Destroy() noexcept
+{
+	parent.downloaders.erase_and_dispose(parent.downloaders.iterator_to(*this),
+					     DeleteDisposer());
+}
+
+void
+UPnPDeviceDirectory::Downloader::OnHeaders(unsigned status,
+					   std::multimap<std::string, std::string> &&)
+{
+	if (status != 200) {
+		Destroy();
+		return;
+	}
+}
+
+void
+UPnPDeviceDirectory::Downloader::OnData(ConstBuffer<void> _data)
+{
+	data.append((const char *)_data.data, _data.size);
+}
+
+void
+UPnPDeviceDirectory::Downloader::OnEnd()
+{
+	AtScopeExit(this) { Destroy(); };
+
+	ContentDirectoryDescriptor d(std::move(id),
+				     std::chrono::steady_clock::now(),
+				     expires);
+
+	try {
+		d.Parse(url, data.c_str());
+	} catch (const std::exception &e) {
+		LogError(e);
+	}
+
+	parent.LockAdd(std::move(d));
+}
+
+void
+UPnPDeviceDirectory::Downloader::OnError(std::exception_ptr e) noexcept
+{
+	LogError(e);
+	Destroy();
+}
+
 // The service type string we are looking for.
 static constexpr char ContentDirectorySType[] = "urn:schemas-upnp-org:service:ContentDirectory:1";
 
@@ -39,7 +102,7 @@ static bool
 isCDService(const char *st) noexcept
 {
 	constexpr size_t sz = sizeof(ContentDirectorySType) - 3;
-	return memcmp(ContentDirectorySType, st, sz) == 0;
+	return strncmp(ContentDirectorySType, st, sz) == 0;
 }
 
 // The type of device we're asking for in search
@@ -50,7 +113,7 @@ static bool
 isMSDevice(const char *st) noexcept
 {
 	constexpr size_t sz = sizeof(MediaServerDType) - 3;
-	return memcmp(MediaServerDType, st, sz) == 0;
+	return strncmp(MediaServerDType, st, sz) == 0;
 }
 
 static void
@@ -106,67 +169,30 @@ UPnPDeviceDirectory::LockRemove(const std::string &id)
 	}
 }
 
-inline void
-UPnPDeviceDirectory::Explore()
-{
-	for (;;) {
-		std::unique_ptr<DiscoveredTask> tsk;
-		if (!queue.take(tsk)) {
-			queue.workerExit();
-			return;
-		}
-
-		// Device signals its existence and well-being. Perform the
-		// UPnP "description" phase by downloading and decoding the
-		// description document.
-		char *buf;
-		// LINE_SIZE is defined by libupnp's upnp.h...
-		char contentType[LINE_SIZE];
-		int code = UpnpDownloadUrlItem(tsk->url.c_str(), &buf, contentType);
-		if (code != UPNP_E_SUCCESS) {
-			continue;
-		}
-
-		AtScopeExit(buf){ free(buf); };
-
-		// Update or insert the device
-		ContentDirectoryDescriptor d(std::move(tsk->device_id),
-					     std::chrono::steady_clock::now(),
-					     tsk->expires);
-
-		try {
-			d.Parse(tsk->url, buf);
-		} catch (const std::exception &e) {
-			LogError(e);
-		}
-
-		LockAdd(std::move(d));
-	}
-}
-
-void *
-UPnPDeviceDirectory::Explore(void *ctx)
-{
-	UPnPDeviceDirectory &directory = *(UPnPDeviceDirectory *)ctx;
-	directory.Explore();
-	return (void*)1;
-}
-
 inline int
-UPnPDeviceDirectory::OnAlive(Upnp_Discovery *disco)
+UPnPDeviceDirectory::OnAlive(Upnp_Discovery *disco) noexcept
 {
 	if (isMSDevice(disco->DeviceType) ||
 	    isCDService(disco->ServiceType)) {
-		DiscoveredTask *tp = new DiscoveredTask(disco);
-		if (queue.put(tp))
-			return UPNP_E_FINISH;
+		auto *downloader = new Downloader(*this, *disco);
+
+		try {
+			downloader->Start();
+		} catch (...) {
+			BlockingCall(GetEventLoop(), [downloader](){
+					downloader->Destroy();
+				});
+
+			LogError(std::current_exception());
+			return UPNP_E_SUCCESS;
+		}
 	}
 
 	return UPNP_E_SUCCESS;
 }
 
 inline int
-UPnPDeviceDirectory::OnByeBye(Upnp_Discovery *disco)
+UPnPDeviceDirectory::OnByeBye(Upnp_Discovery *disco) noexcept
 {
 	if (isMSDevice(disco->DeviceType) ||
 	    isCDService(disco->ServiceType)) {
@@ -182,7 +208,7 @@ UPnPDeviceDirectory::OnByeBye(Upnp_Discovery *disco)
 // Example: ContentDirectories appearing and disappearing from the network
 // We queue a task for our worker thread(s)
 int
-UPnPDeviceDirectory::Invoke(Upnp_EventType et, void *evp)
+UPnPDeviceDirectory::Invoke(Upnp_EventType et, void *evp) noexcept
 {
 	switch (et) {
 	case UPNP_DISCOVERY_SEARCH_RESULT:
@@ -212,39 +238,41 @@ UPnPDeviceDirectory::ExpireDevices()
 	const auto now = std::chrono::steady_clock::now();
 	bool didsomething = false;
 
-	for (auto it = directories.begin();
-	     it != directories.end();) {
-		if (now > it->expires) {
-			it = directories.erase(it);
-			didsomething = true;
-		} else {
-			it++;
-		}
-	}
+	directories.remove_if([now, &didsomething](const ContentDirectoryDescriptor &d){
+			bool expired = now > d.expires;
+			if (expired)
+				didsomething = true;
+			return expired;
+		});
 
 	if (didsomething)
 		Search();
 }
 
-UPnPDeviceDirectory::UPnPDeviceDirectory(UpnpClient_Handle _handle,
-					 UPnPDiscoveryListener *_listener)
-	:handle(_handle),
-	 listener(_listener),
-	 queue("DiscoveredQueue")
+UPnPDeviceDirectory::UPnPDeviceDirectory(EventLoop &event_loop,
+					 UpnpClient_Handle _handle,
+					 UPnPDiscoveryListener *_listener) noexcept
+	:curl(event_loop), handle(_handle),
+	 listener(_listener)
 {
 }
 
-UPnPDeviceDirectory::~UPnPDeviceDirectory()
+UPnPDeviceDirectory::~UPnPDeviceDirectory() noexcept
 {
-	/* this destructor exists here just so it won't get inlined */
+	BlockingCall(GetEventLoop(), [this](){
+			downloaders.clear_and_dispose(DeleteDisposer());
+		});
+}
+
+inline EventLoop &
+UPnPDeviceDirectory::GetEventLoop() noexcept
+{
+	return curl->GetEventLoop();
 }
 
 void
 UPnPDeviceDirectory::Start()
 {
-	if (!queue.start(1, Explore, this))
-		throw std::runtime_error("Discover work queue start failed");
-
 	Search();
 }
 
@@ -278,11 +306,10 @@ UPnPDeviceDirectory::GetDirectories()
 	ExpireDevices();
 
 	std::vector<ContentDirectoryService> out;
-	for (auto dit = directories.begin();
-	     dit != directories.end(); dit++) {
-		for (const auto &service : dit->device.services) {
+	for (const auto &descriptor : directories) {
+		for (const auto &service : descriptor.device.services) {
 			if (isCDService(service.serviceType.c_str())) {
-				out.emplace_back(dit->device, service);
+				out.emplace_back(descriptor.device, service);
 			}
 		}
 	}