Merge pull request #60 from robohouse-delft/interrupt-subscription

Add thread-safe way to interrupt waitForSubscriptionEvent().

Author: jontje

include/abb_librws/rws_client.h

@@ -574,6 +574,21 @@ class RWSClient : public POCOClient
    * \return RWSResult containing the result.
    */
   RWSResult endSubscription();
+
+  /**
+   * \brief Force close the active subscription connection.
+   *
+   * This will cause waitForSubscriptionEvent() to return or throw.
+   * It does not delete the subscription from the controller.
+   *
+   * The preferred way to close the subscription is to request the robot controller to end it via
+   * endSubscription(). This function can be used to force the connection to close immediately in
+   * case the robot controller is not responding.
+   *
+   * This function blocks until an active waitForSubscriptionEvent() has finished.
+   *
+   */
+  void forceCloseSubscription();
 
   /**
    * \brief A method for logging out the currently active RWS session.

include/abb_librws/rws_interface.h

@@ -512,6 +512,20 @@ class RWSInterface
    */
   bool endSubscription();
 
+  /**
+   * \brief Froce close the active subscription connection.
+   *
+   * This will cause waitForSubscriptionEvent() to return or throw.
+   * It does not delete the subscription from the controller.
+   *
+   * The preferred way to close the subscription is to request the robot controller to end it via
+   * endSubscription(). This function can be used to force the connection to close immediately in
+   * case the robot controller is not responding.
+   *
+   * This function blocks until an active waitForSubscriptionEvent() has finished.
+   */
+  void forceCloseSubscription();
+
   /**
    * \brief A method for registering a user as local.
    *

include/abb_librws/rws_poco_client.h

@@ -345,6 +345,19 @@ class POCOClient
    */
   POCOResult webSocketRecieveFrame();
 
+  /**
+   * \brief Forcibly shut down the websocket connection.
+   *
+   * The connection is shut down immediately.
+   * Subsequently, the function will block until a current call to webSocketRecieveFrame() has finished,
+   * before cleaning up the local state.
+   *
+   * Note that since mutexes do not guarantee the order of acquisition for multiple contenders,
+   * it is undefined how many calls to webSocketRecieveFrame() will still attempt to use the shut down
+   * connection before the local state is cleaned. Those invocation will throw a runtime error.
+   */
+  void webSocketShutdown();
+
   /**
    * \brief A method for retrieving a substring in a string.
    *
@@ -420,10 +433,28 @@ class POCOClient
    */
   Poco::Mutex http_mutex_;
 
+  /**
+   * \brief A mutex for protecting the client's WebSocket pointer.
+   *
+   * This mutex must be held while setting or invalidating the p_websocket_ member.
+   * Note that the websocket_use_mutex_ must also be held while invalidating the pointer,
+   * since someone may be using it otherwise.
+   *
+   * If acquiring both websocket_connect_mutex_ and websocket_use_mutex_,
+   * the connect mutex must be acquired first.
+   */
+  Poco::Mutex websocket_connect_mutex_;
+
   /**
    * \brief A mutex for protecting the client's WebSocket resources.
+   *
+   * This mutex must be held while using the p_websocket_ member,
+   * and while invalidating the pointer.
+   *
+   * If acquiring both websocket_connect_mutex_ and websocket_use_mutex_,
+   * the connect mutex must be acquired first.
    */
-  Poco::Mutex websocket_mutex_;
+  Poco::Mutex websocket_use_mutex_;
 
   /**
    * \brief HTTP credentials for the remote server's access authentication process.

src/rws_client.cpp

@@ -490,6 +490,11 @@ RWSClient::RWSResult RWSClient::endSubscription()
   return result;
 }
 
+void RWSClient::forceCloseSubscription()
+{
+  webSocketShutdown();
+}
+
 RWSClient::RWSResult RWSClient::logout()
 {
   uri_ = Resources::LOGOUT;

src/rws_interface.cpp

@@ -374,6 +374,11 @@ bool RWSInterface::endSubscription()
   return rws_client_.endSubscription().success;
 }
 
+void RWSInterface::forceCloseSubscription()
+{
+  rws_client_.webSocketShutdown();
+}
+
 bool RWSInterface::registerLocalUser(std::string username,
                                      std::string application,
                                      std::string location)

src/rws_poco_client.cpp

@@ -334,8 +334,16 @@ POCOClient::POCOResult POCOClient::webSocketConnect(const std::string uri,
   try
   {
     result.addHTTPRequestInfo(request);
-    p_websocket_ = new WebSocket(http_client_session_, request, response);
-    p_websocket_->setReceiveTimeout(Poco::Timespan(timeout));
+    {
+      // We must have at least websocket_connect_mutext_.
+      // If a connection already exists, we must also have websocket_use_mutex_.
+      // If not, nobody should have the mutex anyway, so we should get it immediately.
+      ScopedLock<Mutex> connect_lock(websocket_connect_mutex_);
+      ScopedLock<Mutex> use_lock(websocket_use_mutex_);
+      p_websocket_ = new WebSocket(http_client_session_, request, response);
+
+      p_websocket_->setReceiveTimeout(Poco::Timespan(timeout));
+    }
 
     result.addHTTPResponseInfo(response);
     result.status = POCOResult::OK;
@@ -372,7 +380,7 @@ POCOClient::POCOResult POCOClient::webSocketConnect(const std::string uri,
 POCOClient::POCOResult POCOClient::webSocketRecieveFrame()
 {
   // Lock the object's mutex. It is released when the method goes out of scope.
-  ScopedLock<Mutex> lock(websocket_mutex_);
+  ScopedLock<Mutex> lock(websocket_use_mutex_);
 
   // Result of the communication.
   POCOResult result;
@@ -451,6 +459,26 @@ POCOClient::POCOResult POCOClient::webSocketRecieveFrame()
   return result;
 }
 
+void POCOClient::webSocketShutdown()
+{
+  // Make sure nobody is connecting while we're closing.
+  ScopedLock<Mutex> connect_lock(websocket_connect_mutex_);
+
+  // Make sure there is actually a connection to close.
+  if (!webSocketExist())
+  {
+    return;
+  }
+
+  // Shut down the socket. This should make webSocketReceiveFrame() return as soon as possible.
+  p_websocket_->shutdown();
+
+  // Also acquire the websocket lock before invalidating the pointer,
+  // or we will break running calls to webSocketRecieveFrame().
+  ScopedLock<Mutex> use_lock(websocket_use_mutex_);
+  p_websocket_ = Poco::SharedPtr<Poco::Net::WebSocket>();
+}
+
 /************************************************************
  * Auxiliary methods
  */