TcpClient listens to reply with poll on a separate thread

.gitignore

@@ -21,3 +21,5 @@ gdb.out
 html/*
 test/testCppUtils
 test/testCppUtils.out
+*.kate-swp
+

include/TcpClient.hpp

@@ -1,6 +1,8 @@
 #ifndef TCP_CLIENT_HPP
 #define TCP_CLIENT_HPP
 
+#include "Thread.hpp"
+
 #include <string>
 
 #include <sys/types.h>
@@ -16,29 +18,44 @@ public:
   TcpClient ( const std::string host,
               const std::string port );
 
-  ~TcpClient();
+  virtual ~TcpClient();
 
   bool connect();
   bool disconnect();
 
   bool send(const std::string msg);
-  bool receive(std::string &reply);
 
 
 private:
 
+  virtual void msgArrived(const std::string) = 0;
+  virtual void onDisconnect() = 0;
+
+  class WatcherThread : public Thread
+  {
+  public:
+    WatcherThread( TcpClient &data );
+
+  private:
+    void* run();
+    bool receive();
+
+    TcpClient &m_tcpClient;
+  };
+
   bool openSocket();
   bool closeSocket();
 
   bool connectToHost();
-  bool getHostInfo(struct addrinfo **servinfo);
-  void printHostDetails(struct addrinfo *servinfo);
-  bool connectToFirstAddress(struct addrinfo *servinfo);
-
-  int         m_socket;
-  std::string m_host;
-  std::string m_port;
-  bool        m_connected;
+  bool getHostInfo(struct addrinfo **servinfo) const;
+  void printHostDetails(struct addrinfo *servinfo) const;
+  bool connectToFirstAddress(struct addrinfo *servinfo) const;
+
+  int           m_socket;
+  std::string   m_host;
+  std::string   m_port;
+  bool          m_connected;
+  WatcherThread m_watcher;
 
 };
 

other/tcpclient_main.cpp

@@ -7,21 +7,50 @@
 #include <iostream>
 #include <string>
 
+
+class PrinterTcpClient : public TcpClient
+{
+public:
+  PrinterTcpClient ( const std::string host,
+                     const std::string port )
+    : TcpClient(host, port)
+  {
+    TRACE;
+  }
+
+private:
+
+  void msgArrived( const std::string msg)
+  {
+    TRACE;
+    LOG( Logger::DEBUG, std::string("Got msg: ").append(msg).c_str() );
+  }
+
+  void onDisconnect()
+  {
+    TRACE;
+    LOG( Logger::DEBUG, "What shall I do..." );
+  }
+
+};
+
 int main( int argc, char * argv[] )
 {
   Logger::createInstance();
   Logger::init(std::cout);
   Logger::setLogLevel(Logger::FINEST);
 
-  TcpClient tcpclient("localhost", "4455");
+  PrinterTcpClient tcpclient("localhost", "4455");
 
   tcpclient.connect();
 
+  sleep(2);
   tcpclient.send("madao");
+  sleep(2);
 
-  std::string reply;
-  tcpclient.receive(reply);
-  std::cout << "got reply \"" << reply << "\"" << std::endl;
+//   std::string reply;
+//   tcpclient.receive(reply);
+//   std::cout << "got reply \"" << reply << "\"" << std::endl;
 
   tcpclient.disconnect();
 

src/TcpClient.cpp

@@ -7,6 +7,8 @@
 #include <sys/socket.h>
 #include <arpa/inet.h> // inet_ntop
 
+#include <poll.h>
+
 
 TcpClient::TcpClient( const std::string host,
                       const std::string port )
@@ -14,8 +16,11 @@ TcpClient::TcpClient( const std::string host,
   , m_host(host)
   , m_port(port)
   , m_connected(false)
+  , m_watcher(*this)
 {
   TRACE;
+
+  m_watcher.start();
 }
 
 TcpClient::~TcpClient()
@@ -23,6 +28,9 @@ TcpClient::~TcpClient()
   TRACE;
 
   disconnect();
+
+  m_watcher.stop();
+  m_watcher.join();
 }
 
 
@@ -66,22 +74,6 @@ bool TcpClient::send(const std::string msg)
 }
 
 
-bool TcpClient::receive(std::string &reply)
-{
-  TRACE;
-
-  char buffer[256];
-  ssize_t n = read(m_socket, buffer, 255);
-  if (n == -1) {
-    LOG( Logger::ERR, errnoToString("ERROR reading from socket. ").c_str() );
-    return false;
-  }
-  reply = std::string(buffer, n);
-
-  return true;
-}
-
-
 bool TcpClient::openSocket()
 {
   TRACE;
@@ -136,7 +128,7 @@ bool TcpClient::connectToHost()
 }
 
 
-bool TcpClient::getHostInfo(struct addrinfo **servinfo)
+bool TcpClient::getHostInfo(struct addrinfo **servinfo) const
 {
   TRACE;
 
@@ -166,7 +158,7 @@ bool TcpClient::getHostInfo(struct addrinfo **servinfo)
 }
 
 
-void TcpClient::printHostDetails(struct addrinfo *servinfo)
+void TcpClient::printHostDetails(struct addrinfo *servinfo) const
 {
   TRACE;
 
@@ -196,7 +188,7 @@ void TcpClient::printHostDetails(struct addrinfo *servinfo)
 }
 
 
-bool TcpClient::connectToFirstAddress(struct addrinfo *servinfo)
+bool TcpClient::connectToFirstAddress(struct addrinfo *servinfo) const
 {
   TRACE;
 
@@ -224,3 +216,68 @@ bool TcpClient::connectToFirstAddress(struct addrinfo *servinfo)
                                 " connection refused.").c_str() );
   return false;
 }
+
+
+TcpClient::WatcherThread::WatcherThread( TcpClient &data )
+  : m_tcpClient(data)
+{
+  TRACE;
+}
+
+
+void* TcpClient::WatcherThread::run()
+{
+  TRACE;
+
+  while ( m_isRunning ) {
+
+    struct timespec tm = {0,1000};
+    nanosleep(&tm, &tm) ;
+    if ( m_tcpClient.m_connected ) {
+
+      pollfd fds[1] ;
+      fds[0].fd       = m_tcpClient.m_socket ;
+      fds[0].events   = POLLIN | POLLPRI ;
+      fds[0].revents  = 0 ;
+
+      int ret = poll( fds , 1, 1000) ;
+      if ( ret == -1 ) {
+        LOG( Logger::ERR, errnoToString("ERROR at polling. ").c_str() );
+        m_tcpClient.m_connected = false;
+        m_tcpClient.onDisconnect();
+      }
+
+      if ( ret != 0 && !receive() ) {
+          m_tcpClient.m_connected = false;
+          m_tcpClient.onDisconnect();
+        }
+
+    }
+  }
+
+  return 0;
+}
+
+
+bool TcpClient::WatcherThread::receive()
+{
+  TRACE;
+
+  char buffer[256];
+  int len = recv( m_tcpClient.m_socket, buffer , 256, 0) ;
+
+  if (len == -1) {
+    LOG( Logger::ERR, errnoToString("ERROR reading from socket. ").c_str() );
+    return false;
+  }
+
+  if (len == 0) {
+    LOG( Logger::DEBUG, "Connection closed by peer." );
+    return false;
+  }
+
+  std::string msg(buffer, len);
+  m_tcpClient.msgArrived(msg);
+
+  return true;
+}