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


/***************************************************************************
 *   Copyright (C) December 6, 2008 by Mattias Runge                             *
 *   mattias@runge.se                                                      *
 *   asyncsocket.cpp                                            *
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 *   This program is distributed in the hope that it will be useful,       *
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of        *
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the         *
 *   GNU General Public License for more details.                          *
 *                                                                         *
 *   You should have received a copy of the GNU General Public License     *
 *   along with this program; if not, write to the                         *
 *   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;
    myForceReconnect = false;
 
    Thread<AsyncSocket>();
}
 
AsyncSocket::~AsyncSocket()
{
    if (mySocket != -1)
    {
        ::close(mySocket);
        mySocket = -1;
    }
   
    stop();
}
 
void AsyncSocket::run()
{
    SyslogStream &slog = SyslogStream::getInstance();
 
    reconnectLoop();
 
    char buf[MAXBUFFER + 1];
    string data;
    int status;
    int rc;
    int timeSince = time(NULL) + 10;
   
    AsyncSocket::mySemaphore.lock();
 
    try
    {
        while (1)
        {
            rc = mySemaphore.wait(5);
 
            if (myForceReconnect)
            {
                slog << "Disconnected from server.\n";
                reconnectLoop();
                timeSince = time(NULL) + 10;
                continue;
            }
 
            //cout << "Socket awoken...\n";
           
            if (rc == ETIMEDOUT)
            {
                /* Socket timed out, this means we have not received anything in some time
                and we should check the connection */
 
                setEvent(ASYNCSOCKET_EVENT_INACTIVITY);
                timeSince = time(NULL);
            }
            else
            {
                while (myOutQueue.size() > 0)
                {
                    data = myOutQueue.pop();
 
                    //slog << "Sending: " << data << "\n";
 
                    status = ::send(mySocket, data.c_str(), data.size(), 0);
 
                    //slog << "Status: " << status << "\n";
 
                    if (status == -1)
                    {
                        switch (errno)
                        {
                            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).)");
 
                            case EAGAIN:
                            //slog << "The socket is marked non-blocking and the requested operation would block.\n";
                            break;
 
                            case EBADF:
                            throw new SocketException("An invalid descriptor was specified.");
 
                            case ECONNRESET:
                            throw new SocketException("Connection reset by peer.");
 
                            case EDESTADDRREQ:
                            throw new SocketException("The socket is not connection-mode, and no peer address is set.");
 
                            case EFAULT:
                            throw new SocketException("An invalid user space address was specified for an argument.");
 
                            case EINTR:
                            throw new SocketException("A signal occurred before any data was transmitted; see signal(7).");
 
                            case EINVAL:
                            throw new SocketException("1Invalid argument passed.");
 
                            case EISCONN:
                            throw new SocketException("The connection-mode socket was connected already but a recipient was specified. (Now either this error is returned, or the recipient specification is ignored.)");
 
                            case EMSGSIZE:
                            throw new SocketException("The socket type requires that message be sent atomically, and the size of the message to be sent made this impossible.");
 
                            case ENOBUFS:
                            throw new SocketException("The output queue for a network interface was full.  This generally indicates that the interface has stopped sending, but may be caused by transient congestion.  (Normally, this does not occur in Linux.  Packets are just silently dropped when a device queue overflows.)");
 
                            case ENOMEM:
                            throw new SocketException("No memory available.");
 
                            case ENOTCONN:
                            throw new SocketException("The socket is not connected, and no target has been given.");
 
                            case ENOTSOCK:
                            throw new SocketException("The argument s is not a socket.");
 
                            case EOPNOTSUPP:
                            throw new SocketException("Some bit in the flags argument is inappropriate for the socket type.");
 
                            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";
                            break;
                        }
                    }
                }
 
                memset(buf, 0, MAXBUFFER + 1);
 
                status = ::recv(mySocket, buf, MAXBUFFER, 0);
 
                //slog << "Receiving: " << buf << "\n";
 
                if (status == -1)
                {
                    switch (errno)
                    {
                        case EAGAIN:
                        //slog << "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.\n";
                        break;
 
                        case EBADF:
                        throw new SocketException("The argument s is an invalid descriptor.");
 
                        case ECONNREFUSED:
                        throw new SocketException("A remote host refused to allow the network connection (typically because it is not running the requested service).");
 
                        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();
                        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";
                        break;
                    }
                }
                else if (status == 0)
                {
                    slog << "Disconnected from server.\n";
                    reconnectLoop();
                }
                else if (status > 0)
                {
                    data = buf;
 
                    myInQueue.push(data);
 
                    setEvent(ASYNCSOCKET_EVENT_DATA);
                   
                    timeSince = time(NULL);
                }
               
                if (timeSince + 10 < time(NULL))
                {
                    setEvent(ASYNCSOCKET_EVENT_INACTIVITY);
                    timeSince = time(NULL);
                }
            }
        }
    }
    catch (SocketException *e)
    {
        slog << "Exception: " << e->getDescription() << "\n";
        mySemaphore.unlock();
        setEvent(ASYNCSOCKET_EVENT_DIED);
        stop();
    }
 
    close();
 
    mySemaphore.unlock();
}
 
void AsyncSocket::reconnectLoop()
{
    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 << "Will try again in " << myReconnectTimeout << " seconds\n";
            sleep(myReconnectTimeout);
        }
    }
}
 
void AsyncSocket::connect()
{
 
 
    close();
 
 
   
    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);
 
    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)
    {
        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).)");
 
            case EPERM:
            throw new SocketException("The user tried to connect to a broadcast address without having the socket broadcast  flag  enabled  or  the  connection request failed because of a local firewall rule.");
 
            case EADDRINUSE:
            throw new SocketException("Local address is already in use.");
 
            case EAFNOSUPPORT:
            throw new SocketException("The passed address didn't have the correct address family in its sa_family field.");
 
            case EAGAIN:
            throw new SocketException("No more free local ports or insufficient entries in the routing cache.  For AF_INET see the net.ipv4.ip_local_port_range sysctl in ip(7) on how to increase the number of local ports.");
 
            case EALREADY:
            throw new SocketException("The socket is non-blocking and a previous connection attempt has not yet been completed.");
 
            case EBADF:
            throw new SocketException("The file descriptor is not a valid index in the descriptor table.");
 
            case ECONNREFUSED:
            throw new SocketException("No-one listening on the remote address.");
 
            case EFAULT:
            throw new SocketException("The socket structure address is outside the user's address space.");
 
            case EINPROGRESS:
            throw new SocketException("The socket is non-blocking and the connection cannot be completed immediately.  It is possible to select(2)  or  poll(2) for  completion  by  selecting the socket for writing.  After select(2) indicates writability, use getsockopt(2) to read the SO_ERROR option at level SOL_SOCKET to determine whether connect() completed  successfully  (SO_ERROR  is  zero)  or unsuccessfully (SO_ERROR is one of the usual error codes listed here, explaining the reason for the failure).");
 
            case EINTR:
            throw new SocketException("The system call was interrupted by a signal that was caught; see signal(7).");
 
            case EISCONN:
            throw new SocketException("The socket is already connected.");
 
            case ENETUNREACH:
            throw new SocketException("Network is unreachable.");
 
            case ENOTSOCK:
            throw new SocketException("The file descriptor is not associated with a socket.");
 
            case ETIMEDOUT:
            throw new SocketException("Timeout  while  attempting  connection.  The server may be too busy to accept new connections.  Note that for IP sockets the timeout may be very long when syncookies are enabled on the server.");
 
            default:
            throw new SocketException("Other errors may be generated by the underlying protocol modules. : " + itos(errno));
        }
    }
 
    struct sigaction saio;
    saio.sa_handler = AsyncSocket::signalHandler;
    sigemptyset(&saio.sa_mask);
    saio.sa_flags = 0;
    saio.sa_restorer = NULL;
    sigaction(SIGIO, &saio, NULL);
 
    fcntl(mySocket, F_SETOWN, getpid());
    int flags = fcntl(mySocket, F_GETFL);
 
    if (flags < 0)
        throw new SocketException("Async socket fcntl failed");
 
    fcntl(mySocket, F_SETFL, flags | O_NONBLOCK | FASYNC);
 
    myForceReconnect = false;
}
 
void AsyncSocket::close()
{
    if (mySocket != -1)
    {
        ::close(mySocket);
        mySocket = -1;
        setEvent(ASYNCSOCKET_EVENT_CLOSED);
    }
}
 
void AsyncSocket::setReconnectTimeout(unsigned int timeout)
{
    myReconnectTimeout = timeout;
}
 
void AsyncSocket::startEvent()
{
    myEventSemaphore.lock();
}
 
int AsyncSocket::getEvent()
{
    int event = myEvent;
    myEvent = ASYNCSOCKET_EVENT_NONE;
    return event;
}
 
void AsyncSocket::waitForEvent()
{
    myEventSemaphore.wait();
}
 
void AsyncSocket::stopEvent()
{
    myEventSemaphore.unlock();
}
 
void AsyncSocket::setAddress(string address, int port)
{
    myAddress = address;
    myPort = port;
}
 
bool AsyncSocket::availableData()
{
    return (myInQueue.size() > 0);
}
 
string AsyncSocket::getData()
{
    return myInQueue.pop();
}
 
bool AsyncSocket::sendData(string data)
{
    myOutQueue.push(data);
    mySemaphore.broadcast();
    return true;
}
 
void AsyncSocket::setEvent(int event)
{
    myEventSemaphore.lock();
    myEvent = event;
    myEventSemaphore.unlock();
    myEventSemaphore.broadcast();
}
 
void AsyncSocket::forceReconnect()
{
    myForceReconnect = true;
    mySemaphore.broadcast();
}
 
void AsyncSocket::signalHandler(int signum)
{
    //FIXME: We must know which socket is ready to read by using select... 
    //cout << "DEBUG: signalHandler signum: " << signum << endl;
    mySemaphore.broadcast();
}
 

Rev 974	Rev 975
1	/***************************************************************************	1	/***************************************************************************
2	* Copyright (C) December 6, 2008 by Mattias Runge *	2	* Copyright (C) December 6, 2008 by Mattias Runge *
3	* mattias@runge.se *	3	* mattias@runge.se *
4	* asyncsocket.cpp *	4	* asyncsocket.cpp *
5	* *	5	* *
6	* This program is free software; you can redistribute it and/or modify *	6	* This program is free software; you can redistribute it and/or modify *
7	* it under the terms of the GNU General Public License as published by *	7	* it under the terms of the GNU General Public License as published by *
8	* the Free Software Foundation; either version 2 of the License, or *	8	* the Free Software Foundation; either version 2 of the License, or *
9	* (at your option) any later version. *	9	* (at your option) any later version. *
10	* *	10	* *
11	* This program is distributed in the hope that it will be useful, *	11	* This program is distributed in the hope that it will be useful, *
12	* but WITHOUT ANY WARRANTY; without even the implied warranty of *	12	* but WITHOUT ANY WARRANTY; without even the implied warranty of *
13	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the *	13	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the *
14	* GNU General Public License for more details. *	14	* GNU General Public License for more details. *
15	* *	15	* *
16	* You should have received a copy of the GNU General Public License *	16	* You should have received a copy of the GNU General Public License *
17	* along with this program; if not, write to the *	17	* along with this program; if not, write to the *
18	* Free Software Foundation, Inc., *	18	* Free Software Foundation, Inc., *
19	* 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. *	19	* 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. *
20	***************************************************************************/	20	***************************************************************************/
21		21
22	#include "asyncsocket.h"	22	#include "asyncsocket.h"
23	#include <iostream>	23	#include <iostream>
24	Semaphore AsyncSocket::mySemaphore;	24	Semaphore AsyncSocket::mySemaphore;
25		25
26	AsyncSocket::AsyncSocket()	26	AsyncSocket::AsyncSocket()
27	{	27	{
28	mySocket = -1;	28	mySocket = -1;
29	myReconnectTimeout = 10;	29	myReconnectTimeout = 10;
30	myForceReconnect = false;	30	myForceReconnect = false;
31		31
32	Thread<AsyncSocket>();	32	Thread<AsyncSocket>();
33	}	33	}
34		34
35	AsyncSocket::~AsyncSocket()	35	AsyncSocket::~AsyncSocket()
36	{	36	{
37	if (mySocket != -1)	37	if (mySocket != -1)
38	{	38	{
39	::close(mySocket);	39	::close(mySocket);
40	mySocket = -1;	40	mySocket = -1;
41	}	41	}
42		42
43	stop();	43	stop();
44	}	44	}
45		45
46	void AsyncSocket::run()	46	void AsyncSocket::run()
47	{	47	{
48	SyslogStream &slog = SyslogStream::getInstance();	48	SyslogStream &slog = SyslogStream::getInstance();
49		49
50	reconnectLoop();	50	reconnectLoop();
51		51
52	char buf[MAXBUFFER + 1];	52	char buf[MAXBUFFER + 1];
53	string data;	53	string data;
54	int status;	54	int status;
55	int rc;	55	int rc;
56	int timeSince = time(NULL) + 10;	56	int timeSince = time(NULL) + 10;
57		57
58	AsyncSocket::mySemaphore.lock();	58	AsyncSocket::mySemaphore.lock();
59		59
60	try	60	try
61	{	61	{
62	while (1)	62	while (1)
63	{	63	{
64	rc = mySemaphore.wait(5);	64	rc = mySemaphore.wait(5);
65		65
66	if (myForceReconnect)	66	if (myForceReconnect)
67	{	67	{
68	slog << "Disconnected from server.\n";	68	slog << "Disconnected from server.\n";
69	reconnectLoop();	69	reconnectLoop();
70	timeSince = time(NULL) + 10;	70	timeSince = time(NULL) + 10;
71	continue;	71	continue;
72	}	72	}
73		73
74	//cout << "Socket awoken...\n";	74	//cout << "Socket awoken...\n";
75		75
76	if (rc == ETIMEDOUT)	76	if (rc == ETIMEDOUT)
77	{	77	{
78	/* Socket timed out, this means we have not received anything in some time	78	/* Socket timed out, this means we have not received anything in some time
79	and we should check the connection */	79	and we should check the connection */
80		80
81	setEvent(ASYNCSOCKET_EVENT_INACTIVITY);	81	setEvent(ASYNCSOCKET_EVENT_INACTIVITY);
82	timeSince = time(NULL);	82	timeSince = time(NULL);
83	}	83	}
84	else	84	else
85	{	85	{
86	while (myOutQueue.size() > 0)	86	while (myOutQueue.size() > 0)
87	{	87	{
88	data = myOutQueue.pop();	88	data = myOutQueue.pop();
89		89
90	//slog << "Sending: " << data << "\n";	90	//slog << "Sending: " << data << "\n";
91		91
92	status = ::send(mySocket, data.c_str(), data.size(), 0);	92	status = ::send(mySocket, data.c_str(), data.size(), 0);
93		93
94	//slog << "Status: " << status << "\n";	94	//slog << "Status: " << status << "\n";
95		95
96	if (status == -1)	96	if (status == -1)
97	{	97	{
98	switch (errno)	98	switch (errno)
99	{	99	{
100	case EACCES:	100	case EACCES:
101	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).)");	101	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).)");
102		102
103	case EAGAIN:	103	case EAGAIN:
104	//slog << "The socket is marked non-blocking and the requested operation would block.\n";	104	//slog << "The socket is marked non-blocking and the requested operation would block.\n";
105	break;	105	break;
106		106
107	case EBADF:	107	case EBADF:
108	throw new SocketException("An invalid descriptor was specified.");	108	throw new SocketException("An invalid descriptor was specified.");
109		109
110	case ECONNRESET:	110	case ECONNRESET:
111	throw new SocketException("Connection reset by peer.");	111	throw new SocketException("Connection reset by peer.");
112		112
113	case EDESTADDRREQ:	113	case EDESTADDRREQ:
114	throw new SocketException("The socket is not connection-mode, and no peer address is set.");	114	throw new SocketException("The socket is not connection-mode, and no peer address is set.");
115		115
116	case EFAULT:	116	case EFAULT:
117	throw new SocketException("An invalid user space address was specified for an argument.");	117	throw new SocketException("An invalid user space address was specified for an argument.");
118		118
119	case EINTR:	119	case EINTR:
120	throw new SocketException("A signal occurred before any data was transmitted; see signal(7).");	120	throw new SocketException("A signal occurred before any data was transmitted; see signal(7).");
121		121
122	case EINVAL:	122	case EINVAL:
123	throw new SocketException("1Invalid argument passed.");	123	throw new SocketException("1Invalid argument passed.");
124		124
125	case EISCONN:	125	case EISCONN:
126	throw new SocketException("The connection-mode socket was connected already but a recipient was specified. (Now either this error is returned, or the recipient specification is ignored.)");	126	throw new SocketException("The connection-mode socket was connected already but a recipient was specified. (Now either this error is returned, or the recipient specification is ignored.)");
127		127
128	case EMSGSIZE:	128	case EMSGSIZE:
129	throw new SocketException("The socket type requires that message be sent atomically, and the size of the message to be sent made this impossible.");	129	throw new SocketException("The socket type requires that message be sent atomically, and the size of the message to be sent made this impossible.");
130		130
131	case ENOBUFS:	131	case ENOBUFS:
132	throw new SocketException("The output queue for a network interface was full. This generally indicates that the interface has stopped sending, but may be caused by transient congestion. (Normally, this does not occur in Linux. Packets are just silently dropped when a device queue overflows.)");	132	throw new SocketException("The output queue for a network interface was full. This generally indicates that the interface has stopped sending, but may be caused by transient congestion. (Normally, this does not occur in Linux. Packets are just silently dropped when a device queue overflows.)");
133		133
134	case ENOMEM:	134	case ENOMEM:
135	throw new SocketException("No memory available.");	135	throw new SocketException("No memory available.");
136		136
137	case ENOTCONN:	137	case ENOTCONN:
138	throw new SocketException("The socket is not connected, and no target has been given.");	138	throw new SocketException("The socket is not connected, and no target has been given.");
139		139
140	case ENOTSOCK:	140	case ENOTSOCK:
141	throw new SocketException("The argument s is not a socket.");	141	throw new SocketException("The argument s is not a socket.");
142		142
143	case EOPNOTSUPP:	143	case EOPNOTSUPP:
144	throw new SocketException("Some bit in the flags argument is inappropriate for the socket type.");	144	throw new SocketException("Some bit in the flags argument is inappropriate for the socket type.");
145		145
146	case EPIPE:	146	case EPIPE:
147	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.");	147	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.");
148		148
149	default:	149	default:
150	slog << "Unknow exception: " << errno << "\n";	150	slog << "Unknow exception: " << errno << "\n";
151	break;	151	break;
152	}	152	}
153	}	153	}
154	}	154	}
155		155
156	memset(buf, 0, MAXBUFFER + 1);	156	memset(buf, 0, MAXBUFFER + 1);
157		157
158	status = ::recv(mySocket, buf, MAXBUFFER, 0);	158	status = ::recv(mySocket, buf, MAXBUFFER, 0);
159		159
160	//slog << "Receiving: " << buf << "\n";	160	//slog << "Receiving: " << buf << "\n";
161		161
162	if (status == -1)	162	if (status == -1)
163	{	163	{
164	switch (errno)	164	switch (errno)
165	{	165	{
166	case EAGAIN:	166	case EAGAIN:
167	//slog << "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.\n";	167	//slog << "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.\n";
168	break;	168	break;
169		169
170	case EBADF:	170	case EBADF:
171	throw new SocketException("The argument s is an invalid descriptor.");	171	throw new SocketException("The argument s is an invalid descriptor.");
172		172
173	case ECONNREFUSED:	173	case ECONNREFUSED:
174	throw new SocketException("A remote host refused to allow the network connection (typically because it is not running the requested service).");	174	throw new SocketException("A remote host refused to allow the network connection (typically because it is not running the requested service).");
175		175
176	case EFAULT:	176	case EFAULT:
177	throw new SocketException("The receive buffer pointer(s) point outside the process's address space.");	177	throw new SocketException("The receive buffer pointer(s) point outside the process's address space.");
178		178
179	case EINTR:	179	case EINTR:
180	throw new SocketException("The receive was interrupted by delivery of a signal before any data were available; see signal(7).");	180	throw new SocketException("The receive was interrupted by delivery of a signal before any data were available; see signal(7).");
181		181
182	case EINVAL:	182	case EINVAL:
183	//throw new SocketException("2Invalid argument passed.");	183	//throw new SocketException("2Invalid argument passed.");
184	slog << "Disconnected from server.\n";	184	slog << "Disconnected from server.\n";
185	reconnectLoop();	185	reconnectLoop();
186	break;	186	break;
187		187
188	case ENOMEM:	188	case ENOMEM:
189	throw new SocketException("Could not allocate memory for recvmsg().");	189	throw new SocketException("Could not allocate memory for recvmsg().");
190		190
191	case ENOTCONN:	191	case ENOTCONN:
192	throw new SocketException("The socket is associated with a connection-oriented protocol and has not been connected (see connect(2) and accept(2)).");	192	throw new SocketException("The socket is associated with a connection-oriented protocol and has not been connected (see connect(2) and accept(2)).");
193		193
194	case ENOTSOCK:	194	case ENOTSOCK:
195	throw new SocketException("The argument s does not refer to a socket.");	195	throw new SocketException("The argument s does not refer to a socket.");
196		196
197	default:	197	default:
198	slog << "Unknow exception: " << errno << "\n";	198	slog << "Unknow exception: " << errno << "\n";
199	break;	199	break;
200	}	200	}
201	}	201	}
202	else if (status == 0)	202	else if (status == 0)
203	{	203	{
204	slog << "Disconnected from server.\n";	204	slog << "Disconnected from server.\n";
205	reconnectLoop();	205	reconnectLoop();
206	}	206	}
207	else if (status > 0)	207	else if (status > 0)
208	{	208	{
209	data = buf;	209	data = buf;
210		210
211	myInQueue.push(data);	211	myInQueue.push(data);
212		212
213	setEvent(ASYNCSOCKET_EVENT_DATA);	213	setEvent(ASYNCSOCKET_EVENT_DATA);
214		214
215	timeSince = time(NULL);	215	timeSince = time(NULL);
216	}	216	}
217		217
218	if (timeSince + 10 < time(NULL))	218	if (timeSince + 10 < time(NULL))
219	{	219	{
220	setEvent(ASYNCSOCKET_EVENT_INACTIVITY);	220	setEvent(ASYNCSOCKET_EVENT_INACTIVITY);
221	timeSince = time(NULL);	221	timeSince = time(NULL);
222	}	222	}
223	}	223	}
224	}	224	}
225	}	225	}
226	catch (SocketException *e)	226	catch (SocketException *e)
227	{	227	{
228	slog << "Exception: " << e->getDescription() << "\n";	228	slog << "Exception: " << e->getDescription() << "\n";
229	mySemaphore.unlock();	229	mySemaphore.unlock();
230	setEvent(ASYNCSOCKET_EVENT_DIED);	230	setEvent(ASYNCSOCKET_EVENT_DIED);
231	stop();	231	stop();
232	}	232	}
233		233
234	close();	234	close();
235		235
236	mySemaphore.unlock();	236	mySemaphore.unlock();
237	}	237	}
238		238
239	void AsyncSocket::reconnectLoop()	239	void AsyncSocket::reconnectLoop()
240	{	240	{
241	SyslogStream &slog = SyslogStream::getInstance();	241	SyslogStream &slog = SyslogStream::getInstance();
242		242
243	while (1)	243	while (1)
244	{	244	{
245	try	245	try
246	{	246	{
247	slog << "Trying to connect...\n";	247	slog << "Trying to connect...\n";
248	connect();	248	connect();
249	slog << "Connection established.\n";	249	slog << "Connection established.\n";
250	break;	250	break;
251	}	251	}
252	catch (SocketException *e)	252	catch (SocketException *e)
253	{	253	{
254	slog << "Could not connect: " << e->getDescription() << "\n";	254	slog << "Could not connect: " << e->getDescription() << "\n";
255	slog << "Will try again in " << myReconnectTimeout << " seconds\n";	255	slog << "Will try again in " << myReconnectTimeout << " seconds\n";
256	sleep(myReconnectTimeout);	256	sleep(myReconnectTimeout);
257	}	257	}
258	}	258	}
259	}	259	}
260		260
261	void AsyncSocket::connect()	261	void AsyncSocket::connect()
262	{	262	{
263	if (mySocket != -1)	-
264	{	-
265	::close(~~mySocket~~);	263	close();
266	mySocket = -1;	-
267	}	-
268		264
269	mySocket = ::socket(AF_INET, SOCK_STREAM, 0);	265	mySocket = ::socket(AF_INET, SOCK_STREAM, 0);
270		266
271	int on = 1;	267	int on = 1;
272	int status = setsockopt(mySocket, SOL_SOCKET, SO_REUSEADDR, (const char*)&on, sizeof(on));	268	int status = setsockopt(mySocket, SOL_SOCKET, SO_REUSEADDR, (const char*)&on, sizeof(on));
273	if (status == -1)	269	if (status == -1)
274	{	270	{
275	throw new SocketException("Connect:Reuseaddress: " + itos(errno));	271	throw new SocketException("Connect:Reuseaddress: " + itos(errno));
276	}	272	}
277		273
278	///FIXME: Verify that this works	274	///FIXME: Verify that this works
279	status = setsockopt(mySocket, SOL_SOCKET, SO_KEEPALIVE, (const char*)&on, sizeof(on));	275	status = setsockopt(mySocket, SOL_SOCKET, SO_KEEPALIVE, (const char*)&on, sizeof(on));
280	if (status == -1)	276	if (status == -1)
281	{	277	{
282	throw new SocketException("Connect:Keepalive: " + itos(errno));	278	throw new SocketException("Connect:Keepalive: " + itos(errno));
283	}	279	}
284		280
285	///FIXME: Verify that this works	281	///FIXME: Verify that this works
286	status = setsockopt(mySocket, SOL_TCP, TCP_KEEPIDLE, (const char*)&on, sizeof(on));	282	status = setsockopt(mySocket, SOL_TCP, TCP_KEEPIDLE, (const char*)&on, sizeof(on));
287	if (status == -1)	283	if (status == -1)
288	{	284	{
289	throw new SocketException("Connect:Keepidle: " + itos(errno));	285	throw new SocketException("Connect:Keepidle: " + itos(errno));
290	}	286	}
291		287
292	memset(&myAddressStruct, 0, sizeof(myAddressStruct));	288	memset(&myAddressStruct, 0, sizeof(myAddressStruct));
293		289
294	myAddressStruct.sin_family = AF_INET;	290	myAddressStruct.sin_family = AF_INET;
295	myAddressStruct.sin_port = htons(myPort);	291	myAddressStruct.sin_port = htons(myPort);
296		292
297	status = inet_pton(AF_INET, myAddress.c_str(), &myAddressStruct.sin_addr);	293	status = inet_pton(AF_INET, myAddress.c_str(), &myAddressStruct.sin_addr);
298		294
299	if (status == -1)	295	if (status == -1)
300	{	296	{
301	if (errno == EAFNOSUPPORT)	297	if (errno == EAFNOSUPPORT)
302	throw new SocketException("Connect: EAFNOSUPPORT");	298	throw new SocketException("Connect: EAFNOSUPPORT");
303	}	299	}
304		300
305	status = ::connect(mySocket, (sockaddr*)&myAddressStruct, sizeof(myAddressStruct));	301	status = ::connect(mySocket, (sockaddr*)&myAddressStruct, sizeof(myAddressStruct));
306		302
307	if (status == -1)	303	if (status == -1)
308	{	304	{
309	switch (errno)	305	switch (errno)
310	{	306	{
311	case EACCES:	307	case EACCES:
312	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).)");	308	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).)");
313		309
314	case EPERM:	310	case EPERM:
315	throw new SocketException("The user tried to connect to a broadcast address without having the socket broadcast flag enabled or the connection request failed because of a local firewall rule.");	311	throw new SocketException("The user tried to connect to a broadcast address without having the socket broadcast flag enabled or the connection request failed because of a local firewall rule.");
316		312
317	case EADDRINUSE:	313	case EADDRINUSE:
318	throw new SocketException("Local address is already in use.");	314	throw new SocketException("Local address is already in use.");
319		315
320	case EAFNOSUPPORT:	316	case EAFNOSUPPORT:
321	throw new SocketException("The passed address didn't have the correct address family in its sa_family field.");	317	throw new SocketException("The passed address didn't have the correct address family in its sa_family field.");
322		318
323	case EAGAIN:	319	case EAGAIN:
324	throw new SocketException("No more free local ports or insufficient entries in the routing cache. For AF_INET see the net.ipv4.ip_local_port_range sysctl in ip(7) on how to increase the number of local ports.");	320	throw new SocketException("No more free local ports or insufficient entries in the routing cache. For AF_INET see the net.ipv4.ip_local_port_range sysctl in ip(7) on how to increase the number of local ports.");
325		321
326	case EALREADY:	322	case EALREADY:
327	throw new SocketException("The socket is non-blocking and a previous connection attempt has not yet been completed.");	323	throw new SocketException("The socket is non-blocking and a previous connection attempt has not yet been completed.");
328		324
329	case EBADF:	325	case EBADF:
330	throw new SocketException("The file descriptor is not a valid index in the descriptor table.");	326	throw new SocketException("The file descriptor is not a valid index in the descriptor table.");
331		327
332	case ECONNREFUSED:	328	case ECONNREFUSED:
333	throw new SocketException("No-one listening on the remote address.");	329	throw new SocketException("No-one listening on the remote address.");
334		330
335	case EFAULT:	331	case EFAULT:
336	throw new SocketException("The socket structure address is outside the user's address space.");	332	throw new SocketException("The socket structure address is outside the user's address space.");
337		333
338	case EINPROGRESS:	334	case EINPROGRESS:
339	throw new SocketException("The socket is non-blocking and the connection cannot be completed immediately. It is possible to select(2) or poll(2) for completion by selecting the socket for writing. After select(2) indicates writability, use getsockopt(2) to read the SO_ERROR option at level SOL_SOCKET to determine whether connect() completed successfully (SO_ERROR is zero) or unsuccessfully (SO_ERROR is one of the usual error codes listed here, explaining the reason for the failure).");	335	throw new SocketException("The socket is non-blocking and the connection cannot be completed immediately. It is possible to select(2) or poll(2) for completion by selecting the socket for writing. After select(2) indicates writability, use getsockopt(2) to read the SO_ERROR option at level SOL_SOCKET to determine whether connect() completed successfully (SO_ERROR is zero) or unsuccessfully (SO_ERROR is one of the usual error codes listed here, explaining the reason for the failure).");
340		336
341	case EINTR:	337	case EINTR:
342	throw new SocketException("The system call was interrupted by a signal that was caught; see signal(7).");	338	throw new SocketException("The system call was interrupted by a signal that was caught; see signal(7).");
343		339
344	case EISCONN:	340	case EISCONN:
345	throw new SocketException("The socket is already connected.");	341	throw new SocketException("The socket is already connected.");
346		342
347	case ENETUNREACH:	343	case ENETUNREACH:
348	throw new SocketException("Network is unreachable.");	344	throw new SocketException("Network is unreachable.");
349		345
350	case ENOTSOCK:	346	case ENOTSOCK:
351	throw new SocketException("The file descriptor is not associated with a socket.");	347	throw new SocketException("The file descriptor is not associated with a socket.");
352		348
353	case ETIMEDOUT:	349	case ETIMEDOUT:
354	throw new SocketException("Timeout while attempting connection. The server may be too busy to accept new connections. Note that for IP sockets the timeout may be very long when syncookies are enabled on the server.");	350	throw new SocketException("Timeout while attempting connection. The server may be too busy to accept new connections. Note that for IP sockets the timeout may be very long when syncookies are enabled on the server.");
355		351
356	default:	352	default:
357	throw new SocketException("Other errors may be generated by the underlying protocol modules. : " + itos(errno));	353	throw new SocketException("Other errors may be generated by the underlying protocol modules. : " + itos(errno));
358	}	354	}
359	}	355	}
360		356
361	struct sigaction saio;	357	struct sigaction saio;
362	saio.sa_handler = AsyncSocket::signalHandler;	358	saio.sa_handler = AsyncSocket::signalHandler;
363	sigemptyset(&saio.sa_mask);	359	sigemptyset(&saio.sa_mask);
364	saio.sa_flags = 0;	360	saio.sa_flags = 0;
365	saio.sa_restorer = NULL;	361	saio.sa_restorer = NULL;
366	sigaction(SIGIO, &saio, NULL);	362	sigaction(SIGIO, &saio, NULL);
367		363
368	fcntl(mySocket, F_SETOWN, getpid());	364	fcntl(mySocket, F_SETOWN, getpid());
369	int flags = fcntl(mySocket, F_GETFL);	365	int flags = fcntl(mySocket, F_GETFL);
370		366
371	if (flags < 0)	367	if (flags < 0)
372	throw new SocketException("Async socket fcntl failed");	368	throw new SocketException("Async socket fcntl failed");
373		369
374	fcntl(mySocket, F_SETFL, flags \| O_NONBLOCK \| FASYNC);	370	fcntl(mySocket, F_SETFL, flags \| O_NONBLOCK \| FASYNC);
375		371
376	myForceReconnect = false;	372	myForceReconnect = false;
377	}	373	}
378		374
379	void AsyncSocket::close()	375	void AsyncSocket::close()
380	{	376	{
381	if (mySocket != -1)	377	if (mySocket != -1)
382	{	378	{
383	::close(mySocket);	379	::close(mySocket);
384	mySocket = -1;	380	mySocket = -1;
385	setEvent(ASYNCSOCKET_EVENT_CLOSED);	381	setEvent(ASYNCSOCKET_EVENT_CLOSED);
386	}	382	}
387	}	383	}
388		384
389	void AsyncSocket::setReconnectTimeout(unsigned int timeout)	385	void AsyncSocket::setReconnectTimeout(unsigned int timeout)
390	{	386	{
391	myReconnectTimeout = timeout;	387	myReconnectTimeout = timeout;
392	}	388	}
393		389
394	void AsyncSocket::startEvent()	390	void AsyncSocket::startEvent()
395	{	391	{
396	myEventSemaphore.lock();	392	myEventSemaphore.lock();
397	}	393	}
398		394
399	int AsyncSocket::getEvent()	395	int AsyncSocket::getEvent()
400	{	396	{
401	int event = myEvent;	397	int event = myEvent;
402	myEvent = ASYNCSOCKET_EVENT_NONE;	398	myEvent = ASYNCSOCKET_EVENT_NONE;
403	return event;	399	return event;
404	}	400	}
405		401
406	void AsyncSocket::waitForEvent()	402	void AsyncSocket::waitForEvent()
407	{	403	{
408	myEventSemaphore.wait();	404	myEventSemaphore.wait();
409	}	405	}
410		406
411	void AsyncSocket::stopEvent()	407	void AsyncSocket::stopEvent()
412	{	408	{
413	myEventSemaphore.unlock();	409	myEventSemaphore.unlock();
414	}	410	}
415		411
416	void AsyncSocket::setAddress(string address, int port)	412	void AsyncSocket::setAddress(string address, int port)
417	{	413	{
418	myAddress = address;	414	myAddress = address;
419	myPort = port;	415	myPort = port;
420	}	416	}
421		417
422	bool AsyncSocket::availableData()	418	bool AsyncSocket::availableData()
423	{	419	{
424	return (myInQueue.size() > 0);	420	return (myInQueue.size() > 0);
425	}	421	}
426		422
427	string AsyncSocket::getData()	423	string AsyncSocket::getData()
428	{	424	{
429	return myInQueue.pop();	425	return myInQueue.pop();
430	}	426	}
431		427
432	bool AsyncSocket::sendData(string data)	428	bool AsyncSocket::sendData(string data)
433	{	429	{
434	myOutQueue.push(data);	430	myOutQueue.push(data);
435	mySemaphore.broadcast();	431	mySemaphore.broadcast();
436	return true;	432	return true;
437	}	433	}
438		434
439	void AsyncSocket::setEvent(int event)	435	void AsyncSocket::setEvent(int event)
440	{	436	{
441	myEventSemaphore.lock();	437	myEventSemaphore.lock();
442	myEvent = event;	438	myEvent = event;
443	myEventSemaphore.unlock();	439	myEventSemaphore.unlock();
444	myEventSemaphore.broadcast();	440	myEventSemaphore.broadcast();
445	}	441	}
446		442
447	void AsyncSocket::forceReconnect()	443	void AsyncSocket::forceReconnect()
448	{	444	{
449	myForceReconnect = true;	445	myForceReconnect = true;
450	mySemaphore.broadcast();	446	mySemaphore.broadcast();
451	}	447	}
452		448
453	void AsyncSocket::signalHandler(int signum)	449	void AsyncSocket::signalHandler(int signum)
454	{	450	{
455	//FIXME: We must know which socket is ready to read by using select...	451	//FIXME: We must know which socket is ready to read by using select...
456	//cout << "DEBUG: signalHandler signum: " << signum << endl;	452	//cout << "DEBUG: signalHandler signum: " << signum << endl;
457	mySemaphore.broadcast();	453	mySemaphore.broadcast();
458	}	454	}
459		455

Subversion Repositories HomeAutomation

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