WebSVN – HomeAutomation – Diff – /trunk/PcSoftware/Atom/Socket/asyncsocket.cpp

  *   Free Software Foundation, Inc.,                                       *
  *   59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.             *
  ***************************************************************************/
 #include "asyncsocket.h"
-#include <iostream>
 Semaphore AsyncSocket::mySemaphore;
 AsyncSocket::AsyncSocket()
+{
     mySocket = -1;
-    myReconnectTimeout = 10;
+    myReconnectTimeout = 0;
     myForceReconnect = false;
     Thread<AsyncSocket>();
 }
 AsyncSocket::~AsyncSocket()
+{
     if (mySocket != -1)
     {
         ::close(mySocket);
         mySocket = -1;
+    }
     stop();
 void AsyncSocket::run()
+{
     SyslogStream &slog = SyslogStream::getInstance();
+    if (myReconnectTimeout == 0)
+    {
+        connect();
+    }
+    else
+    {
-    reconnectLoop();
+        reconnectLoop();
+    }
     char buf[MAXBUFFER + 1];
     string data;
     int status;
     int rc;
     int timeSince = time(NULL) + 10;
-    AsyncSocket::mySemaphore.lock();
     try
+    {
         while (1)
+        {
+            AsyncSocket::mySemaphore.lock();
+            if (myReconnectTimeout == 0)
+            {
-            rc = mySemaphore.wait(5);
+                rc = mySemaphore.wait();
+            }
+            else
+            {
+                rc = mySemaphore.wait(10);
+            }
+            AsyncSocket::mySemaphore.unlock();
             if (myForceReconnect)
+            {
                 slog << "Disconnected from server.\n";
                 reconnectLoop();
                             case EPIPE:
                             throw new SocketException("The  local end has been shut down on a connection oriented socket. In this case the process will also receive a SIGPIPE unless MSG_NOSIGNAL is set.");
                             default:
-                            slog << "Unknow exception: " << errno << "\n";
+                            slog << "Unknow exception: " + itos(errno) + "\n";
                             break;
+                        }
                     }
                 }
                 memset(buf, 0, MAXBUFFER + 1);
                 status = ::recv(mySocket, buf, MAXBUFFER, 0);
-                //slog << "Receiving: " << buf << "\n";
                 if (status == -1)
+                {
                     switch (errno)
+                    {
                         case EAGAIN:
                         case EFAULT:
                         throw new SocketException("The receive buffer pointer(s) point outside the process's address space.");
                         case EINTR:
                         throw new SocketException("The receive was interrupted by delivery of a signal before any data were available; see signal(7).");
                         case EINVAL:
                         //throw new SocketException("2Invalid argument passed.");
                         slog << "Disconnected from server.\n";
                         reconnectLoop();
+                        timeSince = time(NULL) + 10;
                         break;
                         case ENOMEM:
                         throw new SocketException("Could not allocate memory for recvmsg().");
                         case ENOTCONN:
                         throw new SocketException("The socket is associated with a connection-oriented protocol and has not been connected (see connect(2) and  accept(2)).");
                         case ENOTSOCK:
                         throw new SocketException("The argument s does not refer to a socket.");
                         default:
-                        slog << "Unknow exception: " << errno << "\n";
+                        slog << "Unknow exception: " + itos(errno) + "\n";
                         break;
+                    }
+                }
                 else if (status == 0)
+                {
                     slog << "Disconnected from server.\n";
                     reconnectLoop();
+                    timeSince = time(NULL);
+                }
                 else if (status > 0)
+                {
+                    timeSince = time(NULL) + 10;
                     data = buf;
+                    //slog << "Receiving: " + data + "\n";
                     myInQueue.push(data);
                     setEvent(ASYNCSOCKET_EVENT_DATA);
-                    timeSince = time(NULL);
+                }
                 if (timeSince + 10 < time(NULL))
+                {
                     setEvent(ASYNCSOCKET_EVENT_INACTIVITY);
+            }
+        }
+    }
     catch (SocketException *e)
+    {
-        slog << "Exception: " << e->getDescription() << "\n";
+        slog << "Exception: " + e->getDescription() + "\n";
-        mySemaphore.unlock();
+        //mySemaphore.unlock();
         setEvent(ASYNCSOCKET_EVENT_DIED);
         stop();
     }
     close();
-    mySemaphore.unlock();
+    //mySemaphore.unlock();
+}
 void AsyncSocket::reconnectLoop()
+{
+    if (myReconnectTimeout == 0)
+    {
+        throw new SocketException("Connection is closed.");
+    }
     SyslogStream &slog = SyslogStream::getInstance();
     while (1)
+    {
         try
+        {
-            slog << "Trying to connect...\n";
             connect();
-            slog << "Connection established.\n";
             break;
+        }
         catch (SocketException *e)
+        {
-            slog << "Could not connect: " << e->getDescription() << "\n";
+            slog << "Could not connect: " + e->getDescription() + "\n";
-            slog << "Will try again in " << myReconnectTimeout << " seconds\n";
+            slog << "Will try again in " + itos(myReconnectTimeout) + " seconds\n";
             sleep(myReconnectTimeout);
+        }
+    }
+}
 void AsyncSocket::connect()
+{
+    SyslogStream &slog = SyslogStream::getInstance();
     close();
+    slog << "Trying to connect...\n";
     mySocket = ::socket(AF_INET, SOCK_STREAM, 0);
     int on = 1;
     int status = setsockopt(mySocket, SOL_SOCKET, SO_REUSEADDR, (const char*)&on, sizeof(on));
     if (status == -1)
+    {
         throw new SocketException("Connect:Reuseaddress: " + itos(errno));
     }
     ///FIXME: Verify that this works
     status = setsockopt(mySocket, SOL_SOCKET, SO_KEEPALIVE, (const char*)&on, sizeof(on));
     if (status == -1)
+    {
         throw new SocketException("Connect:Keepalive: " + itos(errno));
     }
     ///FIXME: Verify that this works
     status = setsockopt(mySocket, SOL_TCP, TCP_KEEPIDLE, (const char*)&on, sizeof(on));
     if (status == -1)
+    {
         throw new SocketException("Connect:Keepidle: " + itos(errno));
     }
     memset(&myAddressStruct, 0, sizeof(myAddressStruct));
     myAddressStruct.sin_family = AF_INET;
     myAddressStruct.sin_port = htons(myPort);
+    struct hostent *hptr = gethostbyname(myAddress.c_str());
+    if (hptr == NULL)
+    {
+        throw new SocketException("Connect: Could not resolv ip address");
+    }
+    memcpy(&myAddressStruct.sin_addr, hptr->h_addr, hptr->h_length);
+    /*
     status = inet_pton(AF_INET, myAddress.c_str(), &myAddressStruct.sin_addr);
     if (status == -1)
+    {
         if (errno == EAFNOSUPPORT)
             throw new SocketException("Connect: EAFNOSUPPORT");
+    }
+    */
     status = ::connect(mySocket, (sockaddr*)&myAddressStruct, sizeof(myAddressStruct));
     if (status == -1)
+    {
         throw new SocketException("Async socket fcntl failed");
     fcntl(mySocket, F_SETFL, flags | O_NONBLOCK | FASYNC);
     myForceReconnect = false;
+    slog << "Connection established.\n";
+}
 void AsyncSocket::close()
+{
     if (mySocket != -1)

Subversion Repositories HomeAutomation

(root)/trunk/PcSoftware/Atom/Socket/asyncsocket.cpp @ 976 – Rev 975 → 976