See detailled log

- Builds on windows again
- All HTTP requests are no passed into interpreter
- New response element to reply with data
- Moved basichttp URL
- New HTTP servlet invoker to register additional URLs
- More bugfixes than I care to mention

7 files changed, 411 insertions, 66 deletions

diff --git a/test/src/scxml-test-framework-client.cpp b/test/src/scxml-test-framework-client.cpp
index 841df0f..1eeab9c 100644
--- a/test/src/scxml-test-framework-client.cpp
+++ b/test/src/scxml-test-framework-client.cpp
@@ -1,5 +1,5 @@
 #include "uscxml/Interpreter.h"
-#include "uscxml/plugins/ioprocessor/basichttp/libevent/EventIOProcessor.h"
+#include "uscxml/server/HTTPServer.h"
 #include <sstream>
 
 extern "C" {
@@ -20,40 +20,19 @@ extern "C" {
  {"load":"http://localhost:9999/scxml-test-framework/test/targetless-transition/test3.scxml"}
 */
 
-class TestIOProcessor : public uscxml::EventIOProcessor, public uscxml::InterpreterMonitor {
+class TestIOProcessor : public uscxml::HTTPServlet, public uscxml::InterpreterMonitor {
 public:
 
-
   static int lastToken;
-  static std::map<std::string, std::pair<uscxml::Interpreter*, evhttp_request*> > _interpreters;
+  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 onStableConfiguration(uscxml::Interpreter* interpreter) {
-    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;
-    
-    struct evbuffer *databuf = evbuffer_new();
-    evbuffer_add(databuf, replyString.str().c_str(), replyString.str().length());
-    evhttp_send_reply(_interpreters[interpreter->getName()].second, 200, "OK", databuf);
-    evbuffer_free(databuf);
-
+  virtual void beforeCompletion(uscxml::Interpreter* interpreter) {
+    _interpreters[interpreter->getName()].second.curlReq = NULL;
   }
 
-  virtual void beforeCompletion(uscxml::Interpreter* interpreter) {}
   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) {}
@@ -65,6 +44,7 @@ public:
     }
     std::cout << std::endl;
   }
+  
   virtual void afterEnteringStates(uscxml::Interpreter* interpreter) {
     std::cout << "After entering states: ";
     for (int i = 0; i < interpreter->getConfiguration().size(); i++) {
@@ -72,6 +52,7 @@ public:
     }
     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++) {
@@ -84,6 +65,7 @@ public:
     }
     std::cout << std::endl;
   }
+  
   virtual void afterExitingStates(uscxml::Interpreter* interpreter) {
     std::cout << "After exiting states: ";
     for (int i = 0; i < interpreter->getConfiguration().size(); i++) {
@@ -92,37 +74,46 @@ public:
     std::cout << std::endl;
   }
   
-  virtual void httpRecvReq(struct evhttp_request *req) {
-    
-    std::cout << "---- received:" << std::endl;
-
-    if (evhttp_request_get_command(req) != EVHTTP_REQ_POST)
+  virtual void onStableConfiguration(uscxml::Interpreter* interpreter) {
+    if (alreadyAnswered)
       return;
     
-    evhttp_request_own(req);
-
-    struct evkeyval *header;
-    struct evkeyvalq *headers;
-    headers = evhttp_request_get_input_headers(req);
+    Arabica::XPath::NodeSet<std::string> configuration = interpreter->getConfiguration();
     
-    for (header = headers->tqh_first; header;
-         header = header->next.tqe_next) {
-//      std::cout << header->key << ": " << header->value << std::endl;
+    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);
+    
+  }
 
-    std::string content;
-    struct evbuffer *buf;
-		buf = evhttp_request_get_input_buffer(req);
-		while (evbuffer_get_length(buf)) {
-			int n;
-			char cbuf[128];
-			n = evbuffer_remove(buf, cbuf, sizeof(buf)-1);
-			if (n > 0) {
-				content.append(cbuf, n);
-			}
-		}
+  void httpRecvRequest(const uscxml::HTTPServer::Request& request) {
+    
+//    uscxml::HTTPServer::Reply httpReply(request);
+//    uscxml::HTTPServer::reply(httpReply);
+//    return;
     
-    uscxml::Data jsonReq = uscxml::Data::fromJSON(content);
+    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;
     
     
@@ -130,6 +121,19 @@ public:
     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++);
+//        } else {
+//          interpreterIter++;
+//        }
+      }
+
+      
       uscxml::Interpreter* interpreter = uscxml::Interpreter::fromURI(filename);
       if (interpreter) {
         std::string token = uscxml::toStr(lastToken++);
@@ -137,7 +141,7 @@ public:
         interpreter->setName(token);
         interpreter->addMonitor(this);
         interpreter->start();
-        _interpreters[token] = std::make_pair(interpreter, req);
+        _interpreters[token] = std::make_pair(interpreter, request);
       }
       return;
     }
@@ -151,33 +155,27 @@ public:
       event.name = jsonReq.compound["event"].compound["name"].atom;
       std::cout << "Sending event " << event << std::endl;
 //      evhttp_request_free(_interpreters[token].second);
-      _interpreters[token].second = req;
+      alreadyAnswered = false;
+      _interpreters[token].second = request;
       _interpreters[token].first->receive(event);
     }
-    
-  }
-  
-  std::string getPath() {
-    return "test";
+   
   }
-  
+    
   void setURL(const std::string& url) {
     std::cout << "Listening at " << url << std::endl;
-    _url = url;
   }
 };
 
 int TestIOProcessor::lastToken;
-std::map<std::string, std::pair<uscxml::Interpreter*, evhttp_request*> > TestIOProcessor::_interpreters;
+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::EventIOServer::registerProcessor(testServer);
+  uscxml::HTTPServer::registerServlet("test", testServer);
 
   while(true)
     tthread::this_thread::sleep_for(tthread::chrono::milliseconds(20));
   
-//	uscxml::Interpreter* interpreter = uscxml::Interpreter::fromURI(argv[1]);
-//	interpreter->dump();
-//	interpreter->interpret();
 }
\ No newline at end of file
diff --git a/test/src/test-communication.cpp b/test/src/test-communication.cpp.old
index a0cdbbd..a0cdbbd 100644
--- a/test/src/test-communication.cpp
+++ b/test/src/test-communication.cpp.old
diff --git a/test/src/test-completion.cpp b/test/src/test-completion.cpp.old
index 67d8708..67d8708 100644
--- a/test/src/test-completion.cpp
+++ b/test/src/test-completion.cpp.old
diff --git a/test/src/test-curl-multi-api.cpp b/test/src/test-curl-multi-api.cpp
new file mode 100644
index 0000000..fac06dc
--- /dev/null
+++ b/test/src/test-curl-multi-api.cpp
@@ -0,0 +1,347 @@
+#include <curl/curl.h>
+#include <glog/logging.h>
+#include <string>
+#include <iostream>
+#include <sstream>
+#include <map>
+#include <set>
+#include <boost/shared_ptr.hpp>
+#include <boost/enable_shared_from_this.hpp>
+
+#include "uscxml/concurrency/tinythread.h"
+
+// use arabica URL parser
+#include <io/uri.hpp>
+
+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) {};
+};
+
+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()) {
+//      (*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;
+};
+
+class URL {
+public:
+	URL() : _impl() {}
+	URL(const std::string url) : _impl(new URLImpl(url)) {}
+	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; }
+	bool operator!=(const URL& other) const {
+		return _impl != other._impl;
+	}
+	URL& operator= (const URL& other)       {
+		_impl = other._impl;
+		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); }
+
+  void addMonitor(URLMonitor* monitor)    { _impl->addMonitor(monitor); }
+  void removeMonitor(URLMonitor* monitor) { _impl->removeMonitor(monitor); }
+
+  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); }
+
+	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;
+	tthread::recursive_mutex _mutex;
+  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) {}
+}
\ No newline at end of file
diff --git a/test/src/test-ecmascript-v8.cpp b/test/src/test-ecmascript-v8.cpp.old
index 9ce39d9..9ce39d9 100644
--- a/test/src/test-ecmascript-v8.cpp
+++ b/test/src/test-ecmascript-v8.cpp.old
diff --git a/test/src/test-execution.cpp b/test/src/test-execution.cpp.old
index 272ce41..272ce41 100644
--- a/test/src/test-execution.cpp
+++ b/test/src/test-execution.cpp.old
diff --git a/test/src/test-prolog-swi.cpp b/test/src/test-prolog-swi.cpp.old
index 9e44f12..9e44f12 100644
--- a/test/src/test-prolog-swi.cpp
+++ b/test/src/test-prolog-swi.cpp.old