From 3a517b28906fceb1d6867d29ff37403200626fa6 Mon Sep 17 00:00:00 2001 From: Dan Goodliffe Date: Sun, 30 Aug 2015 04:13:57 +0100 Subject: Migrate CurlMultiHandle and extend CurlStream for it. --- libadhocutil/curlHandle.cpp | 5 ++ libadhocutil/curlHandle.h | 2 + libadhocutil/curlMultiHandle.cpp | 92 +++++++++++++++++++++++++++++++++++++ libadhocutil/curlMultiHandle.h | 38 +++++++++++++++ libadhocutil/curlStream.h | 1 + libadhocutil/unittests/testCurl.cpp | 30 ++++++++++++ 6 files changed, 168 insertions(+) create mode 100644 libadhocutil/curlMultiHandle.cpp create mode 100644 libadhocutil/curlMultiHandle.h 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 +#include +#include "runtimeContext.h" +#include "curlStream.h" + +class RunningCurl : public CurlStreamSource { + public: + RunningCurl(const std::string & url, const boost::function & c) : + CurlStreamSource(url), + consumer(c) + { + } + + void Callback() override + { + typedef boost::reference_wrapper rc_ref; + boost::iostreams::stream curlstrm(boost::ref(*this)); + consumer(curlstrm); + } + + private: + const boost::function consumer; +}; + +CurlMultiHandle::CurlMultiHandle() +{ +} + +CurlMultiHandle::~CurlMultiHandle() +{ +} + +CurlHandlePtr +CurlMultiHandle::addCurl(const std::string & url, const boost::function & 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 +#include +#include +#include "intrusivePtrBase.h" +#include "visibility.h" +#include "curlHandle.h" + +class RunningCurl; +typedef boost::intrusive_ptr RunningCurlPtr; + +class DLL_PUBLIC CurlMultiHandle : public IntrusivePtrBase { + public: + typedef std::set CURLs; + typedef boost::function 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 Running; + + DLL_PRIVATE void addRunner(CURLM * curlm, Running & running, CurlMultiHandle::CURLs & curls); + + CURLs curls; +}; +typedef boost::intrusive_ptr 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 +#include #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 & 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 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"]); +} + -- cgit v1.2.3