reconnect client

Source/CarlaServer/source/carla/CarlaServer.cpp

@@ -14,7 +14,7 @@ namespace carla {
     _pimpl(std::make_unique<Pimpl>(worldPort, writePort, readPort)) {}
 
   CarlaServer::~CarlaServer() {}
-
+  
   void CarlaServer::init(uint32_t LevelCount) {
     _pimpl->sendWorld(static_cast<uint32_t>(Mode::NUMBER_OF_MODES), LevelCount);
   }

Source/CarlaServer/source/carla/server/CarlaCommunication.cpp

@@ -16,26 +16,27 @@ namespace server {
   }
 
   // This is the thread that sends a string over the TCP socket.
-  static void serverWorkerThread(TCPServer &server, const Reward &rwd) {
+  static void serverWorkerThread(TCPServer &server, thread::AsyncReaderJobQueue<Reward> &thr, const Reward &rwd) {
+      std::string message;
+      bool correctSerialize = rwd.SerializeToString(&message);
 
-    std::string message;
-    bool correctSerialize = rwd.SerializeToString(&message);
+      TCPServer::error_code error;
 
-    TCPServer::error_code error;
+      if (correctSerialize) {
+        server.writeString(message, error);
+        if (error) { logTCPError("Failed to send", error); }
+      } else {
+        logTCPError("Falied to serialize", error);
+      }
 
-    if (correctSerialize) {
-      server.writeString(message, error);
-      if (error) { logTCPError("Failed to send", error); }
-    } else {
-      logTCPError("Falied to serialize", error);
-    }
+      if (!server.Connected()) thr.restart();
 
   }
 
   //TODO:
   // Sortida amb google protocol
   // This is the thread that listens for string over the TCP socket.
-  static std::string clientWorkerThread(TCPServer &server) {
+  static std::string clientWorkerThread(TCPServer &server, thread::AsyncWriterJobQueue<std::string> &thr) {
 
     //if (!server.Connected()) server.AcceptSocket();
 
@@ -43,16 +44,19 @@ namespace server {
     TCPServer::error_code error;
 
     server.readString(message, error);
+
     if (error && (error != boost::asio::error::eof)) { // eof is expected.
       logTCPError("Failed to read", error);
       return std::string();
     }
 
+    if (!server.Connected()) thr.restart();
+
     return message;
   }
 
   // This is the thread that listens & sends a string over the TCP world socket.
-  static std::string worldReceiveThread(TCPServer &server) {
+  static std::string worldReceiveThread(TCPServer &server, thread::AsyncReadWriteJobQueue<std::string, std::string> &thr) {
     std::string message;
     TCPServer::error_code error;
 
@@ -63,11 +67,14 @@ namespace server {
       logTCPError("Failed to read world", error);
       return std::string();
     }
+   
+    if (!server.Connected()) thr.restart();
+
     return message;
 
   }
 
-  static void worldSendThread(TCPServer &server, const std::string &message) {
+  static void worldSendThread(TCPServer &server, thread::AsyncReadWriteJobQueue<std::string, std::string> &thr, const std::string &message) {
     TCPServer::error_code error;
     //message = message.size + message;
 
@@ -78,6 +85,8 @@ namespace server {
       logTCPError("Failed to send world", error);
     }
 
+    if (!server.Connected()) thr.restart();
+
   }
 
   static void Connect(TCPServer &server) {
@@ -89,16 +98,16 @@ namespace server {
     _server(writePort),
     _client(readPort),
     _worldThread {
-    [this]() { return worldReceiveThread(this->_world); },
-    [this](const std::string & msg) { worldSendThread(this->_world, msg); },
+    [this]() { return worldReceiveThread(this->_world, this->_worldThread); },
+    [this](const std::string & msg) { worldSendThread(this->_world, this->_worldThread, msg); },
     [this]() { Connect(this->_world); }
   },
   _serverThread {
-    [this](const Reward &rwd) { serverWorkerThread(this->_server, rwd); },
+    [this](const Reward &rwd) { serverWorkerThread(this->_server, this->_serverThread, rwd); },
     [this]() { Connect(this->_server); }
   },
   _clientThread {
-    [this]() { return clientWorkerThread(this->_client); },
+    [this]() { return clientWorkerThread(this->_client, this->_clientThread); },
     [this]() { Connect(this->_client); }
   }
   {

Source/CarlaServer/source/carla/server/CarlaCommunication.h

@@ -44,6 +44,8 @@ namespace server {
     thread::AsyncWriterJobQueue<std::string> _clientThread;
 
     thread::AsyncReadWriteJobQueue<std::string, std::string> _worldThread;
+
+    bool communicationEnabled;
   };
 
 }

Source/CarlaServer/source/carla/server/TCPServer.cpp

@@ -67,7 +67,6 @@ namespace carla {
             if ((boost::asio::error::eof == error) ||
                 (boost::asio::error::connection_reset == error))
             {
-              std::cout<<"Client disconected"<<std::endl;
               _connected = false;
             }
         }
@@ -86,7 +85,6 @@ namespace carla {
                 if ((boost::asio::error::eof == error) ||
                 (boost::asio::error::connection_reset == error))
                 {
-                  std::cerr<<"Client disconected"<<std::endl;
                   _connected = false;
                 }
                 else

Source/CarlaServer/source/carla/server/TCPServer.h

@@ -35,13 +35,13 @@ namespace server {
 
   private:
 
-    boost::asio::io_service _service;
+   boost::asio::io_service _service;
 
-    boost::asio::ip::tcp::acceptor _acceptor;
+   boost::asio::ip::tcp::acceptor _acceptor;
 
-	boost::asio::ip::tcp::socket _socket;
+	 boost::asio::ip::tcp::socket _socket;
 
-	bool _connected;
+	 bool _connected;
   };
 
 } // namespace server

Source/CarlaServer/source/carla/thread/AsyncReadWriteJobQueue.h

@@ -26,6 +26,7 @@ namespace thread {
         ReadingJob && readingJob,
         ConnectJob && connectJob) :
       _done(false),
+      _restart(true),
       _writeJob(std::move(writingJob)),
       _readJob(std::move(readingJob)),
       _connectJob(std::move(connectJob)),
@@ -44,19 +45,27 @@ namespace thread {
       _readQueue.push(item);
     }
 
+    void restart(){
+      _restart = true;
+    }
+
   private:
 
     void workerThread() {
-      _connectJob();
-      while (!_done) {
-        R value;
-        _readQueue.wait_and_pop(value);
-        _readJob(value);
-        _writeQueue.push(_writeJob());
+      while(!_done){
+        _restart = false; 
+        _connectJob();
+        while (!_restart && !_done) {
+          R value;
+          _readQueue.wait_and_pop(value);
+          _readJob(value);
+          _writeQueue.push(_writeJob());
+        }
       }
     }
 
     std::atomic_bool _done;
+    std::atomic_bool _restart;
 
     WritingJob _writeJob;
     ReadingJob _readJob;

Source/CarlaServer/source/carla/thread/AsyncReaderJobQueue.h

@@ -24,6 +24,7 @@ namespace thread {
 
 	explicit AsyncReaderJobQueue(Job &&job, ConnectJob &&connectionJob) :
 		_done(false),
+    _restart(true),
 		_job(std::move(job)),
 		_connectionJob(std::move(connectionJob)),
         _queue(),
@@ -40,19 +41,26 @@ namespace thread {
       _queue.push(item);
     }
 
-  private:
+    void restart(){
+      _restart = true;
+    }
 
+  private:
     void workerThread() {
-		_connectionJob();
-      while (!_done) {
-        T value;
-        _queue.wait_and_pop(value);
-        _job(value);
-		//Sleep(10);
+      while (!_done){
+        _restart = false;
+    		_connectionJob();
+        while (!_restart && !_done) {
+          T value;
+          _queue.wait_and_pop(value);
+          _job(value);
+    		  //Sleep(10);
+        }
       }
     }
 
     std::atomic_bool _done;
+    std::atomic_bool _restart;
 
     Job _job;
 	ConnectJob _connectionJob;

Source/CarlaServer/source/carla/thread/AsyncWriterJobQueue.h

@@ -22,6 +22,7 @@ namespace thread {
 
 	explicit AsyncWriterJobQueue(Job &&job, ConnectJob &&connectJob) :
 		_done(false),
+    _restart(true),
 		_job(std::move(job)),
 		_connectJob(std::move(connectJob)),
         _queue(),
@@ -36,17 +37,25 @@ namespace thread {
       return _queue.try_pop(value);
     }
 
+    void restart(){
+      _restart = true;
+    }
+
   private:
 
     void workerThread() {
-		_connectJob();
-      while (!_done) {
-        _queue.push(_job());
-		//Sleep(10);
+      while (!_done){
+        _restart = false;
+		    _connectJob();
+        while (!_restart && !_done) {
+          _queue.push(_job());
+		      //Sleep(10);
+        }
       }
     }
 
     std::atomic_bool _done;
+    std::atomic_bool _restart;
 
     Job _job;
 	ConnectJob _connectJob;