Migrate CurlMultiHandle and extend CurlStream for it.

6 files changed, 168 insertions, 0 deletions

diff --git a/libadhocutil/curlHandle.cpp b/libadhocutil/curlHandle.cpp
index 6315bce..3afd473 100644
--- a/libadhocutil/curlHandle.cpp
+++ b/libadhocutil/curlHandle.cpp
@@ -56,6 +56,11 @@ CurlHandle::perform()
 	checkCurlCode(curl_easy_perform(curl_handle));
 }
 
+CurlHandle::operator CURL *() const
+{
+	return curl_handle;
+}
+
 void
 CurlHandle::checkCurlCode(CURLcode res) const
 {
diff --git a/libadhocutil/curlHandle.h b/libadhocutil/curlHandle.h
index 75e4b0a..b5a5d63 100644
--- a/libadhocutil/curlHandle.h
+++ b/libadhocutil/curlHandle.h
@@ -19,6 +19,8 @@ class DLL_PUBLIC CurlHandle : public virtual IntrusivePtrBase {
 		void appendPost(const char *, const char *);
 		void perform();
 
+		operator CURL *() const;
+
 	protected:
 		void checkCurlCode(CURLcode res) const;
 
diff --git a/libadhocutil/curlMultiHandle.cpp b/libadhocutil/curlMultiHandle.cpp
new file mode 100644
index 0000000..ff509ce
--- /dev/null
+++ b/libadhocutil/curlMultiHandle.cpp
@@ -0,0 +1,92 @@
+#include "curlMultiHandle.h"
+#include <boost/iostreams/stream.hpp>
+#include <map>
+#include "runtimeContext.h"
+#include "curlStream.h"
+
+class RunningCurl : public CurlStreamSource {
+	public:
+		RunningCurl(const std::string & url, const boost::function<void(std::istream &)> & c) :
+			CurlStreamSource(url),
+			consumer(c)
+		{
+		}
+
+		void Callback() override
+		{
+			typedef boost::reference_wrapper<RunningCurl> rc_ref;
+			boost::iostreams::stream<rc_ref> curlstrm(boost::ref(*this));
+			consumer(curlstrm);
+		}
+
+	private:
+		const boost::function<void(std::istream &)> consumer;
+};
+
+CurlMultiHandle::CurlMultiHandle()
+{
+}
+
+CurlMultiHandle::~CurlMultiHandle()
+{
+}
+
+CurlHandlePtr
+CurlMultiHandle::addCurl(const std::string & url, const boost::function<void(std::istream &)> & c)
+{
+	RunningCurl * css = new RunningCurl(url, c);
+	curls.insert(css);
+	return css;
+}
+
+void
+CurlMultiHandle::addRunner(CURLM * curlm, Running & running, CurlMultiHandle::CURLs & curls)
+{
+	auto runner = *curls.begin();
+	curl_multi_add_handle(curlm, *runner);
+	running[*runner] = runner;
+	runner->SwapContext();
+	curls.erase(runner);
+}
+
+void
+CurlMultiHandle::performAll()
+{
+	if (!curls.empty()) {
+		Running running;
+		CURLM * curlm = curl_multi_init();
+
+		while (!curls.empty() && running.size() < 5) {
+			addRunner(curlm, running, curls);
+		}
+		CURLMcode code;
+		int act = running.size();
+		while (act) {
+			while ((code = curl_multi_perform(curlm, &act)) == CURLM_CALL_MULTI_PERFORM) ;
+			// Has anything finished
+			CURLMsg * msg;
+			int msgs = 0;
+			while ((msg = curl_multi_info_read(curlm, &msgs))) {
+				if (msg->msg == CURLMSG_DONE) {
+					curl_multi_remove_handle(curlm, msg->easy_handle);
+					running.erase(msg->easy_handle);
+					if (!curls.empty()) {
+						addRunner(curlm, running, curls);
+						act += 1;
+					}
+				}
+			}
+			// Wait for something to happen
+			fd_set r, w, e;
+			int maxfd = 0;
+			struct timeval to = { 0, 100000 };
+			FD_ZERO(&r);
+			FD_ZERO(&w);
+			FD_ZERO(&e);
+			curl_multi_fdset(curlm, &r, &w, &e, &maxfd);
+			select(act, &r, &w, &e, &to);
+		}
+		curl_multi_cleanup(curlm);
+	}
+}
+
diff --git a/libadhocutil/curlMultiHandle.h b/libadhocutil/curlMultiHandle.h
new file mode 100644
index 0000000..0f70988
--- /dev/null
+++ b/libadhocutil/curlMultiHandle.h
@@ -0,0 +1,38 @@
+#ifndef ADHOCUTIL_CURLMULTIHANDLE_H
+#define ADHOCUTIL_CURLMULTIHANDLE_H
+
+#include <boost/function.hpp>
+#include <set>
+#include <map>
+#include "intrusivePtrBase.h"
+#include "visibility.h"
+#include "curlHandle.h"
+
+class RunningCurl;
+typedef boost::intrusive_ptr<RunningCurl> RunningCurlPtr;
+
+class DLL_PUBLIC CurlMultiHandle : public IntrusivePtrBase {
+	public:
+		typedef std::set<RunningCurlPtr> CURLs;
+		typedef boost::function<void(std::istream &)> Consumer;
+
+		CurlMultiHandle();
+		~CurlMultiHandle();
+
+		CurlMultiHandle(const CurlMultiHandle &) = delete;
+		void operator=(const CurlMultiHandle &) = delete;
+
+		CurlHandlePtr addCurl(const std::string &, const Consumer &);
+		void performAll();
+
+	private:
+		typedef std::map<CURL *, RunningCurlPtr> Running;
+
+		DLL_PRIVATE void addRunner(CURLM * curlm, Running & running, CurlMultiHandle::CURLs & curls);
+
+		CURLs curls;
+};
+typedef boost::intrusive_ptr<CurlMultiHandle> CurlMultiHandlePtr;
+
+#endif
+
diff --git a/libadhocutil/curlStream.h b/libadhocutil/curlStream.h
index cb003bb..efe8c4e 100644
--- a/libadhocutil/curlStream.h
+++ b/libadhocutil/curlStream.h
@@ -15,6 +15,7 @@ class DLL_PUBLIC CurlStreamSource : public boost::iostreams::source, public Curl
 		std::streamsize read(char * target, std::streamsize targetSize);
 
 	private:
+		friend class CurlMultiHandle;
 		DLL_PRIVATE void Callback() override;
 
 		DLL_PRIVATE static size_t recvWrapper(void * data, size_t sz, size_t nm, void * css);
diff --git a/libadhocutil/unittests/testCurl.cpp b/libadhocutil/unittests/testCurl.cpp
index 1c0ed7f..41ee0a1 100644
--- a/libadhocutil/unittests/testCurl.cpp
+++ b/libadhocutil/unittests/testCurl.cpp
@@ -1,7 +1,9 @@
 #define BOOST_TEST_MODULE Curl
 #include <boost/test/unit_test.hpp>
 
+#include <boost/bind.hpp>
 #include "curlHandle.h"
+#include "curlMultiHandle.h"
 #include "curlStream.h"
 #include "definedDirs.h"
 #include "net.h"
@@ -54,3 +56,31 @@ BOOST_AUTO_TEST_CASE( fetch_missing_stream )
 	}, AdHoc::Net::CurlException);
 }
 
+static
+void
+mapFileToName(std::map<std::string, std::string> & map, const std::string & file, std::istream & curlstrm)
+{
+	std::string tok;
+	curlstrm >> tok; // #define
+	curlstrm >> tok; // BOOST_TEST_MODULE
+	curlstrm >> tok; // name :)
+	map[file] = tok;
+}
+
+BOOST_AUTO_TEST_CASE( fetch_multi )
+{
+	CurlMultiHandle cmh;
+	std::map<std::string, std::string> files;
+	cmh.addCurl("file://" + RootDir.string() + "/testBuffer.cpp",
+			boost::bind(&mapFileToName, boost::ref(files), "testBuffer.cpp", _1));
+	cmh.addCurl("file://" + RootDir.string() + "/testCurl.cpp",
+			boost::bind(&mapFileToName, boost::ref(files), "testCurl.cpp", _1));
+	cmh.addCurl("file://" + RootDir.string() + "/testLocks.cpp",
+			boost::bind(&mapFileToName, boost::ref(files), "testLocks.cpp", _1));
+	cmh.performAll();
+	BOOST_REQUIRE_EQUAL(3, files.size());
+	BOOST_REQUIRE_EQUAL("Locks", files["testLocks.cpp"]);
+	BOOST_REQUIRE_EQUAL("Buffer", files["testBuffer.cpp"]);
+	BOOST_REQUIRE_EQUAL("Curl", files["testCurl.cpp"]);
+}
+