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

+{
     myId = socketCount++;
     mySocket = -1;
     myReconnectTimeout = 0;
     myForceReconnect = false;
-    Thread<AsyncSocket>();
+}
 AsyncSocket::~AsyncSocket()
 {
-    if (mySocket != -1)
-    {
-        ::close(mySocket);
-    }
     stop();
+    silentClose();
+}
 void AsyncSocket::run()
+{
-    SyslogStream &slog = SyslogStream::getInstance();
+    // If connection is already up then we should not connect again
     if (!isConnected())
     {
+        // If we have choosen to not use automatic reconnect do not start reconnect loop
         if (myReconnectTimeout == 0)
+        {
+            // Connect to somewhere
             connect();
         }
         else
         {
+            // Start the reconnect loop
+            reconnectLoop();
+        }
+    }
+    bool loop = true;
+    try
+    {
+        while (loop)
+        {
+            // If we have triggered a forced reconnect do it here
+            if (myForceReconnect)
+            {
+                myForceReconnect = false;
-            reconnectLoop();
+                reconnectLoop();
+            }
+            // Receive data
+            loop = receiveData();
+        }
+    }
+    catch (SocketException *e)
+    {
+        // Something bad happend and we can not continue
+        eventAdd(SocketEvent::TYPE_CONNECTION_DIED, e->getDescription());
+    }
+    // Clean up socket if we would want to restart
+    silentClose();
+}
+bool AsyncSocket::receiveData()
+{
-    char buf[MAXBUFFER + 1];
+    char buffer[MAXBUFFER + 1];
-    string data;
+    memset(buffer, 0, MAXBUFFER + 1);
-    int status;
+    int status = ::recv(mySocket, buffer, MAXBUFFER, 0);
-    bool loop = true;
+    if (status == -1)
-    try
-    {
-        while (loop)
+    {
-            if (myForceReconnect)
-            {
-                reconnectLoop();
-            }
-            memset(buf, 0, MAXBUFFER + 1);
-            status = ::recv(mySocket, buf, MAXBUFFER, 0);
-            if (status == -1)
-            {
-                switch (errno)
+        switch (errno)
-                {
+        {
-                    case EAGAIN:
+            case EAGAIN:
-                    throw new SocketException("The socket is marked non-blocking and the receive operation would block, or a receive timeout had been set and the timeout expired before data was received.");
+            throw new SocketException("The socket is marked non-blocking and the receive operation would block, or a receive timeout had been set and the timeout expired before data was received.");
-                    break;
-                    case EBADF:
+            case EBADF:
-                    throw new SocketException("The argument s is an invalid descriptor.");
+            throw new SocketException("The argument s is an invalid descriptor.");
-                    case ECONNREFUSED:
+            case ECONNREFUSED:
-                    throw new SocketException("A remote host refused to allow the network connection (typically because it is not running the requested service).");
+            throw new SocketException("A remote host refused to allow the network connection (typically because it is not running the requested service).");
-                    case EFAULT:
+            case EFAULT:
-                    throw new SocketException("The receive buffer pointer(s) point outside the process's address space.");
+            throw new SocketException("The receive buffer pointer(s) point outside the process's address space.");
-                    case EINTR:
+            case EINTR:
-                    throw new SocketException("The receive was interrupted by delivery of a signal before any data were available; see signal(7).");
+            throw new SocketException("The receive was interrupted by delivery of a signal before any data were available; see signal(7).");
-                    case EINVAL:
+            case EINVAL:
-                    throw new SocketException("Invalid argument passed.");
+            throw new SocketException("Invalid argument passed.");
-                    break;
-                    case ENOMEM:
+            case ENOMEM:
-                    throw new SocketException("Could not allocate memory for recvmsg().");
+            throw new SocketException("Could not allocate memory for recvmsg().");
-                    case ENOTCONN:
+            case ENOTCONN:
-                    case ENOTSOCK:
+            case ENOTSOCK:
-                    throw new SocketException("The argument s does not refer to a socket.");
+            throw new SocketException("The argument s does not refer to a socket.");
-                    case ECONNRESET:
+            case ECONNRESET:
-                    setEvent(ASYNCSOCKET_EVENT_RESET);
+            eventAdd(SocketEvent::TYPE_CONNECTION_RESET);
+            // If we have automatic reconnect we want to start it now
-                    if (myReconnectTimeout > 0)
+            if (myReconnectTimeout == 0)
-                    {
+            {
-                        reconnectLoop();
+                reconnectLoop();
+            }
-                    else
+            else
-                    {
+            {
+                // Otherwise we would like to end the main loop
-                        loop = false;
+                return false;
+            }
-                    break;
+            break;
-                    default:
+            default:
-                    throw new SocketException("Unknow exception: " + itos(errno));
+            throw new SocketException("Unknow exception: " + itos(errno));
-                    break;
+            break;
+        }
+    }
-            else if (status == 0)
+    else if (status == 0)
+    {
+        // Remote host have done a normal shutdown
-                setEvent(ASYNCSOCKET_EVENT_CLOSED);
+        eventAdd(SocketEvent::TYPE_CONNECTION_CLOSED);
-                if (myReconnectTimeout > 0)
+        if (myReconnectTimeout > 0)
+        {
-                    reconnectLoop();
+            reconnectLoop();
+        }
-                else
+        else
+        {
-                    loop = false;
+            return false;
-                }
+        }
+    }
-            else if (status > 0)
+    else if (status > 0)
-            {
+    {
-                data = buf;
+        // We have received data
+        eventAdd(SocketEvent::TYPE_DATA, buffer);
+    }
-                //slog << "Received: " + data + "\n";
-                myInQueue.push(data);
+    return true;
-                setEvent(ASYNCSOCKET_EVENT_DATA);
-            }
+}
-        }
+void AsyncSocket::silentClose()
-    }
+{
-    catch (SocketException *e)
+    if (mySocket != -1)
     {
-        slog << "Exception: " + e->getDescription() + "\n";
+        ::close(mySocket);
-        setEvent(ASYNCSOCKET_EVENT_DIED);
+        mySocket = -1;
+    }
+}
+void AsyncSocket::close()
+{
-    close();
+    silentClose();
+    eventAdd(SocketEvent::TYPE_CONNECTION_CLOSED);
+}
 void AsyncSocket::reconnectLoop()
 {
-    SyslogStream &slog = SyslogStream::getInstance();
-    while (1)
+    while (true)
     {
         try
         {
             connect();
-            break;
+            return;
+        }
         catch (SocketException *e)
         {
-            setEvent(ASYNCSOCKET_EVENT_CONNECT_FAILED);
+            eventAdd(SocketEvent::TYPE_CONNECTION_FAILED, e->getDescription());
-            setEvent(ASYNCSOCKET_EVENT_WAITING_TO_RECONNECT);
+            eventAdd(SocketEvent::TYPE_WAITING_RECONNECT);
-            slog << "Could not connect: " + e->getDescription() + "\n";//FIXME: Remove these
-            slog << "Will try again in " + itos(myReconnectTimeout) + " seconds\n";
             sleep(myReconnectTimeout);
         }
     }
 }
 void AsyncSocket::create()
+{
-    close();
+    silentClose();
     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)
+    {
-        close();
+        silentClose();
         throw new SocketException("Create:Reuseaddress: " + itos(errno));
     }
 }
 void AsyncSocket::startListen()
+{
     create();
     myAddressStruct.sin_family = AF_INET;
     int status = ::bind(mySocket, (struct sockaddr *)&myAddressStruct, sizeof(myAddressStruct));
     if (status == -1)
+    {
-        close();
+        silentClose();
         switch (errno)
+        {
             case EACCES:
             throw new SocketException("The address is protected, and the user is not the superuser.");
     status = ::listen(mySocket, MAXCONNECTIONS);
     if (status == -1)
+    {
-        close();
+        silentClose();
         switch (errno)
+        {
             case EADDRINUSE:
             throw new SocketException("Another socket is already listening on the same port.");
     return false;
+}
 void AsyncSocket::connect()
+{
-    SyslogStream &slog = SyslogStream::getInstance();
+    eventAdd(SocketEvent::TYPE_CONNECTING);
     create();
     memset(&myAddressStruct, 0, sizeof(myAddressStruct));
     myAddressStruct.sin_family = AF_INET;
     myAddressStruct.sin_port = htons(myPort);
-    //slog << "Trying to resolv " + myAddress + "\n";
     struct hostent *hptr = gethostbyname(myAddress.c_str());
     if (hptr == NULL)
+    {
-        close();
+        silentClose();
         throw new SocketException("Connect: Could not resolv ip address");
+    }
     memcpy(&myAddressStruct.sin_addr, hptr->h_addr, hptr->h_length);
     int status = ::connect(mySocket, (sockaddr*)&myAddressStruct, sizeof(myAddressStruct));
     if (status == -1)
+    {
-        close();
+        silentClose();
         switch (errno)
+        {
             case EACCES:
-            throw new SocketException("For Unix domain sockets, which are identified by pathname: Write permission is denied on the socket file, or search per- mission is denied for one of the directories in the path prefix.  (See also path_resolution(7).)");
+            throw new SocketException("For Unix domain sockets, which are identified by pathname: Write permission is denied on the socket file, or search per- mission is denied for one of the directories in the path prefix. (See also path_resolution(7).)");
             default:
             throw new SocketException("Other errors may be generated by the underlying protocol modules. : " + itos(errno));
+        }
+    }
-    myForceReconnect = false;
-    setEvent(ASYNCSOCKET_EVENT_CONNECTED);
+    eventAdd(SocketEvent::TYPE_CONNECTED);
-}
-void AsyncSocket::close()
-{
-    if (mySocket != -1)
-    {
-        ::close(mySocket);
-        mySocket = -1;
-        setEvent(ASYNCSOCKET_EVENT_CLOSED);
-    }
-}
-bool AsyncSocket::isConnected()
-{
-    return mySocket != -1;
-}
-void AsyncSocket::setReconnectTimeout(unsigned int timeout)
-{
-    myReconnectTimeout = timeout;
-}
-void AsyncSocket::startEvent()
-{
-    myEventSemaphore.lock();
-}
-bool AsyncSocket::availableEvent()
-{
-    return (myEventQueue.size() > 0);
-}
-int AsyncSocket::getEvent()
-{
-    return myEventQueue.pop();
-}
-void AsyncSocket::waitForEvent()
-{
-    myEventSemaphore.wait();
-}
-void AsyncSocket::stopEvent()
-{
-    myEventSemaphore.unlock();
-}
-void AsyncSocket::setAddress(string address, int port)
-{
-    myAddress = address;
-    myPort = port;
-}
-void AsyncSocket::setPort(int port)
-{
-    myPort = port;
-}
-void AsyncSocket::setSocket(int socket)
-{
-    mySocket = socket;
-}
-int AsyncSocket::getSocket()
-{
-    return mySocket;
-}
-bool AsyncSocket::availableData()
-{
-    return (myInQueue.size() > 0);
-}
-string AsyncSocket::getData()
-{
-    return myInQueue.pop();
 }
-bool AsyncSocket::sendData(string data)
+void AsyncSocket::sendData(string data)
-{
-    //cout << "SendData on " + myAddress + ":" + itos(myPort) + ".\n";
-    sendDataDirect(data);
-    //myOutQueue.push(data);
-    //mySemaphore.broadcast();
-    return true;
-}
-void AsyncSocket::sendDataDirect(string data)
+{
-    SyslogStream &slog = SyslogStream::getInstance();
-    //slog << "Sending: " << data << "\n";
     int status = ::send(mySocket, data.c_str(), data.size(), 0);
-    //slog << "Status: " << status << "\n";
-    try
-    {
-        if (status == -1)
+    if (status == -1)
+    {
-            switch (errno)
+        switch (errno)
+        {
-                case EACCES:
+            case EACCES:
-                throw new SocketException("(For  Unix  domain sockets, which are identified by pathname) Write permission is denied on the destination socket file, or search permission is denied for one of the directories the path prefix.  (See path_resolution(7).)");
+            throw new SocketException("(For Unix domain sockets, which are identified by pathname) Write permission is denied on the destination socket file, or search permission is denied for one of the directories the path prefix. (See path_resolution(7).)");
-                case EAGAIN:
+            case EAGAIN:
-                //slog << "The socket is marked non-blocking and the requested operation would block.\n";
+            throw new SocketException("The socket is marked non-blocking and the requested operation would block.");
-                break;
-                case EBADF:
+            case EBADF:
-                throw new SocketException("An invalid descriptor was specified.");
+            throw new SocketException("An invalid descriptor was specified.");
-                case ECONNRESET:
+            case ECONNRESET:
-                case EPIPE:
+            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.");
+            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:
+            default:
-                slog << "Unknow exception: " + itos(errno) + "\n";
+            throw new SocketException("Unknow exception: " + itos(errno));
-                break;
-            }
         }
-    }
-    catch (SocketException *e)
-    {
-        slog << "sendDataDirect: Exception: " + e->getDescription() + "\n";
-    }
-}
-void AsyncSocket::setEvent(int event)
-{
-    myEventQueue.push(event);
-    myEventSemaphore.broadcast();
+    }
+}
-void AsyncSocket::forceReconnect()
+void AsyncSocket::eventAdd(unsigned int eventType)
+{
-    ///FIXME: Threadsafe?
-    myForceReconnect = true;
+    eventAdd(eventType, "");
+}
+void AsyncSocket::eventAdd(unsigned int eventType, string eventData)
+{
+    SocketEvent socketEvent(eventType, eventData);
+    myEventQueue.push(socketEvent);
+    myEventSemaphore.broadcast();
+}

Subversion Repositories HomeAutomation

(root)/trunk/PcSoftware/Atom/Socket/asyncsocket.cpp @ 984 – Rev 983 → 984