Add a workaround for MultipartStream constant busy state (RobotWebToo…

…ls#83) * Add a workaround for MultipartStream constant busy state * Remove C++11 features
dfautomation · Jan 28, 2019 · c3d08d4 · c3d08d4
1 parent 243d0fb
commit c3d08d4
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 8 deletions.
diff --git a/include/web_video_server/multipart_stream.h b/include/web_video_server/multipart_stream.h
@@ -9,6 +9,11 @@
 namespace web_video_server
 {
 
+struct PendingFooter {
+  ros::Time timestamp;
+  boost::weak_ptr<std::string> contents;
+};
+
 class MultipartStream {
 public:
   MultipartStream(async_web_server_cpp::HttpConnectionPtr& connection,
@@ -17,7 +22,7 @@ class MultipartStream {
 
   void sendInitialHeader();
   void sendPartHeader(const ros::Time &time, const std::string& type, size_t payload_size);
-  void sendPartFooter();
+  void sendPartFooter(const ros::Time &time);
   void sendPartAndClear(const ros::Time &time, const std::string& type, std::vector<unsigned char> &data);
   void sendPart(const ros::Time &time, const std::string& type, const boost::asio::const_buffer &buffer,
 		async_web_server_cpp::HttpConnection::ResourcePtr resource);
@@ -29,7 +34,7 @@ class MultipartStream {
   const std::size_t max_queue_size_;
   async_web_server_cpp::HttpConnectionPtr connection_;
   std::string boundry_;
-  std::queue<boost::weak_ptr<const void> > pending_footers_;
+  std::queue<PendingFooter> pending_footers_;
 };
 
 }

diff --git a/src/multipart_stream.cpp b/src/multipart_stream.cpp
@@ -32,10 +32,13 @@ void MultipartStream::sendPartHeader(const ros::Time &time, const std::string& t
   connection_->write(async_web_server_cpp::HttpReply::to_buffers(*headers), headers);
 }
 
-void MultipartStream::sendPartFooter() {
+void MultipartStream::sendPartFooter(const ros::Time &time) {
   boost::shared_ptr<std::string> str(new std::string("\r\n--"+boundry_+"\r\n"));
+  PendingFooter pf;
+  pf.timestamp = time;
+  pf.contents = str;
   connection_->write(boost::asio::buffer(*str), str);
-  if (max_queue_size_ > 0) pending_footers_.push(str);
+  if (max_queue_size_ > 0) pending_footers_.push(pf);
 }
 
 void MultipartStream::sendPartAndClear(const ros::Time &time, const std::string& type,
@@ -44,7 +47,7 @@ void MultipartStream::sendPartAndClear(const ros::Time &time, const std::string&
   {
     sendPartHeader(time, type, data.size());
     connection_->write_and_clear(data);
-    sendPartFooter();
+    sendPartFooter(time);
   }
 }
 
@@ -55,14 +58,24 @@ void MultipartStream::sendPart(const ros::Time &time, const std::string& type,
   {
     sendPartHeader(time, type, boost::asio::buffer_size(buffer));
     connection_->write(buffer, resource);
-    sendPartFooter();
+    sendPartFooter(time);
   }
 }
 
 bool MultipartStream::isBusy() {
-  while (!pending_footers_.empty() && pending_footers_.front().expired())
+  ros::Time currentTime = ros::Time::now();
+  while (!pending_footers_.empty())
   {
-    pending_footers_.pop();
+    if (pending_footers_.front().contents.expired()) {
+      pending_footers_.pop();
+    } else {
+      ros::Time footerTime = pending_footers_.front().timestamp;
+      if ((currentTime - footerTime).toSec() > 0.5) {
+        pending_footers_.pop();
+      } else {
+        break;
+      }
+    }
   }
   return !(max_queue_size_ == 0 || pending_footers_.size() < max_queue_size_);
 }