diff options
Diffstat (limited to 'test')
-rw-r--r-- | test/src/scxml-test-framework-client.cpp | 280 | ||||
-rw-r--r-- | test/src/test-curl-multi-api.cpp | 633 | ||||
-rw-r--r-- | test/src/test-url.cpp | 34 |
3 files changed, 502 insertions, 445 deletions
diff --git a/test/src/scxml-test-framework-client.cpp b/test/src/scxml-test-framework-client.cpp index 1eeab9c..58f1057 100644 --- a/test/src/scxml-test-framework-client.cpp +++ b/test/src/scxml-test-framework-client.cpp @@ -16,155 +16,155 @@ extern "C" { content-type: application/json content-length: 92 Connection: keep-alive - + {"load":"http://localhost:9999/scxml-test-framework/test/targetless-transition/test3.scxml"} */ class TestIOProcessor : public uscxml::HTTPServlet, public uscxml::InterpreterMonitor { public: - static int lastToken; - static bool alreadyAnswered; // we need this for delayed events - static std::map<std::string, std::pair<uscxml::Interpreter*, uscxml::HTTPServer::Request> > _interpreters; - - TestIOProcessor() {} - - virtual void beforeCompletion(uscxml::Interpreter* interpreter) { - _interpreters[interpreter->getName()].second.curlReq = NULL; - } - - virtual void afterCompletion(uscxml::Interpreter* interpreter) {} - virtual void beforeMicroStep(uscxml::Interpreter* interpreter) {} - virtual void beforeTakingTransitions(uscxml::Interpreter* interpreter, const Arabica::XPath::NodeSet<std::string>& transitions) {} - - virtual void beforeEnteringStates(uscxml::Interpreter* interpreter, const Arabica::XPath::NodeSet<std::string>& statesToEnter) { - std::cout << "Entering states: "; - for (int i = 0; i < statesToEnter.size(); i++) { - std::cout << ATTR(statesToEnter[i], "id") << ", "; - } - std::cout << std::endl; - } - - virtual void afterEnteringStates(uscxml::Interpreter* interpreter) { - std::cout << "After entering states: "; - for (int i = 0; i < interpreter->getConfiguration().size(); i++) { - std::cout << ATTR(interpreter->getConfiguration()[i], "id") << ", "; - } - std::cout << std::endl; - } - - virtual void beforeExitingStates(uscxml::Interpreter* interpreter, const Arabica::XPath::NodeSet<std::string>& statesToExit) { - std::cout << "Configuration: "; - for (int i = 0; i < interpreter->getConfiguration().size(); i++) { - std::cout << ATTR(interpreter->getConfiguration()[i], "id") << ", "; - } - std::cout << std::endl; - std::cout << "Exiting states: "; - for (int i = 0; i < statesToExit.size(); i++) { - std::cout << ATTR(statesToExit[i], "id") << ", "; - } - std::cout << std::endl; - } - - virtual void afterExitingStates(uscxml::Interpreter* interpreter) { - std::cout << "After exiting states: "; - for (int i = 0; i < interpreter->getConfiguration().size(); i++) { - std::cout << ATTR(interpreter->getConfiguration()[i], "id") << ", "; - } - std::cout << std::endl; - } - - virtual void onStableConfiguration(uscxml::Interpreter* interpreter) { - if (alreadyAnswered) - return; - - Arabica::XPath::NodeSet<std::string> configuration = interpreter->getConfiguration(); - - uscxml::Data reply; - reply.compound["sessionToken"] = uscxml::Data(interpreter->getName()); - std::string seperator; - for (size_t i = 0; i < configuration.size(); i++) { - if (uscxml::Interpreter::isAtomic(configuration[i])) - reply.compound["nextConfiguration"].array.push_back(uscxml::Data(ATTR(configuration[i], "id"), uscxml::Data::VERBATIM)); - } - - std::cout << "---- reply:" << std::endl; - std::cout << reply << std::endl; - - std::stringstream replyString; - replyString << reply; - - alreadyAnswered = true; - - uscxml::HTTPServer::Request httpRequest = _interpreters[interpreter->getName()].second; - uscxml::HTTPServer::Reply httpReply(httpRequest); - httpReply.content = replyString.str(); - uscxml::HTTPServer::reply(httpReply); - - } - - void httpRecvRequest(const uscxml::HTTPServer::Request& request) { - + static int lastToken; + static bool alreadyAnswered; // we need this for delayed events + static std::map<std::string, std::pair<uscxml::Interpreter*, uscxml::HTTPServer::Request> > _interpreters; + + TestIOProcessor() {} + + virtual void beforeCompletion(uscxml::Interpreter* interpreter) { + _interpreters[interpreter->getName()].second.curlReq = NULL; + } + + virtual void afterCompletion(uscxml::Interpreter* interpreter) {} + virtual void beforeMicroStep(uscxml::Interpreter* interpreter) {} + virtual void beforeTakingTransitions(uscxml::Interpreter* interpreter, const Arabica::XPath::NodeSet<std::string>& transitions) {} + + virtual void beforeEnteringStates(uscxml::Interpreter* interpreter, const Arabica::XPath::NodeSet<std::string>& statesToEnter) { + std::cout << "Entering states: "; + for (int i = 0; i < statesToEnter.size(); i++) { + std::cout << ATTR(statesToEnter[i], "id") << ", "; + } + std::cout << std::endl; + } + + virtual void afterEnteringStates(uscxml::Interpreter* interpreter) { + std::cout << "After entering states: "; + for (int i = 0; i < interpreter->getConfiguration().size(); i++) { + std::cout << ATTR(interpreter->getConfiguration()[i], "id") << ", "; + } + std::cout << std::endl; + } + + virtual void beforeExitingStates(uscxml::Interpreter* interpreter, const Arabica::XPath::NodeSet<std::string>& statesToExit) { + std::cout << "Configuration: "; + for (int i = 0; i < interpreter->getConfiguration().size(); i++) { + std::cout << ATTR(interpreter->getConfiguration()[i], "id") << ", "; + } + std::cout << std::endl; + std::cout << "Exiting states: "; + for (int i = 0; i < statesToExit.size(); i++) { + std::cout << ATTR(statesToExit[i], "id") << ", "; + } + std::cout << std::endl; + } + + virtual void afterExitingStates(uscxml::Interpreter* interpreter) { + std::cout << "After exiting states: "; + for (int i = 0; i < interpreter->getConfiguration().size(); i++) { + std::cout << ATTR(interpreter->getConfiguration()[i], "id") << ", "; + } + std::cout << std::endl; + } + + virtual void onStableConfiguration(uscxml::Interpreter* interpreter) { + if (alreadyAnswered) + return; + + Arabica::XPath::NodeSet<std::string> configuration = interpreter->getConfiguration(); + + uscxml::Data reply; + reply.compound["sessionToken"] = uscxml::Data(interpreter->getName()); + std::string seperator; + for (size_t i = 0; i < configuration.size(); i++) { + if (uscxml::Interpreter::isAtomic(configuration[i])) + reply.compound["nextConfiguration"].array.push_back(uscxml::Data(ATTR(configuration[i], "id"), uscxml::Data::VERBATIM)); + } + + std::cout << "---- reply:" << std::endl; + std::cout << reply << std::endl; + + std::stringstream replyString; + replyString << reply; + + alreadyAnswered = true; + + uscxml::HTTPServer::Request httpRequest = _interpreters[interpreter->getName()].second; + uscxml::HTTPServer::Reply httpReply(httpRequest); + httpReply.content = replyString.str(); + uscxml::HTTPServer::reply(httpReply); + + } + + void httpRecvRequest(const uscxml::HTTPServer::Request& request) { + // uscxml::HTTPServer::Reply httpReply(request); // uscxml::HTTPServer::reply(httpReply); // return; - - std::cout << "---- received:" << std::endl; - evhttp_request_own(request.curlReq); - - std::cout << request.content << std::endl; - uscxml::Data jsonReq = uscxml::Data::fromJSON(request.content); - std::cout << jsonReq << std::endl; - - - // is this a load request? - if (jsonReq.compound.find("load") != jsonReq.compound.end()) { - std::string filename = jsonReq.compound["load"].atom; - std::cout << "Starting Interpreter with " << filename << std::endl; - alreadyAnswered = false; - - std::map<std::string, std::pair<uscxml::Interpreter*, uscxml::HTTPServer::Request> >::iterator interpreterIter = _interpreters.begin(); - while(interpreterIter != _interpreters.end()) { + + std::cout << "---- received:" << std::endl; + evhttp_request_own(request.curlReq); + + std::cout << request.content << std::endl; + uscxml::Data jsonReq = uscxml::Data::fromJSON(request.content); + std::cout << jsonReq << std::endl; + + + // is this a load request? + if (jsonReq.compound.find("load") != jsonReq.compound.end()) { + std::string filename = jsonReq.compound["load"].atom; + std::cout << "Starting Interpreter with " << filename << std::endl; + alreadyAnswered = false; + + std::map<std::string, std::pair<uscxml::Interpreter*, uscxml::HTTPServer::Request> >::iterator interpreterIter = _interpreters.begin(); + while(interpreterIter != _interpreters.end()) { // if (interpreterIter->second.second.curlReq == NULL) { - delete interpreterIter->second.first; - _interpreters.erase(interpreterIter++); + delete interpreterIter->second.first; + _interpreters.erase(interpreterIter++); // } else { // interpreterIter++; // } - } - - - uscxml::Interpreter* interpreter = uscxml::Interpreter::fromURI(filename); - if (interpreter) { - std::string token = uscxml::toStr(lastToken++); - assert(_interpreters.find(token) == _interpreters.end()); - interpreter->setName(token); - interpreter->addMonitor(this); - interpreter->start(); - _interpreters[token] = std::make_pair(interpreter, request); - } - return; - } - - if(jsonReq.compound.find("event") != jsonReq.compound.end()) { - assert(jsonReq.compound["event"].compound.find("sessionToken") != jsonReq.compound["event"].compound.end()); - std::string token = jsonReq.compound["event"].compound["sessionToken"].atom; - assert(_interpreters.find(token) != _interpreters.end()); - uscxml::Event event; - event.type = uscxml::Event::INTERNAL; - event.name = jsonReq.compound["event"].compound["name"].atom; - std::cout << "Sending event " << event << std::endl; + } + + + uscxml::Interpreter* interpreter = uscxml::Interpreter::fromURI(filename); + if (interpreter) { + std::string token = uscxml::toStr(lastToken++); + assert(_interpreters.find(token) == _interpreters.end()); + interpreter->setName(token); + interpreter->addMonitor(this); + interpreter->start(); + _interpreters[token] = std::make_pair(interpreter, request); + } + return; + } + + if(jsonReq.compound.find("event") != jsonReq.compound.end()) { + assert(jsonReq.compound["event"].compound.find("sessionToken") != jsonReq.compound["event"].compound.end()); + std::string token = jsonReq.compound["event"].compound["sessionToken"].atom; + assert(_interpreters.find(token) != _interpreters.end()); + uscxml::Event event; + event.type = uscxml::Event::INTERNAL; + event.name = jsonReq.compound["event"].compound["name"].atom; + std::cout << "Sending event " << event << std::endl; // evhttp_request_free(_interpreters[token].second); - alreadyAnswered = false; - _interpreters[token].second = request; - _interpreters[token].first->receive(event); - } - - } - - void setURL(const std::string& url) { - std::cout << "Listening at " << url << std::endl; - } + alreadyAnswered = false; + _interpreters[token].second = request; + _interpreters[token].first->receive(event); + } + + } + + void setURL(const std::string& url) { + std::cout << "Listening at " << url << std::endl; + } }; int TestIOProcessor::lastToken; @@ -172,10 +172,10 @@ bool TestIOProcessor::alreadyAnswered; std::map<std::string, std::pair<uscxml::Interpreter*, uscxml::HTTPServer::Request> > TestIOProcessor::_interpreters; int main(int argc, char** argv) { - TestIOProcessor* testServer = new TestIOProcessor(); - uscxml::HTTPServer::registerServlet("test", testServer); + TestIOProcessor* testServer = new TestIOProcessor(); + uscxml::HTTPServer::registerServlet("test", testServer); + + while(true) + tthread::this_thread::sleep_for(tthread::chrono::milliseconds(20)); - while(true) - tthread::this_thread::sleep_for(tthread::chrono::milliseconds(20)); - }
\ No newline at end of file diff --git a/test/src/test-curl-multi-api.cpp b/test/src/test-curl-multi-api.cpp index fac06dc..67735d2 100644 --- a/test/src/test-curl-multi-api.cpp +++ b/test/src/test-curl-multi-api.cpp @@ -17,108 +17,124 @@ class URL; class URLMonitor { public: - virtual void downloadStarted(const URL& url) {}; - virtual void downloadCompleted(const URL& url) {}; - virtual void downloadFailed(const URL& url, int errorCode) {}; - virtual void headerChunkReceived(const URL& url, const std::string& headerChunk) {}; - virtual void contentChunkReceived(const URL& url, const std::string& contentChunk) {}; + virtual void downloadStarted(const URL& url) {}; + virtual void downloadCompleted(const URL& url) {}; + virtual void downloadFailed(const URL& url, int errorCode) {}; + virtual void headerChunkReceived(const URL& url, const std::string& headerChunk) {}; + virtual void contentChunkReceived(const URL& url, const std::string& contentChunk) {}; }; class URLImpl : public boost::enable_shared_from_this<URLImpl> { public: - URLImpl(const std::string& url) : _handle(NULL), _uri(url), _isDownloaded(false) { - _handle = curl_easy_init(); - if (_handle != NULL) { - CURLcode curlError; - curlError = curl_easy_setopt(_handle, CURLOPT_URL, _uri.as_string().c_str()); - if (curlError != CURLE_OK) - LOG(ERROR) << "Cannot set url to " << _uri.as_string() << ": " << curl_easy_strerror(curlError); - - curlError = curl_easy_setopt(_handle, CURLOPT_WRITEDATA, this); - if (curlError != CURLE_OK) - LOG(ERROR) << "Cannot register this as write userdata: " << curl_easy_strerror(curlError); - - curlError = curl_easy_setopt(_handle, CURLOPT_WRITEFUNCTION, URLImpl::writeHandler); - if (curlError != CURLE_OK) - LOG(ERROR) << "Cannot set write callback: " << curl_easy_strerror(curlError); - - curlError = curl_easy_setopt(_handle, CURLOPT_HEADERFUNCTION, URLImpl::headerHandler); - if (curlError != CURLE_OK) - LOG(ERROR) << "Cannot request header from curl: " << curl_easy_strerror(curlError); - - curlError = curl_easy_setopt(_handle, CURLOPT_HEADERDATA, this); - if (curlError != CURLE_OK) - LOG(ERROR) << "Cannot register this as header userdata: " << curl_easy_strerror(curlError); - } else { - LOG(ERROR) << "curl_easy_init returned NULL, this is bad!"; - } - } - - ~URLImpl() { - if (_handle != NULL) - curl_easy_cleanup(_handle); - } - - static size_t writeHandler(void *ptr, size_t size, size_t nmemb, void *userdata) { - URLImpl* url = (URLImpl*)userdata; - url->_content.write((char*)ptr, size * nmemb); - return size * nmemb; - } - - static size_t headerHandler(void *ptr, size_t size, size_t nmemb, void *userdata) { - URLImpl* url = (URLImpl*)userdata; - url->_header.write((char*)ptr, size * nmemb); - return size * nmemb; - } - - void addMonitor(URLMonitor* monitor) { _monitors.insert(monitor); } - void removeMonitor(URLMonitor* monitor) { _monitors.erase(monitor); } - - const bool isAbsolute() const { return _uri.is_absolute(); } - const std::string scheme() const { return _uri.scheme(); } - const std::string host() const { return _uri.host(); } - const std::string port() const { return _uri.port(); } - const std::string path() const { return _uri.path(); } - const std::string asString() const { return _uri.as_string(); } - - void downloadStarted() { - std::cout << "Starting download of " << asString() << std::endl; - _content.str(""); - _content.clear(); - _header.str(""); - _header.clear(); - monIter_t monIter = _monitors.begin(); - while(monIter != _monitors.end()) { + URLImpl(const std::string& url) : _handle(NULL), _uri(url), _isDownloaded(false) { + _handle = curl_easy_init(); + if (_handle != NULL) { + CURLcode curlError; + curlError = curl_easy_setopt(_handle, CURLOPT_URL, _uri.as_string().c_str()); + if (curlError != CURLE_OK) + LOG(ERROR) << "Cannot set url to " << _uri.as_string() << ": " << curl_easy_strerror(curlError); + + curlError = curl_easy_setopt(_handle, CURLOPT_WRITEDATA, this); + if (curlError != CURLE_OK) + LOG(ERROR) << "Cannot register this as write userdata: " << curl_easy_strerror(curlError); + + curlError = curl_easy_setopt(_handle, CURLOPT_WRITEFUNCTION, URLImpl::writeHandler); + if (curlError != CURLE_OK) + LOG(ERROR) << "Cannot set write callback: " << curl_easy_strerror(curlError); + + curlError = curl_easy_setopt(_handle, CURLOPT_HEADERFUNCTION, URLImpl::headerHandler); + if (curlError != CURLE_OK) + LOG(ERROR) << "Cannot request header from curl: " << curl_easy_strerror(curlError); + + curlError = curl_easy_setopt(_handle, CURLOPT_HEADERDATA, this); + if (curlError != CURLE_OK) + LOG(ERROR) << "Cannot register this as header userdata: " << curl_easy_strerror(curlError); + } else { + LOG(ERROR) << "curl_easy_init returned NULL, this is bad!"; + } + } + + ~URLImpl() { + if (_handle != NULL) + curl_easy_cleanup(_handle); + } + + static size_t writeHandler(void *ptr, size_t size, size_t nmemb, void *userdata) { + URLImpl* url = (URLImpl*)userdata; + url->_content.write((char*)ptr, size * nmemb); + return size * nmemb; + } + + static size_t headerHandler(void *ptr, size_t size, size_t nmemb, void *userdata) { + URLImpl* url = (URLImpl*)userdata; + url->_header.write((char*)ptr, size * nmemb); + return size * nmemb; + } + + void addMonitor(URLMonitor* monitor) { + _monitors.insert(monitor); + } + void removeMonitor(URLMonitor* monitor) { + _monitors.erase(monitor); + } + + const bool isAbsolute() const { + return _uri.is_absolute(); + } + const std::string scheme() const { + return _uri.scheme(); + } + const std::string host() const { + return _uri.host(); + } + const std::string port() const { + return _uri.port(); + } + const std::string path() const { + return _uri.path(); + } + const std::string asString() const { + return _uri.as_string(); + } + + void downloadStarted() { + std::cout << "Starting download of " << asString() << std::endl; + _content.str(""); + _content.clear(); + _header.str(""); + _header.clear(); + monIter_t monIter = _monitors.begin(); + while(monIter != _monitors.end()) { // (*monIter)->downloadStarted(URL(shared_from_this())); - monIter++; - } - } - - void downloadCompleted() { - std::cout << "Finished loading " << asString() << " with " << _content.str().size() << " bytes" << std::endl; - _isDownloaded = true; - } - - void downloadFailed(int errorCode) { - std::cout << "FAILED!" << strerror(errorCode) << std::endl; - } - - std::string getHeader(bool forceReload = false) { - return _header.str(); - } - - std::string getContent(bool forceReload = false) { - return _content.str(); - } - - std::stringstream _content; - std::stringstream _header; - CURL* _handle; - Arabica::io::URI _uri; - bool _isDownloaded; - - std::set<URLMonitor*> _monitors; - typedef std::set<URLMonitor*>::iterator monIter_t; + monIter++; + } + } + + void downloadCompleted() { + std::cout << "Finished loading " << asString() << " with " << _content.str().size() << " bytes" << std::endl; + _isDownloaded = true; + } + + void downloadFailed(int errorCode) { + std::cout << "FAILED!" << strerror(errorCode) << std::endl; + } + + std::string getHeader(bool forceReload = false) { + return _header.str(); + } + + std::string getContent(bool forceReload = false) { + return _content.str(); + } + + std::stringstream _content; + std::stringstream _header; + CURL* _handle; + Arabica::io::URI _uri; + bool _isDownloaded; + + std::set<URLMonitor*> _monitors; + typedef std::set<URLMonitor*>::iterator monIter_t; }; class URL { @@ -128,10 +144,16 @@ public: URL(boost::shared_ptr<URLImpl> const impl) : _impl(impl) { } URL(const URL& other) : _impl(other._impl) { } virtual ~URL() {}; - - operator bool() const { return _impl; } - bool operator< (const URL& other) const { return _impl < other._impl; } - bool operator==(const URL& other) const { return _impl == other._impl; } + + operator bool() const { + return _impl; + } + bool operator< (const URL& other) const { + return _impl < other._impl; + } + bool operator==(const URL& other) const { + return _impl == other._impl; + } bool operator!=(const URL& other) const { return _impl != other._impl; } @@ -140,208 +162,243 @@ public: return *this; } - std::string getHeader() { return _impl->getHeader(); } - std::string getContent() { return _impl->getContent(); } - - const bool toAbsoluteCwd() { return _impl->toAbsoluteCwd(); } - const bool toAbsolute(const std::string& baseUrl) { return _impl->toAbsolute(baseUrl); } - const bool toAbsolute(const URL& baseUrl) { return _impl->toAbsolute(baseUrl.asString()); } - const std::string asLocalFile(const std::string& suffix, bool reload = false) { return _impl->asLocalFile(suffix, reload); } + std::string getHeader() { + return _impl->getHeader(); + } + std::string getContent() { + return _impl->getContent(); + } - void addMonitor(URLMonitor* monitor) { _impl->addMonitor(monitor); } - void removeMonitor(URLMonitor* monitor) { _impl->removeMonitor(monitor); } + const bool toAbsoluteCwd() { + return _impl->toAbsoluteCwd(); + } + const bool toAbsolute(const std::string& baseUrl) { + return _impl->toAbsolute(baseUrl); + } + const bool toAbsolute(const URL& baseUrl) { + return _impl->toAbsolute(baseUrl.asString()); + } + const std::string asLocalFile(const std::string& suffix, bool reload = false) { + return _impl->asLocalFile(suffix, reload); + } - const bool isAbsolute() const { return _impl->isAbsolute(); } - const std::string scheme() const { return _impl->scheme(); } - const std::string host() const { return _impl->host(); } - const std::string port() const { return _impl->port(); } - const std::string path() const { return _impl->path(); } - const std::string asString() const { return _impl->asString(); } + void addMonitor(URLMonitor* monitor) { + _impl->addMonitor(monitor); + } + void removeMonitor(URLMonitor* monitor) { + _impl->removeMonitor(monitor); + } - friend class URLFetcher; + const bool isAbsolute() const { + return _impl->isAbsolute(); + } + const std::string scheme() const { + return _impl->scheme(); + } + const std::string host() const { + return _impl->host(); + } + const std::string port() const { + return _impl->port(); + } + const std::string path() const { + return _impl->path(); + } + const std::string asString() const { + return _impl->asString(); + } + + friend class URLFetcher; friend std::ostream & operator<<(std::ostream &stream, const URL& p); protected: - void downloadStarted() { return _impl->downloadStarted(); } - void downloadCompleted() { return _impl->downloadCompleted(); } - void downloadFailed(int errorCode) { return _impl->downloadFailed(errorCode); } + void downloadStarted() { + return _impl->downloadStarted(); + } + void downloadCompleted() { + return _impl->downloadCompleted(); + } + void downloadFailed(int errorCode) { + return _impl->downloadFailed(errorCode); + } boost::shared_ptr<URLImpl> _impl; }; class URLFetcher { public: - URLFetcher() { - _multiHandle = curl_multi_init(); - start(); - } - - ~URLFetcher() { - curl_multi_cleanup(_multiHandle); - stop(); - } - - void fetchURL(URL& url) { - tthread::lock_guard<tthread::recursive_mutex> lock(_mutex); - url.downloadStarted(); - _handlesToURLs[url._impl->_handle] = url; - curl_multi_add_handle(_multiHandle, url._impl->_handle); - _condVar.notify_all(); - } - - void breakURL(URL& url) { - tthread::lock_guard<tthread::recursive_mutex> lock(_mutex); - if (_handlesToURLs.find(url._impl->_handle) != _handlesToURLs.end()) { - url.downloadFailed(0); - curl_multi_remove_handle(_multiHandle, url._impl->_handle); - _handlesToURLs.erase(url._impl->_handle); - } - } - - void start() { - tthread::lock_guard<tthread::recursive_mutex> lock(_mutex); - if (!_isStarted) { - _isStarted = true; - _thread = new tthread::thread(URLFetcher::run, this); - } - } - - void stop() { - tthread::lock_guard<tthread::recursive_mutex> lock(_mutex); - if (_isStarted) { - _isStarted = false; - _thread->join(); - delete _thread; - } - } - - static void run(void* instance) { - URLFetcher* THIS = (URLFetcher*)instance; - THIS->_mutex.lock(); - while(THIS->_isStarted) { - if(THIS->_handlesToURLs.size() > 0) { - THIS->_mutex.unlock(); - THIS->perform(); - THIS->_mutex.lock(); - } - THIS->_condVar.wait(THIS->_mutex); - } - THIS->_mutex.unlock(); - } - - void perform() { - - CURLMsg *msg; /* for picking up messages with the transfer status */ - int msgsLeft; /* how many messages are left */ - int stillRunning; - - { - tthread::lock_guard<tthread::recursive_mutex> lock(_mutex); - curl_multi_perform(_multiHandle, &stillRunning); - } - - do { - struct timeval timeout; - int rc; /* select() return code */ - - fd_set fdread, fdwrite, fdexcep; - FD_ZERO(&fdread); FD_ZERO(&fdwrite); FD_ZERO(&fdexcep); - - int maxfd = -1; - long curlTimeOut = -1; - - /* set a suitable timeout to play around with */ - timeout.tv_sec = 1; - timeout.tv_usec = 0; - - { - tthread::lock_guard<tthread::recursive_mutex> lock(_mutex); - curl_multi_timeout(_multiHandle, &curlTimeOut); - } - - if(curlTimeOut >= 0) { - timeout.tv_sec = curlTimeOut / 1000; - if(timeout.tv_sec > 1) - timeout.tv_sec = 1; - else - timeout.tv_usec = (curlTimeOut % 1000) * 1000; - } - - /* get file descriptors from the transfers */ - { - tthread::lock_guard<tthread::recursive_mutex> lock(_mutex); - curl_multi_fdset(_multiHandle, &fdread, &fdwrite, &fdexcep, &maxfd); - } - - rc = select(maxfd+1, &fdread, &fdwrite, &fdexcep, &timeout); - - switch(rc) { - case -1: - /* select error */ - break; - case 0: /* timeout */ - default: /* action */ - { - tthread::lock_guard<tthread::recursive_mutex> lock(_mutex); - curl_multi_perform(_multiHandle, &stillRunning); - } - break; - } - - { - tthread::lock_guard<tthread::recursive_mutex> lock(_mutex); - while ((msg = curl_multi_info_read(_multiHandle, &msgsLeft))) { - if (msg->msg == CURLMSG_DONE) { - _handlesToURLs[msg->easy_handle].downloadCompleted(); - curl_multi_remove_handle(_multiHandle, msg->easy_handle); - _handlesToURLs.erase(msg->easy_handle); - } else { - switch (msg->data.result) { - case CURLM_OK: - break; - case CURLM_BAD_HANDLE: - case CURLM_BAD_EASY_HANDLE: - case CURLM_OUT_OF_MEMORY: - case CURLM_INTERNAL_ERROR: - case CURLM_BAD_SOCKET: - case CURLM_UNKNOWN_OPTION: - case CURLM_LAST: - _handlesToURLs[msg->easy_handle].downloadFailed(msg->data.result); - curl_multi_remove_handle(_multiHandle, msg->easy_handle); - _handlesToURLs.erase(msg->easy_handle); - default: - break; - } - } - } - } - } while(stillRunning && _isStarted); - - } - - tthread::condition_variable _condVar; - tthread::thread* _thread; + URLFetcher() { + _multiHandle = curl_multi_init(); + start(); + } + + ~URLFetcher() { + curl_multi_cleanup(_multiHandle); + stop(); + } + + void fetchURL(URL& url) { + tthread::lock_guard<tthread::recursive_mutex> lock(_mutex); + url.downloadStarted(); + _handlesToURLs[url._impl->_handle] = url; + curl_multi_add_handle(_multiHandle, url._impl->_handle); + _condVar.notify_all(); + } + + void breakURL(URL& url) { + tthread::lock_guard<tthread::recursive_mutex> lock(_mutex); + if (_handlesToURLs.find(url._impl->_handle) != _handlesToURLs.end()) { + url.downloadFailed(0); + curl_multi_remove_handle(_multiHandle, url._impl->_handle); + _handlesToURLs.erase(url._impl->_handle); + } + } + + void start() { + tthread::lock_guard<tthread::recursive_mutex> lock(_mutex); + if (!_isStarted) { + _isStarted = true; + _thread = new tthread::thread(URLFetcher::run, this); + } + } + + void stop() { + tthread::lock_guard<tthread::recursive_mutex> lock(_mutex); + if (_isStarted) { + _isStarted = false; + _thread->join(); + delete _thread; + } + } + + static void run(void* instance) { + URLFetcher* THIS = (URLFetcher*)instance; + THIS->_mutex.lock(); + while(THIS->_isStarted) { + if(THIS->_handlesToURLs.size() > 0) { + THIS->_mutex.unlock(); + THIS->perform(); + THIS->_mutex.lock(); + } + THIS->_condVar.wait(THIS->_mutex); + } + THIS->_mutex.unlock(); + } + + void perform() { + + CURLMsg *msg; /* for picking up messages with the transfer status */ + int msgsLeft; /* how many messages are left */ + int stillRunning; + + { + tthread::lock_guard<tthread::recursive_mutex> lock(_mutex); + curl_multi_perform(_multiHandle, &stillRunning); + } + + do { + struct timeval timeout; + int rc; /* select() return code */ + + fd_set fdread, fdwrite, fdexcep; + FD_ZERO(&fdread); + FD_ZERO(&fdwrite); + FD_ZERO(&fdexcep); + + int maxfd = -1; + long curlTimeOut = -1; + + /* set a suitable timeout to play around with */ + timeout.tv_sec = 1; + timeout.tv_usec = 0; + + { + tthread::lock_guard<tthread::recursive_mutex> lock(_mutex); + curl_multi_timeout(_multiHandle, &curlTimeOut); + } + + if(curlTimeOut >= 0) { + timeout.tv_sec = curlTimeOut / 1000; + if(timeout.tv_sec > 1) + timeout.tv_sec = 1; + else + timeout.tv_usec = (curlTimeOut % 1000) * 1000; + } + + /* get file descriptors from the transfers */ + { + tthread::lock_guard<tthread::recursive_mutex> lock(_mutex); + curl_multi_fdset(_multiHandle, &fdread, &fdwrite, &fdexcep, &maxfd); + } + + rc = select(maxfd+1, &fdread, &fdwrite, &fdexcep, &timeout); + + switch(rc) { + case -1: + /* select error */ + break; + case 0: /* timeout */ + default: { /* action */ + tthread::lock_guard<tthread::recursive_mutex> lock(_mutex); + curl_multi_perform(_multiHandle, &stillRunning); + } + break; + } + + { + tthread::lock_guard<tthread::recursive_mutex> lock(_mutex); + while ((msg = curl_multi_info_read(_multiHandle, &msgsLeft))) { + if (msg->msg == CURLMSG_DONE) { + _handlesToURLs[msg->easy_handle].downloadCompleted(); + curl_multi_remove_handle(_multiHandle, msg->easy_handle); + _handlesToURLs.erase(msg->easy_handle); + } else { + switch (msg->data.result) { + case CURLM_OK: + break; + case CURLM_BAD_HANDLE: + case CURLM_BAD_EASY_HANDLE: + case CURLM_OUT_OF_MEMORY: + case CURLM_INTERNAL_ERROR: + case CURLM_BAD_SOCKET: + case CURLM_UNKNOWN_OPTION: + case CURLM_LAST: + _handlesToURLs[msg->easy_handle].downloadFailed(msg->data.result); + curl_multi_remove_handle(_multiHandle, msg->easy_handle); + _handlesToURLs.erase(msg->easy_handle); + default: + break; + } + } + } + } + } while(stillRunning && _isStarted); + + } + + tthread::condition_variable _condVar; + tthread::thread* _thread; tthread::recursive_mutex _mutex; - bool _isStarted; - - std::map<CURL*, URL> _handlesToURLs; - CURLM* _multiHandle; + bool _isStarted; + + std::map<CURL*, URL> _handlesToURLs; + CURLM* _multiHandle; }; int main(int argc, char** argv) { - URLFetcher fetcher; - URL heise("http://www.heise.de"); - URL localFile("file:///Users/sradomski/Desktop/scxml.xsd"); - URL slashdot("http://slashdot.org"); - URL asdf("daf://localhost:234"); - URL bahn("http://www.bahn.de"); - - fetcher.fetchURL(heise); - fetcher.fetchURL(localFile); - fetcher.fetchURL(asdf); - fetcher.fetchURL(slashdot); - fetcher.fetchURL(bahn); - - while(1) {} + URLFetcher fetcher; + URL heise("http://www.heise.de"); + URL localFile("file:///Users/sradomski/Desktop/scxml.xsd"); + URL slashdot("http://slashdot.org"); + URL asdf("daf://localhost:234"); + URL bahn("http://www.bahn.de"); + + fetcher.fetchURL(heise); + fetcher.fetchURL(localFile); + fetcher.fetchURL(asdf); + fetcher.fetchURL(slashdot); + fetcher.fetchURL(bahn); + + while(1) {} }
\ No newline at end of file diff --git a/test/src/test-url.cpp b/test/src/test-url.cpp index e3ca1e3..d4ae9bb 100644 --- a/test/src/test-url.cpp +++ b/test/src/test-url.cpp @@ -9,23 +9,23 @@ using namespace boost; int main(int argc, char** argv) { - { - Data data = Data::fromJSON("asdf"); - std::cout << data << std::endl; - } - { - Data data = Data::fromJSON("[ '1', '2', '3', '4' ]"); - std::cout << data << std::endl; - } - { - Data data = Data::fromJSON("{'foo1': 'bar2', 'foo3': { 'foo4': 'bar5' }, 'foo6': 'bar7', 'foo8': { 'foo9': 'foo10': { 'foo11': 'bar12' } } }"); - std::cout << data << std::endl; - } - { - Data data = Data::fromJSON("{\"firstName\": \"John\", \"lastName\": \"Smith\", \"age\": 25, \"address\": { \"streetAddress\": \"21 2nd Street\", \"city\": \"New York\",\"state\": \"NY\",\"postalCode\": 10021},\"phoneNumber\": [{\"type\": \"home\",\"number\": \"212 555-1234\"},{ \"type\": \"fax\",\"number\": \"646 555-4567\"}]}"); - std::cout << data << std::endl; - } - + { + Data data = Data::fromJSON("asdf"); + std::cout << data << std::endl; + } + { + Data data = Data::fromJSON("[ '1', '2', '3', '4' ]"); + std::cout << data << std::endl; + } + { + Data data = Data::fromJSON("{'foo1': 'bar2', 'foo3': { 'foo4': 'bar5' }, 'foo6': 'bar7', 'foo8': { 'foo9': 'foo10': { 'foo11': 'bar12' } } }"); + std::cout << data << std::endl; + } + { + Data data = Data::fromJSON("{\"firstName\": \"John\", \"lastName\": \"Smith\", \"age\": 25, \"address\": { \"streetAddress\": \"21 2nd Street\", \"city\": \"New York\",\"state\": \"NY\",\"postalCode\": 10021},\"phoneNumber\": [{\"type\": \"home\",\"number\": \"212 555-1234\"},{ \"type\": \"fax\",\"number\": \"646 555-4567\"}]}"); + std::cout << data << std::endl; + } + { URL url("http://www.heise.de/index.html"); std::cout << url.asString() << std::endl; |